|
|
@@ -1,286 +1,286 @@
|
|
|
-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 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::core::Core;
|
|
|
-use crate::exchange_disguise::on_special_depth;
|
|
|
-
|
|
|
-// 1交易、0参考 bybit 合约 启动
|
|
|
-pub async fn bybit_swap_run(bool_v1: Arc<AtomicBool>,
|
|
|
- is_trade: bool,
|
|
|
- _core_arc: Arc<Mutex<Core>>,
|
|
|
- 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 = _core_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 = _core_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_core_clone = _core_arc.clone();
|
|
|
- spawn(async move {
|
|
|
- let mut interval = time::interval(Duration::from_secs(30));
|
|
|
- loop {
|
|
|
- interval.tick().await;
|
|
|
- {
|
|
|
- let mut core = position_core_clone.lock().await;
|
|
|
- core.update_position_rest_swap().await;
|
|
|
- }
|
|
|
-
|
|
|
- }
|
|
|
- });
|
|
|
- }
|
|
|
-}
|
|
|
-
|
|
|
-
|
|
|
-
|
|
|
-async fn on_private_data(bot_arc_clone: Arc<Mutex<Core>>, 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_unlock_core();
|
|
|
-
|
|
|
- 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 core = bot_arc_clone.lock().await;
|
|
|
- core.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 core = bot_arc_clone.lock().await;
|
|
|
- core.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 core = bot_arc_clone.lock().await;
|
|
|
- core.update_position(positions).await;
|
|
|
- }
|
|
|
- }
|
|
|
-}
|
|
|
-
|
|
|
-async fn on_public_data(bot_arc_clone: Arc<Mutex<Core>>, 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_unlock_core();
|
|
|
-
|
|
|
- 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 core = bot_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]);
|
|
|
- }
|
|
|
-}
|
|
|
-
|
|
|
-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);
|
|
|
-}
|
|
|
+// 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 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::core::Core;
|
|
|
+// use crate::exchange_disguise::on_special_depth;
|
|
|
+//
|
|
|
+// // 1交易、0参考 bybit 合约 启动
|
|
|
+// pub async fn bybit_swap_run(bool_v1: Arc<AtomicBool>,
|
|
|
+// is_trade: bool,
|
|
|
+// _core_arc: Arc<Mutex<Core>>,
|
|
|
+// 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 = _core_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 = _core_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_core_clone = _core_arc.clone();
|
|
|
+// spawn(async move {
|
|
|
+// let mut interval = time::interval(Duration::from_secs(30));
|
|
|
+// loop {
|
|
|
+// interval.tick().await;
|
|
|
+// {
|
|
|
+// let mut core = position_core_clone.lock().await;
|
|
|
+// core.update_position_rest_swap().await;
|
|
|
+// }
|
|
|
+//
|
|
|
+// }
|
|
|
+// });
|
|
|
+// }
|
|
|
+// }
|
|
|
+//
|
|
|
+//
|
|
|
+//
|
|
|
+// async fn on_private_data(bot_arc_clone: Arc<Mutex<Core>>, 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_unlock_core();
|
|
|
+//
|
|
|
+// 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 core = bot_arc_clone.lock().await;
|
|
|
+// core.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 core = bot_arc_clone.lock().await;
|
|
|
+// core.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 core = bot_arc_clone.lock().await;
|
|
|
+// core.update_position(positions).await;
|
|
|
+// }
|
|
|
+// }
|
|
|
+// }
|
|
|
+//
|
|
|
+// async fn on_public_data(bot_arc_clone: Arc<Mutex<Core>>, 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_unlock_core();
|
|
|
+//
|
|
|
+// 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 core = bot_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]);
|
|
|
+// }
|
|
|
+// }
|
|
|
+//
|
|
|
+// 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);
|
|
|
+// }
|
