as-rust/strategy/src/predictor.rs

use std::cmp::max;
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, Clone)]
pub struct Predictor {
    pub depth_vec: Vec<Depth>,                                                  // 深度队列
    pub record_vec: VecDeque<Record>,                                           // 蜡烛队列
    pub spread_vec: VecDeque<Decimal>,                                          // 价差队列
    pub trade_price_long_vec: FixedTimeRangeDeque<Decimal>,
    pub trade_233_vec: FixedTimeRangeDeque<Trade>,
    pub trade_0_vec: FixedTimeRangeDeque<Trade>,

    pub prices: Vec<Vec<VecDeque<Decimal>>>,                                    // [[[做市所], [参考所0]], ...]
    pub ks: Vec<Decimal>,
    pub bs: Vec<Decimal>,

    pub mid_price: Decimal,                                                     // 中间价
    pub fair_price: Decimal,
    pub ask_price: Decimal,                                                     // 卖一价
    pub bid_price: Decimal,                                                     // 买一价
    pub last_price: 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 balance: Decimal,                                                       // 初始余额
    pub prev_balance: Decimal,

    pub ask_delta: Decimal,                                                     // δa
    pub bid_delta: Decimal,                                                     // δb

    pub mid_price_vec: Vec<Decimal>,                                            //
    pub fair_price_std_vec: Vec<Decimal>,                                       // 公平价格列表，标准化之后的
    pub price_avg_times_vec: Vec<Decimal>,                                      // 公平所与做市所的价格倍率的平均值
    pub price_avg_times_long_vec: Vec<Decimal>,                                 // 公平所与做市所的价格倍率的平均值

    pub is_ready: bool,                                                         // 是否已准备好

    pub last_update_time: Decimal,                                              // 最后更新时间（depth）
    pub last_index: Decimal,                                                    // 最后更新的index

    pub prev_insert_time: Decimal,
    pub prev_save_time: Decimal,
    pub init_time: Decimal,
    pub fitting_delay: Decimal,
    pub prev_fitting_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 = 15 * 60_000_000;
    const TRADE_LONG_RANGE_MICROS: i64 = 10 * 60_000_000;
    // const SPREAD_RANGE_MICROS: i64 = 15 * 60_000_000;
    const TRADE_SHORT_RANGE_MICROS: i64 = 2 * 60_000_000;
    // const ONE_MILLION: Decimal = dec!(1_000_000);
    // const TWENTY_THOUSAND: Decimal = dec!(20_000);
    const DONT_VIEW: Decimal = dec!(14142135623730951);

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

        let account_port = params.port.clone();
        tokio::spawn(async move {
            let len = 17usize;
            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] == Self::DONT_VIEW {
                        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!(30000) {
                    continue;
                }

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

                let path = format!("./db/{}.html", account_port);
                utils::write_to_file(&temp_html_str, path).await;
                prev_save_time = Decimal::from(Utc::now().timestamp_millis());
            }
        });

        let predictor = Self {
            // 接针版本
            depth_vec: vec![Depth::new(); params.ref_exchange.len()],
            fair_price_std_vec: vec![Decimal::ZERO; params.ref_exchange.len()],
            mid_price_vec: vec![Decimal::ZERO; params.ref_exchange.len()],
            price_avg_times_vec: vec![Decimal::ZERO; params.ref_exchange.len()],
            price_avg_times_long_vec: vec![Decimal::ZERO; params.ref_exchange.len()],

            prices: vec![vec![VecDeque::new(); 2]; params.ref_exchange.len()],
            ks: vec![Decimal::ZERO; params.ref_exchange.len()],
            bs: vec![Decimal::ZERO; params.ref_exchange.len()],

            record_vec: VecDeque::new(),

            trade_price_long_vec: FixedTimeRangeDeque::new(Self::TRADE_LONG_RANGE_MICROS),
            trade_233_vec: FixedTimeRangeDeque::new(Self::TRADE_SHORT_RANGE_MICROS),
            trade_0_vec: FixedTimeRangeDeque::new(Self::TRADE_SHORT_RANGE_MICROS),
            spread_vec: VecDeque::new(),
            mid_price: Default::default(),
            fair_price: Default::default(),
            ask_price: Default::default(),
            bid_price: Default::default(),
            last_price: Default::default(),
            optimal_ask_price: Default::default(),
            optimal_bid_price: Default::default(),

            ask_delta: Default::default(),
            bid_delta: Default::default(),

            is_ready: false,

            inventory: Default::default(),
            pos_avg_price: Default::default(),
            pos_amount: Default::default(),
            balance: Default::default(),
            prev_balance: Default::default(),

            last_update_time: Default::default(),
            last_index: Default::default(),

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

            fitting_delay: Default::default(),
            prev_fitting_time: Default::default(),
            params,

            debug_sender: tx,
        };

        predictor
    }

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

        if index == 233  {
            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;

            // 拟合k与b
            for (mid_index, mp) in self.mid_price_vec.iter().enumerate() {
                if mp.is_zero() {
                    continue
                }

                self.prices[mid_index][0].push_back(self.mid_price);
                self.prices[mid_index][1].push_back(mp.clone());

                // 长度限制
                if self.prices[mid_index][0].len() > 10000 {
                    self.prices[mid_index][0].pop_front();
                    self.prices[mid_index][1].pop_front();
                }

                // 拟合，60s拟合一次
                let before_fitting = Utc::now().timestamp_millis();
                if Decimal::from(before_fitting) - self.prev_fitting_time > dec!(60_000) || self.prices[mid_index][0].len() < 1000 {
                    if let Some((k, b)) = self.linear_least_squares(mid_index).await {
                        self.ks[mid_index] = k;
                        self.bs[mid_index] = b;

                        self.fitting_delay = Decimal::from(Utc::now().timestamp_millis() - before_fitting);
                        self.prev_fitting_time = Decimal::from(before_fitting)
                    } else {
                        return;
                    }
                }
            }
        } else {
            self.update_fair_price(depth, index).await;

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

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

        self.processor(depth.time, false).await;
    }

    pub async fn on_trade(&mut self, trade: &Trade, index: usize) {
        // index == 233代表做市所
        // index == 0代表参考所

        self.last_price = trade.price;

        self.trade_price_long_vec.push_back(trade.price);

        if index == 233 {
            self.trade_233_vec.push_back(trade.clone());
        } else if index == 0 {
            self.trade_0_vec.push_back(trade.clone());
        }

        // self.processor().await;
    }

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

    pub async fn on_record(&mut self, _record: &Record) {}

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

        let prev_pos_amount = self.pos_amount;
        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_pos_amount != self.pos_amount {
            self.processor(update_time, true).await;
        }
    }

    pub async fn on_balance(&mut self, balance: Decimal) {
        self.balance = balance;
    }

    pub fn get_real_rate(price_vec: &FixedTimeRangeDeque<Decimal>) -> Decimal {
        let last_fair_price = price_vec.deque.iter().last().unwrap();
        let min_price = price_vec.deque.iter().min().unwrap();
        let max_price = price_vec.deque.iter().max().unwrap();
        let up_rate = (last_fair_price - min_price) / min_price;
        let down_rate = (max_price - last_fair_price) / max_price;

        if up_rate > down_rate {
            up_rate
        } else {
            -down_rate
        }
    }

    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];

        // https://quant.stackexchange.com/questions/50651/how-to-understand-micro-price-aka-weighted-mid-price
        // let total = a1.value + b1.value;
        // let fair_price = (a1.price + b1.price) / Decimal::TWO;

        // self.fair_price_vec[index] = a1.price * b1.value / total + b1.price * a1.value / total;
        let mut mp = (a1.price + b1.price) / Decimal::TWO;
        mp.rescale(self.mid_price.scale());
        self.mid_price_vec[index] = mp;

        // 生成fp
        self.fair_price_std_vec[index] = mp * self.ks[index] + self.bs[index];
        self.fair_price_std_vec[index].rescale(self.mid_price.scale());


        // 生成最终用于挂单的公平价格
        let fair_price_sum: Decimal = self.fair_price_std_vec.iter().sum();
        let fair_price_count = self.fair_price_std_vec.iter()
            .filter(|&&value| value != Decimal::new(0, 0)) // 过滤掉0
            .count();
        if fair_price_count != 0 {
            self.fair_price = fair_price_sum / Decimal::from(fair_price_count);

            let mut spread_abs = ((self.fair_price - self.mid_price) / self.mid_price).abs();
            spread_abs.rescale(5);

            self.spread_vec.push_back(spread_abs);
            if self.spread_vec.len() > 1000 {
                self.spread_vec.pop_front();
            }

            let opt_abs_value = self.spread_vec.iter().max().unwrap().clone();

            self.params.open = max(max(self.params.min_open, dec!(0.0006)), opt_abs_value);
        }
    }

    pub async fn update_delta(&mut self) {
        if self.mid_price.is_zero() {
            return;
        }

        let is_close_long = self.inventory > Decimal::ZERO;
        let is_close_short = self.inventory < Decimal::ZERO;

        if is_close_long {
            self.ask_delta = dec!(0);
            self.bid_delta = dec!(-2);

            self.optimal_ask_price = self.fair_price + self.fair_price * self.params.close;
            self.optimal_bid_price = Self::DONT_VIEW;
        } else if is_close_short {
            self.bid_delta = dec!(0);
            self.ask_delta = dec!(-2);

            self.optimal_bid_price = self.fair_price - self.fair_price * self.params.close;
            self.optimal_ask_price = Self::DONT_VIEW;
        } else {
            if self.fair_price > self.mid_price {
                self.bid_delta = dec!(0);
                self.ask_delta = dec!(-2);

                self.optimal_bid_price = self.fair_price - self.fair_price * self.params.open;
                self.optimal_ask_price = Self::DONT_VIEW;
            } else if self.fair_price < self.mid_price {
                self.ask_delta = dec!(0);
                self.bid_delta = dec!(-2);

                self.optimal_ask_price = self.fair_price + self.fair_price * self.params.open;
                self.optimal_bid_price = Self::DONT_VIEW;
            } else {
                self.bid_delta = dec!(0);
                self.ask_delta = dec!(0);

                self.optimal_bid_price = self.fair_price - self.fair_price * self.params.open;
                self.optimal_ask_price = self.fair_price + self.fair_price * self.params.open;
            }
        }

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

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

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

        for fair_price in &self.fair_price_std_vec {
            if fair_price.is_zero() {
                return;
            }
        }

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

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

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

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

    // 最小二乘法拟合函数，支持VecDeque
    pub async fn linear_least_squares(&self, index: usize) -> Option<(Decimal, Decimal)> {
        let x = &self.prices[index][1];
        let y = &self.prices[index][0];

        // 检查数组长度是否相同
        if x.len() != y.len() {
            return None;
        }
        let n = x.len();
        if n == 0 {
            return None;
        }

        let mut sum_x = Decimal::zero();
        let mut sum_y = Decimal::zero();
        let mut sum_xx = Decimal::zero();
        let mut sum_xy = Decimal::zero();

        // 遍历VecDeque中的元素
        for (xi, yi) in x.iter().zip(y.iter()) {
            sum_x += xi;
            sum_y += yi;
            let xi_sq = xi * xi;
            sum_xx += xi_sq;
            sum_xy += xi * yi;
        }

        // 计算分子和分母
        let numerator = sum_xy - (sum_x * sum_y) / Decimal::from(n);
        let denominator = sum_xx - (sum_x * sum_x) / Decimal::from(n);

        // 如果分母为0，返回None
        if denominator == Decimal::zero() {
            return None;
        }

        let k = numerator / denominator;
        let mean_x = sum_x / Decimal::from(n);
        let mean_y = sum_y / Decimal::from(n);
        let b = mean_y - k * mean_x;

        Some((k, b))
    }

    // #[instrument(skip(self), level="TRACE")]
    async fn processor(&mut self, data_time: Decimal, is_hard_update: bool) {
        self.check_ready();
        if !self.is_ready {
            return;
        }

        self.update_delta().await;

        // let cci_arc = self.cci_arc.clone();
        let now = data_time;
        let mid_price = self.mid_price;
        let ask_price = if self.params.ref_exchange.len() > 0 {
            self.fair_price_std_vec[0]
        } else {
            Self::DONT_VIEW
        };
        let bid_price = if self.params.ref_exchange.len() > 1 {
            self.fair_price_std_vec[1]
        } else {
            Self::DONT_VIEW
        };
        let optimal_ask_price = self.optimal_ask_price;
        let optimal_bid_price = self.optimal_bid_price;
        let last_price = Self::DONT_VIEW;
        let fair_price = self.fair_price;

        let total_amount_0: Decimal = self.trade_233_vec.deque.iter().map(|trade| trade.value).sum();
        let total_amount_1: Decimal = self.trade_0_vec.deque.iter().map(|trade| trade.value).sum();

        let spread = Self::DONT_VIEW;
        let spread_max = self.bs[0];
        let spread_min = self.ks[0];
        // 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 inventory = self.inventory;

        let sigma_square = if total_amount_0 + total_amount_1 == Decimal::ZERO {
            Decimal::ZERO
        } else {
            total_amount_0 / (total_amount_0 + total_amount_1)
        };

        let gamma = self.balance;
        let kappa = self.fitting_delay;

        let flow_ratio = Decimal::ZERO;

        let need_append = now - self.prev_insert_time > dec!(500);
        if !need_append && !is_hard_update {
            return;
        }
        if !is_hard_update {
            self.prev_insert_time = Decimal::from(Utc::now().timestamp_millis())
        }
        let pos_avg_price = self.pos_avg_price;

        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,
            fair_price,
            pos_avg_price
        ]).unwrap();
    }

    // #[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![]
    }
}