给参考数据和交易数据安个家

strategy/src/avellaneda_stoikov.rs

@@ -142,7 +142,7 @@ impl AvellanedaStoikov {
         }
     }
 
-    pub async fn on_depth(&mut self, depth: &Depth) {
+    pub async fn on_depth(&mut self, depth: &Depth, _index: u32) {
         self.depth_vec.push_back(depth.clone());
 
         self.ask_price = depth.asks[0].price;
@@ -185,11 +185,6 @@ impl AvellanedaStoikov {
         if self.record_vec.len() > 4 {
             self.record_vec.pop_front();
         }
-
-        // 如果蜡烛数量足够，则更新mfi
-        if self.record_vec.len() >= 4 {
-            self.update_mfi();
-        }
     }
 
     pub async fn update_inventory(&mut self, inventory: &Decimal, min_amount_value: &Decimal) {
@@ -265,25 +260,25 @@ impl AvellanedaStoikov {
         }
     }
 
-    fn calc_flow_ratio(_prev_flow_ratio: &Decimal, min_volume: &Decimal, trades: &mut FixedTimeRangeDeque<Trade>) -> Decimal {
-        let mut flow_in_value = Decimal::ZERO;
-        let mut flow_out_value = Decimal::ZERO;
-        for trade_iter in trades.deque.iter() {
-            if trade_iter.size > Decimal::ZERO {
-                flow_in_value += trade_iter.value;
-            } else {
-                flow_out_value += trade_iter.value;
-            }
-        }
-
-        // 使用EMA來更新資金流，確保平滑性
-        if flow_out_value + flow_in_value > *min_volume {
-            let now = (flow_in_value - flow_out_value) / (flow_out_value + flow_in_value);
-            now
-        } else {
-            Decimal::ZERO
-        }
-    }
+    // fn calc_flow_ratio(_prev_flow_ratio: &Decimal, min_volume: &Decimal, trades: &mut FixedTimeRangeDeque<Trade>) -> Decimal {
+    //     let mut flow_in_value = Decimal::ZERO;
+    //     let mut flow_out_value = Decimal::ZERO;
+    //     for trade_iter in trades.deque.iter() {
+    //         if trade_iter.size > Decimal::ZERO {
+    //             flow_in_value += trade_iter.value;
+    //         } else {
+    //             flow_out_value += trade_iter.value;
+    //         }
+    //     }
+    //
+    //     // 使用EMA來更新資金流，確保平滑性
+    //     if flow_out_value + flow_in_value > *min_volume {
+    //         let now = (flow_in_value - flow_out_value) / (flow_out_value + flow_in_value);
+    //         now
+    //     } else {
+    //         Decimal::ZERO
+    //     }
+    // }
 
     pub fn check_ready(&mut self) {
         if self.is_ready {
@@ -325,21 +320,12 @@ impl AvellanedaStoikov {
     // #[instrument(skip(self), level="TRACE")]
     async fn processor(&mut self) {
         self.update_t_diff();
-        // info!(?self.t_diff);
-        self.update_flow_ratio();
-        // info!(?self.flow_ratio_long);
         self.update_sigma_square();
-        // info!(?self.sigma_square);
         self.update_gamma();
-        // info!(?self.gamma);
         self.update_kappa();
-        // info!(?self.kappa);
         self.update_ref_price();
-        // info!(?self.ref_price);
         self.update_delta();
-        // info!(?self.ask_delta, ?self.bid_delta);
         self.update_optimal_ask_and_bid();
-        // info!("=============================================");
 
         self.check_ready();
         if !self.is_ready {

strategy/src/binance_usdt_swap.rs

@@ -120,7 +120,7 @@ async fn on_data(core_arc: Arc<Mutex<Core>>, multiplier: &Decimal, run_symbol: &
             let depth = ExchangeStructHandler::book_ticker_handle(ExchangeEnum::BinanceSwap, response, multiplier);
             trace_stack.on_after_format();
 
-            on_depth(core_arc, &response.label, &mut trace_stack, &depth).await;
+            on_depth(core_arc, &response.label, &mut trace_stack, &depth, 1).await;
         }
         "ACCOUNT_UPDATE" => {
             let account = ExchangeStructHandler::account_info_handle(ExchangeEnum::BinanceSwap, response, run_symbol);

strategy/src/bitget_usdt_swap.rs

@@ -185,12 +185,12 @@ async fn on_public_data(core_arc: Arc<Mutex<Core>>, mul: &Decimal, response: &Re
                 };
 
                 trace_stack.on_after_format();
-                on_depth(core_arc, &response.label, &mut trace_stack, &result_depth).await;
+                on_depth(core_arc, &response.label, &mut trace_stack, &result_depth, 0).await;
             }
             // 全量
             else {
                 trace_stack.on_after_format();
-                on_depth(core_arc, &response.label, &mut trace_stack, &depth).await;
+                on_depth(core_arc, &response.label, &mut trace_stack, &depth, 0).await;
 
                 depth_asks.clear();
                 depth_asks.append(&mut depth.asks);

strategy/src/bybit_usdt_swap.rs

@@ -151,12 +151,12 @@ async fn on_public_data(core_arc: Arc<Mutex<Core>>, mul: &Decimal, response: &Re
                 };
 
                 trace_stack.on_after_format();
-                on_depth(core_arc, &response.label, &mut trace_stack, &result_depth).await;
+                on_depth(core_arc, &response.label, &mut trace_stack, &result_depth, 0).await;
             }
             // 全量
             else {
                 trace_stack.on_after_format();
-                on_depth(core_arc, &response.label, &mut trace_stack, &depth).await;
+                on_depth(core_arc, &response.label, &mut trace_stack, &depth, 0).await;
 
                 depth_asks.clear();
                 depth_asks.append(&mut depth.asks);

strategy/src/core.rs

@@ -621,7 +621,7 @@ impl Core {
     }
 
     // #[instrument(skip(self, depth, name_ref, trace_stack), level="TRACE")]
-    pub async fn on_depth(&mut self, depth: &Depth, name_ref: &String, trace_stack: &mut TraceStack) {
+    pub async fn on_depth(&mut self, depth: &Depth, name_ref: &String, trace_stack: &mut TraceStack, index: u32) {
         // ================================ 刷新更新间隔 =========================================
         let now_time = depth.time.to_i64().unwrap();
         if self.market_update_time.contains_key(name_ref) && *self.market_update_time.get(name_ref).unwrap() != 0i64 {
@@ -639,7 +639,7 @@ impl Core {
         // ================================ 在系统已经准备就绪的情况下，更新相关参数 =========================================
         if self.mode_signal == 0 && self.ready == 1 {
             // 更新预定价格
-            self.avellaneda_stoikov.on_depth(depth).await;
+            self.avellaneda_stoikov.on_depth(depth, index).await;
             // 触发事件撤单逻辑
             // 更新策略时间
             self.strategy.local_time = Utc::now().timestamp_millis();

strategy/src/exchange_disguise.rs

@@ -124,11 +124,11 @@ pub async fn run_reference_exchange(is_shutdown_arc: Arc<AtomicBool>,
 //     }
 // }
 
-pub async fn on_depth(core_arc: Arc<Mutex<Core>>, label: &String, trace_stack: &mut TraceStack, depth: &Depth) {
+pub async fn on_depth(core_arc: Arc<Mutex<Core>>, label: &String, trace_stack: &mut TraceStack, depth: &Depth, index: u32) {
     let mut core = core_arc.lock().await;
     trace_stack.on_after_unlock_core();
 
-    core.on_depth(depth, &label, trace_stack).await;
+    core.on_depth(depth, &label, trace_stack, index).await;
 }
 
 pub async fn on_trade(core_arc: Arc<Mutex<Core>>, label: &String, trace_stack: &mut TraceStack, trade: &Trade) {

strategy/src/gate_usdt_swap.rs

@@ -111,7 +111,7 @@ async fn on_data(core_arc: Arc<Mutex<Core>>,
             let depth = ExchangeStructHandler::book_ticker_handle(ExchangeEnum::GateSwap, response, multiplier);
             trace_stack.on_after_format();
 
-            on_depth(core_arc, &response.label, &mut trace_stack, &depth).await;
+            on_depth(core_arc, &response.label, &mut trace_stack, &depth, 0).await;
         }
         "futures.trades" => {
             trace_stack.set_source("gate_usdt_swap.trades".to_string());