as-rust/strategy/src/core.rs

use tokio::time::Instant;
use std::collections::{BTreeMap, HashMap};
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::prelude::{ToPrimitive};
use rust_decimal_macros::dec;
use tokio::spawn;
use tokio::sync::mpsc::{Sender};
use tokio::sync::{Mutex};
use tokio::task::JoinHandle;
use tokio::time::sleep;
use tracing::{error, info, warn};
use global::cci::CentralControlInfo;
use global::params::Params;
use global::trace_stack::TraceStack;
use standard::{Account, Depth, Market, Order, OrderCommand, Platform, Position, PositionModeEnum, Record, SpecialTicker, Ticker, Trade};
use standard::exchange::{Exchange};
use standard::exchange::ExchangeEnum::{BinanceSwap, BybitSwap, BitgetSwap, GateSwap};

use crate::model::{LocalPosition, OrderInfo, TokenParam};
use crate::predictor::Predictor;
use crate::strategy::Strategy;
use crate::utils;
use crate::utils::clip;


pub struct Core {
    pub params: Params,
    // 启动时间
    pub start_time: i64,
    // 币对
    pub symbol: String,
    // 基础货币
    pub base: String,
    // 报价货币
    pub quote: String,
    //
    pub strategy: Strategy,
    // 本地挂单时间表
    pub local_order_infos: HashMap<String, Vec<String>>,
    // 本地挂单表
    pub local_orders: HashMap<String, OrderInfo>,
    // 本地订单缓存队列
    pub local_orders_backup: HashMap<String, OrderInfo>,
    // 本地订单缓存cid队列
    pub local_orders_backup_cid: Vec<String>,
    // 本地已处理cid缓存队列
    pub handled_orders_cid: Vec<String>,
    // 本地利润值
    pub local_profit: Decimal,
    // 本地U保证金
    pub local_cash: Decimal,
    // 本地币保证金
    pub local_coin: Decimal,
    // 仓位信息
    pub local_position: LocalPosition,
    // 仓位信息-来自订单
    pub local_position_by_orders: LocalPosition,
    //
    pub local_buy_amount: Decimal,
    pub local_sell_amount: Decimal,
    pub local_buy_value: Decimal,
    pub local_sell_value: Decimal,
    pub local_cancel_log: HashMap<String, i64>,
    pub local_cancel_cache: HashMap<String, OrderInfo>,
    pub interval: u64,
    pub exchange: String,
    pub exit_msg: String,
    // 仓位检查结果序列
    pub position_check_series: Vec<i8>,
    // 止损大小
    pub stop_loss: Decimal,
    // 资金使用率
    pub used_pct: Decimal,
    // 启停信号 0 表示运行 大于1开始倒计时 1时停机
    pub mode_signal: i8,
    // 交易盘口订单流更新时间
    pub trade_order_update_time: i64,
    // onTick触发时间记录
    pub on_tick_event_time: i64,
    // 盘口ticker信息
    pub tickers: HashMap<String, SpecialTicker>,
    // 盘口 depth信息
    pub depths: HashMap<String, Vec<Decimal>>,
    // 行情更新延迟监控(风控)
    pub market_update_time: HashMap<String, i64>,
    pub market_update_interval: HashMap<String, Decimal>,
    pub ref_num: i8,
    pub ref_name: Vec<String>,
    pub trade_name: String,
    pub ready: i8,
    pub predictor: Predictor,
    pub market: Market,
    pub platform_rest: Box<dyn Platform + Send + Sync>,
    // 市场最优买卖价
    pub max_buy_min_sell_cache: HashMap<String, Vec<Decimal>>,
    // 最近一次的depth信息
    pub local_depths: HashMap<String, Vec<Decimal>>,
    pub is_update: HashMap<String, bool>,
    pub running: Arc<AtomicBool>,
    pub hold_coin: Decimal,

    // 打印限频
    pub prev_log_ready_timestamp: i64,
    pub log_ready_log_interval: i64,

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

    // 老版的trader_msg留下来的
    pub agg_market: Vec<Decimal>,
    pub ref_price: Vec<Vec<Decimal>>,
    pub predict: Decimal,
}

impl Core {
    pub async fn new(exchange: String,
                     params: Params,
                     exchange_params: BTreeMap<String, String>,
                     order_sender: Sender<Order>,
                     error_sender: Sender<Error>,
                     running: Arc<AtomicBool>,
                     cci_arc: Arc<Mutex<CentralControlInfo>>) -> Core {
        let symbol = params.pair.clone();
        let pairs: Vec<&str> = params.pair.split('_').collect();
        let mut core_obj = Core {
            params: params.clone(),
            start_time: 0,
            symbol: symbol.clone(),
            base: pairs[0].to_string(),
            quote: pairs[1].to_string(),
            // 现货底仓
            hold_coin: clip(params.hold_coin, Decimal::ZERO, Decimal::ONE_HUNDRED * Decimal::ONE_HUNDRED),
            strategy: Strategy::new(&params, true),
            local_order_infos: Default::default(),
            local_orders: Default::default(),
            local_orders_backup: Default::default(),
            local_orders_backup_cid: Default::default(),
            handled_orders_cid: Default::default(),
            local_profit: Default::default(),
            local_cash: Default::default(),
            local_coin: Default::default(),
            local_position: LocalPosition {
                long_pos: Default::default(),
                short_pos: Default::default(),
                long_avg: Default::default(),
                short_avg: Default::default(),
            },
            local_position_by_orders: LocalPosition {
                long_pos: Default::default(),
                short_pos: Default::default(),
                long_avg: Default::default(),
                short_avg: Default::default(),
            },
            local_buy_amount: Default::default(),
            local_sell_amount: Default::default(),
            local_buy_value: Default::default(),
            local_sell_value: Default::default(),
            local_cancel_log: Default::default(),
            local_cancel_cache: Default::default(),
            interval: params.interval,
            exchange: params.exchange.clone(),
            exit_msg: "正常退出".to_string(),
            position_check_series: Default::default(),
            stop_loss: params.stop_loss,
            used_pct: dec!(1),
            mode_signal: 0,
            trade_order_update_time: Utc::now().timestamp_millis(),
            on_tick_event_time: Utc::now().timestamp_millis(),
            tickers: Default::default(),
            depths: Default::default(),
            market_update_time: Default::default(),
            market_update_interval: Default::default(),
            ref_num: params.ref_exchange.len() as i8,
            ref_name: Default::default(),
            trade_name: "".to_string(),
            ready: 0,
            predictor: Predictor::new(cci_arc.clone(), params.clone()),
            market: Market {
                symbol: symbol.clone(),
                base_asset: "".to_string(),
                quote_asset: "".to_string(),
                tick_size: Default::default(),
                price_precision: Default::default(),
                amount_precision: Default::default(),
                min_qty: Default::default(),
                max_qty: Default::default(),
                min_notional: Default::default(),
                max_notional: Default::default(),
                multiplier: Default::default(),
                amount_size: Default::default(),
            },
            platform_rest: match exchange.as_str() {
                // "kucoin_usdt_swap" => {
                //     Exchange::new(KucoinSwap, symbol, params.colo != 0i8, exchange_params, order_sender, error_sender).await
                // }
                "gate_usdt_swap" => {
                    Exchange::new(GateSwap, symbol, params.colo != 0i8, exchange_params, order_sender, error_sender).await
                }
                // "gate_usdt_spot" => {
                //     Exchange::new(GateSpot, symbol, params.colo != 0i8, exchange_params, order_sender, error_sender).await
                // }
                "binance_usdt_swap" => {
                    Exchange::new(BinanceSwap, symbol, params.colo != 0i8, exchange_params, order_sender, error_sender).await
                }
                // "binance_spot" => {
                //     Exchange::new(BinanceSpot, symbol, params.colo != 0i8, exchange_params, order_sender, error_sender).await
                // }
                // "bitget_spot" => {
                //     Exchange::new(BitgetSpot, symbol, params.colo != 0i8, exchange_params, order_sender, error_sender).await
                // }
                "bitget_usdt_swap" => {
                    Exchange::new(BitgetSwap, symbol, params.colo != 0i8, exchange_params, order_sender, error_sender).await
                }
                // "okex_usdt_swap" => {
                //     Exchange::new(OkxSwap, symbol, params.colo != 0i8, exchange_params, order_sender, error_sender).await
                // }
                "bybit_usdt_swap" => {
                    Exchange::new(BybitSwap, symbol, params.colo != 0i8, exchange_params, order_sender, error_sender).await
                }
                // "coinex_usdt_swap" => {
                //     Exchange::new(CoinexSwap, symbol, params.colo != 0i8, exchange_params, order_sender, error_sender).await
                // }
                // "htx_usdt_swap" => {
                //     Exchange::new(HtxSwap, symbol, params.colo != 0i8, exchange_params, order_sender, error_sender).await
                // }
                _ => {
                    error!("203未找到对应的交易所rest枚举！");
                    panic!("203未找到对应的交易所rest枚举！");
                }
            },
            max_buy_min_sell_cache: Default::default(),
            local_depths: Default::default(),
            is_update: Default::default(),
            running,
            prev_log_ready_timestamp: 0,
            log_ready_log_interval: 10 * 1000,
            cci_arc,
            agg_market: vec![],
            ref_price: vec![],
            predict: Default::default(),
        };
        for i in 0..=params.ref_exchange.len() - 1 {
            // 拼接不会消耗原字符串
            let tickers_key: String = format!("{}{}{}{}", params.ref_exchange[i], "@", params.ref_pair[i], "@ref");
            let ref_name_element = tickers_key.clone();
            let depths_key: String = tickers_key.clone();
            let market_update_time_key = tickers_key.clone();
            let market_update_interval_key = tickers_key.clone();
            let max_buy_min_sell_cache_key = tickers_key.clone();

            core_obj.tickers.insert(tickers_key, SpecialTicker::new());
            core_obj.ref_name.push(ref_name_element);
            core_obj.depths.insert(depths_key, Default::default());
            core_obj.market_update_time.insert(market_update_time_key, Default::default());
            core_obj.market_update_interval.insert(market_update_interval_key, Default::default());
            core_obj.max_buy_min_sell_cache.insert(max_buy_min_sell_cache_key, vec![Decimal::ZERO, Decimal::ZERO]);
        }
        let name = format!("{}{}{}", core_obj.exchange.clone(), "@", core_obj.symbol);
        let market_update_time_key = name.clone();
        let market_update_interval_key = name.clone();
        let tickers_key = name.clone();
        let depths_key = name.clone();
        let max_buy_min_sell_cache_key = name.clone();
        core_obj.trade_name = name;
        core_obj.market_update_time.insert(market_update_time_key, Default::default());
        core_obj.market_update_interval.insert(market_update_interval_key, Default::default());
        core_obj.tickers.insert(tickers_key, SpecialTicker::new());
        core_obj.depths.insert(depths_key, Default::default());
        core_obj.max_buy_min_sell_cache.insert(max_buy_min_sell_cache_key, vec![Decimal::ZERO, Decimal::ZERO]);
        // broker.newWs
        let mut price_alpha: Vec<Decimal> = Vec::new();
        for ref_pair_str in params.ref_pair {
            if params.pair.contains("1000") && !ref_pair_str.contains("1000") {
                price_alpha.push(dec!(1000.0));
            } else if !params.pair.contains("1000") && ref_pair_str.contains("1000") {
                price_alpha.push(dec!(0.001))
            } else {
                price_alpha.push(dec!(1.0));
            }
        }
        info!("价格系数：{:?}", price_alpha);

        return core_obj;
    }

    // #[instrument(skip(self, data, trace_stack), level="TRACE")]
    pub async fn update_order(&mut self, data: Vec<OrderInfo>, trace_stack: TraceStack) {
        for order in data {
            self.update_local_order(order, trace_stack.clone()).await;
        }
    }

    // #[instrument(skip(self, data, trace_stack), level="TRACE")]
    pub async fn update_local_order(&mut self, data: OrderInfo, trace_stack: TraceStack) {
        // if data.filled != Decimal::ZERO {
        //     // info!("\n\n");
        //     // info!("接收到订单信息①：{:?}", data);
        // }
        /*
         更新订单
            首先直接复写本地订单
            1、如果是开仓单
                如果新增: 增加本地订单
                如果取消: 删除本地订单 查看是否完全成交 如果是部分成交 则按已成交量发送平仓订单 修改本地仓位
                如果成交: 删除本地订单 发送平仓订单 修改本地仓位
            2、如果是平仓单
                如果新增: 增加本地订单
                如果取消: 删除本地订单 查看是否完全成交 如果是部分成交 则按未成交量发送平仓订单 修改本地仓位
                如果成交: 删除本地订单 修改本地仓位
            NEW 可以从 ws / rest 来
            REMOVE 主要从 ws 来 必须包含 filled 和 filled_price 用于本地仓位推算 定期rest查过旧订单
            为了防止下单失败依然有订单成交 本地需要做一个缓存
        */
        // 触发订单更新
        self.trade_order_update_time = Utc::now().timestamp_millis();

        // 更新跟踪
        if self.local_orders.contains_key(&data.client_id) {
            let mut order_info = self.local_orders.get(&data.client_id).unwrap().clone();

            if data.trace_stack.after_network != 0 { order_info.trace_stack = data.trace_stack.clone() }

            self.local_orders.insert(data.client_id.clone(), order_info);
        }

        // 新增订单推送 仅需要cid oid信息
        if data.status == "NEW" {
            // 更新oid信息 更新订单 loceltime信息(尤其是查单返回new的情况 必须更新 否则会误触发风控)

            if self.local_orders.contains_key(&data.client_id) {
                let mut order_info = self.local_orders.get(&data.client_id).unwrap().clone();
                order_info.order_id = data.order_id;
                order_info.local_time = Utc::now().timestamp_millis();
                self.local_orders.insert(data.client_id.clone(), order_info);
            }
        } else if data.status == "REMOVE" {
            // 如果在撤单记录中 说明此订单结束生命周期 可以移除记录
            if self.local_cancel_log.contains_key(&data.client_id) {
                self.local_cancel_log.remove(&data.client_id);
            }
            // 在cid缓存队列中 说明是本策略的订单
            if self.local_orders_backup.contains_key(&data.client_id) {
                // 不在已处理cid缓存队列中 说明还没参与过仓位计算 则执行订单计算
                if self.handled_orders_cid.contains(&data.client_id) {
                    // info!("订单已经参与过仓位计算 拒绝重复进行计算, 订单号：{}", data.client_id);
                } else {
                    // 添加进已处理队列
                    self.handled_orders_cid.push(data.client_id.clone());
                    // 提取成交信息 方向 价格 量
                    let filled = data.filled;
                    let side = self.local_orders_backup.get(&data.client_id).unwrap().side.clone();
                    let mut filled_price = Decimal::ZERO;
                    if data.filled_price > filled_price {
                        filled_price = data.filled_price;
                    } else {
                        filled_price = self.local_orders_backup.get(&data.client_id).unwrap().price.clone();
                    }
                    // 只有开仓成交才触发onPosition
                    // 如果漏推送 rest补充的订单查询信息过来 可能会导致 kd kk 推送出现计算分母为0的情况
                    if filled > Decimal::ZERO {
                        let mut filled_order = data.clone();
                        filled_order.side = side.clone();

                        // info!("移除本地订单：{:?}, local_by_orders: {:?}", filled_order, self.local_position_by_orders);
                        // 如果是现货交易 还需要修改equity
                        if self.exchange.contains("spot") {
                            // 现货必须考虑fee 买入fee单位为币 卖出fee单位为u
                            let fee = data.fee;
                            // buy  开多
                            if side == "kd" {
                                self.local_buy_amount += filled - fee;
                                self.local_buy_value += (filled - fee) * filled_price;
                                let new_long_pos = self.local_position_by_orders.long_pos + filled - fee;
                                if new_long_pos == Decimal::ZERO {
                                    self.local_position_by_orders.long_avg = Decimal::ZERO;
                                    self.local_position_by_orders.long_pos = Decimal::ZERO;
                                } else {
                                    self.local_position_by_orders.long_avg = (self.local_position_by_orders.long_pos * self.local_position_by_orders.long_avg +
                                        filled * filled_price) / new_long_pos;
                                    self.local_position_by_orders.long_pos = new_long_pos;
                                }
                                self.local_cash -= filled * filled_price;
                                self.local_coin = filled - fee;

                                // sell 平多
                            } else if side == "pd" {
                                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);
                                let new_long_pos = self.local_position_by_orders.long_pos - filled;
                                if new_long_pos == Decimal::ZERO {
                                    self.local_position_by_orders.long_avg = Decimal::ZERO;
                                    self.local_position_by_orders.long_pos = Decimal::ZERO;
                                } else {
                                    self.local_position_by_orders.long_pos = new_long_pos;
                                }
                                self.local_cash += filled * filled_price - fee;
                                self.local_coin -= filled;

                            // buy 平空
                            } else if side == "pk" {
                                self.local_buy_amount += filled - fee;
                                self.local_buy_value += (filled - fee) * filled_price;
                                self.local_profit += filled * (self.local_position_by_orders.short_avg - filled_price);
                                let new_short_pos = self.local_position_by_orders.short_pos - filled - fee;
                                if new_short_pos == Decimal::ZERO {
                                    self.local_position_by_orders.short_avg = Decimal::ZERO;
                                    self.local_position_by_orders.short_pos = Decimal::ZERO;
                                } else {
                                    self.local_position_by_orders.short_pos = new_short_pos;
                                }
                                self.local_cash -= filled * filled_price;
                                self.local_coin += filled - fee;

                            // sell 开空
                            } else if side == "kk" {
                                self.local_sell_amount += filled;
                                self.local_sell_value += filled * filled_price;
                                let new_short_pos = self.local_position_by_orders.short_pos + filled;
                                if new_short_pos == Decimal::ZERO {
                                    self.local_position_by_orders.short_avg = Decimal::ZERO;
                                    self.local_position_by_orders.short_pos = Decimal::ZERO;
                                } else {
                                    self.local_position_by_orders.short_avg = (self.local_position_by_orders.short_pos * self.local_position_by_orders.short_avg
                                        + filled * filled_price) / new_short_pos;
                                    self.local_position_by_orders.short_pos = new_short_pos;
                                }
                                self.local_cash += filled * filled_price - fee;
                                self.local_coin -= filled;
                            } else {
                                info!("错误的仓位方向{}", side);
                            }

                        // 合约订单流仓位计算
                        } else {
                            // buy 开多
                            if side == "kd" {
                                self.local_buy_amount += filled;
                                self.local_buy_value += filled * filled_price;
                                let new_long_pos = self.local_position_by_orders.long_pos + filled;
                                if new_long_pos == Decimal::ZERO {
                                    self.local_position_by_orders.long_avg = Decimal::ZERO;
                                    self.local_position_by_orders.long_pos = Decimal::ZERO;
                                } else {
                                    self.local_position_by_orders.long_avg = (self.local_position_by_orders.long_pos *
                                        self.local_position_by_orders.long_avg + filled * filled_price) / new_long_pos;
                                    self.local_position_by_orders.long_pos = self.local_position_by_orders.long_pos + filled;
                                }

                            // sell 开空
                            } else if side == "kk" {
                                self.local_sell_amount += filled;
                                self.local_sell_value += filled * filled_price;
                                let new_short_pos = self.local_position_by_orders.short_pos + filled;
                                if new_short_pos == Decimal::ZERO {
                                    self.local_position_by_orders.short_avg = Decimal::ZERO;
                                    self.local_position_by_orders.short_pos = Decimal::ZERO;
                                } else {
                                    self.local_position_by_orders.short_avg = (self.local_position_by_orders.short_pos * self.local_position_by_orders.short_avg + filled * filled_price) / new_short_pos;
                                    self.local_position_by_orders.short_pos = self.local_position_by_orders.short_pos + filled;
                                }

                            // sell 平多
                            } else if side == "pd" {
                                self.local_sell_amount += filled;
                                self.local_sell_value += filled * filled_price;
                                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;
                                }

                            // buy 平空
                            } else if side == "pk" {
                                self.local_buy_amount += filled;
                                self.local_buy_value += filled * filled_price;
                                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;
                                }
                            } else {
                                error!("错误的仓位方向{}", side);
                            }
                            // 统计合约交易手续费 正fee为扣手续费 负fee为返佣
                            if data.fee > Decimal::ZERO {
                                self.local_profit -= data.fee;
                            }
                        }
                        // info!("成交单耗时数据：{}", time_record.to_string());
                        info!("更新推算仓位 {:?}", self.local_position_by_orders);
                        let place_time = self.local_order_infos.get(&data.client_id).unwrap()[4].parse::<i64>().unwrap();
                        info!("当前时间-订单委托时间={}ms, cid={}", Utc::now().timestamp_millis() - place_time, data.client_id.clone());
                        if let Some(oi) = self.local_cancel_cache.get(&data.client_id) {
                            info!("取消报文：{}", oi.symbol);
                            info!("当前时间-订单首次取消时间={}ms", Utc::now().timestamp_millis() - oi.local_time)
                        } else {
                            info!("没有提交过/没有提交成功过取消订单")
                        }
                        info!("委托时fp={}, mp={}, op={}",
                            self.local_order_infos.get(&data.client_id).unwrap()[5],
                            self.local_order_infos.get(&data.client_id).unwrap()[6],
                            self.local_order_infos.get(&data.client_id).unwrap()[7]
                        );
                        // 本地计算利润
                        self._print_local_trades_summary().await;
                        // 打印各类信息
                        self.strategy.local_orders = self.local_orders.clone();
                        self.strategy._print_summary();
                        info!("------------------------------------------------------------------------------------");
                    }
                    // 这里主要是为了抓提交过的取消订单又成交的情况
                    else if filled.is_zero() {
                        // 提交取消订单成功
                        if !self.local_cancel_cache.contains_key(&data.client_id) {
                            self.local_cancel_cache.insert(data.client_id.clone(), data.clone());
                        }
                    }
                    // 每次有订单变动就触发一次策略
                    if self.mode_signal == 0 && self.ready == 1 {
                        // 更新交易数据
                        self.update_trade_msg();
                        // 触发策略挂单逻辑
                        // 更新策略时间
                        self.strategy.local_time = Utc::now().timestamp_millis();
                        // trace_stack.on_before_strategy();
                        let mut order = self.strategy.do_strategy(&mut self.predictor, &self.local_orders, &self.local_position, &self.local_coin, &self.local_cash);
                        // trace_stack.on_after_strategy();
                        // 记录指令触发信息
                        if order.is_not_empty() {
                            // info!("触发onOrder");
                            self._update_local_orders(&mut order);
                            //交易所处理订单信号
                            let mut platform_rest_fb = self.platform_rest.clone_box();
                            let mut ts = trace_stack.clone();
                            ts.set_order_command(order.to_string());
                            // ts.on_before_send_thread();
                            // ts.on_before_send();
                            // info!("update_order 订单指令：{:?}", order);
                            spawn(async move {
                                platform_rest_fb.command_order(&mut order, &mut ts).await;
                            });
                        }
                    }
                }
            } else {
                // info!("订单不属于本策略 拒绝进行仓位计算: {}", data.client_id);
            }

            if self.local_orders.contains_key(&data.client_id) {
                // 成交数量不为空，则打印耗时追踪
                if data.filled > Decimal::ZERO {
                    let local_order = self.local_orders.get(&data.client_id).unwrap();
                    info!("订单耗时追踪：{:?}", local_order.trace_stack.to_string());
                }

                // debug!("删除本地订单, client_id:{:?}", data);
                self.local_orders.remove(&data.client_id);
            } else {
                // debug!("该订单不在本地挂单表中, order:{:?}", data);
            }
        } else {
            error!("未知的订单事件类型：{:?}", data);
        }
    }

    // #[instrument(skip(self), level="TRACE")]
    pub async fn _print_local_trades_summary(&mut self) {
        // 计算本地累计利润
        let local_buy_amount = self.local_buy_amount.round_dp(5);
        let local_buy_value = self.local_buy_value.round_dp(5);
        let local_sell_amount = self.local_sell_amount.round_dp(5);
        let local_sell_value = self.local_sell_value.round_dp(5);

        if self.strategy.mp > Decimal::ZERO {
            let unrealized = (local_buy_amount - local_sell_amount) * self.strategy.mp;
            let realized = local_sell_value - local_buy_value;
            let local_profit = (unrealized + realized).round_dp(5);
            self.strategy.local_profit = local_profit;
            info!("买量 {}，卖量 {}，买额{}，卖额{}", local_buy_amount, local_sell_amount, local_buy_value, local_sell_value);

            // 更新账户余额
            let mut now_balance = (self.local_cash + self.local_coin * self.predictor.mid_price) / self.used_pct;
            now_balance.rescale(4);

            self.predictor.on_balance(now_balance).await;
        }
    }

    // 检测初始数据是否齐全
    // #[instrument(skip(self), level="TRACE")]
    pub fn check_ready(&mut self) {
        // 检查 ticker 行情
        // for i in &self.ref_name {
        //     if self.tickers.is_empty() || !self.tickers.contains_key(i) {
        //         self.log_ready_status(format!("529参考盘口ticker未准备好: {:?}", self.tickers));
        //         return;
        //     } else {
        //         if self.tickers.get(i).unwrap().buy == dec!(0) || self.tickers.get(i).unwrap().sell == dec!(0) {
        //             self.log_ready_status(format!("533参考盘口ticker未准备好: {:?}", self.tickers));
        //             return;
        //         }
        //     }
        // }
        // if self.tickers.contains_key(&self.trade_name) {
        //     if self.tickers.get(&self.trade_name).unwrap().buy == dec!(0) || self.tickers.get(&self.trade_name).unwrap().sell == dec!(0) {
        //         self.log_ready_status(format!("540交易盘口ticker未准备好: {:?}", self.tickers));
        //         return;
        //     }
        // } else {
        //     self.log_ready_status(format!("544交易盘口ticker未准备好: {:?}", self.tickers));
        //     return;
        // }
        // // 检查 market 行情
        // self.agg_market = self.get_all_market_data();
        // if self.agg_market.len() != LENGTH * (1usize + self.ref_num as usize) {
        //     self.log_ready_status(format!("550聚合行情未准备好: market长度：{}, 检验数: {}", self.agg_market.len(), LENGTH * (1usize + self.ref_num as usize)));
        //     return;
        // } else {
        //     // 如果准备就绪，则可以开始交易
        //     info!("----------------------------------聚合行情准备就绪，可以开始交易---------------------------------");
        //     self.predictor.market_info_handler(&self.agg_market);
        //     self.ready = 1;
        // }
        self.ready = 1;
    }

    // #[instrument(skip(self, msg), level="TRACE")]
    pub fn log_ready_status(&mut self, msg: String) {
        // 隔一会再打印未准备就绪的台词
        let now_timestamp = Utc::now().timestamp_millis();
        if now_timestamp - self.prev_log_ready_timestamp > self.log_ready_log_interval {
            self.prev_log_ready_timestamp = now_timestamp;
            info!("{}", msg);
        }
    }

    pub async fn on_trade(&mut self, trade: &Trade, _name_ref: &String, trace_stack: &mut TraceStack, index: usize) {
        self.predictor.on_trade(trade, index).await;

        // 产生交易信号
        if trade.value > Decimal::ONE_HUNDRED {
            let mut orders = self.strategy.do_strategy(&mut self.predictor, &self.local_orders, &self.local_position, &self.local_coin, &self.local_cash);
            trace_stack.on_after_strategy();

            if orders.is_not_empty() {
                let mut platform_rest_fb = self.platform_rest.clone_box();
                // 先更新本地记录再发单。
                self._update_local_orders(&mut orders);
                // info!("订单指令：{:?}", orders);
                let mut ts = trace_stack.clone();
                spawn(async move {
                    platform_rest_fb.command_order(&mut orders, &mut ts).await;
                });
            }
        }
    }

    pub async fn on_ticker(&mut self, ticker: &Ticker, _trace_stack: &mut TraceStack) {
        self.predictor.on_ticker(ticker).await;
    }

    pub async fn on_record(&mut self, record: &Record) {
        self.predictor.on_record(record).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, index: usize) {
        // ================================ 刷新更新间隔 =========================================
        let now_time = depth.time.to_i64().unwrap();
        if self.market_update_time.contains_key(name_ref) && *self.market_update_time.get(name_ref).unwrap() != 0i64 {
            let interval = Decimal::from(now_time - self.market_update_time.get(name_ref).unwrap());
            if *self.market_update_interval.get(name_ref).unwrap() == dec!(0) {
                self.market_update_interval.insert(name_ref.clone(), interval);
            } else {
                let value = self.market_update_interval.get(name_ref).unwrap();
                self.market_update_interval.insert(name_ref.clone(), value * dec!(0.999) + interval * dec!(0.001));
            }
        }
        self.market_update_time.insert(name_ref.clone(), now_time);

        // ================================ 在系统已经准备就绪的情况下，更新相关参数 =========================================
        if self.mode_signal == 0 && self.ready == 1 {
            // 更新公平价格
            self.predictor.on_depth(depth, index).await;
            // 触发事件撤单逻辑
            // 更新策略时间
            self.strategy.local_time = Utc::now().timestamp_millis();

            // 产生交易信号
            let mut orders = self.strategy.do_strategy(&mut self.predictor, &self.local_orders, &self.local_position, &self.local_coin, &self.local_cash);
            trace_stack.on_after_strategy();

            if orders.is_not_empty() {
                let mut platform_rest_fb = self.platform_rest.clone_box();
                // 先更新本地记录再发单。
                self._update_local_orders(&mut orders);
                // info!("订单指令：{:?}", orders);
                let mut ts = trace_stack.clone();
                spawn(async move {
                    platform_rest_fb.command_order(&mut orders, &mut ts).await;
                });

                // 更新账户余额到中控
                {
                    // 更新账户余额
                    let mut now_balance = (self.local_cash + self.local_coin * self.predictor.mid_price) / self.used_pct;
                    now_balance.rescale(4);
                    let mut cci = self.cci_arc.lock().await;
                    cci.now_balance = now_balance;
                }
            }
        }

        // 同步更新核心数据到中控数据显示和策略层
        {
            let mut unrealized_pn_l = self.local_profit;
            unrealized_pn_l.rescale(4);

            let mut now_price = self.strategy.mp;
            now_price.rescale(8);

            let mut cci = self.cci_arc.lock().await;
            cci.unrealized_pn_l = unrealized_pn_l;
            cci.now_price = now_price;
        }
    }

    // #[instrument(skip(self, data), level="TRACE")]
    pub async fn update_position(&mut self, data: Vec<Position>) {
        if data.is_empty() {
            return;
        }
        let mut position = LocalPosition::new();
        let mut pos_avg_price = Decimal::ZERO;
        let mut update_time = Decimal::ZERO;
        for pos in &data {
            pos_avg_price = pos.price;
            update_time = pos.update_time;

            if pos.position_mode == PositionModeEnum::Long {
                position.long_pos = pos.amount;
                position.long_avg = pos.price;
            } else if pos.position_mode == PositionModeEnum::Short {
                position.short_pos = pos.amount.abs();
                position.short_avg = pos.price;
            }
        }
        // 更新仓位信息
        if position != self.local_position {
            info!("收到新的仓位推送: {:?}", data);
            info!("更新本地仓位: {:?}", position);
            info!("\n\n");
            self.local_position = position;
        }

        // 持仓相关信息
        let mut pos = self.local_position_by_orders.long_pos - self.local_position_by_orders.short_pos;
        if !self.exchange.contains("spot") {
            pos = self.local_position.long_pos - self.local_position.short_pos;
        }
        pos.rescale(8);
        self.predictor.on_inventory(&pos, &pos_avg_price, &self.strategy.min_amount_value, update_time).await;

        let mut entry_price;
        if pos.gt(&Decimal::ZERO) {
            entry_price = self.local_position_by_orders.long_avg;
        } else {
            entry_price = self.local_position_by_orders.short_avg;
        }
        entry_price.rescale(8);
        let mut cci = self.cci_arc.lock().await;
        cci.pos = pos;
        cci.entry_price = entry_price;
    }

    // #[instrument(skip(self), level="TRACE")]
    // pub fn on_agg_market(&mut self) {
    //     /* 处理聚合行情
    //        1. 获取聚合行情
    //        2. 更新预测器
    //        3. 触发tick回测
    //      */
    //     // 更新聚合市场数据
    //     // 更新聚合市场信息
    //     self.agg_market = self.get_all_market_data();
    //     // 更新预测器
    //     self.predictor.market_info_handler(&self.agg_market);
    // }

    // #[instrument(skip(self), level="TRACE")]
    pub fn update_trade_msg(&mut self) {
        // 更新保证金
        self.local_cash = self.local_cash.round_dp(10);
        self.local_coin = self.local_coin.round_dp(10);
        // 使用本地推算仓位
        // self.trade_msg.position = self.local_position_by_orders.clone();
        // self.trade_msg.orders = self.local_orders.clone();
        // 更新 ref
        let mut ref_tickers: BTreeMap<String, Ticker> = BTreeMap::new();
        for i in &self.ref_name {
            let bp = self.tickers.get(i).unwrap().buy;
            let ap = self.tickers.get(i).unwrap().sell;
            ref_tickers.insert(i.clone(), Ticker {
                time: Decimal::ZERO,
                high: Default::default(),
                low: Default::default(),
                sell: ap,
                buy: bp,
                last: Default::default(),
                volume: Default::default(),
                open_interest: Default::default(),
            });
        }
        self.ref_price = self.predictor.get_ref_price(&ref_tickers);
    }

    // 本地记录所有报单信息
    // #[instrument(skip(self, orders), level="TRACE")]
    pub fn _update_local_orders(&mut self, orders: &mut OrderCommand) {
        orders.limits_open.extend(orders.limits_close.clone());

        if !orders.limits_open.is_empty() {
            for j in orders.limits_open.keys() {
                let order_info = OrderInfo {
                    symbol: self.symbol.clone(),
                    amount: Decimal::from_str(orders.limits_open.get(j).unwrap()[0].as_str()).unwrap(),
                    side: orders.limits_open.get(j).unwrap()[1].clone(),
                    price: Decimal::from_str(orders.limits_open.get(j).unwrap()[2].as_str()).unwrap(),
                    client_id: orders.limits_open.get(j).unwrap()[3].clone(),
                    filled_price: Default::default(),
                    filled: Decimal::ZERO,
                    order_id: "".to_string(),
                    local_time: self.strategy.local_time,
                    create_time: self.strategy.local_time,
                    status: "".to_string(),
                    fee: Default::default(),
                    trace_stack: TraceStack::new(0, Instant::now()),
                };
                self.local_order_infos.insert(orders.limits_open.get(j).unwrap()[3].clone(), orders.limits_open.get(j).unwrap().clone());
                // 本地挂单表
                self.local_orders.insert(orders.limits_open.get(j).unwrap()[3].clone(), order_info.clone());
                // 本地缓存表
                self.local_orders_backup.insert(orders.limits_open.get(j).unwrap()[3].clone(), order_info);
                // 本地缓存cid表
                self.local_orders_backup_cid.push(orders.limits_open.get(j).unwrap()[3].clone());
            }
        }
        if !orders.cancel.is_empty() {
            for cancel_key in orders.cancel.keys() {
                let cid = orders.cancel.get(cancel_key).unwrap()[0].clone();
                
                if self.local_cancel_log.contains_key(&cid) {
                    let num = self.local_cancel_log.get(&cid).unwrap() + 1;
                    self.local_cancel_log.insert(cid, num);
                } else {
                    self.local_cancel_log.insert(cid, 0);
                }
            }
        }
        let max_len = 9999usize;
        // 清除过于久远的历史记录
        if self.local_orders_backup_cid.len() > max_len {
            let cid = self.local_orders_backup_cid[0].clone();
            // 判断是否超过1个小时 如果超过则移除历史记录
            if self.local_orders_backup.contains_key(&cid) {
                let local_time = self.local_orders_backup.get(&cid).unwrap().local_time;
                if Utc::now().timestamp_millis() - local_time > 3600000 {
                    self.local_orders_backup.remove(&cid);
                    self.local_orders_backup_cid.retain(|x| *x != cid)
                }
            }
        }
        if self.handled_orders_cid.len() > max_len {
            self.handled_orders_cid.remove(0usize);
        }
    }

    // 获取深度信息
    // #[instrument(skip(self), level="TRACE")]
    pub fn get_all_market_data(&mut self) -> Vec<Decimal> {
        // 只能定时触发 组合市场信息=交易盘口+参考盘口
        let mut market: Vec<Decimal> = Vec::new();
        // 获取交易盘口市场信息
        let mut data: Vec<Decimal> = self.local_depths.get(&self.trade_name).unwrap().clone();
        self.is_update.insert(self.symbol.clone(), true);
        let mut max_min_price = self.max_buy_min_sell_cache.get(&self.trade_name).unwrap().clone();
        market.append(&mut data);
        market.append(&mut max_min_price);

        for i in &self.ref_name {
            // 获取参考盘口市场信息
            data = self.local_depths.get(i).unwrap().clone();
            self.is_update.insert(i.clone(), true);
            max_min_price = self.max_buy_min_sell_cache.get(i).unwrap().clone();
            data.append(&mut max_min_price);
            market.append(&mut data);
        }
        return market;
    }

    // #[instrument(skip(self), level="TRACE")]
    pub async fn get_exchange_info(&mut self) {
        self.market = self.platform_rest.get_self_market();
        info!(?self.market);
    }

    // #[instrument(skip(self, data), level="TRACE")]
    pub async fn update_equity(&mut self, data: Account) {
        /*
           更新保证金信息
           合约一直更新
           现货只有当出现异常时更新
       */
        if self.exchange.contains("spot") {
            return;
        }
        self.local_cash = data.balance * self.used_pct;
    }

    // #[instrument(skip(self), level="TRACE")]
    pub async fn update_equity_rest_swap(&mut self) {
        match self.platform_rest.get_account().await {
            Ok(account) => {
                /*
                   更新保证金信息
                   合约一直更新
                   现货只有当出现异常时更新
               */
                self.local_cash = account.balance * self.used_pct
            }
            Err(e) => {
                info!("获取账户信息错误: {:?}", e);
            }
        }
    }

    // #[instrument(skip(self), level="TRACE")]
    pub async fn update_equity_rest_spot(&mut self) {
        match self.platform_rest.get_spot_account().await {
            Ok(mut val) => {
                // 如果返回的数组里没有交易货币，则补充交易货币
                if !val.iter().any(|a| a.coin.to_uppercase().eq(&self.base.to_uppercase())) {
                    let mut base_coin_account = Account::new();
                    base_coin_account.coin = self.base.to_uppercase();
                    val.push(base_coin_account);
                }

                for account in val {
                    // 交易货币
                    if self.base.to_uppercase() == account.coin {
                        self.local_coin = account.balance;
                    }
                    // 本位货币
                    if self.quote.to_uppercase() == account.coin {
                        self.local_cash = account.balance;
                    }
                }
            }
            Err(err) => {
                error!("获取仓位信息异常: {}", err);
            }
        }
    }

    // #[instrument(skip(self), level="TRACE")]
    pub async fn check_risk(&mut self) {
        // 参数检查的风控
        if self.strategy.start_cash == Decimal::ZERO {
            warn!("请检查交易账户余额");
            warn!(?self.strategy.start_cash);
            return;
        }

        if self.strategy.mp == Decimal::ZERO {
            warn!("请检查最新价格");
            warn!(?self.strategy.mp);
            return;
        }

        // 不是现货执行的回撤风控1
        if !self.exchange.contains("spot") {
            let draw_back = Decimal::ONE - self.strategy.equity / self.strategy.max_equity;

            if draw_back > self.stop_loss {
                let exit_msg = format!("{} 总资金吊灯回撤 {}。当前净值：{}, 最高净值{}，触发止损，准备停机。",
                                       self.params.account_name, draw_back, self.strategy.equity, self.strategy.max_equity);
                warn!(exit_msg);
                self.exit_msg = exit_msg;
                self.stop().await;
            }
        }
        // 回撤风控2
        let draw_back = self.local_profit / self.strategy.start_equity;
        if draw_back < -self.stop_loss {
            let exit_msg = format!("{} 交易亏损，触发止损，准备停机。", self.params.account_name);
            warn!(exit_msg);
            self.exit_msg = exit_msg;
            self.stop().await;
        }
        // 回撤风控3，加浮亏的回撤风控
        // let floating_profit = if !self.local_position.long_pos.is_zero() && !self.local_position.long_avg.is_zero() && !self.strategy.mp.is_zero() {
        //     let profit_price = self.strategy.mp - self.local_position.long_avg;
        //
        //     profit_price * self.local_position.long_pos
        // } else if !self.local_position.short_pos.is_zero() && !self.local_position.short_pos.is_zero() && !self.strategy.mp.is_zero() {
        //     let profit_price = self.local_position.short_avg - self.strategy.mp;
        //
        //     profit_price * self.local_position.short_pos
        // } else {
        //     Decimal::ZERO
        // };
        // let floating_draw_back = floating_profit / self.strategy.start_equity;
        // if floating_draw_back < -self.stop_loss {
        //     let exit_msg = format!("{} 浮动亏损太大{}，当前价格：{}, 准备停机。", self.params.account_name, floating_profit, self.strategy.mp);
        //     warn!(exit_msg);
        //     self.exit_msg = exit_msg;
        //     self.stop().await;
        // }

        // 报单延迟风控，平均延迟允许上限5000ms
        if self.ready == 1 && self.platform_rest.get_request_avg_delay() > dec!(5000) {
            let exit_msg = format!("{} 延迟爆表 触发风控 准备停机。", self.params.account_name);
            warn!(exit_msg);
            self.exit_msg = exit_msg;
            self.stop().await;
        }

        // 仓位异常风控，只在合约模式下执行
        // if !self.exchange.contains("spot") {
        //     let local_position_by_orders_pos = (self.local_position_by_orders.long_pos - self.local_position_by_orders.short_pos).abs();
        //     let local_position_pos = (self.local_position.long_pos - self.local_position.short_pos).abs();
        //
        //     if local_position_by_orders_pos != local_position_pos {
        //         warn!("{}发现仓位异常", self.params.account_name);
        //         warn!(?self.local_position_by_orders, ?self.local_position);
        //         self.position_check_series.push(1);
        //     } else {
        //         self.position_check_series.push(0);
        //     }
        //
        //     // self.position_check_series长度限制
        //     if self.position_check_series.len() > 6 {
        //         self.position_check_series.remove(0);
        //     }
        //
        //     // 连续不符合判定
        //     if self.position_check_series.iter().sum::<i8>() >= 6 {
        //         let exit_msg = format!("{} 合约连续检查本地仓位和推算仓位不符合，退出。", self.params.account_name);
        //         warn!(exit_msg);
        //         self.exit_msg = exit_msg;
        //         self.stop().await;
        //     }
        // }

        // 下单异常风控
        if self.strategy.total_amount == Decimal::ZERO {
            let exit_msg = format!("{} 开仓量为0，退出。", self.params.account_name);
            warn!(exit_msg);
            self.exit_msg = exit_msg;
            self.stop().await;
        }

        // 行情更新异常风控
        let mut exchange_names = self.ref_name.clone();
        exchange_names.push(self.trade_name.clone());
        for exchange_name in exchange_names {
            let now_time_millis = Utc::now().timestamp_millis();
            let last_update_millis = self.market_update_time.get(&exchange_name).unwrap();
            let delay = now_time_millis - last_update_millis;
            let limit = global::public_params::MARKET_DELAY_LIMIT;

            if self.ready == 1 && delay > limit {
                let exit_msg = format!("{} ticker_name:{}, delay:{}ms，行情更新延迟过高，退出。",
                                       self.params.account_name, exchange_name, delay);
                warn!(?now_time_millis, ?last_update_millis, ?limit);
                warn!(exit_msg);
                self.exit_msg = exit_msg;
                self.stop().await;
            }
        }
        let local_orders = self.local_orders.clone();
        // 订单异常风控
        for (client_id, order) in local_orders {
            // 订单长时间停留 怀疑漏单 但未必一定漏 5min
            if Utc::now().timestamp_millis() - order.local_time > 5 * 60 * 1000 {
                let exit_msg = format!("{}订单停留过长，怀疑异常，退出，cid:{}。", self.params.account_name, client_id);
                warn!(exit_msg);
                self.exit_msg = exit_msg;
                self.stop().await;
            }
        }

        // 持仓均价异常风控（浮盈风控）
        if self.strategy.long_pos_bias != Decimal::ZERO {
            if self.strategy.long_hold_value > Decimal::TWO * self.strategy.min_amount_value {
                if self.strategy.long_pos_bias > dec!(4) || self.strategy.long_pos_bias < -Decimal::TWO {
                    let exit_msg = format!("{} long_pos_bias: {}，持仓均价异常(mp: {}, avg: {})，退出。", self.params.account_name, self.strategy.long_pos_bias, self.strategy.mp, self.strategy.pos.long_avg);
                    warn!(exit_msg);
                    self.exit_msg = exit_msg;
                    self.stop().await;
                }
            }
        }
        if self.strategy.short_pos_bias != Decimal::ZERO {
            if self.strategy.short_hold_value > Decimal::TWO * self.strategy.min_amount_value {
                if self.strategy.short_pos_bias > dec!(4) || self.strategy.short_pos_bias < -Decimal::TWO {
                    let exit_msg = format!("{} short_pos_bias: {}，持仓均价异常(mp: {}, avg: {})，退出。", self.params.account_name, self.strategy.short_pos_bias, self.strategy.mp, self.strategy.pos.short_avg);
                    warn!(exit_msg);
                    self.exit_msg = exit_msg;
                    self.stop().await;
                }
            }
        }

        // 订单撤单异常风控
        for (client_id, cancel_delay) in self.local_cancel_log.clone() {
            if cancel_delay > 300 {
                let exit_msg = format!("{} 长时间无法撤销，client_id: {}，退出。", self.params.account_name, client_id);
                warn!(exit_msg);
                warn!(?self.strategy.ref_price, ?self.strategy.mp);
                self.exit_msg = exit_msg;
                self.stop().await;
            }
        }

        // 定价异常风控
        // if self.ready == 1 && (self.predictor.fair_price - self.predictor.mid_price).abs() / self.predictor.mid_price > dec!(0.03) {
        //     let exit_msg = format!("{} 定价偏离过大，怀疑定价异常，退出。", self.params.account_name);
        //     warn!(exit_msg);
        //     warn!(?self.predictor.fair_price, ?self.predictor.mid_price);
        //     self.exit_msg = exit_msg;
        //     self.stop().await;
        // }
    }

    // #[instrument(skip(self), level="TRACE")]
    pub async fn buy_token(&mut self) {
        // 买入平台币
        // 获取U数量，平台币数量
        // 更新账户
        let mut cash = Decimal::ZERO;
        let mut token = Decimal::ZERO;
        let token_param = get_exchange_token(&self.exchange);
        if token_param.token == "***" {
            error!("购买平台币失败，未找到交易所的平台币！");
            return;
        }
        match self.platform_rest.get_spot_account().await {
            Ok(val) => {
                for account in val {
                    if account.coin == "USDT".to_string() {
                        cash += account.balance;
                    }
                    if token_param.token == account.coin {
                        token += account.balance;
                    }
                }
            }
            Err(err) => {
                error!("购买{}-获取账户失败 {}", token_param.token, err);
            }
        }
        info!("持u {} , 持有平台币 {}", cash, token);
        match self.platform_rest.get_ticker_symbol(format!("{}_USDT", token_param.token)).await {
            Ok(val) => {
                let mp = (val.buy + val.sell) / Decimal::TWO;
                let token_value = token * mp;
                if token_value < token_param.limit_value {
                    info!("平台币 {} 数量过少，需要补充。", token_param.token);
                    let need_cash: Decimal = Decimal::TWO * Decimal::ONE_HUNDRED;
                    if cash > need_cash {
                        info!("准备买入{}", token_param.token);
                        match self.platform_rest.take_order_symbol(token_param.token, Decimal::ONE, "t-888", "kd", mp * Decimal::from_str("1.001").unwrap(), Decimal::from_str("50").unwrap() / mp).await {
                            Ok(value) => {
                                info!("买入平台币下单成功: {:?}", value);
                            }
                            Err(error) => {
                                error!("买入平台币下单失败: {}", error)
                            }
                        }
                    } else {
                        info!("账户余额：{}{}，至少需要：{}{}, 不执行买入{}操作！", cash, self.quote, need_cash, self.quote, token_param.token);
                    }
                } else {
                    info!("平台币{}数量充足！", token_param.token);
                }
            }
            Err(err) => {
                error!("购买平台币-获取平台币行情失败 {}", err);
            }
        }
    }

    // #[instrument(skip(self, target_hold_coin), level="TRACE")]
    pub async fn check_position(&mut self, target_hold_coin: Decimal, is_first_clear: bool) -> bool {
        info!("------------------------------------------------------------------------------------------------------------");
        info!("步骤一：检查挂单：");
        let mut is_order_clear = false;
        match self.platform_rest.cancel_orders_all().await {
            Ok(val) => {
                let length = val.len();
                is_order_clear = length == 0;

                info!("已清空所有挂单（{}条）", length);

                for o in val {
                    info!("    {:?}", o);
                }
            }
            Err(err) => {
                warn!("取消所有订单异常({})，启动备用方法。", err);

                match self.platform_rest.cancel_orders().await {
                    Ok(val) => {
                        let length = val.len();
                        is_order_clear = length == 0;

                        info!("清空所有挂单（{}条）：{:?}", length, val);
                    }
                    Err(exc) => {
                        error!("清空当前币对订单异常: {}", exc);
                    }
                }
            }
        }
        info!("挂单检查完毕。");
        info!("");

        // 如果不是第一次检查，留500ms给交易所撮合
        if !is_first_clear {
            sleep(Duration::from_millis(500)).await;
        }

        info!("步骤二：检查仓位：");
        let is_position_clear;
        if self.exchange.contains("spot") { // 现货
            is_position_clear = self.check_position_spot(target_hold_coin.clone()).await == 0;
            info!("检查遗漏仓位（现货），目标持仓：{}USDT", target_hold_coin);
        } else { // 合约
            is_position_clear = self.check_position_swap().await == 0;
            info!("遗漏仓位检查完毕（合约）！");
        }
        info!("------------------------------------------------------------------------------------------------------------");
        info!("");

        return is_order_clear && is_position_clear;
    }

    // #[instrument(skip(self, target_hold_coin), level="TRACE")]
    pub async fn check_position_spot(&mut self, target_hold_coin: Decimal) -> usize {
        let mut length = 0;

        match self.platform_rest.get_spot_account().await {
            Ok(mut val) => {
                // 如果返回的数组里没有交易货币，则补充交易货币
                if !val.iter().any(|a| a.coin.to_uppercase().eq(&self.base.to_uppercase())) {
                    let mut base_coin_account = Account::new();
                    base_coin_account.coin = self.base.to_uppercase();
                    val.push(base_coin_account);
                }

                // 仓位补货、卖货
                for account in val {
                    let coin_name = account.coin.to_uppercase();
                    if check_coin(&self.exchange, &coin_name) {
                        continue;
                    }
                    let symbol = format!("{}_USDT", coin_name);
                    let mut _hold_coin = Decimal::ZERO;
                    // 如果是交易币，则设定仓位目标
                    if coin_name.eq(self.base.to_uppercase().as_str()) {
                        _hold_coin = target_hold_coin;
                    }
                    let ap;
                    let bp;
                    let mp;
                    match self.platform_rest.get_ticker().await {
                        Ok(ticker) => {
                            ap = ticker.sell;
                            bp = ticker.buy;
                            mp = (ap + bp) / Decimal::TWO;

                            let coin_value = account.balance * mp;
                            let diff = _hold_coin - coin_value;
                            let side;
                            let price = Decimal::ZERO;
                            let amount;
                            if diff > Decimal::from(10) {
                                side = "kd";
                                // price = mp*1.001;
                                amount = diff;
                            } else if diff < Decimal::from(-10) {
                                side = "kk";
                                // price = mp*0.999;
                                amount = -diff / mp;
                            } else {
                                // 不需要调整说明没有仓位了。

                                continue;
                            }
                            // 需要调整说明有仓位。
                            length = 1;

                            info!(?ticker);
                            info!("需要调整现货仓位 {}USDT(目标：{}USDT) 共计{}{}。", diff, _hold_coin, amount, coin_name);
                            let ts = TraceStack::new(0, Instant::now());
                            // ts.on_before_send();
                            // 价格0，市价单
                            match self.platform_rest.take_order_symbol(symbol.clone(), Decimal::ONE, utils::generate_client_id(None).as_str(), side, price, amount).await {
                                Ok(v) => {
                                    // ts.on_after_send();
                                    info!("side: {}, {} 下单，{:?}, {}", side, symbol, v, ts.to_string());
                                    // 执行完当前币对  结束循环
                                    continue;
                                }
                                Err(ex) => {
                                    // ts.on_after_send();
                                    error!("side: {}, {} {}, {}", side, symbol, ex, ts.to_string());
                                    // 执行完当前币对  结束循环
                                    continue;
                                }
                            }
                        }
                        Err(_e) => {
                            error!("清仓中- 获取 {} 币对行情错误，该币对无法调整仓位。", symbol);
                            continue;
                        }
                    }
                }

                // 补仓之后获取最新余额
                self.update_equity_rest_spot().await;
            }
            Err(err) => {
                error!("获取仓位信息异常: {}", err);
            }
        }
        info!("---------------------------遗漏仓位检查完毕（现货）！-----------------------------------");

        return length;
    }

    // #[instrument(skip(self), level="TRACE")]
    pub async fn check_position_swap(&mut self) -> usize {
        let mut length = 0;
        match self.platform_rest.get_positions().await {
            Ok(val) => {
                info!("检查仓位信息");

                for position in val {
                    if position.amount.eq(&Decimal::ZERO) {
                        continue;
                    }
                    length = length + 1;
                    info!("    仓位：{:?}", position);
                    let price = Decimal::ZERO;
                    let side;
                    info!(?position);
                    match position.position_mode {
                        PositionModeEnum::Long => {
                            // pd
                            side = "pd";
                        }
                        PositionModeEnum::Short => {
                            // pk
                            side = "pk";
                        }
                        _ => {
                            error!("    仓位position_mode匹配失败，不做操作！");
                            // 执行完当前币对  结束循环
                            continue;
                        }
                    }
                    // 发起清仓订单
                    let mut ts = TraceStack::new(0, Instant::now());
                    ts.on_before_send();
                    // 市价单
                    match self.platform_rest.take_order_symbol(position.symbol.clone(),
                                                               Decimal::ONE,
                                                               utils::generate_client_id(None).as_str(),
                                                               side,
                                                               price,
                                                               position.amount.abs()).await {
                        Ok(order) => {
                            ts.on_after_send();
                            info!("    {}仓位清除市价下单成功 {:?}, {}", position.symbol.clone(), order, ts.to_string());
                            // 执行完当前币对  结束循环
                            continue;
                        }
                        Err(error) => {
                            // ts.on_after_send();
                            error!("    {}仓位清除市价下单异常 {}, {}", position.symbol.clone(), error, ts.to_string());
                            // 执行完当前币对  结束循环
                            continue;
                        }
                    };
                }
                // match self.platform_rest.get_ticker_symbol(position.symbol.clone()).await {
                //     Ok(ticker) => {
                //         let ap = ticker.sell;
                //         let bp = ticker.buy;
                //         let mp = (ap + bp) / Decimal::TWO;
                //         let price;
                //         let side;
                //         let market_info;
                //         // 获取market
                //         match self.platform_rest.get_market_symbol(position.symbol.clone()).await {
                //             Ok(market) => {
                //                 market_info = market;
                //             }
                //             Err(err) => {
                //                 error!("    {} 获取当前market异常: {}", position.symbol.clone(), err);
                //                 continue;
                //             }
                //         }
                //         info!(?position);
                //         match position.position_mode {
                //             PositionModeEnum::Long => {
                //                 // pd
                //                 price = (mp * dec!(0.9985) / market_info.tick_size).floor() * market_info.tick_size;
                //                 side = "pd";
                //             }
                //             PositionModeEnum::Short => {
                //                 // pk
                //                 price = (mp * dec!(1.0015) / market_info.tick_size).floor() * market_info.tick_size;
                //                 side = "pk";
                //             }
                //             _ => {
                //                 error!("    仓位position_mode匹配失败，不做操作！");
                //                 // 执行完当前币对  结束循环
                //                 continue;
                //             }
                //         }
                //         // 发起清仓订单
                //         info!(?ticker);
                //         let mut ts = TraceStack::new(0, Instant::now());
                //         ts.on_before_send();
                //         match self.platform_rest.take_order_symbol(position.symbol.clone(), Decimal::ONE, utils::generate_client_id(None).as_str(), side, price, position.amount.abs()).await {
                //             Ok(order) => {
                //                 ts.on_after_send();
                //                 info!("    {}仓位清除下单成功 {:?}, {}", position.symbol.clone(), order, ts.to_string());
                //                 // 执行完当前币对  结束循环
                //                 continue;
                //             }
                //             Err(error) => {
                //                 ts.on_after_send();
                //                 error!("    {}仓位清除下单异常 {}, {}", position.symbol.clone(), error, ts.to_string());
                //                 // 执行完当前币对  结束循环
                //                 continue;
                //             }
                //         };
                //     }
                //     Err(err) => {
                //         error!("    {} 获取当前ticker异常: {}", position.symbol.clone(), err)
                //     }
                // }
                // }
            }
            Err(error) => {
                length = 0;

                error!("获取仓位信息异常: {}", error);
            }
        }

        return length;
    }


    // #[instrument(skip(self), level="TRACE")]
    pub async fn stop(&mut self) {
        /*
         *  停机函数
         *  mode_signal 不能小于80
         *  前3秒用于maker平仓
         *  后1秒用于撤maker平仓单
         *  休眠1秒再执行check_position 避免卡单导致漏仓位
        */
        info!("进入停机流程...");
        self.running.store(false, Ordering::Relaxed);
        self.mode_signal = 80;
        // info!("开始退出操作");
        // info!("为避免api失效导致遗漏仓位 建议人工复查");
        // self.check_position().await;
        // // 开启停机信号
        // // sleep(Duration::from_secs(1)).await;
        // info!("双重检查遗漏仓位");
        // self.check_position().await;
        // info!("停机退出  停机原因： {}", self.exit_msg);
        // // 发送交易状态 await self._post_params()
        // // TODO: 向中控发送信号
        // info!("退出进程!");
    }

    // #[instrument(skip(self), level="TRACE")]
    pub async fn exit(&mut self) {
        info!("-------------------------启动退出流程({})----------------------------", self.exit_msg);
        info!("");

        self.clear_position_and_orders(Decimal::ZERO).await;

        info!("订单、仓位清除完毕，为避免api失效导致遗漏仓位，建议人工复查。");
        info!("停机原因：{}。", self.exit_msg);
    }

    // #[instrument(skip(self), level="TRACE")]
    pub async fn before_trade(&mut self) -> bool {
        sleep(Duration::from_secs(1)).await;
        // 获取市场信息
        self.get_exchange_info().await;
        // 获取价格信息
        let ticker = self.platform_rest.get_ticker().await.expect("获取价格信息异常!");
        info!(?ticker);
        let mp = (ticker.buy + ticker.sell) / Decimal::TWO;
        // 获取账户信息
        if self.exchange.contains("spot") {
            self.update_equity_rest_spot().await;
        } else {
            self.update_equity_rest_swap().await;
        }
        // 更新中控账户相关信息
        {
            let mut now_balance = self.local_cash / self.used_pct;
            now_balance.rescale(4);

            let mut cci = self.cci_arc.lock().await;
            cci.now_balance = now_balance;
        }
        // 初始资金
        let start_cash = self.local_cash.clone();
        let start_coin = self.local_coin.clone();
        if start_cash.is_zero() && start_coin.is_zero() {
            self.exit_msg = format!("{}{}{}{}", "初始余额为零 cash: ", start_cash, " coin: ", start_coin);
            // 停止程序
            self.stop().await;
            return false;
        }
        info!("初始cash: {start_cash} 初始coin: {start_coin}");
        // 初始化策略基础信息
        if mp <= Decimal::ZERO {
            self.exit_msg = format!("{}{}", "初始价格获取错误: ", mp);
            // 停止程序
            self.stop().await;
            return false;
        } else {
            info!("初始价格为 {}", mp);
        }
        self.strategy.mp = mp.clone();
        self.strategy.start_cash = start_cash.clone();
        self.strategy.start_coin = start_coin.clone();
        self.strategy.start_equity = start_cash + start_coin * mp;
        self.strategy.max_equity = self.strategy.start_equity.clone();
        self.strategy.equity = self.strategy.start_equity.clone();
        self.strategy.total_amount = self.strategy.equity * self.strategy.lever_rate / self.strategy.mp;
        // 获取数量精度
        self.strategy.step_size = self.market.amount_size.clone();
        if self.strategy.step_size > Decimal::ONE {
            self.strategy.step_size = self.strategy.step_size.trunc();
        }
        // 获取价格精度
        self.strategy.tick_size = self.market.tick_size.clone();
        if self.strategy.tick_size > Decimal::ONE {
            self.strategy.tick_size = self.strategy.tick_size.trunc();
        }
        // 获取最小下单价值
        self.strategy.min_amount_value = self.market.min_notional;

        if self.strategy.step_size.is_zero() || self.strategy.tick_size.is_zero() || self.strategy.min_amount_value.is_zero() {
            self.exit_msg = format!("交易精度未正常获取 step_size:{}, tick_size:{}, min_amount_value:{}", self.strategy.step_size, self.strategy.tick_size, self.strategy.min_amount_value);
            // 停止程序
            self.stop().await;
            return false;
        } else {
            info!("数量精度 {}", self.strategy.step_size);
            info!("价格精度 {}", self.strategy.tick_size);
            info!("最小下单价值 {}", self.strategy.min_amount_value);
        }
        let grid = Decimal::from(self.params.grid.clone());
        // 计算下单数量
        let long_one_hand_value: Decimal = start_cash * self.params.lever_rate / grid;
        let short_one_hand_value: Decimal;
        let long_one_hand_amount: Decimal = (long_one_hand_value / mp / &self.strategy.step_size).floor() * self.strategy.step_size;
        let short_one_hand_amount: Decimal;

        if self.exchange.contains("spot") {
            short_one_hand_value = start_coin * mp * self.params.lever_rate / grid;
            short_one_hand_amount = (short_one_hand_value / mp / self.strategy.step_size).floor() * self.strategy.step_size;
        } else {
            short_one_hand_value = start_cash * self.params.lever_rate / grid;
            short_one_hand_amount = (short_one_hand_value / mp / self.strategy.step_size).floor() * self.strategy.step_size;
        }
        info!("最低单手交易下单量为 buy: {}, sell: {}", long_one_hand_amount, short_one_hand_amount);
        info!(?long_one_hand_value, ?short_one_hand_value, ?long_one_hand_amount, ?short_one_hand_amount, ?self.local_cash, ?self.local_coin);
        let hand_min_limit = Decimal::new(5, 0);
        if (long_one_hand_amount.is_zero() && short_one_hand_amount.is_zero()) ||
            (long_one_hand_value < hand_min_limit && short_one_hand_value < hand_min_limit) {
            self.exit_msg = format!("请检查账户余额！初始下单量太少 buy: {} sell: {}", long_one_hand_amount, short_one_hand_amount);
            // 停止程序
            self.stop().await;
            return false;
        }
        // 初始化调度器
        self.local_cash = start_cash;
        self.local_coin = start_coin;
        self.predictor.on_balance(self.local_cash).await;

        // 买入平台币
        if self.exchange.contains("spot") { // 现货
            self.buy_token().await;
        }

        // 清空挂单和仓位
        self.clear_position_and_orders(self.hold_coin).await;
        /*
        ###### 交易前准备就绪 可以开始交易 ######
        self.loop.create_task(self.rest.go())
        self.loop.create_task(self.on_timer())
        self.loop.create_task(self._run_server())
        self.loop.create_task(self.run_stratey())
        self.loop.create_task(self.early_stop_loop())
        */
        return true;
    }

    pub async fn clear_position_and_orders(&mut self, target_hold_coin: Decimal) {
        let mut clear_count = 1;
        let mut total_clear_count = 1;

        loop {
            let is_clear = self.check_position(target_hold_coin, clear_count <= 1).await;
            // 如果N次中间有一次没清理干净，则重新清理
            if !is_clear {
                clear_count = 1
            }

            // 如果连续N次都检查到清理干净，则表明大概率是清理干净了的
            if clear_count >= 1 {
                info!("连续{}次清理完成。", clear_count);
                info!("");
                break
            }

            // 最大次数判定
            if total_clear_count >= 500 {
                info!("清理次数达到上限：{}次，不再执行清理。", total_clear_count);
                info!("");
                break
            }

            clear_count += 1;
            total_clear_count += 1;
            info!("清理指令发送完毕，启动第{}次检查。", clear_count);
            info!("");

            // 每次清理间隔1s
            sleep(Duration::from_secs(1)).await;
        }
    }
}

// #[instrument(skip(core_arc), level="TRACE")]
pub fn run_strategy(core_arc: Arc<Mutex<Core>>) -> JoinHandle<()> {
    spawn(async move {
        //定期触发策略
        info!("定时触发器启动");
        info!("前期准备完成");
        sleep(Duration::from_secs(1)).await;
        loop {
            let start_time = Utc::now().timestamp_millis();
            let mut delay = 1u64;
            {
                let mut core = core_arc.lock().await;
                if core.ready == 1 {
                    // 更新交易信息集合
                    core.update_trade_msg();
                    if core.mode_signal != 0 {
                        if core.mode_signal > 1 {
                            core.mode_signal -= 1;
                        }
                        if core.mode_signal == 1 {
                            return;
                        }
                        // 触发策略  更新策略时间
                        core.strategy.local_time = Utc::now().timestamp_millis();
                        let mut platform_rest_fb = core.platform_rest.clone_box();
                        // 获取信号
                        if core.mode_signal > 20 {
                            // 先执行onExit
                            let mut predictor = core.predictor.clone();
                            let local_orders = core.local_orders.clone();
                            let local_position = core.local_position.clone();
                            let local_cash = core.local_cash.clone();
                            let local_coin = core.local_coin.clone();

                            let mut orders = core.strategy.on_exit(&mut predictor, &local_orders, &local_position, &local_coin, &local_cash);
                            if orders.is_not_empty() {
                                info!("触发onExit");
                                info!(?orders);
                                core._update_local_orders(&mut orders);
                                spawn(async move {
                                    let mut ts = TraceStack::new(0, Instant::now());
                                    platform_rest_fb.command_order(&mut orders, &mut ts).await;
                                });
                            }
                        } else {
                            // 再执行onSleep
                            let mut predictor = core.predictor.clone();
                            let local_orders = core.local_orders.clone();
                            let local_position = core.local_position.clone();
                            let local_cash = core.local_cash.clone();
                            let local_coin = core.local_coin.clone();

                            let mut orders = core.strategy.on_sleep(&mut predictor, &local_orders, &local_position, &local_coin, &local_cash);
                            // 记录指令触发信息
                            if orders.is_not_empty() {
                                info!("触发onSleep");
                                info!(?orders);
                                core._update_local_orders(&mut orders);
                                spawn(async move {
                                    let mut ts = TraceStack::new(0, Instant::now());
                                    platform_rest_fb.command_order(&mut orders, &mut ts).await;
                                });
                            }
                        }
                    }
                } else {
                    core.check_ready();
                }
                // 计算耗时并进行休眠
                let pass_time = (Utc::now().timestamp_millis() - start_time).to_u64().unwrap();
                if pass_time < core.interval {
                    delay = core.interval - pass_time;
                }
            }
            sleep(Duration::from_millis(delay)).await;
        }
    })
}

// 定期触发的系统逻辑
// #[instrument(skip(core_arc), level="TRACE")]
pub fn on_timer(core_arc: Arc<Mutex<Core>>) -> JoinHandle<()> {
    let core_arc_clone = core_arc.clone();

    return spawn(async move {
        tokio::time::sleep(Duration::from_secs(20)).await;

        loop {
            tokio::time::sleep(Duration::from_secs(10)).await;

            let mut core = core_arc_clone.lock().await;
            {
                // 检查风控
                core.check_risk().await;

                // 线程停止信号
                if core.mode_signal == 1 {
                    return;
                }

                // 计算预估成交额
                let total_trade_value = core.local_buy_value + core.local_sell_value;
                let time_diff = Decimal::from(Utc::now().timestamp_millis() - core.start_time);
                let _86400 = dec!(86400);
                let _10000 = dec!(10000);
                let trade_vol_24h = (total_trade_value / time_diff) * _86400;
                core.strategy.trade_vol_24h_w = trade_vol_24h / _10000;
                core.strategy.trade_vol_24h_w.rescale(2);

                // TODO core没有rest
                // info!("Rest报单平均延迟{}ms", core.rest.avg_delay);
                // info!("Rest报单最高延迟{}ms", core.rest.max_delay);
                for (_name, _interval) in &core.market_update_interval {
                    // debug!("WS盘口{}行情平均更新间隔{}ms。", name, interval);
                }
            }
        }
    });
}

// 是不是不用调整仓位的币
pub fn check_coin(exchanges: &String, coin_name: &String) -> bool {
    let mut result = false;
    match exchanges.as_str() {
        "bitget_spot" => {
            result = ["BGB", "USDT"].contains(&coin_name.as_str());
        }
        _ => {}
    }
    result
}

//获取平台币
pub fn get_exchange_token(exchanges: &String) -> TokenParam {
    return match exchanges.as_str() {
        "bitget_spot" => {
            TokenParam {
                token: "BGB".to_string(),
                // 30u
                limit_value: Decimal::TEN * (Decimal::ONE + Decimal::TWO),
            }
        }
        _ => {
            TokenParam {
                token: "***".to_string(),
                limit_value: Decimal::ZERO,
            }
        }
    };
}