完善知情者概率

standard/src/bybit_swap.rs

@@ -22,7 +22,9 @@ struct SwapTicker {
     high_price24h: Decimal,
     low_price24h: Decimal,
     bid1_price: Decimal,
+    bid1_size: Decimal,
     ask1_price: Decimal,
+    ask1_size: Decimal,
     last_price: Decimal,
     volume24h: Decimal
 }

standard/src/lib.rs

@@ -160,6 +160,20 @@ pub struct Trade {
     pub symbol: String,
 }
 
+/// 单位时间的交易数据
+/// - `time(Decimal)`: 交易时间戳(s)
+/// - `size(Decimal)`: 交易量，负数是卖方
+/// - `price(Decimal)`: 成交价格
+/// - `value(Decimal)`: 成交价值（计价币对）
+/// - `symbol(String)`: 成交符号
+#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
+pub struct UnitTradeInfo {
+    pub time: Decimal,
+    pub sell_size: Decimal,
+    pub buy_size: Decimal,
+    pub total_size: Decimal
+}
+
 /// 特殊压缩结构体(订单流)
 /// - `0(Decimal)`: id
 /// - `1(Decimal)`: 交易更新时间戳(ms)

strategy/src/avellaneda_stoikov.rs

@@ -1,15 +1,15 @@
 use std::cmp::{max, min};
-use std::collections::{BTreeMap};
+use std::collections::{BTreeMap, HashMap, VecDeque};
 use std::sync::Arc;
 use chrono::Utc;
 use rust_decimal::prelude::*;
 use rust_decimal_macros::dec;
 use tokio::sync::Mutex;
-use tracing::info;
+use tracing::{error, info};
 use global::cci::CentralControlInfo;
 use global::fixed_time_range_deque::FixedTimeRangeDeque;
 use global::predictor_state::PredictorState;
-use standard::{Depth, Ticker, Trade};
+use standard::{Depth, Ticker, Trade, UnitTradeInfo};
 
 /**
 1. 微价格（加权中间价）作为基价
@@ -27,12 +27,15 @@ pub struct AvellanedaStoikov {
     pub trade_short_vec: FixedTimeRangeDeque<Trade>,                            // 交易队列
     pub spread_vec: FixedTimeRangeDeque<Decimal>,
     pub ticker_vec: FixedTimeRangeDeque<Ticker>,
-    pub vwpin_source_vec: FixedTimeRangeDeque<Decimal>,                          // vpin计算源数据
+    pub vwpin_source_map: HashMap<i64, UnitTradeInfo>,                          // vpin计算源数据
 
     pub volume24h: Decimal,                                                     // 最新的24小时成交数量
     pub mid_price: Decimal,                                                     // 中间价
     pub ask_price: Decimal,                                                     // 卖一价
+    pub ask_size: Decimal,                                                      // 卖一量
     pub bid_price: Decimal,                                                     // 买一价
+    pub bid_size: Decimal,                                                      // 买一量
+
     pub last_price: Decimal,                                                    // 最后成交价
     pub spread: Decimal,                                                        // 市场冲击
     pub spread_max: Decimal,                                                    // 最大市场冲击
@@ -47,6 +50,8 @@ pub struct AvellanedaStoikov {
     pub kappa: Decimal,                                                         // κ 订单簿 流动性 参数
 
     pub vwpin: Decimal,                                                         // vwpin值
+    pub vwpin_avg: Decimal,                                                     // vwpin平均值
+    pub vwpin_vec: VecDeque<Decimal>,                                           // vwpin历史值列表
 
     pub flow_ratio_long: Decimal,                                               // 资金流比例
     pub flow_ratio_short: Decimal,                                              // 资金流比例
@@ -82,12 +87,14 @@ impl AvellanedaStoikov {
             trade_long_vec: FixedTimeRangeDeque::new(Self::TRADE_LONG_RANGE_MICROS),
             ticker_vec: FixedTimeRangeDeque::new(Self::TRADE_LONG_RANGE_MICROS),
             trade_short_vec: FixedTimeRangeDeque::new(Self::TRADE_SHORT_RANGE_MICROS),
-            vwpin_source_vec: FixedTimeRangeDeque::new(Self::MAX_TIME_RANGE_MICROS),
+            vwpin_source_map: Default::default(),
             volume24h: Default::default(),
 
             mid_price: Default::default(),
             ask_price: Default::default(),
+            ask_size: Default::default(),
             bid_price: Default::default(),
+            bid_size: Default::default(),
             last_price: Default::default(),
             spread: Default::default(),
             spread_max: Default::default(),
@@ -116,6 +123,8 @@ impl AvellanedaStoikov {
             level: Default::default(),
             flow_ratio_short: Default::default(),
             vwpin: Default::default(),
+            vwpin_avg: Default::default(),
+            vwpin_vec: VecDeque::new(),
         };
 
         avellaneda_stoikov
@@ -169,25 +178,57 @@ impl AvellanedaStoikov {
         self.depth_vec.push_back(depth.clone());
 
         self.ask_price = depth.asks[0].price;
-        let ask_size = depth.asks[0].size;
+        // let ask_size = depth.asks[0].size;
         self.bid_price = depth.bids[0].price;
-        let bid_size = depth.bids[0].size;
-        self.mid_price = (self.ask_price * bid_size + self.bid_price * ask_size) / (ask_size + bid_size);
-
-        let total_size = ask_size + bid_size;
-
-        let vwpin_item = total_size/self.volume24h * (ask_size - bid_size) / total_size;
-
-        self.vwpin_source_vec.push_back(vwpin_item);
+        // let bid_size = depth.bids[0].size;
+        // self.mid_price = (self.ask_price * bid_size + self.bid_price * ask_size) / (ask_size + bid_size);
+        self.mid_price = (self.ask_price + self.bid_price) / Decimal::TWO;
 
         self.processor().await;
     }
 
-
-
     pub async fn on_trade(&mut self, trade: &Trade) {
         self.trade_long_vec.push_back(trade.clone());
         self.trade_short_vec.push_back(trade.clone());
+        // if trade.size > Decimal::ZERO { // 买
+        //     self.bid_price = trade.price;
+        //     self.bid_size = trade.size;
+        // } else { // 卖
+        //     self.ask_price = trade.price;
+        //     self.ask_size = trade.size.abs();
+        // }
+        // 秒级
+        let time_key = trade.time.to_i64().unwrap() / 1000;
+        // 100秒以前的数据 过期
+        let timeout_key = time_key - 100;
+        if self.vwpin_source_map.contains_key(&time_key) {
+            let value = self.vwpin_source_map.get_mut(&time_key).unwrap();
+            if trade.size > Decimal::ZERO {
+                value.buy_size = value.buy_size + trade.size;
+            } else {
+                value.sell_size = value.sell_size + trade.size.abs();
+            }
+            value.total_size = self.volume24h;
+        } else {
+            let mut unit_trade_info = UnitTradeInfo{
+                time: trade.time,
+                sell_size: Decimal::ZERO,
+                buy_size: Decimal::ZERO,
+                total_size: self.volume24h,
+            };
+            if trade.size > Decimal::ZERO {
+                unit_trade_info.buy_size = trade.size.abs();
+            } else {
+                unit_trade_info.sell_size = trade.size.abs();
+            }
+
+            self.vwpin_source_map.insert(time_key, unit_trade_info);
+        }
+        // 保留100s以内的数据
+        self.vwpin_source_map.retain(|&key, _| key >= timeout_key);
+
+        // 更新vwpin
+        self.update_vwpin();
 
         self.last_price = trade.price;
         self.update_spread();
@@ -196,7 +237,9 @@ impl AvellanedaStoikov {
 
     pub async fn on_ticker(&mut self, ticker: &Ticker) {
         self.ticker_vec.push_back(ticker.clone());
-        self.volume24h = ticker.volume;
+        if ticker.volume > Decimal::ZERO{
+            self.volume24h = ticker.volume;
+        }
 
         self.processor().await;
     }
@@ -208,7 +251,15 @@ impl AvellanedaStoikov {
 
     pub async fn update_inventory(&mut self, inventory: &Decimal, min_amount_value: &Decimal) {
         self.prev_trade_time = Utc::now().timestamp_micros();
-        self.inventory = (inventory / (min_amount_value / self.mid_price)).round();
+        if self.mid_price == Decimal::ZERO {
+            error!("----------mid_price is zero-----------");
+        } else {
+            self.inventory = (inventory / (min_amount_value / self.mid_price)).round();
+        }
+        if min_amount_value.is_zero() {
+            error!("----------min_amount_value is zero-----------");
+        }
+
 
         self.update_level().await;
         self.processor().await;
@@ -261,7 +312,12 @@ impl AvellanedaStoikov {
             //     Decimal::PI * self.gamma * self.sigma_square * self.inventory.abs().powd(Decimal::TWO) * self.t_diff
             // };
             let pos_edge = self.gamma * self.sigma_square * self.inventory.abs().powd(dec!(2)) * self.t_diff;
-
+            if self.gamma == Decimal::ZERO {
+                error!("----------gamma is zero-----------")
+            }
+            if self.kappa == Decimal::ZERO {
+                error!("----------kappa is zero-----------")
+            }
             self.base_delta = self.gamma * self.sigma_square * self.t_diff / Decimal::TWO + (Decimal::ONE / self.gamma) * (Decimal::ONE + self.gamma / self.kappa).ln();
             self.ratio_edge = self.flow_ratio_long * self.sigma_square;
 
@@ -381,7 +437,7 @@ impl AvellanedaStoikov {
         if flow_out_value + flow_in_value > *min_volume {
             // let now = (flow_in_value - flow_out_value) / (flow_out_value + flow_in_value);
             // a * now + (Decimal::ONE - a) * prev_flow_ratio
-            (flow_in_value - flow_out_value) / (flow_out_value + flow_in_value)
+            (flow_in_value - flow_out_value).checked_div(flow_out_value + flow_in_value).unwrap()
         } else {
             Decimal::ZERO
         }
@@ -417,7 +473,7 @@ impl AvellanedaStoikov {
         if flow_out_value + flow_in_value > *min_volume {
             // let now = (flow_in_value - flow_out_value) / (flow_out_value + flow_in_value);
             // a * now + (Decimal::ONE - a) * prev_flow_ratio
-            (flow_in_value - flow_out_value) / (flow_out_value + flow_in_value)
+            (flow_in_value - flow_out_value).checked_div(flow_out_value + flow_in_value).unwrap()
         } else {
             Decimal::ZERO
         }
@@ -429,7 +485,32 @@ impl AvellanedaStoikov {
     }
 
     pub fn update_vwpin(&mut self) {
-        self.vwpin = self.vwpin_source_vec.deque.iter().copied().sum();
+        if self.vwpin_source_map.len() > 0 {
+            let mut basic_vec: Vec<Decimal> = Vec::new();
+
+            for (_key, value) in &self.vwpin_source_map {
+                if value.total_size > Decimal::ZERO {
+                    let total_size = value.sell_size + value.buy_size;
+                    let vwpin_item = total_size / value.total_size * (value.sell_size - value.buy_size).abs() / total_size;
+                    basic_vec.push(vwpin_item);
+                }
+            }
+
+            if basic_vec.len() > 0{
+                let vwpin = basic_vec.iter().copied().sum();
+                self.vwpin = vwpin;
+               // 10s后开始更新平均值
+               if basic_vec.len() > 10 {
+                   // 限制记录历史vwpin值数量，防止数据溢出·
+                   if self.vwpin_vec.len() >= 100 {
+                       self.vwpin_vec.pop_front();
+                   }
+                   self.vwpin_vec.push_back(vwpin);
+                   let count_vwpin: Decimal = self.vwpin_vec.iter().copied().sum();
+                   self.vwpin_avg = count_vwpin / Decimal::from(self.vwpin_vec.len());
+               }
+            }
+        }
     }
 
     pub fn check_ready(&mut self) {
@@ -461,6 +542,10 @@ impl AvellanedaStoikov {
             return;
         }
 
+        if self.vwpin_source_map.len() < 5 {
+            return;
+        }
+
         if self.trade_long_vec.len() < 100 {
             return;
         }
@@ -496,8 +581,7 @@ impl AvellanedaStoikov {
         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.last_price,
+            mid_price: self.mid_price,
             ask_price: self.ask_price,
             bid_price: self.bid_price,
             last_price: self.last_price,
@@ -508,9 +592,10 @@ impl AvellanedaStoikov {
             optimal_bid_price: self.optimal_bid_price,
 
             inventory: self.inventory,
-            sigma_square: self.flow_ratio_long,
+            // sigma_square: self.flow_ratio_long,
+            sigma_square: self.vwpin_avg,
             gamma: self.flow_ratio_short,
-            kappa: self.t_diff,
+            kappa: self.vwpin,
             vwpin: self.vwpin,
 
             flow_ratio: self.flow_ratio_long,

strategy/src/bybit_usdt_swap.rs

@@ -13,7 +13,7 @@ use standard::exchange::ExchangeEnum::BybitSwap;
 use standard::exchange_struct_handler::ExchangeStructHandler;
 use standard::{Depth, OrderBook};
 use crate::core::Core;
-use crate::exchange_disguise::{on_depth, on_trade};
+use crate::exchange_disguise::{on_depth, on_ticker, on_trade};
 use crate::model::OrderInfo;
 
 // 参考 Bybit 合约 启动
@@ -160,8 +160,11 @@ async fn on_public_data(core_arc: Arc<Mutex<Core>>, mul: &Decimal, response: &Re
             }
         }
         "tickers" => {
+            trace_stack.set_source("bybit_usdt_swap.tickers".to_string());
             let ticker = ExchangeStructHandler::ticker_handle(BybitSwap, response).await;
-            info!(?ticker)
+            trace_stack.on_after_format();
+
+            on_ticker(core_arc, &mut trace_stack, &ticker).await;
         }
         _ => {
             error!("未知推送类型");

strategy/src/core.rs

@@ -612,6 +612,10 @@ impl Core {
         self.avellaneda_stoikov.on_trade(trade).await;
     }
 
+    pub async fn on_ticker(&mut self, ticker: &Ticker, _trace_stack: &mut TraceStack) {
+        self.avellaneda_stoikov.on_ticker(ticker).await;
+    }
+
     // #[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) {
         // ================================ 刷新更新间隔 =========================================

strategy/src/exchange_disguise.rs

@@ -3,7 +3,7 @@ use std::sync::Arc;
 use std::sync::atomic::AtomicBool;
 use tokio::sync::Mutex;
 use global::trace_stack::TraceStack;
-use standard::{Depth, Trade};
+use standard::{Depth, Ticker, Trade};
 use crate::binance_usdt_swap::{binance_swap_run, reference_binance_swap_run};
 use crate::bybit_usdt_swap::{bybit_swap_run, reference_bybit_swap_run};
 use crate::coinex_usdt_swap::coinex_swap_run;
@@ -137,6 +137,13 @@ pub async fn on_trade(core_arc: Arc<Mutex<Core>>, label: &String, trace_stack: &
     core.on_trade(trade, &label, trace_stack).await;
 }
 
+pub async fn on_ticker(core_arc: Arc<Mutex<Core>>, trace_stack: &mut TraceStack, ticker: &Ticker) {
+    let mut core = core_arc.lock().await;
+    trace_stack.on_after_unlock_core();
+
+    core.on_ticker(ticker, trace_stack).await;
+}
+
 pub async fn on_order() {}
 
 pub async fn on_position() {}