as-rust/strategy/src/kucoin_swap.rs

use std::collections::{BTreeMap};
use std::sync::Arc;
use std::sync::atomic::{AtomicBool};
use std::time::Duration;

use rust_decimal::Decimal;
use tokio::spawn;
use tokio::sync::Mutex;
use tokio::time::sleep;
use tracing::error;

use exchanges::kucoin_swap_ws::{KucoinSwapLogin, KucoinSwapSubscribeType, KucoinSwapWs, KucoinSwapWsType};
use exchanges::response_base::ResponseData;
use global::trace_stack::TraceStack;
use standard::exchange::ExchangeEnum::KucoinSwap;
use crate::exchange_disguise::on_special_depth;

use crate::model::{OrderInfo};
use crate::core::Core;

// 1交易、0参考 kucoin 合约 启动
pub async fn kucoin_swap_run(is_shutdown_arc :Arc<AtomicBool>,
                             is_trade: bool,
                             core_arc: Arc<Mutex<Core>>,
                             name: String,
                             symbols: Vec<String>,
                             is_colo: bool,
                             exchange_params: BTreeMap<String, String>) {
    let mut symbol_arr = Vec::new();
    for symbol in &symbols {
        let symbol_mapper = standard::utils::symbol_enter_mapper(KucoinSwap,symbol.as_str());
        let new_symbol = symbol_mapper.replace("_", "").to_uppercase() + "M";
        symbol_arr.push(new_symbol);
    }

    // 新增定期获取余额的协程
    let account_core_arc = core_arc.clone();
    spawn(async move {
        loop {
            // 每30秒重新获取一次
            sleep(Duration::from_secs(30)).await;

            {
                let mut core = account_core_arc.lock().await;
                core.update_equity_rest_swap().await;
            }
        }
    });

    let (write_tx, write_rx) = futures_channel::mpsc::unbounded();
    let write_tx_am = Arc::new(Mutex::new(write_tx));
    spawn(async move {
        //创建读写通道
        spawn( async move {
            let mut ws;
            // 交易
            if is_trade {
                let login_params = parse_btree_map_to_kucoin_swap_login(exchange_params);
                ws = KucoinSwapWs::new_label(name.clone(), is_colo, Option::from(login_params), KucoinSwapWsType::Private).await;
                ws.set_subscribe(vec![
                    KucoinSwapSubscribeType::PuContractMarketLevel2Depth50,

                    KucoinSwapSubscribeType::PrContractPosition,
                    KucoinSwapSubscribeType::PrContractMarketTradeOrders
                ]);
            } else { // 参考
                ws = KucoinSwapWs::new_label(name.clone(), is_colo, None, KucoinSwapWsType::Public).await;
                ws.set_subscribe(vec![
                    KucoinSwapSubscribeType::PuContractMarketLevel2Depth50,

                    // KucoinSwapSubscribeType::PuContractMarketExecution
                ]);
            }

            // 数据处理
            let core_arc_clone = Arc::clone(&core_arc);

            let mut update_flag_u = Decimal::ZERO;
            // let mut max_buy = Decimal::ZERO;
            // let mut min_sell = Decimal::ZERO;
            let multiplier = core_arc.lock().await.platform_rest.get_self_market().ct_val;

            let fun = move |data: ResponseData| {
                // 在 async 块之前克隆 Arc
                let core_arc_cc = core_arc_clone.clone();
                let mul = multiplier.clone();

                async move {
                    on_data(core_arc_cc,
                            &mut update_flag_u,
                            &mul,
                            // &mut max_buy,
                            // &mut min_sell,
                            data).await
                }
            };

            // 建立链接
            ws.set_symbols(symbol_arr);
            ws.ws_connect_async(is_shutdown_arc, fun, &write_tx_am, write_rx).await.expect("链接失败(内部一个心跳线程应该已经关闭了)");
        });
    });
}

async fn on_data(core_arc_clone: Arc<Mutex<Core>>,
                 update_flag_u: &mut Decimal,
                 multiplier: &Decimal,
                 // max_buy: &mut Decimal,
                 // min_sell: &mut Decimal,
                 response: ResponseData) {
    let mut trace_stack = TraceStack::new(response.time, response.ins);
    trace_stack.on_after_span_line();

    match response.channel.as_str() {
        "level2" => {
            trace_stack.set_source("kucoin_usdt_swap.level2".to_string());
            let special_depth = standard::handle_info::HandleSwapInfo::handle_special_depth(KucoinSwap, &response);
            trace_stack.on_after_format();

            on_special_depth(core_arc_clone, update_flag_u, &response.label, &mut trace_stack, &special_depth).await
        }
        "tickerV2" => {
            trace_stack.set_source("kucoin_swap.tickerV2".to_string());

            let special_depth = standard::handle_info::HandleSwapInfo::handle_book_ticker(KucoinSwap, &response);
            trace_stack.on_before_network(special_depth.create_at.clone());

            on_special_depth(core_arc_clone, update_flag_u, &response.label, &mut trace_stack, &special_depth).await
        }
        "symbolOrderChange" => {
            trace_stack.set_source("kucoin_swap.symbolOrderChange".to_string());

            let orders = standard::handle_info::HandleSwapInfo::handle_order(KucoinSwap, response, multiplier.clone());
            let mut order_infos:Vec<OrderInfo> = Vec::new();
            for mut order in orders.order {
                if order.status == "NULL" {
                    continue;
                }

                let order_info = OrderInfo::parse_order_to_order_info(&mut order);
                order_infos.push(order_info);
            }

            let mut core = core_arc_clone.lock().await;
            core.update_order(order_infos, trace_stack).await;
        }
        "position.change" => {
            let positions = standard::handle_info::HandleSwapInfo::handle_position(KucoinSwap, &response, multiplier);
            let mut core = core_arc_clone.lock().await;
            core.update_position(positions).await;
        }
        _ => {
            error!("kucoin_swap 未知推送类型");
            error!(?response);
        }
    }
}

fn parse_btree_map_to_kucoin_swap_login(exchange_params: BTreeMap<String, String>) -> KucoinSwapLogin {
    KucoinSwapLogin {
        access_key: exchange_params.get("access_key").unwrap().clone(),
        secret_key: exchange_params.get("secret_key").unwrap().clone(),
        pass_key: exchange_params.get("pass_key").unwrap().clone(),
    }
}