下单风控 亏损时+波动率异常不开单

global/src/clear_log_utils.rs

@@ -1,7 +1,7 @@
 use std::collections::HashMap;
 use std::fmt::Debug;
 use std::io;
-use tracing::{Event, Subscriber, warn};
+use tracing::{Event, info, Subscriber, warn};
 use tracing_appender_timezone::non_blocking::WorkerGuard;
 use tracing_subscriber::{fmt, Layer};
 use tracing_subscriber::layer::{Context, SubscriberExt};
@@ -39,6 +39,7 @@ where
             event.record(&mut visitor);
 
             let msg = format!("account={}, type=error, msg={}", self.account_name.clone(), visitor.message);
+            info!(msg)
             // send_remote_err_log(msg)
         }
     }

global/src/lib.rs

@@ -6,4 +6,5 @@ pub mod export_utils;
 pub mod account_info;
 pub mod cci;
 pub mod clear_position_result;
+pub mod trade;
 pub mod clear_log_utils;

global/src/log_utils.rs

@@ -1,7 +1,7 @@
 use std::collections::HashMap;
 use std::fmt::Debug;
 use std::io;
-use tracing::{Event, Subscriber, warn};
+use tracing::{Event, info, Subscriber, warn};
 use tracing_appender_timezone::non_blocking::WorkerGuard;
 use tracing_subscriber::{fmt, Layer};
 use tracing_subscriber::layer::{Context, SubscriberExt};
@@ -39,6 +39,7 @@ impl<S> Layer<S> for ReportingLayer
             event.record(&mut visitor);
 
             let msg = format!("account={}, type=error, msg={}", self.account_name.clone(), visitor.message);
+            info!(msg)
             // send_remote_err_log(msg)
         }
     }

global/src/trade.rs

@@ -0,0 +1,17 @@
+use rust_decimal::Decimal;
+
+pub struct Trade {
+    // 价格
+    pub price: Decimal,
+    // 时间
+    pub time: i64
+}
+
+impl Trade {
+    pub fn new_by_ticker (price: Decimal, time: i64) -> Trade {
+        Trade{
+            price,
+            time
+        }
+    }
+}

strategy/src/core.rs

@@ -1,13 +1,13 @@
 use tokio::time::Instant;
 use std::cmp::max;
-use std::collections::{BTreeMap, HashMap};
+use std::collections::{BTreeMap, HashMap, VecDeque};
 use std::io::Error;
 use std::str::FromStr;
 use std::sync::{Arc};
 use std::sync::atomic::{AtomicBool, Ordering};
 use std::time::Duration;
 use chrono::{Utc};
-use rust_decimal::Decimal;
+use rust_decimal::{Decimal, MathematicalOps};
 use rust_decimal::prelude::{ToPrimitive};
 use rust_decimal_macros::dec;
 use tokio::spawn;
@@ -20,6 +20,7 @@ use global::cci::CentralControlInfo;
 use global::params::Params;
 use global::public_params::{ASK_PRICE_INDEX, BID_PRICE_INDEX, LENGTH};
 use global::trace_stack::TraceStack;
+use global::trade::Trade;
 use standard::{Account, Market, Order, OrderCommand, Platform, Position, PositionModeEnum, SpecialTicker, Ticker};
 use standard::exchange::{Exchange};
 use standard::exchange::ExchangeEnum::{BinanceSwap, BitgetSwap, BybitSwap, CoinexSwap, GateSwap, HtxSwap, KucoinSwap};
@@ -115,6 +116,12 @@ pub struct Core {
     pub agg_market: Vec<Decimal>,
     pub ref_price: Vec<Vec<Decimal>>,
     pub predict: Decimal,
+
+    // 波动率指标相关数据（最近100条）
+    pub trade_vec: VecDeque<Trade>,  // 行情数据 此处价格取买一卖一中间价，时间为交易所的数据创建时间
+    pub sigma_vec: VecDeque<Decimal>,  // 波动率记录
+    pub is_sigma_abnormal: bool,  // 是否sigma反常
+    pub is_sigma_allow_open: bool, // 是否允许开单
 }
 
 impl Core {
@@ -251,6 +258,10 @@ impl Core {
             agg_market: vec![],
             ref_price: vec![],
             predict: Default::default(),
+            trade_vec: VecDeque::with_capacity(100),
+            sigma_vec: VecDeque::with_capacity(100),
+            is_sigma_abnormal: false,
+            is_sigma_allow_open: true,
         };
         for i in 0..=params.ref_exchange.len() - 1 {
             // 拼接不会消耗原字符串
@@ -439,6 +450,7 @@ impl Core {
                                 info!("错误的仓位方向{}", side);
                             }
                         } else { // 合约订单流仓位计算
+                            let mut this_order_profit = Decimal::ZERO;
                             if side == "kd" { // buy 开多
                                 self.local_buy_amount += filled;
                                 self.local_buy_value += filled * filled_price;
@@ -465,7 +477,8 @@ impl Core {
                             } else if side == "pd" { // sell 平多
                                 self.local_sell_amount += filled;
                                 self.local_sell_value += filled * filled_price;
-                                self.local_profit += filled * (filled_price - self.local_position_by_orders.long_avg);
+                                this_order_profit = filled * (filled_price - self.local_position_by_orders.long_avg);
+                                self.local_profit += this_order_profit;
                                 self.local_position_by_orders.long_pos = self.local_position_by_orders.long_pos - filled;
                                 if self.local_position_by_orders.long_pos == Decimal::ZERO {
                                     self.local_position_by_orders.long_avg = Decimal::ZERO;
@@ -473,7 +486,8 @@ impl Core {
                             } else if side == "pk" { // buy 平空
                                 self.local_buy_amount += filled;
                                 self.local_buy_value += filled * filled_price;
-                                self.local_profit += filled * (self.local_position_by_orders.short_avg - filled_price);
+                                this_order_profit = filled * (self.local_position_by_orders.short_avg - filled_price);
+                                self.local_profit += this_order_profit;
                                 self.local_position_by_orders.short_pos = self.local_position_by_orders.short_pos - filled;
                                 if self.local_position_by_orders.short_pos == Decimal::ZERO {
                                     self.local_position_by_orders.short_avg = Decimal::ZERO;
@@ -485,6 +499,14 @@ impl Core {
                             if data.fee > Decimal::ZERO {
                                 self.local_profit -= data.fee;
                             }
+                            // 订单亏损 并且 波动率异常 不允许开单
+                            if this_order_profit < Decimal::ZERO && self.is_sigma_abnormal {
+                                self.is_sigma_allow_open = false;
+                                info!("交易触发亏损，不允许开单！");
+                                info!("sigma_vec：{:?}" , self.sigma_vec);
+                            } else {
+                                self.is_sigma_allow_open = true;
+                            }
                         }
                         // info!("成交单耗时数据：{}", time_record.to_string());
                         info!("更新推算仓位 {:?}", self.local_position_by_orders);
@@ -510,6 +532,7 @@ impl Core {
                                                               &self.local_coin,
                                                               &self.ref_price,
                                                               &self.predict,
+                                                              &self.is_sigma_allow_open,
                                                               &trace_stack.ins);
                         // trace_stack.on_after_strategy();
                         // 记录指令触发信息
@@ -637,6 +660,20 @@ impl Core {
                 self.on_agg_market();
             }
         } else if *name_ref == self.ref_name[0] { // 判断是否为当前跟踪的盘口
+            // 写入行情数据
+            let ticker = self.tickers.get(name_ref).unwrap();
+            if self.trade_vec.len() == 100 {
+                self.trade_vec.pop_front();
+            }
+            self.trade_vec.push_back(Trade::new_by_ticker(ticker.mid_price.clone(), ticker.create_at/1000));
+            // 更新波动率
+            if self.trade_vec.len() > 99 {
+                self.calc_sigma();
+                // 波动率正常，但还是不开单信号，允许开单
+                if !self.is_sigma_abnormal && !self.is_sigma_allow_open {
+                    self.is_sigma_allow_open = true;
+                }
+            }
             // 判断是否需要触发ontick 对行情进行过滤
             // 过滤条件 价格变化很大 时间间隔很长
             let mut flag = 0;
@@ -665,6 +702,7 @@ impl Core {
                                                        &self.local_coin,
                                                        &self.ref_price,
                                                        &self.predict,
+                                                       &self.is_sigma_allow_open,
                                                        &trace_stack.ins);
                 trace_stack.on_after_strategy();
 
@@ -703,6 +741,87 @@ impl Core {
         }
     }
 
+    pub fn calc_sigma(&mut self) {
+        for (index, trade) in self.trade_vec.iter().enumerate() {
+            if index == 0 {
+                continue
+            }
+
+            // 计算过去至多100条数据的sigma值 sigma^2 = (1 / (tn-t0))*sum((S(tk) - S(tk-1)) ^ 2)
+            let mut sigma_index = index - 1;
+            let t_last = Decimal::from_str(&format!("{}", trade.time)).unwrap();
+
+            let mut _t_first = Decimal::from_str(&format!("{}", trade.time)).unwrap();
+            // 右值
+            let mut total_right = Decimal::ZERO;
+            loop {
+                let flag_trade = self.trade_vec.get(sigma_index).unwrap();
+                let next_trade = self.trade_vec.get(sigma_index + 1).unwrap();
+
+                // 下标合法性判断
+                if sigma_index == 0 || sigma_index + 100 <= index {
+                    _t_first = Decimal::from_str(&format!("{}", flag_trade.time)).unwrap();
+                    break;
+                }
+
+                // 计算差值
+                let diff = Decimal::ONE - flag_trade.price / next_trade.price;
+                total_right += diff * diff;
+
+                sigma_index = sigma_index - 1;
+            }
+            let sigma_square = if _t_first == t_last {
+                let time_diff = Decimal::ONE;
+                (Decimal::ONE / time_diff) * total_right
+            } else {
+                let time_diff = (t_last - _t_first) / Decimal::ONE_THOUSAND;
+                (Decimal::ONE / time_diff) * total_right
+            };
+            let mut sigma = sigma_square.sqrt().unwrap();
+            sigma.rescale(6);
+            if self.sigma_vec.len() == 100 {
+                self.sigma_vec.pop_front();
+            }
+            self.sigma_vec.push_back(sigma);
+        }
+        if self.sigma_vec.len() > 99 {
+            let sigma = match self.sigma_vec.back() {
+                Some(&value) => value,
+                None => Decimal::TEN,
+            };
+
+            // 计算过去至多100个sigma值的平均值
+            let sigma_ma = if self.sigma_vec.len() > 0 {
+                let mut sigma_ma_index = self.sigma_vec.len();
+                let mut sigma_total = Decimal::ZERO;
+                let mut sigma_count = Decimal::ZERO;
+                loop {
+                    if sigma_ma_index == 0 || sigma_ma_index + 99 < self.sigma_vec.len() {
+                        break
+                    }
+                    // 步进
+                    sigma_ma_index -= 1;
+                    // 计算
+                    sigma_total += self.sigma_vec[sigma_ma_index];
+                    sigma_count += Decimal::ONE;
+                }
+                let mut sigma_ma = sigma_total / sigma_count;
+                sigma_ma.rescale(6);
+
+                sigma_ma
+            } else {
+                sigma
+            };
+            // sigma值大于平均值定义为波动率异常
+            if sigma > sigma_ma {
+                self.is_sigma_abnormal = true;
+                // info!("sigma: {}, sigma_ma: {}, sigma_vec: {:?}", sigma, sigma_ma, self.sigma_vec);
+            } else {
+                self.is_sigma_abnormal = false;
+            }
+        }
+    }
+
     // #[instrument(skip(self, data), level="TRACE")]
     pub async fn update_position(&mut self, data: Vec<Position>) {
         if data.is_empty() {

strategy/src/strategy.rs

@@ -1251,6 +1251,7 @@ impl Strategy {
                    local_coin: &Decimal,
                    ref_price: &Vec<Vec<Decimal>>,
                    predict: &Decimal,
+                   is_sigma_allow_open: &bool,
                    _ins: &Instant) -> OrderCommand {
         self.on_time_print();
 
@@ -1279,8 +1280,12 @@ impl Strategy {
         // 下单指令处理逻辑
         self._cancel_open(&mut command, local_orders);              // 撤单命令处理
         self._post_close(&mut command, local_orders);               // 平仓单命令处理
-        self._post_open(&mut command, local_orders);                // 限价单命令处理
-
+        // 波动率是否允许下单
+        if *is_sigma_allow_open {
+            self._post_open(&mut command, local_orders);                // 限价单命令处理
+        } else {
+            info!("波动率异常订单亏损, 不开单！");
+        }
         self._check_local_orders(&mut command, local_orders);       // 固定时间检查超时订单
         self._update_in_cancel(&mut command, local_orders);         // 更新撤单队列，是一个filter
         self._check_request_limit(&mut command);                    // 限制频率，移除不合规则之订单，是一个filter