bybit使用ticker的版本

exchanges/src/bybit_swap_ws.rs

@@ -320,6 +320,7 @@ impl BybitSwapWs {
                 res_data.data_type = json_value["type"].as_str().unwrap().to_string();
             } else if channel.contains("tickers") {
                 res_data.channel = "tickers".to_string();
+                res_data.data["ts"] = json_value["ts"].clone();
             } else if channel.contains("kline") {
                 res_data.channel = "kline".to_string();
             } else if channel.contains("position") {

standard/src/bybit_swap_handle.rs

@@ -1,5 +1,6 @@
 use std::str::FromStr;
 use rust_decimal::Decimal;
+use rust_decimal::prelude::{FromPrimitive, ToPrimitive};
 use rust_decimal_macros::dec;
 use serde_json::{from_value, Value};
 use tokio::time::Instant;
@@ -139,26 +140,22 @@ pub fn format_order_item(order: serde_json::Value, ct_val: Decimal) -> Order {
     debug!("format-order-end, bybit_handle");
     return rst_order;
 }
+
 // 处理特殊Ticket信息
-pub fn handle_special_ticker(res_data: &ResponseData) -> SpecialDepth {
-    format_special_ticker(res_data.data.clone(), res_data.label.clone())
-}
+pub fn handle_ticker(res_data: &ResponseData) -> SpecialDepth {
+    let ap = Decimal::from_str(res_data.data["ask1Price"].as_str().unwrap()).unwrap();
+    let bp = Decimal::from_str(res_data.data["bid1Price"].as_str().unwrap()).unwrap();
+    let aq = Decimal::from_str(res_data.data["ask1Size"].as_str().unwrap()).unwrap();
+    let bq = Decimal::from_str(res_data.data["bid1Size"].as_str().unwrap()).unwrap();
+    let mp = (bp + ap) * dec!(0.5);
 
-pub fn format_special_ticker(data: serde_json::Value, label: String) -> SpecialDepth {
-    let depth_asks = format_depth_items(data["a"].clone());
-    let depth_bids = format_depth_items(data["b"].clone());
-    let t = Decimal::from_str(&data["ts"].to_string()).unwrap();
-    let create_at = data["t"].as_i64().unwrap() * 1000;
+    let t = Decimal::from_i64(res_data.data["ts"].as_i64().unwrap()).unwrap();
+    let create_at = t.to_i64().unwrap();
 
-    let ap = depth_asks[0].price;
-    let bp = depth_bids[0].price;
-    let aq = depth_asks[0].amount;
-    let bq = depth_bids[0].amount;
-    let mp = (bp + ap) * dec!(0.5);
-    let ticker_info = SpecialTicker { sell: ap, buy: bp, mid_price: mp, t, create_at };
+    let ticker_info = SpecialTicker { sell: ap, buy: bp, mid_price: mp, t, create_at: 0 };
     let depth_info = vec![bp, bq, ap, aq];
     SpecialDepth {
-        name: label,
+        name: res_data.label.clone(),
         depth: depth_info,
         ticker: ticker_info,
         t,

standard/src/handle_info.rs

@@ -78,7 +78,7 @@ impl HandleSwapInfo {
             //     bitget_spot_handle::handle_special_ticker(res_data)
             // },
             ExchangeEnum::BybitSwap => {
-                bybit_swap_handle::handle_special_ticker(res_data)
+                bybit_swap_handle::handle_ticker(res_data)
             }
         }
     }

strategy/src/binance_usdt_swap.rs

@@ -52,48 +52,53 @@ pub(crate) async fn reference_binance_swap_run(is_shutdown_arc: Arc<AtomicBool>,
 async fn on_data(core_arc_clone: Arc<Mutex<Core>>,
                  update_flag_u: &mut Decimal,
                  data: ResponseData) {
-    if data.code != "200".to_string() {
+    if data.code.as_str() != "200" {
         return;
     }
 
     let mut trace_stack = TraceStack::new(data.time, data.ins);
     trace_stack.on_after_span_line();
 
-    if data.channel == "aggTrade" {
-        // let trade: OriginalTradeBa = serde_json::from_str(data.data.as_str()).unwrap();
-        // let name = data.label.clone();
+    match data.channel.as_str() {
+        "aggTrade" => {
+            // let trade: OriginalTradeBa = serde_json::from_str(data.data.as_str()).unwrap();
+            // let name = data.label.clone();
 
-        // 订单流逻辑
-        // on_trade(trade.clone(), core_arc_clone.clone()).await;
+            // 订单流逻辑
+            // on_trade(trade.clone(), core_arc_clone.clone()).await;
 
-        // 原本的逻辑
-        // let mut core = core_arc_clone.lock().await;
-        // let str = data.label.clone();
-        // if core.is_update.contains_key(&data.label) && *core.is_update.get(str.as_str()).unwrap() {
-        //     *_max_buy = Decimal::ZERO;
-        //     *_min_sell = Decimal::ZERO;
-        //     core.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
-        // }
-        // core.max_buy_min_sell_cache.insert(data.label, vec![*_max_buy, *_min_sell]);
-    } else if data.channel == "bookTicker" {
-        trace_stack.set_source("binance_usdt_swap.bookTicker".to_string());
-        // 将ticker数据转换为模拟深度
-        let special_depth = standard::handle_info::HandleSwapInfo::handle_book_ticker(BinanceSwap, &data);
-        trace_stack.on_after_format();
+            // 原本的逻辑
+            // let mut core = core_arc_clone.lock().await;
+            // let str = data.label.clone();
+            // if core.is_update.contains_key(&data.label) && *core.is_update.get(str.as_str()).unwrap() {
+            //     *_max_buy = Decimal::ZERO;
+            //     *_min_sell = Decimal::ZERO;
+            //     core.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
+            // }
+            // core.max_buy_min_sell_cache.insert(data.label, vec![*_max_buy, *_min_sell]);
+        }
+        "bookTicker" => {
+            trace_stack.set_source("binance_usdt_swap.bookTicker".to_string());
+            // 将ticker数据转换为模拟深度
+            let special_depth = standard::handle_info::HandleSwapInfo::handle_book_ticker(BinanceSwap, &data);
+            trace_stack.on_after_format();
 
-        on_special_depth(core_arc_clone, update_flag_u, &data.label, &mut trace_stack, &special_depth).await;
-    } else if data.channel == "depth" {
-        trace_stack.set_source("binance_usdt_swap.depth".to_string());
-        // 将depth数据转换为模拟深度
-        let special_depth = standard::handle_info::HandleSwapInfo::handle_special_depth(BinanceSwap, &data);
-        trace_stack.on_after_format();
+            on_special_depth(core_arc_clone, update_flag_u, &data.label, &mut trace_stack, &special_depth).await;
+        }
+        "depth" => {
+            trace_stack.set_source("binance_usdt_swap.depth".to_string());
+            // 将depth数据转换为模拟深度
+            let special_depth = standard::handle_info::HandleSwapInfo::handle_special_depth(BinanceSwap, &data);
+            trace_stack.on_after_format();
 
-        on_special_depth(core_arc_clone, update_flag_u, &data.label, &mut trace_stack, &special_depth).await;
+            on_special_depth(core_arc_clone, update_flag_u, &data.label, &mut trace_stack, &special_depth).await;
+        }
+        _ => {}
     }
 }

strategy/src/bybit_usdt_swap.rs

@@ -6,12 +6,12 @@ use rust_decimal::Decimal;
 use tokio::{spawn, time};
 use tokio::sync::Mutex;
 use tokio::time::Instant;
-use tracing::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 crate::core::Core;
+use crate::exchange_disguise::on_special_depth;
 use crate::model::OrderInfo;
 
 // 1交易、0参考 bybit 合约 启动
@@ -30,11 +30,11 @@ pub async fn bybit_swap_run(is_shutdown_arc: Arc<AtomicBool>,
     ws_public.set_subscribe(vec![
         BybitSwapSubscribeType::PuTickers
     ]);
-    if is_trade {
-        ws_public.set_subscribe(vec![
-            BybitSwapSubscribeType::PuBlicTrade
-        ]);
-    }
+    // if is_trade {
+    //     ws_public.set_subscribe(vec![
+    //         BybitSwapSubscribeType::PuBlicTrade
+    //     ]);
+    // }
     // 挂起公共ws
     let write_tx_am_public = Arc::new(Mutex::new(write_tx_public));
     let is_shutdown_clone_public = Arc::clone(&is_shutdown_arc);
@@ -85,7 +85,7 @@ pub async fn bybit_swap_run(is_shutdown_arc: Arc<AtomicBool>,
                 let run_symbol_clone = run_symbol.clone();
 
                 async move {
-                    on_private_data(core_arc_clone.clone(), &ct_val, data, &run_symbol_clone).await;
+                    on_private_data(core_arc_clone.clone(), &ct_val, &run_symbol_clone, data).await;
                 }
             };
 
@@ -110,114 +110,120 @@ pub async fn bybit_swap_run(is_shutdown_arc: Arc<AtomicBool>,
     }
 }
 
-async fn on_private_data(core_arc_clone: Arc<Mutex<Core>>,
-                         ct_val: &Decimal,
-                         data: ResponseData,
-                         run_symbol: &String) {
+async fn on_private_data(core_arc_clone: Arc<Mutex<Core>>, ct_val: &Decimal, run_symbol: &String, data: ResponseData) {
     let mut trace_stack = TraceStack::new(data.time, data.ins);
     trace_stack.on_after_span_line();
 
-    if data.code != "200".to_string() {
+    if data.code.as_str() != "200" {
         return;
     }
-    if data.channel == "wallet" {
-        let account = standard::handle_info::HandleSwapInfo::handle_account_info(BybitSwap, &data, run_symbol);
-        {
+
+    match data.channel.as_str() {
+        "wallet" => {
+            let account = standard::handle_info::HandleSwapInfo::handle_account_info(BybitSwap, &data, run_symbol);
             let mut core = core_arc_clone.lock().await;
             core.update_equity(account).await;
         }
-    } else if data.channel == "order" {
-        let orders = standard::handle_info::HandleSwapInfo::handle_order(BybitSwap, data.clone(), ct_val.clone());
-        trace_stack.on_after_format();
+        "order" => {
+            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 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: TraceStack::new(0, Instant::now()),
+                };
+                order_infos.push(order_info);
             }
-            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: TraceStack::new(0, Instant::now()),
-            };
-            order_infos.push(order_info);
-        }
 
-        {
             let mut core = core_arc_clone.lock().await;
             core.update_order(order_infos, trace_stack).await;
         }
-    } else if data.channel == "position" {
-        let positions = standard::handle_info::HandleSwapInfo::handle_position(BybitSwap, &data, ct_val);
-        {
+        "position" => {
+            let positions = standard::handle_info::HandleSwapInfo::handle_position(BybitSwap, &data, ct_val);
             let mut core = core_arc_clone.lock().await;
             core.update_position(positions).await;
         }
+        _ => {}
     }
 }
 
-async fn on_public_data(_core_arc_clone: Arc<Mutex<Core>>,
-                        _update_flag_u: &mut Decimal,
-                        data: ResponseData) {
-    let mut trace_stack = TraceStack::new(data.time, data.ins);
-    trace_stack.on_after_span_line();
-
-    if data.code != "200".to_string() {
+async fn on_public_data(core_arc_clone: Arc<Mutex<Core>>, update_flag_u: &mut Decimal, response: ResponseData) {
+    if response.code.as_str() != "200" {
         return;
     }
 
-    info!(?data);
+    let mut trace_stack = TraceStack::new(response.time, response.ins);
+    trace_stack.on_after_span_line();
+
+    match response.channel.as_str() {
+        "tickers" => {
+            trace_stack.set_source("bybit_usdt_swap.tickers".to_string());
 
-    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;
-        // }
-        // 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(core_arc_clone, update_flag_u, &label, &mut trace_stack, &depth).await;
-    } else if data.channel == "trade" {
-        // let mut core = core_arc_clone.lock().await;
-        // let str = data.label.clone();
-        // if core.is_update.contains_key(&data.label) && *core.is_update.get(str.as_str()).unwrap(){
-        //     *max_buy = Decimal::ZERO;
-        //     *min_sell = Decimal::ZERO;
-        //     core.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
-        //     }
-        // }
-        // core.max_buy_min_sell_cache.insert(data.label, vec![*max_buy, *min_sell]);
+            // 处理ticker信息
+            let special_depth = standard::handle_info::HandleSwapInfo::handle_book_ticker(BybitSwap, &response);
+            trace_stack.on_after_format();
+
+            on_special_depth(core_arc_clone, update_flag_u, &response.label, &mut trace_stack, &special_depth).await;
+        }
+        "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;
+            // }
+            // 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(core_arc_clone, update_flag_u, &label, &mut trace_stack, &depth).await;
+        }
+        "trade" => {
+            // let mut core = core_arc_clone.lock().await;
+            // let str = data.label.clone();
+            // if core.is_update.contains_key(&data.label) && *core.is_update.get(str.as_str()).unwrap(){
+            //     *max_buy = Decimal::ZERO;
+            //     *min_sell = Decimal::ZERO;
+            //     core.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
+            //     }
+            // }
+            // core.max_buy_min_sell_cache.insert(data.label, vec![*max_buy, *min_sell]);
+        }
+        _ => {}
     }
 }
 

strategy/src/gate_swap.rs

@@ -98,84 +98,88 @@ async fn on_data(core_arc_clone: Arc<Mutex<Core>>,
                  multiplier: &Decimal,
                  run_symbol: &String,
                  data: ResponseData) {
-    if data.code != "200".to_string() {
+    if data.code.as_str() != "200" {
         return;
     }
 
     let mut trace_stack = TraceStack::new(data.time, data.ins);
     trace_stack.on_after_span_line();
 
-    if data.channel == "futures.order_book" {
-        trace_stack.set_source("gate_usdt_swap.order_book".to_string());
-        let special_depth = standard::handle_info::HandleSwapInfo::handle_special_depth(GateSwap, &data);
-        trace_stack.on_after_format();
-
-        on_special_depth(core_arc_clone, update_flag_u, &data.label, &mut trace_stack, &special_depth).await;
-    } else if data.channel == "futures.book_ticker" {
-        trace_stack.set_source("gate_usdt_swap.book_ticker".to_string());
-        // 将ticker数据转换为模拟深度
-        let special_depth = standard::handle_info::HandleSwapInfo::handle_book_ticker(GateSwap, &data);
-        trace_stack.on_after_format();
-
-        on_special_depth(core_arc_clone, update_flag_u, &data.label, &mut trace_stack, &special_depth).await;
-    } else if data.channel == "futures.balances" {
-        let account = standard::handle_info::HandleSwapInfo::handle_account_info(GateSwap, &data, run_symbol);
-        {
+    match data.channel.as_str() {
+        "futures.order_book" => {
+            trace_stack.set_source("gate_usdt_swap.order_book".to_string());
+            let special_depth = standard::handle_info::HandleSwapInfo::handle_special_depth(GateSwap, &data);
+            trace_stack.on_after_format();
+
+            on_special_depth(core_arc_clone, update_flag_u, &data.label, &mut trace_stack, &special_depth).await;
+        }
+        "futures.book_ticker" => {
+            trace_stack.set_source("gate_usdt_swap.book_ticker".to_string());
+            // 将ticker数据转换为模拟深度
+            let special_depth = standard::handle_info::HandleSwapInfo::handle_book_ticker(GateSwap, &data);
+            trace_stack.on_after_format();
+
+            on_special_depth(core_arc_clone, update_flag_u, &data.label, &mut trace_stack, &special_depth).await;
+        }
+        "futures.balances" => {
+            let account = standard::handle_info::HandleSwapInfo::handle_account_info(GateSwap, &data, run_symbol);
             let mut core = core_arc_clone.lock().await;
             core.update_equity(account).await;
         }
-    } else if data.channel == "futures.orders" {
-        trace_stack.set_source("gate_swap.orders".to_string());
-        let orders = standard::handle_info::HandleSwapInfo::handle_order(GateSwap, data.clone(), multiplier.clone());
-
-        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: TraceStack::new(0, Instant::now()),
-            };
-            order_infos.push(order_info);
-        }
+        "futures.orders" => {
+            trace_stack.set_source("gate_swap.orders".to_string());
+            let orders = standard::handle_info::HandleSwapInfo::handle_order(GateSwap, data.clone(), multiplier.clone());
+
+            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: TraceStack::new(0, Instant::now()),
+                };
+                order_infos.push(order_info);
+            }
 
-        {
-            let mut core = core_arc_clone.lock().await;
-            core.update_order(order_infos, trace_stack).await;
+            {
+                let mut core = core_arc_clone.lock().await;
+                core.update_order(order_infos, trace_stack).await;
+            }
         }
-    } else if data.channel == "futures.positions" {
-        let positions = standard::handle_info::HandleSwapInfo::handle_position(GateSwap, &data, multiplier);
-        {
+        "futures.positions" => {
+            let positions = standard::handle_info::HandleSwapInfo::handle_position(GateSwap, &data, multiplier);
             let mut core = core_arc_clone.lock().await;
             core.update_position(positions).await;
         }
-    } else if data.channel == "futures.trades" {
-        // let mut core = core_arc_clone.lock().await;
-        // let str = data.label.clone();
-        // if core.is_update.contains_key(&data.label) && *core.is_update.get(str.as_str()).unwrap(){
-        //     *max_buy = Decimal::ZERO;
-        //     *min_sell = Decimal::ZERO;
-        //     core.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
-        //     }
-        // }
-        // core.max_buy_min_sell_cache.insert(data.label, vec![*max_buy, *min_sell]);
+        "futures.trades" => {
+            // let mut core = core_arc_clone.lock().await;
+            // let str = data.label.clone();
+            // if core.is_update.contains_key(&data.label) && *core.is_update.get(str.as_str()).unwrap(){
+            //     *max_buy = Decimal::ZERO;
+            //     *min_sell = Decimal::ZERO;
+            //     core.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
+            //     }
+            // }
+            // core.max_buy_min_sell_cache.insert(data.label, vec![*max_buy, *min_sell]);
+        }
+        _ => {}
     }
 }