as-rust/src/core_libs.rs

use strategy::core::Core;
use std::collections::BTreeMap;
use std::io::Error;
use strategy::{exchange_disguise, core};
use std::sync::Arc;
use std::sync::atomic::{AtomicBool};
use std::time::Duration;
use chrono::Utc;
use tokio::sync::{mpsc, Mutex};
use tokio::time::Instant;
use tracing::{error, info};
use global::cci::CentralControlInfo;
use global::params::Params;
use global::trace_stack::TraceStack;
use standard::{Order, PriceOrder};
use strategy::model::OrderInfo;

pub async fn init(params: Params,
                  ws_running: Arc<AtomicBool>,
                  running: Arc<AtomicBool>,
                  cci_arc: Arc<Mutex<CentralControlInfo>>) -> Arc<Mutex<Core>> {
    // 封装
    let mut exchange_params:BTreeMap<String, String> = BTreeMap::new();
    exchange_params.insert("access_key".to_string(), params.access_key.clone());
    exchange_params.insert("secret_key".to_string(), params.secret_key.clone());
    exchange_params.insert("pass_key".to_string(), params.pass_key.clone());

    let (order_sender, mut order_receiver) = mpsc::channel::<Order>(100);
    let (price_order_sender, mut price_order_receiver) = mpsc::channel::<PriceOrder>(100);
    let (error_sender, mut error_receiver) = mpsc::channel::<Error>(100);

    let mut core_obj = Core::new(params.exchange.clone(),
                                   params.clone(),
                                   exchange_params.clone(),
                                   order_sender.clone(),
                                   price_order_sender.clone(),
                                   error_sender.clone(),
                                   running.clone(),
                                   cci_arc.clone()).await;
    let ref_name = core_obj.ref_name[0].clone();
    let trade_name = core_obj.trade_name.clone();

    info!("core初始化……");
    core_obj.before_trade().await;
    let core_arc = Arc::new(Mutex::new(core_obj));

    // 参考交易所
    exchange_disguise::run_reference_exchange(ws_running.clone(),
                                              params.ref_exchange.get(0).unwrap().clone(),
                                              core_arc.clone(),
                                              ref_name,
                                              params.ref_pair.clone(),
                                              params.colo != 0i8,
                                              exchange_params.clone()).await;
    // 交易交易所
    exchange_disguise::run_transactional_exchange(ws_running.clone(),
                                                  params.exchange,
                                                  core_arc.clone(),
                                                  trade_name,
                                                  vec![params.pair.clone()],
                                                  params.colo != 0i8,
                                                  exchange_params.clone()).await;
    // 启动定期触发的系统逻辑
    core::on_timer(core_arc.clone());
    // 启动策略逻辑
    core::run_strategy(core_arc.clone());
    info!("core初始化完成。");

    let order_handler_core_arc = core_arc.clone();
    tokio::spawn(async move {
        loop {
            tokio::time::sleep(Duration::from_millis(1)).await;

            match order_receiver.recv().await {
                Some(order) => {
                    // 刚下的订单有可能会成交，所以有几率触发后续订单逻辑，所以这里也是一个订单触发逻辑
                    let trace_stack = TraceStack::new(Utc::now().timestamp_micros(), Instant::now());

                    if order.status != "NULL" {
                        // trace_stack.on_before_format();
                        let mut core = order_handler_core_arc.lock().await;
                        // let mut delay_time_lock_instance = delay_time_lock.lock().await;
                        let order_info = OrderInfo {
                            symbol: "".to_string(),
                            amount: order.amount.abs(),
                            side: "".to_string(),
                            price: order.price,
                            client_id: order.custom_id,
                            filled_price: order.avg_price,
                            filled: order.deal_amount.abs(),
                            order_id: order.id,
                            local_time: 0,
                            create_time: 0,
                            status: order.status,
                            fee: Default::default(),
                            trace_stack: order.trace_stack.clone(),
                        };
                        // trace_stack.on_after_format();

                        core.update_local_order(order_info.clone(), trace_stack).await;
                    }
                },
                None => {
                    error!("Order channel has been closed!");
                    break
                }
            }
        }
    });

    let price_order_handler_core_arc = core_arc.clone();
    tokio::spawn(async move {
        loop {
            tokio::time::sleep(Duration::from_millis(1)).await;

            match price_order_receiver.recv().await {
                Some(price_order) => {
                    let mut core = price_order_handler_core_arc.lock().await;

                    core.update_local_price_order(price_order).await;
                },
                None => {
                    error!("Price order channel has been closed!");
                    break
                }
            }
        }
    });

    let _error_handler_core_arc = core_arc.clone();
    tokio::spawn(async move {
        loop {
            tokio::time::sleep(Duration::from_millis(1)).await;

            match error_receiver.recv().await {
                Some(_error) => {
                    // let mut core = _error_handler_core_arc.lock().await;
                    // error!("main: 订单出现错误{:?}", _error);
                    // core.strategy._print_summary();
                },
                None => {
                    error!("Error channel has been closed!");
                    break
                }
            }
        }
    });

    return core_arc;
}