use tokio::time::Instant; 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::{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::public_params::{ASK_PRICE_INDEX, BID_PRICE_INDEX, LENGTH}; use global::trace_stack::TraceStack; use standard::{Account, Market, Order, OrderCommand, Platform, Position, PositionModeEnum, SpecialTicker, Ticker}; use standard::exchange::{Exchange}; use standard::exchange::ExchangeEnum::{BinanceSwap, BybitSwap, 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_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 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, pub hold_coin: Decimal, // 打印限频 pub prev_log_ready_timestamp: i64, pub log_ready_log_interval: i64, // 中控 pub cci_arc: Arc>, // 中控信息汇集 // 老版的trader_msg留下来的 pub agg_market: Vec, pub ref_price: Vec>, pub predict: Decimal, } impl Core { pub async fn new(exchange: String, params: Params, exchange_params: BTreeMap, order_sender: Sender, error_sender: Sender, running: Arc, cci_arc: Arc>) -> 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(¶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, exit_msg: "正常退出".to_string(), position_check_series: Default::default(), stop_loss: params.stop_loss, used_pct: dec!(0.95), 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(), 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(), 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 // } // "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 } _ => { 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 = 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); core_obj.predictor = Predictor::new(core_obj.ref_name.len()) .alpha(price_alpha) .gamma(params.gamma); return core_obj; } // #[instrument(skip(self, data, trace_stack), level="TRACE")] pub async fn update_order(&mut self, data: Vec, 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) { // 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 + 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; } 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 - 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; } 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(); info!("------------------------------------------------------------------------------------"); } // 每次有订单变动就触发一次策略 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.on_tick(&self.local_orders, &self.local_position_by_orders, &self.agg_market, &self.local_cash, &self.local_coin, &self.ref_price, &self.predict, &trace_stack.ins); // 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 { // debug!("订单不属于本策略 拒绝进行仓位计算: {}", 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 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); } } // 检测初始数据是否齐全 // #[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; } } // #[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); } } // #[instrument(skip(self, depth, name_ref, trace_stack), level="TRACE")] pub async fn on_depth_update(&mut self, depth: &Vec, name_ref: &String, trace_stack: &mut TraceStack) { // 要从回调传入的深度信息中获取data.name let now_time = Utc::now().timestamp_millis(); 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); // 判断是否需要触发on_depth // 是否是交易盘口 if *name_ref == self.trade_name { // 允许交易 if self.mode_signal == 0 && self.ready == 1 { self.on_agg_market(); } } else if *name_ref == self.ref_name[0] { // 判断是否为当前跟踪的盘口 // 判断是否需要触发ontick 对行情进行过滤 // 过滤条件 价格变化很大 时间间隔很长 let mut flag = 0; let bid_price_rate = (depth[BID_PRICE_INDEX] - self.depths.get(name_ref).unwrap()[BID_PRICE_INDEX]).abs() / depth[BID_PRICE_INDEX]; let ask_price_rate = (depth[ASK_PRICE_INDEX] - self.depths.get(name_ref).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(); } // 允许交易 if self.mode_signal == 0 && self.ready == 1 && flag == 1 { // 更新交易数据 self.update_trade_msg(); // 触发事件撤单逻辑 // 更新策略时间 self.strategy.local_time = Utc::now().timestamp_millis(); // 产生交易信号 let mut orders = self.strategy.on_tick(&self.local_orders, &self.local_position_by_orders, &self.agg_market, &self.local_cash, &self.local_coin, &self.ref_price, &self.predict, &trace_stack.ins); 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.strategy.equity / 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) { 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: {:?}", data); info!("更新本地仓位:{:?}", position); 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); 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 = 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: 0, high: Default::default(), low: Default::default(), sell: ap, buy: bp, last: Default::default(), volume: 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_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 { // 只能定时触发 组合市场信息=交易盘口+参考盘口 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; } // #[instrument(skip(self), level="TRACE")] pub async fn get_exchange_info(&mut self) { self.market = self.platform_rest.get_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(val) => { /* 更新保证金信息 合约一直更新 现货只有当出现异常时更新 */ self.local_cash = val.balance * self.used_pct } Err(e) => { info!("获取账户信息错误: {:?}", e); } } } pub async fn update_position_rest_swap(&mut self) { let position = self.platform_rest.get_position().await; match position { Ok(val) => { // info!("bybit_swap：定时获取的仓位信息"); self.update_position(val).await; } Err(err) => { error!("bybit_swap：定时获取仓位信息错误!\nget_position:res_data={:?}", err); } } } // #[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; } // 报单延迟风控，平均延迟允许上限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 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 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: {}，持仓均价异常，退出。", 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.ready == 1 && (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; } } // #[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) -> bool { let mut is_clear = false; info!("------------------------------------------------------------------------------------------------------------"); info!("步骤一：检查挂单："); match self.platform_rest.cancel_orders_all().await { Ok(val) => { let length = val.len(); is_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_clear = length == 0; info!("清空所有挂单（{}条）：{:?}", length, val); } Err(exc) => { error!("清空当前币对订单异常: {}", exc); } } } } info!("挂单检查完毕。"); info!(""); info!("步骤二：检查仓位："); if self.exchange.contains("spot") { // 现货 is_clear = is_clear && (self.check_position_spot(target_hold_coin.clone()).await == 0); info!("检查遗漏仓位（现货），目标持仓：{}USDT", target_hold_coin); } else { // 合约 is_clear = is_clear && (self.check_position_swap().await == 0); info!("遗漏仓位检查完毕（合约）！"); } info!("------------------------------------------------------------------------------------------------------------"); info!(""); return is_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!("检查仓位信息（{}条仓位信息，部分交易所会返回0持仓的）：", length); 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 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!("-------------------------启动退出流程----------------------------"); info!(""); // 循环清空仓位，如若彻底清空，才进行退出。 let mut clear_count = 1; while !self.check_position(Decimal::ZERO).await { sleep(Duration::from_secs(1)).await; clear_count += 1; info!("清理指令发送完毕，启动第{}次检查。", clear_count); info!(""); } 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("获取价格信息异常!"); 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(); } 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); 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; // 买入平台币 if self.exchange.contains("spot") { // 现货 self.buy_token().await; } // 清空挂单和仓位 self.check_position(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; } } // #[instrument(skip(core_arc), level="TRACE")] pub fn run_strategy(core_arc: Arc>) -> JoinHandle<()> { return 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 local_orders = core.local_orders.clone(); let position = core.local_position_by_orders.clone(); let agg_market = core.agg_market.clone(); let local_cash = core.local_cash.clone(); let local_coin = core.local_coin.clone(); let ref_price = core.ref_price.clone(); let predict = core.predict.clone(); let mut orders = core.strategy.on_exit(&local_orders, &position, &agg_market, &local_cash, &local_coin, &ref_price, &predict); 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 local_orders = core.local_orders.clone(); let position = core.local_position_by_orders.clone(); let agg_market = core.agg_market.clone(); let local_cash = core.local_cash.clone(); let local_coin = core.local_coin.clone(); let ref_price = core.ref_price.clone(); let predict = core.predict.clone(); let mut orders = core.strategy.on_sleep(&local_orders, &position, &agg_market, &local_cash, &local_coin, &ref_price, &predict); // 记录指令触发信息 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>) -> 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 trade_vol_24h = (total_trade_value / time_diff) * dec!(86400); core.strategy.trade_vol_24h_w = trade_vol_24h / dec!(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, } } }; }