use std::cmp::max; 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::{Receiver, Sender}; use tokio::sync::{Mutex}; use tokio::task::JoinHandle; use tokio::time::sleep; use tracing::{debug, error, info, warn}; use global::params::Params; use global::public_params::{ASK_PRICE_INDEX, BID_PRICE_INDEX, LENGTH}; use standard::{Account, Market, Order, OrderCommand, Platform, Position, PositionModeEnum, SpecialTicker, Ticker}; use standard::exchange::{Exchange}; use standard::exchange::ExchangeEnum::{BinanceSpot, BinanceSwap, GateSpot, GateSwap, KucoinSwap}; use crate::model::{LocalPosition, OrderInfo, TraderMsg}; use crate::predictor::Predictor; use crate::strategy::Strategy; pub struct Quant { pub params: Params, // 启动时间 pub start_time: i64, // 币对 pub symbol: String, // 基础货币 pub base: String, // 报价货币 pub quote: String, // pub strategy: Strategy, // 本地挂单表 pub local_orders: HashMap, // 本地订单缓存队列 pub local_orders_backup: HashMap, // 本地订单缓存cid队列 pub local_orders_backup_cid: Vec, // 本地已处理cid缓存队列 pub handled_orders_cid: Vec, // 本地利润值 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, pub interval: u64, pub exchange: String, pub trade_msg: TraderMsg, pub exit_msg: String, // 仓位检查结果序列 pub position_check_series: Vec, // 止损大小 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, // 盘口 depth信息 pub depths: HashMap>, // 行情更新延迟监控(风控) pub market_update_time: HashMap, pub market_update_interval: HashMap, pub ref_num: i8, pub ref_name: Vec, pub trade_name: String, pub ready: i8, pub predictor: Predictor, pub market: Market, pub platform_rest:Box, // 市场最优买卖价 pub max_buy_min_sell_cache: HashMap>, // 最近一次的depth信息 pub local_depths: HashMap>, pub is_update: HashMap, pub running: Arc, } impl Quant { pub async fn new(exchange: String, params: Params, exchange_params: BTreeMap, order_sender: Sender, error_sender: Sender, running: Arc) -> Quant { let symbol = params.pair.clone(); let pairs: Vec<&str> = params.pair.split('_').collect(); let mut quant_obj = Quant { params: params.clone(), start_time: 0, symbol: symbol.clone(), base: pairs[0].to_string(), quote: pairs[1].to_string(), strategy: Strategy::new(¶ms, true), 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(), interval: params.interval, exchange: params.exchange, trade_msg: TraderMsg::new(), exit_msg: "正常退出".to_string(), position_check_series: Default::default(), stop_loss: params.stop_loss, used_pct: params.used_pct, 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 { loop_count: 0, market_info_list: vec![], mid_price_list: vec![], ref_mid_price_per_exchange_per_frame: vec![], ref_exchange_length: 0, data_length_max: 0, alpha: vec![], gamma: Default::default(), avg_spread_list: vec![], }, market: Market { symbol: symbol.clone(), base_asset: "".to_string(), quote_asset: "".to_string(), tick_size: Default::default(), amount_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(), ct_val: Default::default(), }, platform_rest: match exchange.as_str() { "kucoin_usdt_swap" => { Exchange::new(KucoinSwap, symbol, false, exchange_params, order_sender, error_sender).await }, "gate_usdt_swap" => { Exchange::new(GateSwap, symbol, false, exchange_params, order_sender, error_sender).await }, "gate_usdt_spot" => { Exchange::new(GateSpot, symbol, false, exchange_params, order_sender, error_sender).await }, "binance_usdt_swap" => { Exchange::new(BinanceSwap, symbol, false, exchange_params, order_sender, error_sender).await }, "binance_usdt_spot" => { Exchange::new(BinanceSpot, symbol, false, 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, }; 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(); quant_obj.tickers.insert(tickers_key, SpecialTicker { sell: Default::default(), buy: Default::default(), mid_price: Default::default(), }); quant_obj.ref_name.push(ref_name_element); quant_obj.depths.insert(depths_key, Default::default()); quant_obj.market_update_time.insert(market_update_time_key, Default::default()); quant_obj.market_update_interval.insert(market_update_interval_key, Default::default()); quant_obj.max_buy_min_sell_cache.insert(max_buy_min_sell_cache_key, vec![Decimal::ZERO, Decimal::ZERO]); } let name = format!("{}{}{}", quant_obj.exchange.clone(), "@", quant_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(); quant_obj.trade_name = name; quant_obj.market_update_time.insert(market_update_time_key, Default::default()); quant_obj.market_update_interval.insert(market_update_interval_key, Default::default()); quant_obj.tickers.insert(tickers_key, SpecialTicker { sell: Default::default(), buy: Default::default(), mid_price: Default::default(), }); quant_obj.depths.insert(depths_key, Default::default()); quant_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 = 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); quant_obj.predictor = Predictor::new(quant_obj.ref_name.len()) .alpha(price_alpha) .gamma(params.gamma); return quant_obj; } pub async fn handle_signals(quant_arc: Arc>, mut rx: Receiver) { spawn(async move{ loop { sleep(Duration::from_millis(1)).await; match rx.try_recv() { Ok(val)=>{ // 只处理这两种订单回执 if ["NEW", "REMOVE"].contains(&val.status.as_str()){ let mut local_order_info = OrderInfo{ symbol: "".to_string(), amount: Default::default(), side: "".to_string(), price: Default::default(), client_id: "".to_string(), filled_price: Default::default(), filled: Default::default(), order_id: "".to_string(), local_time: 0, create_time: 0, status: "".to_string(), fee: Decimal::ZERO, trace_stack: Default::default(), }; if val.status== "NEW" { local_order_info.client_id = val.custom_id; local_order_info.order_id = val.id; } else if val.status == "REMOVE" { local_order_info.client_id = val.custom_id; } let mut bot = quant_arc.lock().await; // let mut time_delay = time_record.lock().await; // 写入本地订单缓存 bot.update_local_order(local_order_info); } }, Err(e) => { info!("订单回执消费失败！{}", e); return; } } } }); } pub fn update_order(&mut self, data: Vec){ for order in data { self.update_local_order(order); } } pub fn update_local_order(&mut self, data: OrderInfo) { 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(); // 新增订单推送 仅需要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); } if self.local_orders.contains_key(&data.client_id) { debug!("删除本地订单, client_id:{:?}", data); self.local_orders.remove(&data.client_id); } else { debug!("该订单不在本地挂单表中, order:{:?}", data); } // 在cid缓存队列中 说明是本策略的订单 if self.local_orders_backup.contains_key(&data.client_id) { // 不在已处理cid缓存队列中 说明还没参与过仓位计算 则执行订单计算 if self.handled_orders_cid.contains(&data.client_id) { debug!("订单已经参与过仓位计算 拒绝重复进行计算, 订单号：{}", 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); if self.exchange.contains("spot") { // 如果是现货交易 还需要修改equity // 现货必须考虑fee 买入fee单位为币 卖出fee单位为u let fee = data.fee; if side == "kd" { // buy 开多 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.clone(); 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; } else if side == "pd" { // sell 平多 self.local_sell_amount += filled; self.local_sell_value += filled * filled_price; self.local_profit += filled * (filled_price - self.local_position_by_orders.long_avg); 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; } else if side == "pk" { // buy 平空 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; 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; } else if side == "kk" { // sell 开空 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 { // 合约订单流仓位计算 if side == "kd" { // buy 开多 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; } } else if side == "kk" { // sell 开空 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; } } else if side == "pd" { // sell 平多 self.local_sell_amount += filled; self.local_sell_value += filled * filled_price; self.local_profit += filled * (filled_price - self.local_position_by_orders.long_avg); 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; } } else if side == "pk" { // buy 平空 self.local_buy_amount += filled; self.local_buy_value += filled * filled_price; self.local_profit += filled * (self.local_position_by_orders.short_avg - filled_price); 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); // 本地计算利润 self._print_local_trades_summary(); // 打印各类信息 self.strategy.local_orders = self.local_orders.clone(); self.strategy._print_summary(); } // 每次有订单变动就触发一次策略 if self.mode_signal == 0 && self.ready == 1 { // 更新交易数据 self.update_trade_msg(); // 触发策略挂单逻辑 // 更新策略时间 self.strategy.local_time = Utc::now().timestamp_millis(); // time_record.strategy_start = Utc::now().timestamp_micros(); let order = self.strategy.on_time(&self.trade_msg); // time_record.strategy_end = Utc::now().timestamp_micros(); // 记录指令触发信息 if order.is_not_empty() { // info!("触发onOrder"); self._update_local_orders(&order); // time_record.quant_end = Utc::now().timestamp_micros(); // time_record.source = "508 orderUpdate".to_string(); // time_record.order_command = order.to_string(); //交易所处理订单信号 let mut platform_rest_fb = self.platform_rest.clone_box(); // info!("订单指令：{:?}", order); spawn(async move{ // info!("update_local_order订单指令：{:?}", order); platform_rest_fb.command_order(order).await; }); } } } } else { debug!("订单不属于本策略 拒绝进行仓位计算: {}", data.client_id); } } else { error!("未知的订单事件类型：{:?}", data); } } pub 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); } } // 检测初始数据是否齐全 pub fn check_ready(&mut self) { // 检查 ticker 行情 for i in &self.ref_name { if self.tickers.is_empty() || !self.tickers.contains_key(i) { info!("529参考盘口ticker未准备好: {:?}", self.tickers); return; } else { if self.tickers.get(i).unwrap().buy == dec!(0) || self.tickers.get(i).unwrap().sell == dec!(0) { info!("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) { info!("540交易盘口ticker未准备好: {:?}", self.tickers); return; } } else { info!("544交易盘口ticker未准备好: {:?}", self.tickers); return; } // 检查 market 行情 let all_market: Vec = self.get_all_market_data(); if all_market.len() != LENGTH * (1usize + self.ref_num as usize) { info!("550聚合行情未准备好: market长度：{}, 检验数: {}", all_market.len(), LENGTH * (1usize + self.ref_num as usize)); return; } else { info!("553聚合行情准备就绪"); self.trade_msg.market = all_market; self.predictor.market_info_handler(&self.trade_msg.market); } self.ready = 1 } pub fn _update_depth(&mut self, depth: Vec, name :String) { // 要从回调传入的深度信息中获取data.name let market_update_interval_key = name.clone(); let market_update_time_key = name.clone(); let depths1_key = name.clone(); let depths2_key = name.clone(); let depths3_key = name.clone(); let now_time = Utc::now().timestamp_millis(); if self.market_update_time.contains_key(&name) && *self.market_update_time.get(&name).unwrap() != 0i64 { let interval = Decimal::from(now_time - self.market_update_time.get(&name).unwrap()); if *self.market_update_interval.get(&name).unwrap() == dec!(0) { self.market_update_interval.insert(market_update_interval_key, interval); } else { let value = self.market_update_interval.get(&name).unwrap(); self.market_update_interval.insert(market_update_interval_key, value * dec!(0.999) + interval * dec!(0.001)); } } self.market_update_time.insert(market_update_time_key, now_time); // 初始化depths if self.depths.get(&name).unwrap().is_empty() { self.depths.insert(depths1_key, depth.clone()); } // 判断是否需要触发ondepth // 是否是交易盘口 if name == self.trade_name { // 更新depths self.depths.insert(depths2_key, depth.clone()); // 允许交易 if self.mode_signal == 0 && self.ready == 1 { self.on_agg_market(); } } else if name == self.ref_name[0] { // 判断是否为当前跟踪的盘口 // 判断是否需要触发ontick 对行情进行过滤 // 过滤条件 价格变化很大 时间间隔很长 let mut flag = 0; let bid_price_rate = (depth[BID_PRICE_INDEX] - self.depths.get(&name).unwrap()[BID_PRICE_INDEX]).abs() / depth[BID_PRICE_INDEX]; let ask_price_rate = (depth[ASK_PRICE_INDEX] - self.depths.get(&name).unwrap()[ASK_PRICE_INDEX]).abs() / depth[ASK_PRICE_INDEX]; let rate = dec!(0.0002); if bid_price_rate > rate || ask_price_rate > rate || Utc::now().timestamp_millis() - self.on_tick_event_time > 50 { // 允许交易 flag = 1; // 更新ontick触发时间记录 self.on_tick_event_time = Utc::now().timestamp_millis(); } // 更新depths self.depths.insert(depths3_key, depth); // 允许交易 if self.mode_signal == 0 && self.ready == 1 && flag == 1 { // 更新交易数据 self.update_trade_msg(); // 触发事件撤单逻辑 // 更新策略时间 self.strategy.local_time = Utc::now().timestamp_millis(); // time_record.strategy_start = Utc::now().timestamp_micros(); // 产生交易信号 let orders = self.strategy.on_time(&self.trade_msg); // time_record.strategy_end = Utc::now().timestamp_micros(); if orders.is_not_empty() { debug!("触发onTick"); self._update_local_orders(&orders); // time_record.quant_end = Utc::now().timestamp_micros(); //异步交易所处理订单信号 let mut platform_rest_fb = self.platform_rest.clone_box(); // time_record.source = "646 depth".to_string(); // time_record.order_command = orders.to_string(); // info!("订单指令：{:?}", orders); spawn(async move{ // info!("_update_depth订单指令：{:?}", orders); platform_rest_fb.command_order(orders).await; }); } } } } pub fn update_position(&mut self, data: Vec) { if data.is_empty() { return; } let mut position = LocalPosition::new(); for pos in &data { 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!("收到新的仓位推送, position: {:?}, local_position: {:?}", data, position); info!("更新本地仓位:{:?}", self.local_position); self.local_position = position; } } pub fn _update_ticker(&mut self, data: SpecialTicker, name: String) { self.tickers.insert(name, data); } pub fn on_agg_market(&mut self) { /* 处理聚合行情 1. 获取聚合行情 2. 更新预测器 3. 触发tick回测 */ // 更新聚合市场数据 let agg_market = self.get_all_market_data(); // 更新聚合市场信息 self.trade_msg.market = agg_market; // 更新预测器 self.predictor.market_info_handler(&self.trade_msg.market); } pub fn update_trade_msg(&mut self) { // 更新保证金 self.trade_msg.cash = self.local_cash.round_dp(10); self.trade_msg.coin = self.local_coin.round_dp(10); let position = self.local_position_by_orders.clone(); // 使用本地推算仓位 self.trade_msg.position = position; let orders = self.local_orders.clone(); self.trade_msg.orders = orders; // 更新 ref let mut ref_tickers: BTreeMap = BTreeMap::new(); for i in &self.ref_name { let bp = self.tickers.get(i).unwrap().buy.clone(); let ap = self.tickers.get(i).unwrap().sell.clone(); ref_tickers.insert(i.clone(), Ticker { time: 0, high: Default::default(), low: Default::default(), sell: ap, buy: bp, last: Default::default(), volume: Default::default(), }); } let ref_price: Vec> = self.predictor.get_ref_price(&ref_tickers); self.trade_msg.ref_price = ref_price; } // 本地记录所有报单信息 pub fn _update_local_orders(&mut self, orders: &OrderCommand) { let mut limits = HashMap::new(); limits.extend(orders.clone().limits_open); limits.extend(orders.clone().limits_close); if !limits.is_empty() { for j in limits.keys() { let order_info = OrderInfo { symbol: self.symbol.clone(), amount: Decimal::from_str(limits.get(j).unwrap()[0].as_str()).unwrap(), side: limits.get(j).unwrap()[1].clone(), price: Decimal::from_str(limits.get(j).unwrap()[2].as_str()).unwrap(), client_id: limits.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: Default::default(), }; // 本地挂单表 self.local_orders.insert(limits.get(j).unwrap()[3].clone(), order_info.clone()); // 本地缓存表 self.local_orders_backup.insert(limits.get(j).unwrap()[3].clone(), order_info); // 本地缓存cid表 self.local_orders_backup_cid.push(limits.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); } } // 获取深度信息 pub fn get_all_market_data(&mut self) -> Vec { // 只能定时触发 组合市场信息=交易盘口+参考盘口 let mut market: Vec = Vec::new(); // 获取交易盘口市场信息 let mut data: Vec = 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; } pub async fn get_exchange_info(&mut self,) { self.market = self.platform_rest.get_self_market(); } pub fn update_equity(&mut self, data: Account) { /* 更新保证金信息 合约一直更新 现货只有当出现异常时更新 */ if self.exchange.contains("spot") { return; } self.local_cash = data.balance * self.used_pct } pub async fn update_equity_rest(&mut self) { match self.platform_rest.get_account().await { Ok(val) => { /* 更新保证金信息 合约一直更新 现货只有当出现异常时更新 */ if self.exchange.contains("spot") { return; } self.local_cash = val.balance * self.used_pct }, Err(e) => { info!("获取账户信息错误: {:?}", e); } } } 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; } // 报单延迟风控，平均延迟允许上限5000ms if 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 long_diff = (self.local_position.long_pos - self.local_position_by_orders.long_pos).abs(); let short_diff = (self.local_position.short_pos - self.local_position_by_orders.short_pos).abs(); let diff_pos = max(long_diff, short_diff); let diff_pos_value = diff_pos * self.strategy.mp; if diff_pos_value > self.strategy._min_amount_value { 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() > 30 { self.position_check_series.remove(0); } // 连续不符合判定 if self.position_check_series.iter().sum::() >= 30 { 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 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: {}，持仓均价异常，退出。", self.params.account_name, self.strategy.long_pos_bias); 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: {}，持仓均价异常，退出。", self.params.account_name, self.strategy.long_pos_bias); 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.strategy.ref_price - self.strategy.mp).abs() / self.strategy.mp > dec!(0.03) { let exit_msg = format!("{} 定价偏离过大，怀疑定价异常，退出。", self.params.account_name); warn!(exit_msg); warn!(?self.strategy.ref_price, ?self.strategy.mp); self.exit_msg = exit_msg; self.stop().await; } } pub async fn check_position(&mut self){ info!("清空挂单！"); match self.platform_rest.cancel_orders().await{ Ok(val)=>{ info!(?val); }, Err(e)=>{ error!("清空挂单异常: {}", e); } }; info!("检查遗漏仓位！"); match self.platform_rest.get_positions().await { Ok(val)=>{ for position in val { if !position.symbol.eq_ignore_ascii_case(self.symbol.as_str()){ continue; } if position.amount.eq(&Decimal::ZERO) { continue; } match self.platform_rest.get_ticker().await { Ok(ticker)=>{ let ap = ticker.sell; let bp = ticker.buy; let mp = ( ap + bp ) / Decimal::TWO; let price; let side; info!(?position); match position.position_mode { PositionModeEnum::Long => { // pd price = (mp*dec!(0.999)/self.market.tick_size).floor()*self.market.tick_size; side = "pd"; }, PositionModeEnum::Short => { // pk price = (mp*dec!(1.001)/self.market.tick_size).floor()*self.market.tick_size; side = "pk"; } _ => { info!("仓位匹配失败，不做操作！"); // 执行完当前币对 结束循环 break; } } match self.platform_rest.take_order("t-123", side, price, position.amount.abs()).await { Ok(order)=>{ info!("清仓下单，{:?}", order); // 执行完当前币对 结束循环 break; }, Err(error)=>{ error!("清仓下单异常：{}", error); // 执行完当前币对 结束循环 break; } }; }, Err(err)=>{ error!("获取当前ticker异常: {}",err) } } } }, Err(error)=>{ error!("获取仓位信息异常: {}", error); } } } pub async fn stop(&mut self){ /* * 停机函数 * mode_signal 不能小于80 * 前6秒用于maker平仓 * 后2秒用于撤maker平仓单 * 休眠2秒再执行check_position 避免卡单导致漏仓位 */ info!("进入停机流程..."); self.mode_signal = 80; sleep(Duration::from_secs(10)).await; info!("开始退出操作"); info!("为避免api失效导致遗漏仓位 建议人工复查"); self.check_position().await; // 开启停机信号 sleep(Duration::from_secs(3)).await; info!("双重检查遗漏仓位"); self.check_position().await; info!("停机退出 停机原因： {}", self.exit_msg); // 发送交易状态 await self._post_params() // TODO: 向中控发送信号 self.running.store(false, Ordering::Relaxed); info!("退出进程!"); } pub async fn exit(&mut self, delay: i8){ info!("预约退出操作 delay:{}", delay); if delay > 0i8 { sleep(Duration::from_secs(delay as u64)).await; } info!("开始退出操作"); info!("为避免api失效导致遗漏仓位 建议人工复查"); self.check_position().await; // 开启停机信号 sleep(Duration::from_secs(3)).await; info!("双重检查遗漏仓位"); self.check_position().await; info!("停机退出 停机原因： {}", self.exit_msg); // 发送交易状态 await self._post_params() // TODO: 向中控发送信号 self.running.store(false, Ordering::Relaxed); info!("退出进程!"); } 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("获取价格信息异常!"); let mp = (ticker.buy + ticker.sell) / Decimal::TWO; // 获取账户信息 self.update_equity_rest().await; // 初始资金 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(); } if self.strategy.step_size.is_zero() || self.strategy.tick_size.is_zero() { self.exit_msg = format!("{}{}{}{}", "交易精度未正常获取 step_size: ", self.strategy.step_size, " tick_size:", self.strategy.tick_size); // 停止程序 self.stop().await; return false; } else { info!("数量精度 {}", self.strategy.step_size); info!("价格精度 {}", self.strategy.tick_size); } 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); 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: ", long_one_hand_amount, " sell: ", short_one_hand_amount); // 停止程序 self.stop().await; return false; } // 初始化调度器 self.local_cash = start_cash; self.local_coin = start_coin; // 清空挂单和仓位 self.check_position().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 fn run_strategy(quant_arc: Arc>) -> JoinHandle<()>{ return spawn(async move { //定期触发策略 info!("定时触发器启动"); info!("前期准备完成"); sleep(Duration::from_secs(10)).await; loop { let start_time = Utc::now().timestamp_millis(); let mut delay = 1u64; { let mut quant = quant_arc.lock().await; if quant.ready == 1 { // 更新交易信息集合 quant.update_trade_msg(); if quant.mode_signal != 0 { if quant.mode_signal > 1 { quant.mode_signal -= 1; } if quant.mode_signal == 1 { return; } // 触发策略 更新策略时间 quant.strategy.local_time = Utc::now().timestamp_millis(); let trade_msg = quant.trade_msg.clone(); let mut platform_rest_fb = quant.platform_rest.clone_box(); // 获取信号 if quant.mode_signal > 20 { // 先执行onExit let orders = quant.strategy.on_exit(&trade_msg); if orders.is_not_empty() { info!("触发onExit"); info!(?orders); quant._update_local_orders(&orders); spawn(async move { platform_rest_fb.command_order(orders).await; }); } } else { // 再执行onSleep let orders = quant.strategy.on_sleep(&trade_msg); // 记录指令触发信息 if orders.is_not_empty() { info!("触发onSleep"); info!(?orders); quant._update_local_orders(&orders); spawn(async move { platform_rest_fb.command_order(orders).await; }); } } } } else { quant.check_ready(); } // 计算耗时并进行休眠 let pass_time = (Utc::now().timestamp_millis() - start_time).to_u64().unwrap(); if pass_time < quant.interval { delay = quant.interval - pass_time; } } sleep(Duration::from_millis(delay)).await; } }); } // 定期触发的系统逻辑 pub fn on_timer(quant_arc: Arc>) -> JoinHandle<()> { let quant_arc_clone = quant_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 quant = quant_arc_clone.lock().await; { // 检查风控 quant.check_risk().await; // 线程停止信号 if quant.mode_signal == 1 { return } // 计算预估成交额 let total_trade_value = quant.local_buy_value + quant.local_sell_value; let time_diff = Decimal::from(Utc::now().timestamp_millis() - quant.start_time); let trade_vol_24h = (total_trade_value / time_diff) * dec!(86400); quant.strategy.trade_vol_24h_w = trade_vol_24h / dec!(10000); quant.strategy.trade_vol_24h_w.rescale(2); // TODO quant没有rest // info!("Rest报单平均延迟{}ms", quant.rest.avg_delay); // info!("Rest报单最高延迟{}ms", quant.rest.max_delay); for (name, interval) in &quant.market_update_interval { debug!("WS盘口{}行情平均更新间隔{}ms。", name, interval); } } } }); }