as-rust/strategy/src/predictor.rs

use std::cmp::{max, min};
use std::collections::{BTreeMap, VecDeque};
use std::sync::Arc;
use chrono::{Utc};
use futures_channel::mpsc::UnboundedSender;
use futures_util::StreamExt;
use rust_decimal::prelude::*;
use rust_decimal_macros::dec;
use tokio::sync::{Mutex};
use tracing::{info};
use global::cci::CentralControlInfo;
use global::fixed_time_range_deque::FixedTimeRangeDeque;
use global::params::Params;
use standard::{Depth, Record, Ticker, Trade};
use crate::utils;

#[derive(Debug)]
pub struct Predictor {
    pub depth_vec: Vec<Depth>,                                                  // 深度队列
    pub volume_vec: Vec<Decimal>,                                               // 深度队列
    pub trade_long_vec: FixedTimeRangeDeque<Trade>,                             // 交易队列
    pub trade_short_vec: FixedTimeRangeDeque<Trade>,                            // 交易队列
    pub spread_vec: Vec<Decimal>,                                               // 市场冲击队列
    pub record_vec: VecDeque<Record>,                                           // 蜡烛队列

    pub mid_price: Decimal,                                                     // 中间价
    pub ask_price: Decimal,                                                     // 卖一价
    pub bid_price: Decimal,                                                     // 买一价
    pub last_price: Decimal,                                                    // 最后成交价
    pub spread: Decimal,                                                        // 市场冲击
    pub spread_max: Decimal,                                                    // 最大市场冲击
    pub spread_best: Decimal,                                                   // 最佳市场冲击
    pub optimal_ask_price: Decimal,                                             // 卖出挂单价
    pub optimal_bid_price: Decimal,                                             // 买入挂单价

    pub inventory: Decimal,                                                     // 库存，也就是q
    pub pos_amount: Decimal,                                                    // 原始持仓量
    pub pos_avg_price: Decimal,                                                 // 原始持仓价格
    pub level: Decimal,                                                         // martin
    pub sigma_square: Decimal,                                                  // σ^2，波动性的平方
    pub gamma: Decimal,                                                         // γ，库存风险厌恶参数
    pub kappa: Decimal,                                                         // κ 订单簿 流动性 参数

    pub flow_ratio: Decimal,                                                    // 资金流比例
    pub money_flow_index: Decimal,                                              // MFI
    pub long_trade_len_dec: Decimal,
    pub short_trade_len_dec: Decimal,
    pub error_rate: Decimal,                                                    // 犯错概率（预估）
    pub dir: Decimal,                                                           // 行情方向

    pub ask_delta: Decimal,                                                     // δa
    pub bid_delta: Decimal,                                                     // δb
    pub base_delta: Decimal,                                                    // 基础挂单距离
    pub ratio_edge: Decimal,                                                    // 资金流修正的挂单距离

    pub fair_price_vec: Vec<Decimal>,                                           // 预定价格队列
    pub fair_price: Decimal,                                                    // 预定价格
    pub fair_price_when_ordering: Decimal,                                      // 下单时的预定价格
    pub price_times_avg: Decimal,                                               // 公平所与做市所的价格倍率的平均值
    pub is_regressed: bool,                                                     // 做市所的价格是否已经回归

    pub cci_arc: Arc<Mutex<CentralControlInfo>>,                                // 中控信息

    pub is_ready: bool,
    pub prev_trade_time: i64,                                                   // 上次交易时间，也就是t
    pub close_price: Decimal,                                                   // 平仓价格
    pub t_diff: Decimal,                                                        // (T-t)
    pub last_update_time: Decimal,
    pub last_index: Decimal,

    pub prev_insert_time: Decimal,
    pub prev_save_time: Decimal,

    pub params: Params,

    pub debug_sender: UnboundedSender<Vec<Decimal>>,
}

impl Predictor {
    // 时间窗口大小（微秒）
    // const MAX_TIME_RANGE_MICROS: i64 = 3 * 60_000_000;
    const TIME_DIFF_RANGE_MICROS: i64 = 10 * 60_000_000;
    const TRADE_LONG_RANGE_MICROS: i64 = 3 * 60_000_000;
    // const SPREAD_RANGE_MICROS: i64 = 15 * 60_000_000;
    const TRADE_SHORT_RANGE_MICROS: i64 = 30_000_000;
    // const ONE_MILLION: Decimal = dec!(1_000_000);
    // const TWENTY_THOUSAND: Decimal = dec!(20_000);
    const IRA: Decimal = dec!(1);

    pub fn new(cci_arc: Arc<Mutex<CentralControlInfo>>, params: Params) -> Self {
        // 创建数据通道
        // 创建一个无界通道
        let (tx, mut rx) = futures_channel::mpsc::unbounded::<Vec<Decimal>>();

        tokio::spawn(async move {
            let len = 16usize;
            let mut prev_save_time = Decimal::from(Utc::now().timestamp_millis());
            let mut debugs: Vec<VecDeque<Option<Decimal>>> = vec![VecDeque::new(); len];

            while let Some(value) = rx.next().await {
                // 数据填充到对应位置
                for i in 0..len {
                    if value[i] == dec!(14142135623730951) {
                        debugs[i].push_back(None);
                    } else {
                        debugs[i].push_back(Some(value[i]));
                    }
                }

                // 长度限制
                if debugs[0].len() > 500_000 {
                    for i in 0..len {
                        debugs[i].pop_front(); // 从前面移除元素
                    }
                }

                let now = Decimal::from(Utc::now().timestamp_millis());
                if now - prev_save_time < dec!(60000) {
                    continue;
                }

                let debugs_clone = debugs.clone();
                let temp_html_str = tokio::task::spawn_blocking(move || {
                    utils::build_html_file(&debugs_clone)
                }).await.unwrap();

                utils::write_to_file(&temp_html_str, "./db/db.html".to_string()).await;
                prev_save_time = Decimal::from(Utc::now().timestamp_millis());
            }
        });

        let predictor = Self {
            // 接针版本
            depth_vec: vec![Depth::new(); 10],
            fair_price_vec: vec![Decimal::ZERO; 10],
            volume_vec: vec![Decimal::ZERO; 10],

            // 老的队列
            spread_vec: vec![],
            trade_long_vec: FixedTimeRangeDeque::new(Self::TRADE_LONG_RANGE_MICROS),
            trade_short_vec: FixedTimeRangeDeque::new(Self::TRADE_SHORT_RANGE_MICROS),
            record_vec: VecDeque::new(),

            mid_price: Default::default(),
            ask_price: Default::default(),
            bid_price: Default::default(),
            last_price: Default::default(),
            spread: Default::default(),
            spread_max: Default::default(),
            spread_best: Default::default(),
            optimal_ask_price: Default::default(),
            optimal_bid_price: Default::default(),

            inventory: Default::default(),
            gamma: Default::default(),
            sigma_square: Default::default(),
            ask_delta: Default::default(),
            bid_delta: Default::default(),
            base_delta: Default::default(),

            ratio_edge: Default::default(),
            kappa: Default::default(),
            fair_price: Default::default(),
            fair_price_when_ordering: Default::default(),

            price_times_avg: Default::default(),
            is_regressed: false,
            cci_arc,

            is_ready: false,
            prev_trade_time: Utc::now().timestamp_micros(),
            close_price: Default::default(),
            t_diff: Default::default(),
            level: Default::default(),
            flow_ratio: Default::default(),
            money_flow_index: Default::default(),
            pos_amount: Default::default(),
            error_rate: Default::default(),
            long_trade_len_dec: Default::default(),
            short_trade_len_dec: Default::default(),
            last_update_time: Default::default(),
            last_index: Default::default(),
            pos_avg_price: Default::default(),

            prev_insert_time: Default::default(),
            prev_save_time: Decimal::from(Utc::now().timestamp_millis()),
            dir: Default::default(),
            params,

            debug_sender: tx,
        };

        predictor
    }

    // 更新最大市场冲击
    pub fn update_spread_max(&mut self) {
        self.spread_max = if let Some(&max_value) = self.spread_vec.iter().max() {
            max_value
        } else {
            Decimal::NEGATIVE_ONE
        };
    }

    // 更新最佳市场冲击
    // pub fn update_spread_best(&mut self) {
    //     self.spread_best = self.spread_max;
    //     let mut max_count = 0usize;
    //
    //     for (spread, count) in self.spread_count_map.iter() {
    //         // info!(?spread, ?count);
    //
    //         if *count < max_count {
    //             continue
    //         }
    //
    //         self.spread_best = *spread;
    //         max_count = *count;
    //     }
    //     // info!("======================")
    // }

    pub fn update_spread(&mut self) {
        // if self.trade_long_vec.len() > 0 {
        let prev_depth_0 = &self.depth_vec[0];
        if prev_depth_0.time.is_zero() {
            return;
        }

        let prev_mid_price = (prev_depth_0.asks[0].price + prev_depth_0.bids[0].price) / Decimal::TWO;
        let now_spread = self.mid_price - prev_mid_price;

        if !now_spread.is_zero() {
            self.spread = now_spread;
            self.spread_vec.push(self.spread.abs());

            // 看空方向
            if now_spread < self.spread_max * Decimal::NEGATIVE_ONE && !self.spread_max.is_zero() {
                self.dir = Decimal::NEGATIVE_ONE
            }

            // 看多方向
            if now_spread > self.spread_max && !self.spread_max.is_zero() {
                self.dir = Decimal::ONE
            }
            // if last_trade_price > first_trade_price {
            //     self.spread_long_vec.push(self.spread);
            // } else {
            //     self.spread_short_vec.push(self.spread);
            // }

            while self.spread_vec.len() > 1_000 && self.inventory.is_zero() {
                self.spread_vec.remove(0);
            }

            self.update_spread_max();
            // self.update_spread_best();
        }
        // }
    }

    pub async fn on_depth(&mut self, depth: &Depth, index: usize) {
        self.last_update_time = depth.time;
        self.last_index = Decimal::from(index);
        self.update_fair_price(depth, index).await;

        if index == 0 {
            self.ask_price = depth.asks[0].price;
            self.bid_price = depth.bids[0].price;
            self.mid_price = (self.ask_price + self.bid_price) / Decimal::TWO;

            self.update_spread();
        }

        self.depth_vec[index] = depth.clone();

        if self.mid_price.is_zero() {
            return;
        }
        self.processor().await;
    }

    pub async fn on_trade(&mut self, trade: &Trade, _index: usize) {
        // self.last_update_time = trade.time;

        self.trade_long_vec.push_back(trade.clone());
        self.trade_short_vec.push_back(trade.clone());

        self.long_trade_len_dec = Decimal::from_usize(self.trade_long_vec.len()).unwrap();
        self.short_trade_len_dec = Decimal::from_usize(self.trade_short_vec.len()).unwrap();
        self.error_rate = self.short_trade_len_dec / self.long_trade_len_dec;
        self.error_rate.rescale(4);

        self.last_price = trade.price;
        // self.processor().await;
    }

    pub async fn update_level(&mut self) {
        self.level = (Decimal::NEGATIVE_ONE + (Decimal::ONE + dec!(8) * self.inventory.abs()).sqrt().unwrap()) / Decimal::TWO;
        self.level = min(self.level, dec!(6));
    }

    pub async fn on_ticker(&mut self, _ticker: &Ticker) {}

    pub async fn on_record(&mut self, record: &Record) {
        // 添加新蜡烛
        if self.record_vec.len() == 0 {
            self.record_vec.push_back(record.clone());
        } else {
            let last_record = self.record_vec.back_mut().unwrap();

            if last_record.time == record.time {
                *last_record = record.clone();
            } else if last_record.time < record.time {
                self.record_vec.push_back(record.clone());
            }
        }

        if self.record_vec.len() > 4 {
            self.record_vec.pop_front();
        }

        // 如果蜡烛数量足够，则更新mfi
        if self.record_vec.len() >= 4 {
            self.update_mfi();
        }
    }

    pub fn update_mfi(&mut self) {
        let mut money_flow_in = Decimal::ZERO;
        let mut money_flow_out = Decimal::ZERO;
        let _3 = dec!(3);

        for record in self.record_vec.iter() {
            let typical_price = (record.high + record.low + record.close) / _3;
            let money_flow = typical_price * record.volume;
            if record.close > record.open {
                money_flow_in += money_flow;
            } else if record.close < record.open {
                money_flow_out += money_flow;
            }
        }

        self.money_flow_index = if money_flow_out.is_zero() {
            Decimal::ONE_HUNDRED
        } else {
            let money_flow_ratio = money_flow_in / money_flow_out;

            Decimal::ONE_HUNDRED - Decimal::ONE_HUNDRED / (Decimal::ONE + money_flow_ratio)
        };

        self.update_flow_ratio();
    }

    pub async fn on_inventory(&mut self, pos_amount: &Decimal, pos_avg_price: &Decimal, min_amount_value: &Decimal) {
        if self.mid_price.is_zero() {
            return;
        }

        let prev_inventory = self.inventory;
        self.pos_amount = pos_amount.clone();
        self.pos_avg_price = pos_avg_price.clone();
        self.inventory = (pos_amount / (min_amount_value / self.mid_price)).trunc();
        // 小于1但不为0的情况，需要平完
        if self.inventory.is_zero() && !pos_amount.is_zero() {
            self.inventory = if pos_amount > &Decimal::ZERO {
                Decimal::ONE
            } else {
                Decimal::NEGATIVE_ONE
            };
        }

        if prev_inventory != self.inventory && prev_inventory.is_zero() {
            self.prev_trade_time = Utc::now().timestamp_micros();
            self.close_price = self.fair_price_when_ordering;
        }

        if prev_inventory != self.inventory && self.inventory.is_zero() {
            self.is_regressed = false;
        }

        self.update_level().await;
        self.processor().await;
    }

    pub fn update_sigma_square(&mut self) {
        self.sigma_square = self.fair_price * self.params.open;
        self.sigma_square.rescale(10);
    }

    pub fn update_gamma(&mut self) {
        self.gamma = dec!(0.236) * Self::IRA;
    }

    pub fn update_kappa(&mut self) {
        if self.mid_price > Decimal::ZERO {
            self.kappa = dec!(888) / self.mid_price;
        } else {
            self.kappa = dec!(1);
        }
    }

    pub async fn update_fair_price(&mut self, depth: &Depth, index: usize) {
        if self.mid_price.is_zero() {
            return;
        }

        let a1 = &depth.asks[0];
        let b1 = &depth.bids[0];
        let mid = (a1.price + b1.price) / Decimal::TWO;

        let i_upper = (a1.value - b1.value) / (a1.value + b1.value);
        // let s_upper = (a1.price - b1.price) / mid;
        // let f_t = Decimal::ZERO;
        // let s = (s_upper + f_t) / Decimal::TWO;
        //
        // let a_upper_t = dec!(0.6);
        // let c_t = dec!(0);
        // let theta = a_upper_t * s + c_t;
        // let fair_price = mid + theta * (i_upper * (i_upper.powd(Decimal::TWO) + Decimal::ONE)) / Decimal::TWO;

        let fair_price = mid + (a1.price - b1.price) * i_upper / Decimal::TWO;

        self.fair_price_vec[index] = if self.fair_price_vec[index].is_zero() {
            fair_price
        } else {
            self.fair_price_vec[index] * dec!(0.9) + fair_price * dec!(0.1)
        };
        self.fair_price_vec[index].rescale(self.mid_price.scale());
        self.volume_vec[index] = a1.size + b1.size;

        // 合成公平价格
        if !self.fair_price_vec[0].is_zero() && !self.fair_price_vec[1].is_zero() {
            self.price_times_avg = if self.price_times_avg.is_zero() {
                self.fair_price_vec[1] / self.fair_price_vec[0]
            } else {
                self.price_times_avg * dec!(0.9999) + dec!(0.0001) * self.fair_price_vec[1] / self.fair_price_vec[0]
            };

            // 进行价格归一化处理，公平所的价格有可能是带前缀的
            let fair_price_part0 = self.fair_price_vec[0] * dec!(0.2);
            let fair_price_part1 = (self.fair_price_vec[1] / self.price_times_avg) * dec!(0.8);
            self.fair_price = fair_price_part0 + fair_price_part1;
        }

        // 判断价格是否回归
        if !self.is_regressed && self.inventory > Decimal::ZERO && self.fair_price < self.mid_price * (Decimal::ONE - self.params.open) {
            self.is_regressed = true
        } else if !self.is_regressed && self.inventory < Decimal::ZERO && self.fair_price > self.mid_price * (Decimal::ONE + self.params.open) {
            self.is_regressed = true
        }
    }

    pub fn update_delta(&mut self) {
        // -2表示不想成交
        // -1表示市价成交

        if self.fair_price.is_zero() {
            return;
        }

        let is_open_long = (self.fair_price - self.mid_price) / self.mid_price > self.params.open;
        let is_open_short = (self.mid_price - self.fair_price) / self.mid_price > self.params.open;
        let is_close_long = self.inventory > Decimal::ZERO;
        let is_close_short = self.inventory < Decimal::ZERO;

        self.bid_delta = dec!(-2);
        self.ask_delta = dec!(-2);

        if is_close_long {
            if self.mid_price > self.pos_avg_price {
                // let profit_rate = (self.mid_price - self.pos_avg_price) / self.pos_avg_price;
                // let fill_rate = profit_rate / (self.params.open * dec!(50));
                // let close_rate = (Decimal::ONE - fill_rate) * self.params.open + self.params.close;
                //
                // self.ask_delta = self.mid_price * close_rate * self.t_diff;
                self.ask_delta = self.mid_price * self.params.close;
            } else if self.mid_price > self.fair_price {
                self.ask_delta = self.mid_price * self.params.close;
            }
        } else if is_close_short {
            if self.mid_price < self.pos_avg_price {
                // let profit_rate = (self.pos_avg_price - self.mid_price) / self.pos_avg_price;
                // let fill_rate = profit_rate / (self.params.open * dec!(50));
                // let close_rate = (Decimal::ONE - fill_rate) * self.params.open + self.params.close;
                //
                // self.bid_delta = self.mid_price * close_rate * self.t_diff;
                self.bid_delta = self.mid_price * self.params.close;
            } else if self.mid_price < self.fair_price {
                self.bid_delta = self.mid_price * self.params.close;
            }
        } else if is_open_long {
            let is_open_long_market = (self.fair_price - self.mid_price) / self.mid_price > self.params.open_market;
            self.bid_delta = if is_open_long_market {
                dec!(-1)
            } else {
                dec!(0)
            };
        } else if is_open_short {
            let is_open_short_market = (self.mid_price - self.fair_price) / self.mid_price > self.params.open_market;
            self.ask_delta = if is_open_short_market {
                dec!(-1)
            } else {
                dec!(0)
            }
        }
    }

    pub fn update_optimal_ask_and_bid(&mut self) {
        self.optimal_ask_price = if self.ask_delta == dec!(-1) {
            self.bid_price
        } else if self.ask_delta == dec!(-2) {
            dec!(14142135623730951)
        } else {
            max(self.ask_price + self.ask_delta, self.bid_price)
        };

        self.optimal_bid_price = if self.bid_delta == dec!(-1) {
            self.ask_price
        } else if self.bid_delta == dec!(-2) {
            dec!(14142135623730951)
        } else {
            min(self.bid_price - self.bid_delta, self.ask_price)
        };

        self.optimal_ask_price.rescale(self.mid_price.scale());
        self.optimal_bid_price.rescale(self.mid_price.scale());
    }

    pub fn update_t_diff(&mut self) {
        if self.prev_trade_time > 0 {
            let time_diff_decimal = Decimal::from_i64(Utc::now().timestamp_micros() - self.prev_trade_time).unwrap();
            self.t_diff = max(Decimal::ONE - time_diff_decimal / Decimal::from_i64(Self::TIME_DIFF_RANGE_MICROS).unwrap(), Decimal::ZERO);
        } else {
            self.t_diff = Decimal::ONE;
        }
    }

    // fn calc_flow_ratio(_prev_flow_ratio: &Decimal, min_volume: &Decimal, trades: &mut FixedTimeRangeDeque<Trade>) -> Decimal {
    //     let mut flow_in_value = Decimal::ZERO;
    //     let mut flow_out_value = Decimal::ZERO;
    //     for trade_iter in trades.deque.iter() {
    //         if trade_iter.size > Decimal::ZERO {
    //             flow_in_value += trade_iter.value;
    //         } else {
    //             flow_out_value += trade_iter.value;
    //         }
    //     }
    //
    //     // 使用EMA來更新資金流，確保平滑性
    //     if flow_out_value + flow_in_value > *min_volume {
    //         let now = (flow_in_value - flow_out_value) / (flow_out_value + flow_in_value);
    //         now
    //     } else {
    //         Decimal::ZERO
    //     }
    // }

    pub fn update_flow_ratio(&mut self) {
        self.flow_ratio = if self.money_flow_index > dec!(80) {
            (self.money_flow_index - dec!(80)) / dec!(20)
        } else if self.money_flow_index < dec!(20) {
            Decimal::NEGATIVE_ONE * (dec!(20) - self.money_flow_index) / dec!(20)
        } else {
            Decimal::ZERO
        };
    }

    pub fn check_ready(&mut self) {
        if self.is_ready {
            return;
        }

        if self.mid_price == Decimal::ZERO {
           return;
        }

        if self.fair_price == Decimal::ZERO {
            return;
        }

        if self.ask_price == Decimal::ZERO {
            return;
        }

        if self.bid_price == Decimal::ZERO {
            return;
        }

        if self.optimal_ask_price < self.ask_price {
            return;
        }

        if self.optimal_bid_price > self.bid_price {
            return;
        }

        if self.trade_long_vec.len() < 100 {
            return;
        }

        self.is_ready = true;
        info!("========================================行情数据预热完毕==================================")
    }

    // #[instrument(skip(self), level="TRACE")]
    async fn processor(&mut self) {
        self.update_t_diff();
        self.update_sigma_square();
        self.update_gamma();
        self.update_kappa();
        self.update_delta();
        self.update_optimal_ask_and_bid();

        self.check_ready();
        if !self.is_ready {
            return;
        }

        // let mut smm = Decimal::ZERO;
        // if !self.depth_vec[1].time.is_zero() {
        //     let sma = self.depth_vec[1].asks[0].price;
        //     let smb = self.depth_vec[1].bids[0].price;
        //     smm = (sma + smb) / Decimal::TWO;
        // }


        // let cci_arc = self.cci_arc.clone();
        let now = Decimal::from_i64(Utc::now().timestamp_millis()).unwrap();
        let mid_price = self.mid_price;
        let ask_price = self.ask_price;
        let bid_price = self.bid_price;
        let last_price = self.last_price;

        let spread = self.mid_price;
        let spread_max = self.optimal_ask_price;
        let spread_min = self.optimal_bid_price;
        // let spread = self.price_times_avg;
        // let spread_max = self.fair_price_vec[1] / self.fair_price_vec[0];
        // let spread_min = self.fair_price / self.mid_price;

        let optimal_ask_price = self.optimal_ask_price;
        let optimal_bid_price = self.optimal_bid_price;

        let inventory = self.inventory;
        let sigma_square = if self.is_regressed { Decimal::ONE } else { Decimal::ZERO };
        let gamma = now - self.last_update_time;
        let kappa = self.fair_price / self.mid_price;

        let flow_ratio = Decimal::ZERO;
        let ref_price = self.fair_price;

        let need_append = now - self.prev_insert_time > Decimal::ONE_HUNDRED;
        if !need_append {
            return;
        }

        self.debug_sender.unbounded_send(vec![
            now,
            mid_price,
            ask_price,
            bid_price,
            last_price,
            spread,
            spread_max,
            spread_min,
            optimal_ask_price,
            optimal_bid_price,
            inventory,
            sigma_square,
            gamma,
            kappa,
            flow_ratio,
            ref_price
        ]).unwrap();

        self.prev_insert_time = Decimal::from(Utc::now().timestamp_millis())
    }

    // #[instrument(skip(self, ref_ticker_map), level="TRACE")]
    pub fn get_ref_price(&mut self, _ref_ticker_map: &BTreeMap<String, Ticker>) -> Vec<Vec<Decimal>> {
        vec![]
    }
}