数据服务已经做好了。

Cargo.toml

@@ -22,6 +22,7 @@ ctrlc = "3.2.5"
 serde_json = "1.0.105"
 rust_decimal = { version = "1.32.0", features = ["maths"] }
 rust_decimal_macros = "1.32.0"
+actix-cors = "0.6"
 
 [workspace]
 members=[

global/src/cci.rs

@@ -1,15 +1,9 @@
 use rust_decimal::Decimal;
 use serde_derive::Serialize;
 use crate::fixed_time_range_deque::FixedTimeRangeDeque;
+use crate::predictor_state::PredictorState;
 
-#[derive(Serialize, Clone)]
-pub struct PredictorState {
-    pub update_time: Decimal,
-
-
-}
-
-#[derive(Serialize, Clone)]
+#[derive(Serialize, Clone, Debug)]
 pub struct CentralControlInfo {
     pub now_balance: Decimal,                                       // 钱包余额
     pub unrealized_pn_l: Decimal,                                   // 未实现盈亏
@@ -22,7 +16,7 @@ pub struct CentralControlInfo {
 
 impl CentralControlInfo {
     // 时间窗口大小（微秒）
-    const MAX_TIME_RANGE_MICROS: i64 = 60_000_000_000;
+    const MAX_TIME_RANGE_MICROS: i64 = 10 * 60_000_000;
 
     pub fn new() -> Self {
         Self {

global/src/lib.rs

@@ -9,3 +9,4 @@ pub mod clear_position_result;
 pub mod trade;
 pub mod clear_log_utils;
 pub mod fixed_time_range_deque;
+pub mod predictor_state;

global/src/predictor_state.rs

@@ -0,0 +1,24 @@
+use rust_decimal::Decimal;
+use serde_derive::{Deserialize, Serialize};
+
+#[derive(Serialize, Deserialize, Clone, Debug)]
+pub struct PredictorState {
+    pub update_time: Decimal,                                                   // 该条数据的时间
+
+    pub mid_price: Decimal,                                                     // 中间价
+    pub ask_price: Decimal,                                                     // 卖一价
+    pub bid_price: Decimal,                                                     // 买一价
+    pub optimal_ask_price: Decimal,                                             // 卖出挂单价
+    pub optimal_bid_price: Decimal,                                             // 买入挂单价
+    pub ref_price: Decimal,                                                     // 预定价格
+
+    pub spread: Decimal,                                                        // 市场冲击
+    pub delta_plus: Decimal,                                                    // δa+δb，买卖挂单间距
+    pub spread_max: Decimal,                                                    // 最大市场冲击
+    pub spread_min: Decimal,                                                    // 最小市场冲击
+
+    pub inventory: Decimal,                                                     // 库存，也就是q
+    pub sigma_square: Decimal,                                                  // σ^2，波动性的平方
+    pub gamma: Decimal,                                                         // γ，库存风险厌恶参数
+    pub kappa: Decimal,                                                         // κ 订单簿 流动性 参数
+}

src/server.rs

@@ -1,7 +1,9 @@
 use std::sync::Arc;
 use std::sync::atomic::{AtomicBool, Ordering};
 use std::time::Duration;
+use actix_cors::Cors;
 use actix_web::{web, App, HttpResponse, HttpServer, Responder, post, get};
+use serde_json::json;
 use tokio::sync::Mutex;
 use tracing::{info};
 use global::cci::CentralControlInfo;
@@ -13,14 +15,33 @@ struct Arcs {
     cci_cache_arc: Arc<Mutex<CentralControlInfo>>
 }
 
-// 句柄 GET 请求
+// 账户相关数据
 #[get("/account")]
 async fn get_account(arcs: web::Data<Arcs>) -> impl Responder {
     // --------------------------------数据解锁处理--------------------------------
     let cci = arcs.cci_cache_arc.lock().await;
 
+    let rst = json!({
+        "now_balance": cci.now_balance,
+        "unrealized_pn_l": cci.unrealized_pn_l,
+        "pos": cci.pos,
+        "entry_price": cci.entry_price,
+        "now_price": cci.now_price
+    });
+
+    // --------------------------------回报--------------------------------
+    let json_string = serde_json::to_string(&rst).unwrap();
+    HttpResponse::Ok().content_type("application/json").body(json_string)
+}
+
+// 句柄 GET 请求
+#[get("/predictor_state")]
+async fn get_predictor_state(arcs: web::Data<Arcs>) -> impl Responder {
+    // --------------------------------数据解锁处理--------------------------------
+    let cci = arcs.cci_cache_arc.lock().await;
+
     // --------------------------------回报--------------------------------
-    let json_string = serde_json::to_string(&(cci.clone())).unwrap();
+    let json_string = serde_json::to_string(&cci.predictor_state_vec.deque).unwrap();
     HttpResponse::Ok().content_type("application/json").body(json_string)
 }
 
@@ -69,9 +90,18 @@ pub fn run_server(port: u32, running: Arc<AtomicBool>, cci_arc: Arc<Mutex<Centra
     let server_fut = HttpServer::new(move || {
         let arcs_clone = arcs.clone();
 
+        // 配置 CORS
+        let cors = Cors::permissive()
+            .allow_any_origin()
+            .allow_any_header()
+            .allow_any_method()
+            .max_age(3600); // 设置预检请求的缓存时间
+
         App::new()
+            .wrap(cors)
             .app_data(web::Data::new(arcs_clone))
             .service(get_account)
+            .service(get_predictor_state)
             .service(on_change)
     })
     .bind(addr)

strategy/src/avellaneda_stoikov.rs

@@ -1,7 +1,11 @@
 use std::collections::{BTreeMap};
+use std::sync::Arc;
+use chrono::Utc;
 use rust_decimal::prelude::*;
-use tracing::info;
+use tokio::sync::Mutex;
+use global::cci::CentralControlInfo;
 use global::fixed_time_range_deque::FixedTimeRangeDeque;
+use global::predictor_state::PredictorState;
 use standard::{Depth, Ticker, Trade};
 
 #[derive(Debug)]
@@ -23,15 +27,18 @@ pub struct AvellanedaStoikov {
     pub sigma_square: Decimal,                                                  // σ^2，波动性的平方
     pub gamma: Decimal,                                                         // γ，库存风险厌恶参数
     pub kappa: Decimal,                                                         // κ 订单簿 流动性 参数
+
     pub delta_plus: Decimal,                                                    // δa+δb，买卖挂单间距
     pub ref_price: Decimal,                                                     // 预定价格
+
+    pub cci_arc: Arc<Mutex<CentralControlInfo>>                                 // 中控信息
 }
 
 impl AvellanedaStoikov {
     // 时间窗口大小（微秒）
-    const MAX_TIME_RANGE_MICROS: i64 = 10_000_000_000;
+    const MAX_TIME_RANGE_MICROS: i64 = 10 * 60_000_000;
     
-    pub fn new() -> Self {
+    pub fn new(cci_arc: Arc<Mutex<CentralControlInfo>>) -> Self {
         let avellaneda_stoikov = Self {
             // 分别给与的长度
             depth_vec: FixedTimeRangeDeque::new(Self::MAX_TIME_RANGE_MICROS),
@@ -53,6 +60,8 @@ impl AvellanedaStoikov {
             delta_plus: Default::default(),
             kappa: Default::default(),
             ref_price: Default::default(),
+
+            cci_arc,
         };
 
         avellaneda_stoikov
@@ -87,7 +96,7 @@ impl AvellanedaStoikov {
         }
     }
 
-    pub fn on_depth(&mut self, depth: &Depth) {
+    pub async fn on_depth(&mut self, depth: &Depth) {
         self.depth_vec.push_back(depth.clone());
 
         self.ask_price = depth.asks[0].price;
@@ -98,18 +107,18 @@ impl AvellanedaStoikov {
         self.update_spread(&mid_price_now);
         self.mid_price = mid_price_now;
 
-        self.processor();
+        self.processor().await;
     }
 
-    pub fn on_trade(&mut self, trade: &Trade) {
+    pub async fn on_trade(&mut self, trade: &Trade) {
         self.trade_vec.push_back(trade.clone());
-        self.processor();
+        self.processor().await;
     }
 
-    pub fn update_inventory(&mut self, inventory: Decimal) {
+    pub async fn update_inventory(&mut self, inventory: Decimal) {
         self.inventory = inventory;
 
-        self.processor();
+        self.processor().await;
     }
 
     pub fn update_sigma_square(&mut self) {
@@ -183,19 +192,41 @@ impl AvellanedaStoikov {
     }
 
     // #[instrument(skip(self), level="TRACE")]
-    fn processor(&mut self) {
+    async fn processor(&mut self) {
         self.update_sigma_square();
-        info!(?self.sigma_square);
+        // info!(?self.sigma_square);
         self.update_gamma();
-        info!(?self.gamma);
+        // info!(?self.gamma);
         self.update_kappa();
-        info!(?self.kappa);
+        // info!(?self.kappa);
         self.update_ref_price();
-        info!(?self.ref_price);
+        // info!(?self.ref_price);
         self.update_delta_plus();
-        info!(?self.delta_plus);
+        // info!(?self.delta_plus);
         self.update_optimal_ask_and_bid();
-        info!(?self.spread_max, ?self.mid_price, ?self.ref_price, ?self.inventory);
+        // info!("=============================================");
+
+        let mut cci = self.cci_arc.lock().await;
+        cci.predictor_state_vec.push_back(PredictorState {
+            update_time: Decimal::from_i64(Utc::now().timestamp_millis()).unwrap(),
+
+            mid_price: self.mid_price,
+            ask_price: self.ask_price,
+            bid_price: self.bid_price,
+            spread: self.spread,
+            spread_max: self.spread_max,
+            spread_min: self.spread_min,
+            optimal_ask_price: self.optimal_ask_price,
+            optimal_bid_price: self.optimal_bid_price,
+
+            inventory: self.inventory,
+            sigma_square: self.sigma_square,
+            gamma: self.gamma,
+            kappa: self.kappa,
+
+            delta_plus: self.delta_plus,
+            ref_price: self.ref_price,
+        });
     }
 
     // #[instrument(skip(self, ref_ticker_map), level="TRACE")]

strategy/src/clear_core.rs

@@ -173,7 +173,7 @@ impl ClearCore {
             ref_name: Default::default(),
             trade_name: "".to_string(),
             ready: 0,
-            predictor: AvellanedaStoikov::new(),
+            predictor: AvellanedaStoikov::new(cci_arc.clone()),
             market: Market {
                 symbol: symbol.clone(),
                 base_asset: "".to_string(),
@@ -278,7 +278,6 @@ impl ClearCore {
             }
         }
         info!("价格系数：{:?}", price_alpha);
-        core_obj.predictor = AvellanedaStoikov::new();
 
         return core_obj;
     }

strategy/src/core.rs

@@ -176,7 +176,7 @@ impl Core {
             ref_name: Default::default(),
             trade_name: "".to_string(),
             ready: 0,
-            avellaneda_stoikov: AvellanedaStoikov::new(),
+            avellaneda_stoikov: AvellanedaStoikov::new(cci_arc.clone()),
             market: Market {
                 symbol: symbol.clone(),
                 base_asset: "".to_string(),
@@ -281,7 +281,6 @@ impl Core {
             }
         }
         info!("价格系数：{:?}", price_alpha);
-        core_obj.avellaneda_stoikov = AvellanedaStoikov::new();
 
         return core_obj;
     }
@@ -618,7 +617,7 @@ impl Core {
     }
 
     pub async fn on_trade(&mut self, trade: &Trade, _name_ref: &String, _trace_stack: &mut TraceStack) {
-        self.avellaneda_stoikov.on_trade(trade);
+        self.avellaneda_stoikov.on_trade(trade).await;
     }
 
     // #[instrument(skip(self, depth, name_ref, trace_stack), level="TRACE")]
@@ -640,7 +639,7 @@ impl Core {
         // ================================ 在系统已经准备就绪的情况下，更新相关参数 =========================================
         if self.mode_signal == 0 && self.ready == 1 {
             // 更新预定价格
-            self.avellaneda_stoikov.on_depth(depth);
+            self.avellaneda_stoikov.on_depth(depth).await;
             // 触发事件撤单逻辑
             // 更新策略时间
             self.strategy.local_time = Utc::now().timestamp_millis();
@@ -719,7 +718,7 @@ impl Core {
             pos = self.local_position.long_pos - self.local_position.short_pos;
         }
         pos.rescale(8);
-        self.avellaneda_stoikov.update_inventory(pos);
+        self.avellaneda_stoikov.update_inventory(pos).await;
 
         let mut entry_price;
         if pos.gt(&Decimal::ZERO) {