as-rust/src/quant_libs.rs

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

pub async fn init(params: Params,
                  ws_running: Arc<AtomicBool>,
                  running: Arc<AtomicBool>,
                  cci_arc: Arc<Mutex<CentralControlInfo>>) -> Arc<Mutex<Quant>> {
    // 封装
    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 (error_sender, mut error_receiver) = mpsc::channel::<Error>(100);

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

    info!("quant初始化……");
    quant_obj.before_trade().await;
    let quant_arc = Arc::new(Mutex::new(quant_obj));

    // 参考交易所
    exchange_disguise::run_reference_exchange(ws_running.clone(),
                                              params.ref_exchange.get(0).unwrap().clone(),
                                              quant_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,
                                                  quant_arc.clone(),
                                                  trade_name,
                                                  vec![params.pair.clone()],
                                                  params.colo != 0i8,
                                                  exchange_params.clone()).await;
    // 启动定期触发的系统逻辑
    quant::on_timer(quant_arc.clone());
    // 启动策略逻辑
    quant::run_strategy(quant_arc.clone());
    info!("quant初始化完成。");

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

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

                    if order.status != "NULL" {
                        trace_stack.on_before_format();
                        let mut quant = order_handler_quant_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();

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

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

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

    return quant_arc;
}