as-rust/strategy/src/bybit_usdt_swap.rs

use std::cmp::Ordering;
use std::collections::BTreeMap;
use std::sync::Arc;
use std::sync::atomic::AtomicBool;
use std::time::Duration;
use futures_util::StreamExt;
use rust_decimal::Decimal;
use tokio::{spawn, time};
use tokio::sync::Mutex;
use tracing::{error, info};
use exchanges::bybit_swap_ws::{BybitSwapLogin, BybitSwapSubscribeType, BybitSwapWs, BybitSwapWsType};
use exchanges::response_base::ResponseData;
use global::trace_stack::TraceStack;
use standard::exchange::ExchangeEnum::{BybitSwap};
use standard::handle_info::{DepthParam, format_depth, make_special_depth};
use standard::MarketOrder;
use crate::model::{OrderInfo, OriginalTradeBy};
use crate::quant::Quant;
use crate::exchange_disguise::on_special_depth;

// 1交易、0参考 bybit 合约 启动
pub async fn bybit_swap_run(bool_v1: Arc<AtomicBool>,
                           is_trade: bool,
                           _quant_arc: Arc<Mutex<Quant>>,
                           name: String,
                           symbols: Vec<String>,
                           is_colo: bool,
                           exchange_params: BTreeMap<String, String>) {
    // 启动公共频道
    let (write_tx_public, write_rx_public) = futures_channel::mpsc::unbounded();
    let (read_tx_public, mut read_rx_public) = futures_channel::mpsc::unbounded();

    let mut ws_public = BybitSwapWs::new_label(name.clone(), is_colo, None, BybitSwapWsType::Public);
    ws_public.set_symbols(symbols.clone());
    ws_public.set_subscribe(vec![
        BybitSwapSubscribeType::PuOrderBook50
    ]);
    if is_trade {
        ws_public.set_subscribe(vec![
            BybitSwapSubscribeType::PuBlicTrade
        ]);
    }
    // 挂起公共ws
    let write_tx_am_public = Arc::new(Mutex::new(write_tx_public));
    let bool_clone_public = Arc::clone(&bool_v1);
     spawn(async move {
        ws_public.ws_connect_async(bool_clone_public,
                                   &write_tx_am_public,
                                   write_rx_public,
                                   read_tx_public).await.expect("链接失败(内部一个心跳线程应该已经关闭了)");
    });
    // 消费数据
    let bot_arc_clone = _quant_arc.clone();
    // 接收public数据
    spawn(async move {
        // ticker
        let mut update_flag_u = Decimal::ZERO;
        let mut max_buy = Decimal::ZERO;
        let mut min_sell = Decimal::ZERO;
        let mut depth_asks: Vec<MarketOrder> = Vec::new();
        let mut depth_bids: Vec<MarketOrder> = Vec::new();

        loop {
            if let Some(public_data) = read_rx_public.next().await {
                on_public_data(bot_arc_clone.clone(),
                               &mut update_flag_u,
                               &mut max_buy,
                               &mut min_sell,
                               public_data,
                               &mut depth_asks,
                               &mut depth_bids).await;
            }
        }
    });
    let trade_symbols = symbols.clone();
    // 交易交易所需要启动私有ws
    if is_trade {
        let (write_tx_private, write_rx_private) = futures_channel::mpsc::unbounded();
        let (read_tx_private, mut read_rx_private) = futures_channel::mpsc::unbounded();
        let auth = Some(parse_btree_map_to_bybit_swap_login(exchange_params));

        let mut ws_private = BybitSwapWs::new_label(name.clone(), is_colo, auth, BybitSwapWsType::Private);
        ws_private.set_symbols(trade_symbols);
        ws_private.set_subscribe(vec![
            BybitSwapSubscribeType::PrPosition,
            BybitSwapSubscribeType::PrOrder,
            BybitSwapSubscribeType::PrWallet
        ]);


        // 挂起私有ws
        let write_tx_am_private = Arc::new(Mutex::new(write_tx_private));
        let bool_clone_private = Arc::clone(&bool_v1);
        spawn(async move {
            ws_private.ws_connect_async(bool_clone_private,
                                        &write_tx_am_private,
                                        write_rx_private,
                                        read_tx_private).await.expect("链接失败(内部一个心跳线程应该已经关闭了)");
        });

        // 消费数据
        let bot_arc_clone = _quant_arc.clone();
        // 接收private信息
        spawn(async move {
            let ct_val = bot_arc_clone.clone().lock().await.platform_rest.get_self_market().ct_val;
            let run_symbol = symbols.clone()[0].clone();
            loop {
                if let Some(private_data) = read_rx_private.next().await {
                    on_private_data(bot_arc_clone.clone(),
                                    ct_val,
                                    private_data,
                                    run_symbol.clone()).await;
                }
            }
        });

        // 定时获取仓位信息
        let position_quant_clone = _quant_arc.clone();
        spawn(async move {
            let mut interval = time::interval(Duration::from_secs(30));
            loop {
                interval.tick().await;
                {
                    let mut quant = position_quant_clone.lock().await;
                    quant.update_position_rest_swap().await;
                }

            }
        });
    }
}



async fn on_private_data(bot_arc_clone: Arc<Mutex<Quant>>, ct_val: Decimal, data: ResponseData, run_symbol: String) {
    let mut trace_stack = TraceStack::default();

    trace_stack.on_after_network(data.time);
    trace_stack.on_before_quant();

    if data.code != "200".to_string() {
        return;
    }
    if data.channel == "wallet" {
        let account = standard::handle_info::HandleSwapInfo::handle_account_info(BybitSwap, data, run_symbol.clone());
        {
            let mut quant = bot_arc_clone.lock().await;
            quant.update_equity(account).await;
        }
    } else if data.channel == "order" {
        trace_stack.on_before_format();
        let orders = standard::handle_info::HandleSwapInfo::handle_order(BybitSwap, data.clone(), ct_val.clone());
        trace_stack.on_after_format();

        let mut order_infos:Vec<OrderInfo> = Vec::new();
        for order in orders.order {
            if order.status == "NULL" {
                continue;
            }
            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: Default::default(),
            };
            order_infos.push(order_info);
        }

        {
            let mut quant = bot_arc_clone.lock().await;
            quant.update_order(order_infos, trace_stack);
        }
    } else if data.channel == "position" {
        let positions = standard::handle_info::HandleSwapInfo::handle_position(BybitSwap,data, ct_val.clone());
        {
            let mut quant = bot_arc_clone.lock().await;
            quant.update_position(positions).await;
        }
    }
}

async fn on_public_data(bot_arc_clone: Arc<Mutex<Quant>>, update_flag_u: &mut Decimal, max_buy: &mut Decimal, min_sell: &mut Decimal, data: ResponseData, depth_asks: &mut Vec<MarketOrder>, depth_bids: &mut Vec<MarketOrder>) {
    let mut trace_stack = TraceStack::default();
    trace_stack.on_after_network(data.time);
    trace_stack.on_before_quant();

    if data.code != "200".to_string() {
        return;
    }
    if data.channel == "orderbook" {
        let mut is_update = false;
        let data_type = data.data_type.clone();
        let label = data.label.clone();
        if data_type == "delta"  {
            is_update = true;
        }
        trace_stack.on_before_format();
        let mut depth_format: DepthParam = format_depth(BybitSwap, data);
        // 是增量更新
        if is_update {
            update_order_book(depth_asks, depth_bids, depth_format.depth_asks, depth_format.depth_bids);
        } else { // 全量
            depth_asks.clear();
            depth_asks.append(&mut depth_format.depth_asks);
            depth_bids.clear();
            depth_bids.append(&mut depth_format.depth_bids);

        }
        let depth = make_special_depth(label.clone(), depth_asks, depth_bids, depth_format.t, depth_format.create_at);
        trace_stack.on_before_network(depth_format.create_at.clone());
        trace_stack.on_after_format();

        on_special_depth(bot_arc_clone, update_flag_u, label, trace_stack, depth).await;
    } else if data.channel == "trade" {
        let mut quant = bot_arc_clone.lock().await;
        let str = data.label.clone();
        if quant.is_update.contains_key(&data.label) && *quant.is_update.get(str.as_str()).unwrap(){
            *max_buy = Decimal::ZERO;
            *min_sell = Decimal::ZERO;
            quant.is_update.remove(str.as_str());
        }
        let trades: Vec<OriginalTradeBy> = serde_json::from_str(data.data.as_str()).unwrap();
        for trade in trades {
            if trade.p > *max_buy || *max_buy == Decimal::ZERO{
                *max_buy = trade.p
            }
            if trade.p < *min_sell || *min_sell == Decimal::ZERO{
                *min_sell = trade.p
            }
        }
        quant.max_buy_min_sell_cache.insert(data.label, vec![*max_buy, *min_sell]);
    }
}

fn parse_btree_map_to_bybit_swap_login(exchange_params: BTreeMap<String, String>) -> BybitSwapLogin {
    BybitSwapLogin {
        api_key: exchange_params.get("access_key").unwrap().clone(),
        secret_key: exchange_params.get("secret_key").unwrap().clone(),
    }
}
fn update_order_book(depth_asks: &mut Vec<MarketOrder>, depth_bids: &mut Vec<MarketOrder>, asks : Vec<MarketOrder>, bids: Vec<MarketOrder>) {
    for i in asks {
        let index_of_value = depth_asks.iter().position(|x| x.price == i.price);
        match index_of_value {
            Some(index) => {
                if i.amount == Decimal::ZERO {
                    depth_asks.remove(index);
                } else {
                    depth_asks[index].amount = i.amount.clone();
                }
            },
            None => {
                depth_asks.push(i.clone());
            },
        }
    }
    for i in bids {
        let index_of_value = depth_bids.iter().position(|x| x.price == i.price);
        match index_of_value {
            Some(index) => {
                if i.amount == Decimal::ZERO {
                    depth_bids.remove(index);
                } else {
                    depth_bids[index].amount = i.amount.clone();
                }
            },
            None => {
                depth_bids.push(i.clone());
            },
        }
    }
    depth_asks.sort_by(|a, b| a.price.partial_cmp(&b.price).unwrap_or(Ordering::Equal));
    depth_bids.sort_by(|a, b| b.price.partial_cmp(&a.price).unwrap_or(Ordering::Equal));

    // 限制总长度100
    depth_asks.truncate(100);
    depth_bids.truncate(100);
}