订单流辅助识别主动性加入。

当前版本：
1. 阻止反方向开单；
2. 立马撤掉已开的反方向订单；
3. 在中间状态不开单。

strategy/src/binance_usdt_swap.rs

@@ -10,7 +10,7 @@ use crate::model::{OriginalTradeBa};
 use crate::quant::Quant;
 use exchanges::binance_swap_ws::{BinanceSwapSubscribeType, BinanceSwapWs, BinanceSwapWsType};
 use futures_util::StreamExt;
-use crate::exchange_disguise::on_special_depth;
+use crate::exchange_disguise::{on_special_depth, on_trade};
 
 // 参考 币安 合约 启动
 pub(crate) async fn reference_binance_swap_run(bool_v1 :Arc<AtomicBool>,
@@ -28,7 +28,7 @@ pub(crate) async fn reference_binance_swap_run(bool_v1 :Arc<AtomicBool>,
         ws.set_subscribe(vec![
             // BinanceSwapSubscribeType::PuDepth20levels100ms,
             BinanceSwapSubscribeType::PuBookTicker,
-            // BinanceSwapSubscribeType::PuAggTrade
+            BinanceSwapSubscribeType::PuAggTrade
         ]);
 
         //读取数据
@@ -55,8 +55,8 @@ pub(crate) async fn reference_binance_swap_run(bool_v1 :Arc<AtomicBool>,
 
 async fn on_data(bot_arc_clone: Arc<Mutex<Quant>>,
                  update_flag_u: &mut Decimal,
-                 max_buy: &mut Decimal,
-                 min_sell: &mut Decimal,
+                 _max_buy: &mut Decimal,
+                 _min_sell: &mut Decimal,
                  data: ResponseData) {
     let mut trace_stack = TraceStack::default();
     trace_stack.on_after_network(data.time);
@@ -68,22 +68,25 @@ async fn on_data(bot_arc_clone: Arc<Mutex<Quant>>,
 
     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]);
-        }
+        // let name = data.label.clone();
+
+        // 订单流逻辑
+        on_trade(trade.clone(), bot_arc_clone.clone()).await;
+
+        // // 原本的逻辑
+        // let mut quant = bot_arc_clone.lock().await;
+        // if quant.is_update.contains_key(&data.label) && *quant.is_update.get(name.as_str()).unwrap() {
+        //     *max_buy = Decimal::ZERO;
+        //     *min_sell = Decimal::ZERO;
+        //     quant.is_update.remove(name.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();
         // 将ticker数据转换为模拟深度

strategy/src/exchange_disguise.rs

@@ -11,6 +11,7 @@ use crate::bitget_spot::bitget_spot_run;
 use crate::gate_swap::gate_swap_run;
 use crate::kucoin_spot::kucoin_spot_run;
 use crate::kucoin_swap::kucoin_swap_run;
+use crate::model::OriginalTradeBa;
 use crate::okx_usdt_swap::okex_swap_run;
 use crate::quant::Quant;
 
@@ -93,7 +94,54 @@ pub async fn on_special_depth(bot_arc: Arc<Mutex<Quant>>,
     }
 }
 
-pub async fn on_trade() {}
+pub async fn on_trade(trade: OriginalTradeBa,
+                      bot_arc_clone: Arc<Mutex<Quant>>) {
+    let mut bot = bot_arc_clone.lock().await;
+    // 1. 塞入数据到bot
+    bot.trades.push(trade.clone());
+    // 2. 长度检查
+    if bot.trades.len() > bot.recall_max_count {
+        bot.trades.remove(0);
+    }
+    // 3. 如果少于100条，不进行判断
+    if bot.trades.len() < 100 {
+        return;
+    }
+    // 求最近的多空总和
+    let mut long_sum = Decimal::ZERO;
+    let mut short_sum = Decimal::ZERO;
+    let last_trade_t = trade.t.clone();
+    let mut rev = bot.trades.clone();
+    rev.reverse();
+    for trade_o in rev {
+        // 如果该元素已过期，我们是按时间顺序插入的，说明前面的应该都过期了，跳出循环，停止检测
+        if trade_o.t < last_trade_t - bot.recall_time {
+            continue;
+        }
+
+        // 卖出订单
+        if trade_o.m {
+            short_sum += trade_o.q;
+        } else {
+            long_sum += trade_o.q;
+        }
+    }
+
+    // 做多主动性
+    if (long_sum / (long_sum + short_sum)) > bot.long_volume_rate {
+        if bot.side != "long".to_string() {
+            bot.side = "long".to_string();
+        }
+    } else if (short_sum / (long_sum + short_sum)) > bot.short_volume_rate {
+        if bot.side != "short".to_string() {
+            bot.side = "short".to_string();
+        }
+    } else {
+        if bot.side != "normal".to_string() {
+            bot.side = "normal".to_string();
+        }
+    }
+}
 
 pub async fn on_order() {}
 

strategy/src/model.rs

@@ -42,7 +42,8 @@ pub struct TraderMsg {
     pub orders: HashMap<String, OrderInfo>,
     pub ref_price: Vec<Vec<Decimal>>,
     pub market: Vec<Decimal>,
-    pub predict: Decimal
+    pub predict: Decimal,
+    pub side: String,
 }
 
 impl TraderMsg {
@@ -60,6 +61,7 @@ impl TraderMsg {
             ref_price: Default::default(),
             market: vec![],
             predict: Default::default(),
+            side: "normal".to_string(),
         }
     }
 }
@@ -93,10 +95,15 @@ pub struct OrderInfo {
     pub trace_stack: TraceStack
 }
 
-#[derive(Serialize, Deserialize)]
+#[derive(Serialize, Deserialize, Clone, Debug)]
 pub struct OriginalTradeBa {
     // 成交价格
     pub p: Decimal,
+    // 成交数量
+    pub q: Decimal,
+    // 成交时间
+    #[serde(rename = "T")]
+    pub t: Decimal,
     // 买方是否是做市方。如true，则此次成交是一个主动卖出单，否则是一个主动买入单。
     pub m: bool
 }

strategy/src/quant.rs

@@ -22,7 +22,7 @@ use standard::{Account, Market, Order, OrderCommand, Platform, Position, Positio
 use standard::exchange::{Exchange};
 use standard::exchange::ExchangeEnum::{BinanceSpot, BinanceSwap, BitgetSpot, GateSpot, GateSwap, KucoinSwap, OkxSwap};
 
-use crate::model::{LocalPosition, OrderInfo, TokenParam, TraderMsg};
+use crate::model::{LocalPosition, OrderInfo, OriginalTradeBa, TokenParam, TraderMsg};
 use crate::predictor::Predictor;
 use crate::strategy::Strategy;
 use crate::utils;
@@ -106,6 +106,14 @@ pub struct Quant {
     // 打印限频
     pub prev_log_ready_timestamp: i64,
     pub log_ready_log_interval: i64,
+
+    // 订单流相关参数
+    pub side: String,                                       // 当前方向, [normal, long, short]
+    pub trades: Vec<OriginalTradeBa>,                       // 订单流集合
+    pub recall_time: Decimal,                               // 前置时间
+    pub recall_max_count: usize,                            // 最大回溯条数
+    pub short_volume_rate: Decimal,                         // 主动性跌比率(0.01代表1%)
+    pub long_volume_rate: Decimal,                          // 主动性涨比率(0.01代表1%)
 }
 
 impl Quant {
@@ -221,6 +229,12 @@ impl Quant {
             running,
             prev_log_ready_timestamp: 0,
             log_ready_log_interval: 10 * 1000,
+            side: "normal".to_string(),
+            trades: vec![],
+            recall_time: dec!(1680),
+            recall_max_count: 1680.to_usize().unwrap(),
+            short_volume_rate: dec!(0.8),
+            long_volume_rate: dec!(0.8),
         };
         for i in 0..=params.ref_exchange.len() - 1 {
             // 拼接不会消耗原字符串
@@ -731,6 +745,8 @@ impl Quant {
         }
         let ref_price: Vec<Vec<Decimal>> = self.predictor.get_ref_price(&ref_tickers);
         self.trade_msg.ref_price = ref_price;
+        // 更新主动性方向
+        self.trade_msg.side = self.side.clone();
     }
 
     // 本地记录所有报单信息

strategy/src/strategy.rs

@@ -103,6 +103,9 @@ pub struct Strategy {
     pub post_side: i64,                                             // 交易方向
     pub trade_vol_24h_w: Decimal,                                   // 24小时成交额（单位：万）
     pub grid: Decimal,                                              // 网格数量
+
+    // 订单流相关
+    pub side: String,                                               // 当前主动性方向
 }
 
 impl Strategy {
@@ -194,6 +197,7 @@ impl Strategy {
             post_side: 0,
             trade_vol_24h_w: Default::default(),
             grid: Decimal::from(params.grid),
+            side: "normal".to_string(),
         };
 
         // 交易名字
@@ -340,6 +344,8 @@ impl Strategy {
             // debug!(?max_pos_rate, ?self.max_pos_rate);
         }
 
+        self.side = trader_msg.side.clone();
+
         return true;
     }
 
@@ -413,7 +419,7 @@ impl Strategy {
                              self.local_profit, self.profit, long_pos_leverage, self.long_pos_bias, short_pos_leverage, self.short_pos_bias).as_str());
         msg.push_str(format!("[请求 {:?}, 上限{:?}次/10秒], ", self._req_num_per_window, self.limit_order_requests_num).as_str());
         msg.push_str(format!("[当前参数, 开仓 {:?}, 平仓 {:?}, 方向 {:?}, 参考 {:?}, 模式 {:?}], ",
-                             self.trade_open_dist, self.trade_close_dist, self.post_side, self.ref_name[self.ref_index], self.maker_mode).as_str());
+                             self.trade_open_dist, self.trade_close_dist, self.side, self.ref_name[self.ref_index], self.maker_mode).as_str());
         msg.push_str(format!("[挂单列表，共{:?}单, ", o_num).as_str());
         for (_, order) in &self.local_orders {
             let mut order_value = order.amount * self.mp;
@@ -587,8 +593,10 @@ impl Strategy {
 
             // 判断是否在本地挂单表中
             if let Some(order) = order_some {
-                // 如果订单创建时间大于100ms，才能有撤单操作
-                if self.local_time - order.create_time < 100 {
+                let is_side_equals = (order.side == "kk" && self.side == "short") || (order.side == "kd" && self.side == "long");
+
+                // 如果订单创建时间大于100ms，或有相反方向订单，才能有撤单操作
+                if is_side_equals || self.local_time - order.create_time < 100 {
                     need_limit_cancel = false;
                 }
             }
@@ -1000,19 +1008,21 @@ impl Strategy {
             // 开多订单处理
             if order.side == "kd".to_string() {
                 // 在价格范围内时不处理
-                if order.price <= long_upper && order.price >= long_lower {
+                if order.price <= long_upper && order.price >= long_lower && self.side == "long".to_string() {
                     continue
                 }
                 // debug!(?key, ?order.price, ?long_upper, ?long_lower);
+                info!("kd-{}-{}", self.side, key);
                 command.cancel.insert(key.clone(), value.clone());
             }
 
             // 开空订单处理
             if order.side == "kk".to_string() {
                 // 在价格范围内时不处理
-                if order.price >= short_lower && order.price <= short_upper {
+                if order.price >= short_lower && order.price <= short_upper && self.side == "short".to_string() {
                     continue
                 }
+                info!("kk-{}-{}", self.side, key);
                 // debug!(?key, ?order.price, ?short_lower, ?short_upper);
                 command.cancel.insert(key.clone(), value.clone());
             }
@@ -1120,7 +1130,7 @@ impl Strategy {
 
         // debug!(?self.post_side);
         // 挂多单
-        if self.post_side >= 0 {
+        if self.post_side >= 0 && self.side == "long".to_string() {
             // debug!(?buy_price_list);
             if buy_price_list.len() == 0 {
                 let mut target_buy_price = (long_upper + long_lower) * dec!(0.5);
@@ -1147,7 +1157,7 @@ impl Strategy {
             }
         }
         // 挂空单
-        if self.post_side <= 0 {
+        if self.post_side <= 0 && self.side == "short".to_string() {
             // debug!(?sell_price_list);
             if sell_price_list.len() == 0 {
                 let mut target_sell_price = (short_lower + short_upper) * dec!(0.5);
@@ -1225,8 +1235,7 @@ impl Strategy {
         self._refresh_request_limit();                      // 刷新频率限制
         self._update_request_num(&mut command);             // 统计刷新频率
 
-
-        if command.limits_open.len() > 0 || command.limits_close.len() > 0{
+        if command.limits_open.len() > 0 || command.limits_close.len() > 0 {
             let time = chrono::Utc::now().timestamp_millis();
             let name = self.params.account_name.clone();
             // 参考卖价
@@ -1242,7 +1251,6 @@ impl Strategy {
             });
         }
 
-
         return command;
     }
 }