Merge branch 'dev' of http://git.skyfffire.com/skyfffire/as-rust into dev

strategy/src/binance_spot.rs

@@ -4,12 +4,12 @@ use std::sync::atomic::AtomicBool;
 use std::time::Duration;
 use rust_decimal::Decimal;
 use serde_json::Value;
-use tracing::{error};
 use tokio::spawn;
 use tokio::sync::mpsc::channel;
 use tokio::sync::Mutex;
 use tokio::time::sleep;
 use exchanges::binance_spot_ws::{BinanceSpotSubscribeType, BinanceSpotWs, BinanceSpotWsType};
+use exchanges::response_base::ResponseData;
 use global::trace_stack::TraceStack;
 use standard::exchange::ExchangeEnum::BinanceSpot;
 use standard::SpecialTicker;
@@ -38,79 +38,79 @@ pub async fn reference_binance_spot_run(bool_v1 :Arc<AtomicBool>, quant_arc: Arc
         let mut update_flag_u = 0i64;
         loop {
             sleep(Duration::from_millis(1)).await;
-            let mut trace_stack = TraceStack::default();
 
-            match rx.recv().await {
-                Some(data) => {
-                    trace_stack.on_network(data.time);
-                    trace_stack.on_before_quant();
-
-                    if data.code != "200".to_string() {
-                        continue;
-                    }
-                    if data.channel == "aggTrade" {
-                        let trade: OriginalTradeBa = serde_json::from_str(data.data.as_str()).unwrap();
-                        let str = data.label.clone();
-                        let mut quant = bot_arc_clone.lock().await;
-                        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());
-                        }
-                        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]);
-                        }
-                    } else if data.channel == "bookTicker" {
-                        trace_stack.on_before_format();
-                        let ticker: OriginalTicker = serde_json::from_str(data.data.as_str()).unwrap();
-                        if ticker.u > update_flag_u {
-                            {
-                                let mut quant = bot_arc_clone.lock().await;
-                                // time_delay.quant_start = Utc::now().timestamp_micros();
-                                quant._update_ticker(SpecialTicker{
-                                    sell: ticker.a.clone(),
-                                    buy: ticker.b.clone(),
-                                    mid_price: Default::default(),
-                                }, data.label.clone());
-                                let depth = vec![ticker.b, ticker.B, ticker.a, ticker.A];
-                                trace_stack.on_after_format();
-                                quant._update_depth(depth.clone(), data.label.clone(), &mut trace_stack);
-                                quant.local_depths.insert(data.label.clone(), depth);
-                            }
-                        } else {
-                            update_flag_u = ticker.u;
-                        }
-                    } else if data.channel == "depth" {
-                        trace_stack.on_before_format();
-                        let v: Value = serde_json::from_str(data.data.clone().as_str()).unwrap();
-                        let u = v["lastUpdateId"].as_i64().unwrap();
-                        if u > update_flag_u {
-                            let depth = standard::handle_info::HandleSwapInfo::handle_special_depth(BinanceSpot, data);
-                            trace_stack.on_after_format();
-                            {
-                                let mut quant = bot_arc_clone.lock().await;
-                                quant._update_ticker(depth.ticker.clone(), depth.name.clone());
-                                quant._update_depth(depth.depth.clone(), depth.name.clone(), &mut trace_stack);
-                                quant.local_depths.insert(depth.name, depth.depth);
-                            }
-                        } else {
-                            update_flag_u = u;
-                        }
-
-                    }
+            match rx.try_recv() {
+                Ok(data) => {
+                    on_data(bot_arc_clone.clone(), &mut update_flag_u, &mut max_buy, &mut min_sell, data).await;
                 },
-                None => {
-                    error!("币安现货参考交易所通道错误");
-                    break;
-                }
+                Err(_e) => { }
             }
 
         }
     });
-}
+}
+
+async fn on_data(bot_arc_clone: Arc<Mutex<Quant>>, update_flag_u: &mut i64, max_buy: &mut Decimal, min_sell: &mut Decimal, data: ResponseData) {
+    let mut trace_stack = TraceStack::default();
+    trace_stack.on_network(data.time);
+    trace_stack.on_before_quant();
+
+    if data.code != "200".to_string() {
+        return;
+    }
+    if data.channel == "aggTrade" {
+        let trade: OriginalTradeBa = serde_json::from_str(data.data.as_str()).unwrap();
+        let str = data.label.clone();
+        let mut quant = bot_arc_clone.lock().await;
+        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());
+        }
+        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]);
+        }
+    } else if data.channel == "bookTicker" {
+        trace_stack.on_before_format();
+        let ticker: OriginalTicker = serde_json::from_str(data.data.as_str()).unwrap();
+        if ticker.u > *update_flag_u {
+            {
+                let mut quant = bot_arc_clone.lock().await;
+                // time_delay.quant_start = Utc::now().timestamp_micros();
+                quant._update_ticker(SpecialTicker{
+                    sell: ticker.a.clone(),
+                    buy: ticker.b.clone(),
+                    mid_price: Default::default(),
+                }, data.label.clone());
+                let depth = vec![ticker.b, ticker.B, ticker.a, ticker.A];
+                trace_stack.on_after_format();
+                quant._update_depth(depth.clone(), data.label.clone(), &mut trace_stack);
+                quant.local_depths.insert(data.label.clone(), depth);
+            }
+        } else {
+            *update_flag_u = ticker.u;
+        }
+    } else if data.channel == "depth" {
+        trace_stack.on_before_format();
+        let v: Value = serde_json::from_str(data.data.clone().as_str()).unwrap();
+        let u = v["lastUpdateId"].as_i64().unwrap();
+        if u > *update_flag_u {
+            let depth = standard::handle_info::HandleSwapInfo::handle_special_depth(BinanceSpot, data);
+            trace_stack.on_after_format();
+            {
+                let mut quant = bot_arc_clone.lock().await;
+                quant._update_ticker(depth.ticker.clone(), depth.name.clone());
+                quant._update_depth(depth.depth.clone(), depth.name.clone(), &mut trace_stack);
+                quant.local_depths.insert(depth.name, depth.depth);
+            }
+        } else {
+            *update_flag_u = u;
+        }
+    }
+}

strategy/src/binance_swap.rs

@@ -1,111 +0,0 @@
-use std::collections::BTreeMap;
-use std::sync::Arc;
-use std::sync::atomic::AtomicBool;
-use std::time::Duration;
-use rust_decimal::Decimal;
-use serde_json::Value;
-use tracing::{error};
-use tokio::spawn;
-use tokio::sync::mpsc::channel;
-use tokio::sync::Mutex;
-use tokio::time::sleep;
-use exchanges::binance_swap_ws::{BinanceSubscribeType, BinanceSwapWs, BinanceWsType};
-use global::trace_stack::TraceStack;
-use standard::exchange::ExchangeEnum::BinanceSwap;
-use standard::SpecialTicker;
-use crate::model::{OriginalTicker, OriginalTradeBa};
-use crate::quant::Quant;
-
-// 参考 币安 合约 启动
-pub(crate) async fn reference_binance_swap_run(bool_v1 :Arc<AtomicBool>, quant_arc: Arc<Mutex<Quant>>, name: String, symbols: Vec<String>, exchange_params: BTreeMap<String, String>){
-    let (tx, mut rx) = channel(100);
-    spawn( async move {
-        let mut ba_exc = BinanceSwapWs::new_label(name, false, exchange_params, BinanceWsType::PublicAndPrivate, tx);
-        ba_exc.set_subscribe(vec![
-            BinanceSubscribeType::PuBookTicker,
-            BinanceSubscribeType::PuAggTrade
-        ]);
-        ba_exc.custom_subscribe(bool_v1, symbols.clone()).await;
-    });
-    spawn(async move {
-        // trade
-        let mut max_buy = Decimal::ZERO;
-        let mut min_sell = Decimal::ZERO;
-        // ticker
-        let mut update_flag_u = 0i64;
-        let bot_arc_clone = Arc::clone(&quant_arc);
-        loop {
-            sleep(Duration::from_millis(1)).await;
-            let mut trace_stack = TraceStack::default();
-
-            match rx.recv().await {
-                Some(data) => {
-                    trace_stack.on_network(data.time);
-                    trace_stack.on_before_quant();
-
-                    if data.code != "200".to_string() {
-                        continue;
-                    }
-                    if data.channel == "aggTrade" {
-                        let trade: OriginalTradeBa = serde_json::from_str(data.data.as_str()).unwrap();
-                        let str = data.label.clone();
-                        let mut quant = bot_arc_clone.lock().await;
-                        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());
-                        }
-                        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]);
-                        }
-                    } else if data.channel == "bookTicker" {
-                        trace_stack.on_before_format();
-                        let ticker: OriginalTicker = serde_json::from_str(data.data.as_str()).unwrap();
-                        if ticker.u > update_flag_u {
-                            {
-                                let mut quant = bot_arc_clone.lock().await;
-                                quant._update_ticker(SpecialTicker{
-                                    sell: ticker.a.clone(),
-                                    buy: ticker.b.clone(),
-                                    mid_price: Default::default(),
-                                }, data.label.clone());
-                                let depth = vec![ticker.b, ticker.B, ticker.a, ticker.A];
-                                trace_stack.on_after_format();
-                                quant._update_depth(depth.clone(), data.label.clone(), &mut trace_stack);
-                                quant.local_depths.insert(data.label.clone(), depth);
-                            }
-                        } else {
-                            update_flag_u = ticker.u;
-                        }
-                    } else if data.channel == "depth" {
-                        trace_stack.on_before_format();
-                        let v: Value = serde_json::from_str(data.data.clone().as_str()).unwrap();
-                        let u = v["u"].as_i64().unwrap();
-                        if u > update_flag_u {
-                            let depth = standard::handle_info::HandleSwapInfo::handle_special_depth(BinanceSwap, data);
-                            trace_stack.on_after_format();
-                            {
-                                let mut quant = bot_arc_clone.lock().await;
-                                quant._update_ticker(depth.ticker.clone(), depth.name.clone());
-                                quant._update_depth(depth.depth.clone(), depth.name.clone(), &mut trace_stack);
-                                quant.local_depths.insert(depth.name, depth.depth);
-                            }
-                        } else {
-                            update_flag_u = u;
-                        }
-                    }
-                },
-                None => {
-                    error!("参考交易所通道错误");
-                    break;
-                }
-            }
-        }
-    });
-}

strategy/src/binance_usdt_swap.rs

@@ -0,0 +1,114 @@
+use std::collections::BTreeMap;
+use std::sync::Arc;
+use std::sync::atomic::AtomicBool;
+use std::time::Duration;
+use rust_decimal::Decimal;
+use serde_json::Value;
+use tokio::spawn;
+use tokio::sync::mpsc::channel;
+use tokio::sync::Mutex;
+use tokio::time::sleep;
+use exchanges::binance_swap_ws::{BinanceSubscribeType, BinanceSwapWs, BinanceWsType};
+use exchanges::response_base::ResponseData;
+use global::trace_stack::TraceStack;
+use standard::exchange::ExchangeEnum::BinanceSwap;
+use standard::SpecialTicker;
+use crate::model::{OriginalTicker, OriginalTradeBa};
+use crate::quant::Quant;
+
+// 参考 币安 合约 启动
+pub(crate) async fn reference_binance_swap_run(bool_v1 :Arc<AtomicBool>, quant_arc: Arc<Mutex<Quant>>, name: String, symbols: Vec<String>, exchange_params: BTreeMap<String, String>){
+    let (tx, mut rx) = channel(100);
+    spawn( async move {
+        let mut ba_exc = BinanceSwapWs::new_label(name, false, exchange_params, BinanceWsType::PublicAndPrivate, tx);
+        ba_exc.set_subscribe(vec![
+            BinanceSubscribeType::PuBookTicker,
+            BinanceSubscribeType::PuAggTrade
+        ]);
+        ba_exc.custom_subscribe(bool_v1, symbols.clone()).await;
+    });
+    let bot_arc_clone = Arc::clone(&quant_arc);
+    spawn(async move {
+        // ticker
+        let mut update_flag_u = 0i64;
+        // trade
+        let mut max_buy = Decimal::ZERO;
+        let mut min_sell = Decimal::ZERO;
+        loop {
+            sleep(Duration::from_millis(1)).await;
+
+            match rx.try_recv() {
+                Ok(data) => {
+                    on_data(bot_arc_clone.clone(), &mut update_flag_u, &mut max_buy, &mut min_sell, data).await;
+                },
+                Err(_e) => { }
+            }
+        }
+    });
+}
+
+async fn on_data(bot_arc_clone: Arc<Mutex<Quant>>, update_flag_u: &mut i64, max_buy: &mut Decimal, min_sell: &mut Decimal, data: ResponseData) {
+    let mut trace_stack = TraceStack::default();
+    trace_stack.on_network(data.time);
+    trace_stack.on_before_quant();
+
+    if data.code != "200".to_string() {
+        return;
+    }
+
+    if data.channel == "aggTrade" {
+
+        let trade: OriginalTradeBa = serde_json::from_str(data.data.as_str()).unwrap();
+        let str = data.label.clone();
+        let mut quant = bot_arc_clone.lock().await;
+        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());
+        }
+        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]);
+        }
+    } else if data.channel == "bookTicker" {
+        trace_stack.on_before_format();
+        let ticker: OriginalTicker = serde_json::from_str(data.data.as_str()).unwrap();
+        if ticker.u > *update_flag_u {
+            {
+                let mut quant = bot_arc_clone.lock().await;
+                quant._update_ticker(SpecialTicker{
+                    sell: ticker.a.clone(),
+                    buy: ticker.b.clone(),
+                    mid_price: Default::default(),
+                }, data.label.clone());
+                let depth = vec![ticker.b, ticker.B, ticker.a, ticker.A];
+                trace_stack.on_after_format();
+                quant._update_depth(depth.clone(), data.label.clone(), &mut trace_stack);
+                quant.local_depths.insert(data.label.clone(), depth);
+            }
+        } else {
+            *update_flag_u = ticker.u;
+        }
+    } else if data.channel == "depth" {
+        trace_stack.on_before_format();
+        let v: Value = serde_json::from_str(data.data.clone().as_str()).unwrap();
+        let u = v["u"].as_i64().unwrap();
+        if u > *update_flag_u {
+            let depth = standard::handle_info::HandleSwapInfo::handle_special_depth(BinanceSwap, data);
+            trace_stack.on_after_format();
+            {
+                let mut quant = bot_arc_clone.lock().await;
+                quant._update_ticker(depth.ticker.clone(), depth.name.clone());
+                quant._update_depth(depth.depth.clone(), depth.name.clone(), &mut trace_stack);
+                quant.local_depths.insert(depth.name, depth.depth);
+            }
+        } else {
+            *update_flag_u = u;
+        }
+    }
+}

strategy/src/bitget_spot.rs

@@ -3,12 +3,12 @@ use std::sync::Arc;
 use std::sync::atomic::AtomicBool;
 use std::time::Duration;
 use rust_decimal::Decimal;
-use tracing::{error};
 use tokio::spawn;
 use tokio::sync::mpsc::channel;
 use tokio::sync::Mutex;
 use tokio::time::sleep;
 use exchanges::bitget_spot_ws::{BitgetSpotWs, BitgetSubscribeType, BitgetWsType};
+use exchanges::response_base::ResponseData;
 use global::trace_stack::TraceStack;
 use standard::exchange::ExchangeEnum::BitgetSpot;
 use crate::model::{OrderInfo, OriginalTradeGa};
@@ -66,98 +66,98 @@ pub async fn bitget_spot_run(bool_v1 :Arc<AtomicBool>, type_num: i8, quant_arc:
     });
 
     spawn(async move {
-        let bot_arc_clone = Arc::clone(&quant_arc);
-        let ct_val = bot_arc_clone.lock().await.platform_rest.get_self_market().ct_val;
+        let ct_val = quant_arc.clone().lock().await.platform_rest.get_self_market().ct_val;
         // trade
         let mut max_buy = Decimal::ZERO;
         let mut min_sell = Decimal::ZERO;
-        // let run_symbol = symbols.clone()[0].clone();
 
         loop {
-            sleep(Duration::from_millis(1)).await;
-            let mut trace_stack = TraceStack::default();
+            sleep(Duration::from_micros(100)).await;
 
-            match rx.recv().await {
-                Some(data) => {
-                    trace_stack.on_network(data.time);
-                    trace_stack.on_before_quant();
-
-                    if data.code != "200".to_string() {
-                        continue;
-                    }
-                    if data.channel == "books5" {
-                        trace_stack.on_before_format();
-                        let depth = standard::handle_info::HandleSwapInfo::handle_special_depth(BitgetSpot, data);
-                        trace_stack.on_after_format();
-                        {
-                            let mut quant = bot_arc_clone.lock().await;
-                            quant._update_ticker(depth.ticker.clone(), depth.name.clone());
-                            quant._update_depth(depth.depth.clone(), depth.name.clone(), &mut trace_stack);
-                            quant.local_depths.insert(depth.name, depth.depth);
-                        }
-                    } 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<OriginalTradeGa> = serde_json::from_str(data.data.as_str()).unwrap();
-                        for trade in trades {
-                            if trade.price > max_buy || max_buy == Decimal::ZERO{
-                                max_buy = trade.price
-                            }
-                            if trade.price < min_sell || min_sell == Decimal::ZERO{
-                                min_sell = trade.price
-                            }
-                        }
-                        quant.max_buy_min_sell_cache.insert(data.label, vec![max_buy, min_sell]);
-                    } else if data.channel == "orders" {
-                        trace_stack.on_before_format();
-                        let orders = standard::handle_info::HandleSwapInfo::handle_order(BitgetSpot, data.clone(), ct_val);
-                        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 == "account" {
-                        // info!(?data);
-                        // let account = standard::handle_info::HandleSwapInfo::handle_account_info(BitgetSpot, data, run_symbol.clone());
-                        // {
-                        //     info!(?account);
-                        //     let mut quant = bot_arc_clone.lock().await;
-                        //     quant.update_equity(account);
-                        // }
-                    }
+            match rx.try_recv() {
+                Ok(data) => {
+                    on_data(quant_arc.clone(), ct_val, &mut max_buy, &mut min_sell, data).await;
                 },
-                None => {
-                    error!("参考交易所通道错误");
-                    break;
-                }
+                Err(_e) => {}
             }
         }
     });
-}
+}
+
+async fn on_data(bot_arc_clone: Arc<Mutex<Quant>>, ct_val: Decimal, max_buy: &mut Decimal, min_sell: &mut Decimal, data: ResponseData) {
+    let mut trace_stack = TraceStack::default();
+
+    trace_stack.on_network(data.time);
+    trace_stack.on_before_quant();
+
+    if data.code != "200".to_string() {
+        return;
+    }
+    if data.channel == "books5" {
+        trace_stack.on_before_format();
+        let depth = standard::handle_info::HandleSwapInfo::handle_special_depth(BitgetSpot, data);
+        trace_stack.on_after_format();
+        {
+            let mut quant = bot_arc_clone.lock().await;
+            quant._update_ticker(depth.ticker.clone(), depth.name.clone());
+            quant._update_depth(depth.depth.clone(), depth.name.clone(), &mut trace_stack);
+            quant.local_depths.insert(depth.name, depth.depth);
+        }
+    } 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<OriginalTradeGa> = serde_json::from_str(data.data.as_str()).unwrap();
+        for trade in trades {
+            if trade.price > *max_buy || *max_buy == Decimal::ZERO{
+                *max_buy = trade.price
+            }
+            if trade.price < *min_sell || *min_sell == Decimal::ZERO{
+                *min_sell = trade.price
+            }
+        }
+        quant.max_buy_min_sell_cache.insert(data.label, vec![*max_buy, *min_sell]);
+    } else if data.channel == "orders" {
+        trace_stack.on_before_format();
+        let orders = standard::handle_info::HandleSwapInfo::handle_order(BitgetSpot, data.clone(), ct_val);
+        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 == "account" {
+        // info!(?data);
+        // let account = standard::handle_info::HandleSwapInfo::handle_account_info(BitgetSpot, data, run_symbol.clone());
+        // {
+        //     info!(?account);
+        //     let mut quant = bot_arc_clone.lock().await;
+        //     quant.update_equity(account);
+        // }
+    }
+}

strategy/src/exchange_disguise.rs

@@ -3,7 +3,7 @@ use std::sync::Arc;
 use std::sync::atomic::AtomicBool;
 use tokio::sync::Mutex;
 use crate::binance_spot::reference_binance_spot_run;
-use crate::binance_swap::reference_binance_swap_run;
+use crate::binance_usdt_swap::reference_binance_swap_run;
 use crate::bitget_spot::bitget_spot_run;
 use crate::gate_swap::gate_swap_run;
 use crate::kucoin_swap::kucoin_swap_run;

strategy/src/gate_swap.rs

@@ -4,13 +4,14 @@ use std::sync::atomic::AtomicBool;
 use std::time::Duration;
 use rust_decimal::Decimal;
 use serde_json::Value;
-use tracing::{error, info};
+use tracing::{info};
 use tokio::spawn;
 use tokio::sync::mpsc::channel;
 use tokio::sync::Mutex;
 use tokio::time::sleep;
 use exchanges::gate_swap_rest::GateSwapRest;
 use exchanges::gate_swap_ws::{GateSubscribeType, GateSwapWs, GateWsType};
+use exchanges::response_base::ResponseData;
 use global::trace_stack::TraceStack;
 use standard::exchange::ExchangeEnum::GateSwap;
 use crate::model::{OrderInfo, OriginalTradeGa};
@@ -63,93 +64,95 @@ pub async fn gate_swap_run(bool_v1 :Arc<AtomicBool>, type_num: i8, quant_arc: Ar
         let mut min_sell = Decimal::ZERO;
         loop {
             sleep(Duration::from_millis(1)).await;
-            let mut trace_stack = TraceStack::default();
 
-            match rx.recv().await {
-                Some(data) => {
-                    trace_stack.on_network(data.time);
-                    trace_stack.on_before_quant();
+            match rx.try_recv() {
+                Ok(data) => {
+                    on_data(bot_arc_clone.clone(), multiplier, run_symbol.clone(), &mut max_buy, &mut min_sell, data).await;
+                },
+                Err(_e) => {}
+            }
+        }
+    });
+}
 
-                    if data.code != "200".to_string() {
-                        continue;
-                    }
-                    if data.channel == "futures.order_book" {
-                        trace_stack.on_before_format();
-                        let depth = standard::handle_info::HandleSwapInfo::handle_special_depth(GateSwap, data);
-                        trace_stack.on_after_format();
+async fn on_data(bot_arc_clone: Arc<Mutex<Quant>>, multiplier: Decimal, run_symbol: String, max_buy: &mut Decimal, min_sell: &mut Decimal, data: ResponseData) {
+    let mut trace_stack = TraceStack::default();
+    trace_stack.on_network(data.time);
+    trace_stack.on_before_quant();
 
-                        {
-                            let mut quant = bot_arc_clone.lock().await;
-                            quant._update_ticker(depth.ticker.clone(), depth.name.clone());
-                            quant._update_depth(depth.depth.clone(), depth.name.clone(), &mut trace_stack);
-                            quant.local_depths.insert(depth.name, depth.depth);
-                        }
-                    } else if data.channel == "futures.balances" {
-                        let account = standard::handle_info::HandleSwapInfo::handle_account_info(GateSwap, data, run_symbol.clone());
-                        {
-                            let mut quant = bot_arc_clone.lock().await;
-                            quant.update_equity(account);
-                        }
-                    } else if data.channel == "futures.orders" {
-                        trace_stack.on_before_format();
-                        let orders = standard::handle_info::HandleSwapInfo::handle_order(GateSwap, data.clone(), multiplier.clone());
-                        trace_stack.on_after_format();
+    if data.code != "200".to_string() {
+        return;
+    }
 
-                        let mut order_infos:Vec<OrderInfo> = Vec::new();
-                        for order in orders.order {
-                            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);
-                        }
+    if data.channel == "futures.order_book" {
+        trace_stack.on_before_format();
+        let depth = standard::handle_info::HandleSwapInfo::handle_special_depth(GateSwap, data);
+        trace_stack.on_after_format();
 
-                        {
-                            let mut quant = bot_arc_clone.lock().await;
-                            quant.update_order(order_infos, trace_stack);
-                        }
-                    } else if data.channel == "futures.positions" {
-                        let positions = standard::handle_info::HandleSwapInfo::handle_position(GateSwap,data, multiplier.clone());
-                        {
-                            let mut quant = bot_arc_clone.lock().await;
-                            quant.update_position(positions);
-                        }
-                    } else if data.channel == "futures.trades" {
-                        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<OriginalTradeGa> = serde_json::from_str(data.data.as_str()).unwrap();
-                        for trade in trades {
-                            if trade.price > max_buy || max_buy == Decimal::ZERO{
-                                max_buy = trade.price
-                            }
-                            if trade.price < min_sell || min_sell == Decimal::ZERO{
-                                min_sell = trade.price
-                            }
-                        }
-                        quant.max_buy_min_sell_cache.insert(data.label, vec![max_buy, min_sell]);
-                    }
-                },
-                None => {
-                    error!("交易所通道错误");
-                    break;
-                }
+        {
+            let mut quant = bot_arc_clone.lock().await;
+            quant._update_ticker(depth.ticker.clone(), depth.name.clone());
+            quant._update_depth(depth.depth.clone(), depth.name.clone(), &mut trace_stack);
+            quant.local_depths.insert(depth.name, depth.depth);
+        }
+    } else if data.channel == "futures.balances" {
+        let account = standard::handle_info::HandleSwapInfo::handle_account_info(GateSwap, data, run_symbol.clone());
+        {
+            let mut quant = bot_arc_clone.lock().await;
+            quant.update_equity(account);
+        }
+    } else if data.channel == "futures.orders" {
+        trace_stack.on_before_format();
+        let orders = standard::handle_info::HandleSwapInfo::handle_order(GateSwap, data.clone(), multiplier.clone());
+        trace_stack.on_after_format();
+
+        let mut order_infos:Vec<OrderInfo> = Vec::new();
+        for order in orders.order {
+            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 == "futures.positions" {
+        let positions = standard::handle_info::HandleSwapInfo::handle_position(GateSwap,data, multiplier.clone());
+        {
+            let mut quant = bot_arc_clone.lock().await;
+            quant.update_position(positions);
+        }
+    } else if data.channel == "futures.trades" {
+        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<OriginalTradeGa> = serde_json::from_str(data.data.as_str()).unwrap();
+        for trade in trades {
+            if trade.price > *max_buy || *max_buy == Decimal::ZERO{
+                *max_buy = trade.price
+            }
+            if trade.price < *min_sell || *min_sell == Decimal::ZERO{
+                *min_sell = trade.price
             }
         }
-    });
-}
+        quant.max_buy_min_sell_cache.insert(data.label, vec![*max_buy, *min_sell]);
+    }
+}

strategy/src/kucoin_swap.rs

@@ -3,12 +3,12 @@ use std::sync::Arc;
 use std::sync::atomic::AtomicBool;
 use std::time::Duration;
 use rust_decimal::Decimal;
-use tracing::{error};
 use tokio::spawn;
 use tokio::sync::mpsc::channel;
 use tokio::sync::Mutex;
 use tokio::time::sleep;
 use exchanges::kucoin_swap_ws::{KucoinSubscribeType, KucoinSwapWs, KucoinWsType};
+use exchanges::response_base::ResponseData;
 use global::trace_stack::TraceStack;
 use standard::exchange::ExchangeEnum::KucoinSwap;
 use crate::model::{OrderInfo, OriginalTradeGa};
@@ -70,100 +70,101 @@ pub async fn kucoin_swap_run(bool_v1 :Arc<AtomicBool>, type_num: i8, quant_arc:
         let multiplier = bot_arc_clone.lock().await.platform_rest.get_self_market().ct_val;
         loop {
             sleep(Duration::from_millis(1)).await;
-            let mut trace_stack = TraceStack::default();
 
-            match rx.recv().await {
-                Some(data) => {
-                    trace_stack.on_network(data.time);
-                    trace_stack.on_before_quant();
-
-                    if data.code != "200".to_string() {
-                        continue;
-                    }
-                    if data.channel == "level2" {
-                        trace_stack.on_before_format();
-                        let depth = standard::handle_info::HandleSwapInfo::handle_special_depth(KucoinSwap,data);
-                        trace_stack.on_after_format();
-                        {
-                            let mut quant = bot_arc_clone.lock().await;
-                            // time_delay.quant_start = Utc::now().timestamp_micros();
-                            quant._update_ticker(depth.ticker.clone(), depth.name.clone());
-                            quant._update_depth(depth.depth.clone(), depth.name.clone(), &mut trace_stack);
-                            quant.local_depths.insert(depth.name, depth.depth);
-                        }
-                    } else if data.channel == "tickerV2" {
-                        let ticker = standard::handle_info::HandleSwapInfo::handle_special_ticker(KucoinSwap, data);
-                        {
-                            let mut quant = bot_arc_clone.lock().await;
-                            quant._update_ticker(ticker.ticker, ticker.name);
-                        }
-                    } else if data.channel == "availableBalance.change" {
-                        // 取消原有推送解析，因为推送的信息不准确
-                        // let account = standard::handle_info::HandleSwapInfo::handle_account_info(KucoinSwap, data, run_symbol.clone());
-                        // {
-                        //     let mut quant = bot_arc_clone.lock().await;
-                        //     quant.update_equity(account);
-                        // }
-                    } else if data.channel == "symbolOrderChange" {
-                        trace_stack.on_before_format();
-                        let orders = standard::handle_info::HandleSwapInfo::handle_order(KucoinSwap, data.clone(), multiplier);
-                        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.change" {
-                        let positions = standard::handle_info::HandleSwapInfo::handle_position(KucoinSwap,data, multiplier);
-                        {
-                            let mut quant = bot_arc_clone.lock().await;
-                            quant.update_position(positions);
-                        }
-                    } else if data.channel == "match" {
-                        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 trade: OriginalTradeGa = serde_json::from_str(data.data.as_str()).unwrap();
-                        if trade.price > max_buy || max_buy == Decimal::ZERO {
-                            max_buy = trade.price
-                        }
-                        if trade.price < min_sell || min_sell == Decimal::ZERO {
-                            min_sell = trade.price
-                        }
-                        quant.max_buy_min_sell_cache.insert(data.label, vec![max_buy, min_sell]);
-                    }
+            match rx.try_recv() {
+                Ok(data) => {
+                    on_data(bot_arc_clone.clone(), multiplier, &mut max_buy, &mut min_sell, data).await;
                 },
-                None => {
-                    error!("交易所通道错误");
-                    break;
-                }
+                Err(_e) => { }
             }
         }
     });
-}
+}
+
+async fn on_data(bot_arc_clone: Arc<Mutex<Quant>>, multiplier: Decimal, max_buy: &mut Decimal, min_sell: &mut Decimal, data: ResponseData) {
+    let mut trace_stack = TraceStack::default();
+    trace_stack.on_network(data.time);
+    trace_stack.on_before_quant();
+
+    if data.code != "200".to_string() {
+        return;
+    }
+    if data.channel == "level2" {
+        trace_stack.on_before_format();
+        let depth = standard::handle_info::HandleSwapInfo::handle_special_depth(KucoinSwap,data);
+        trace_stack.on_after_format();
+        {
+            let mut quant = bot_arc_clone.lock().await;
+            // time_delay.quant_start = Utc::now().timestamp_micros();
+            quant._update_ticker(depth.ticker.clone(), depth.name.clone());
+            quant._update_depth(depth.depth.clone(), depth.name.clone(), &mut trace_stack);
+            quant.local_depths.insert(depth.name, depth.depth);
+        }
+    } else if data.channel == "tickerV2" {
+        let ticker = standard::handle_info::HandleSwapInfo::handle_special_ticker(KucoinSwap, data);
+        {
+            let mut quant = bot_arc_clone.lock().await;
+            quant._update_ticker(ticker.ticker, ticker.name);
+        }
+    } else if data.channel == "availableBalance.change" {
+        // 取消原有推送解析，因为推送的信息不准确
+        // let account = standard::handle_info::HandleSwapInfo::handle_account_info(KucoinSwap, data, run_symbol.clone());
+        // {
+        //     let mut quant = bot_arc_clone.lock().await;
+        //     quant.update_equity(account);
+        // }
+    } else if data.channel == "symbolOrderChange" {
+        trace_stack.on_before_format();
+        let orders = standard::handle_info::HandleSwapInfo::handle_order(KucoinSwap, data.clone(), multiplier);
+        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.change" {
+        let positions = standard::handle_info::HandleSwapInfo::handle_position(KucoinSwap,data, multiplier);
+        {
+            let mut quant = bot_arc_clone.lock().await;
+            quant.update_position(positions);
+        }
+    } else if data.channel == "match" {
+        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 trade: OriginalTradeGa = serde_json::from_str(data.data.as_str()).unwrap();
+        if trade.price > *max_buy || *max_buy == Decimal::ZERO {
+            *max_buy = trade.price
+        }
+        if trade.price < *min_sell || *min_sell == Decimal::ZERO {
+            *min_sell = trade.price
+        }
+        quant.max_buy_min_sell_cache.insert(data.label, vec![*max_buy, *min_sell]);
+    }
+}

strategy/src/lib.rs

@@ -5,7 +5,7 @@ mod predictor;
 mod utils;
 pub mod exchange_disguise;
 mod gp_predictor;
-mod binance_swap;
+mod binance_usdt_swap;
 mod binance_spot;
 mod gate_swap;
 mod kucoin_swap;