时序插入ok

strategy/src/core.rs

@@ -382,6 +382,7 @@ impl Core {
                                 }
                                 self.local_cash -= filled * filled_price;
                                 self.local_coin = filled - fee;
+                                self.predictor.on_inventory(&self.local_position_by_orders.long_pos, &self.local_position_by_orders.long_avg, &self.strategy.min_amount_value).await;
 
                                 // sell 平多
                             } else if side == "pd" {
@@ -397,6 +398,7 @@ impl Core {
                                 }
                                 self.local_cash += filled * filled_price - fee;
                                 self.local_coin -= filled;
+                                self.predictor.on_inventory(&self.local_position_by_orders.long_pos, &self.local_position_by_orders.long_avg, &self.strategy.min_amount_value).await;
 
                             // buy 平空
                             } else if side == "pk" {
@@ -412,6 +414,7 @@ impl Core {
                                 }
                                 self.local_cash -= filled * filled_price;
                                 self.local_coin += filled - fee;
+                                self.predictor.on_inventory(&-self.local_position_by_orders.short_pos, &self.local_position_by_orders.short_avg, &self.strategy.min_amount_value).await;
 
                             // sell 开空
                             } else if side == "kk" {
@@ -428,6 +431,8 @@ impl Core {
                                 }
                                 self.local_cash += filled * filled_price - fee;
                                 self.local_coin -= filled;
+                                self.predictor.on_inventory(&-self.local_position_by_orders.short_pos, &self.local_position_by_orders.short_avg, &self.strategy.min_amount_value).await;
+
                             } else {
                                 info!("错误的仓位方向{}", side);
                             }
@@ -694,10 +699,7 @@ impl Core {
             return;
         }
         let mut position = LocalPosition::new();
-        let mut pos_avg_price = Decimal::ZERO;
         for pos in &data {
-            pos_avg_price = pos.price;
-
             if pos.position_mode == PositionModeEnum::Long {
                 position.long_pos = pos.amount;
                 position.long_avg = pos.price;
@@ -720,7 +722,6 @@ impl Core {
             pos = self.local_position.long_pos - self.local_position.short_pos;
         }
         pos.rescale(8);
-        self.predictor.on_inventory(&pos, &pos_avg_price, &self.strategy.min_amount_value).await;
 
         let mut entry_price;
         if pos.gt(&Decimal::ZERO) {

strategy/src/predictor.rs

@@ -78,12 +78,42 @@ impl Predictor {
             let mut debugs: Vec<VecDeque<Option<Decimal>>> = vec![VecDeque::new(); len];
 
             while let Some(value) = rx.next().await {
+                // 获取插入下标
+                // 反向遍历 VecDeque
+                let mut insert_index = 0usize;
+                if debugs[0].len() > 0 {
+                    let mut j = debugs[0].len() - 1;
+                    loop {
+                        if debugs[0][j].unwrap() <= value[0] {
+                            insert_index = j + 1;
+                            break;
+                        }
+
+                        j = j - 1;
+                        if j == 0 {
+                            break;
+                        }
+                    }
+                }
+
                 // 数据填充到对应位置
                 for i in 0..len {
                     if value[i] == Self::DONT_VIEW {
-                        debugs[i].push_back(None);
+                        // 可能会遇见按时序插入的
+                        if debugs[i].len() > 0 && insert_index < debugs.len() {
+                            // 在合适的位置插入新元素
+                            debugs[i].insert(insert_index, None);
+                        } else {
+                            debugs[i].push_back(None);
+                        }
                     } else {
-                        debugs[i].push_back(Some(value[i]));
+                        // 可能会遇见按时序插入的
+                        if debugs[i].len() > 0 && insert_index < debugs.len() {
+                            // 在合适的位置插入新元素
+                            debugs[i].insert(insert_index, Some(value[i]));
+                        } else {
+                            debugs[i].push_back(Some(value[i]));
+                        }
                     }
                 }
 
@@ -95,7 +125,7 @@ impl Predictor {
                 }
 
                 let now = Decimal::from(Utc::now().timestamp_millis());
-                if now - prev_save_time < dec!(60000) {
+                if now - prev_save_time < dec!(30000) {
                     continue;
                 }
 
@@ -171,7 +201,7 @@ impl Predictor {
         if self.mid_price.is_zero() {
             return;
         }
-        self.processor().await;
+        self.processor(depth.time).await;
     }
 
     pub async fn on_trade(&mut self, trade: &Trade, _index: usize) {
@@ -201,7 +231,7 @@ impl Predictor {
             };
         }
 
-        self.processor().await;
+        self.processor(Decimal::from(Utc::now().timestamp_millis())).await;
     }
 
     pub async fn on_balance(&mut self, balance: Decimal) {
@@ -346,7 +376,7 @@ impl Predictor {
     }
 
     // #[instrument(skip(self), level="TRACE")]
-    async fn processor(&mut self) {
+    async fn processor(&mut self, data_time: Decimal) {
         self.check_ready();
         if !self.is_ready {
             return;
@@ -355,7 +385,7 @@ impl Predictor {
         self.update_delta();
 
         // let cci_arc = self.cci_arc.clone();
-        let now = Decimal::from_i64(Utc::now().timestamp_millis()).unwrap();
+        let now = data_time;
         let mid_price = self.mid_price;
         let ask_price = self.fair_price_std_vec[0];
         let bid_price = self.fair_price_std_vec[1];