增加交易和参考交易所的行情记录

strategy/src/core.rs

@@ -25,7 +25,7 @@ use standard::{Account, Market, Order, OrderCommand, Platform, Position, Positio
 use standard::exchange::{Exchange};
 use standard::exchange::ExchangeEnum::{BinanceSwap, BitgetSwap, BybitSwap, CoinexSwap, GateSwap, HtxSwap, KucoinSwap};
 use crate::fix_price::PricingEngine;
-use crate::incremental_json_writer::{FixPriceLog, IncrementalJsonWriter, OrderLog};
+use crate::incremental_json_writer::{FixPriceLog, IncrementalJsonWriter, MarketLog, OrderLog};
 use crate::model::{LocalPosition, OrderInfo, TokenParam};
 use crate::predictor::Predictor;
 use crate::strategy::Strategy;
@@ -133,6 +133,8 @@ pub struct Core {
     pub order_json_writer: IncrementalJsonWriter<OrderLog>, // 下单信息记录
     pub ref_fix_price_writer: IncrementalJsonWriter<FixPriceLog>, // 参考所定价记录
     pub trans_fix_price_writer: IncrementalJsonWriter<FixPriceLog>, // 交易所定价记录
+    pub ref_market_writer: IncrementalJsonWriter<MarketLog>, // 参考交易所行情
+    pub trans_market_writer: IncrementalJsonWriter<MarketLog> // 交易交易所行情
 }
 
 impl Core {
@@ -282,6 +284,8 @@ impl Core {
             order_json_writer: IncrementalJsonWriter::new("test_data/order_log/order_json", 5, 100usize).unwrap(),
             ref_fix_price_writer: IncrementalJsonWriter::new("test_data/ref_price_log/ref_fix_price", 5, 100usize).unwrap(),
             trans_fix_price_writer: IncrementalJsonWriter::new("test_data/trans_price_log/trans_fix_price", 5, 100usize).unwrap(),
+            ref_market_writer: IncrementalJsonWriter::new("test_data/ref_market_log/ref_market", 5, 100usize).unwrap(),
+            trans_market_writer: IncrementalJsonWriter::new("test_data/trans_market_log/trans_market", 5, 100usize).unwrap(),
         };
         for i in 0..=params.ref_exchange.len() - 1 {
             // 拼接不会消耗原字符串
@@ -786,7 +790,7 @@ impl Core {
     }
 
     // #[instrument(skip(self, depth, name_ref, trace_stack), level="TRACE")]
-    pub async fn on_depth_update(&mut self, depth: &Vec<Decimal>, name_ref: &String, trace_stack: &mut TraceStack) {
+    pub async fn on_depth_update(&mut self, depth: &Vec<Decimal>, name_ref: &String, data_tiem: i64, trace_stack: &mut TraceStack) {
         // 要从回调传入的深度信息中获取data.name
         let now_time = Utc::now().timestamp_millis();
         if self.market_update_time.contains_key(name_ref) && *self.market_update_time.get(name_ref).unwrap() != 0i64 {
@@ -807,6 +811,16 @@ impl Core {
             if self.mode_signal == 0 && self.ready == 1 {
                 self.on_agg_market();
             }
+            // 交易盘口 行情记录
+            let trans_market = MarketLog{
+                ask_price: depth[2],
+                ask_amount: depth[3],
+                bid_price: depth[0],
+                bid_amount: depth[1],
+                timestamp: data_tiem,
+                local_timestamp: now_time,
+            };
+            self.trans_market_writer.add_entry(trans_market).unwrap();
         } else if *name_ref == self.ref_name[0] { // 判断是否为当前跟踪的盘口
             // 写入行情数据
             // let ticker = self.tickers.get(name_ref).unwrap();
@@ -874,6 +888,16 @@ impl Core {
             //         }
             //     }
             // }
+            // 参考盘口 行情记录
+            let trans_market = MarketLog{
+                ask_price: depth[2],
+                ask_amount: depth[3],
+                bid_price: depth[0],
+                bid_amount: depth[1],
+                timestamp: data_tiem,
+                local_timestamp: now_time,
+            };
+            self.ref_market_writer.add_entry(trans_market).unwrap();
         }
 
         {
@@ -1751,6 +1775,8 @@ impl Core {
         self.trans_fix_price_writer.flush().unwrap();
         self.order_json_writer.flush().unwrap();
         self.ref_fix_price_writer.flush().unwrap();
+        self.trans_market_writer.flush().unwrap();
+        self.ref_market_writer.flush().unwrap();
         info!("订单、仓位清除完毕，为避免api失效导致遗漏仓位，建议人工复查。");
         info!("停机原因：{}。", self.exit_msg);
     }

strategy/src/exchange_disguise.rs

@@ -124,7 +124,7 @@ pub async fn on_special_depth(core_arc: Arc<Mutex<Core>>,
         core.depths.insert(label.clone(), special_depth.depth.clone());
 
         // 触发depth更新
-        core.on_depth_update(&(special_depth.depth), &label, trace_stack).await;
+        core.on_depth_update(&(special_depth.depth), &label, special_depth.create_at, trace_stack).await;
 
         core.local_depths.insert(special_depth.name.clone(), special_depth.depth.clone());
 

strategy/src/incremental_json_writer.rs

@@ -239,6 +239,22 @@ pub struct FixPriceLog{
     pub record_timestamp: i64
 }
 
+#[derive(Serialize, Debug)]
+pub struct MarketLog{
+    // 卖一价
+    pub ask_price: Decimal,
+    // 卖一量
+    pub ask_amount: Decimal,
+    // 买一价
+    pub bid_price: Decimal,
+    // 买一量
+    pub bid_amount: Decimal,
+    // 行情时间
+    pub timestamp: i64,
+    // 记录时间
+    pub local_timestamp: i64
+}
+
 // --------------------- 示例数据结构 ---------------------
 #[derive(Serialize, Debug)]
 struct LogEntry {