标准层不报错了，待整理。

Cargo.toml

@@ -6,3 +6,20 @@ edition = "2021"
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
+standard = { path="./standard" }
+exchanges = { path="./exchanges" }
+global = { path="./global" }
+
+tokio = { version = "1.31.0", features = ["full"] }
+chrono = "0.4.26"
+tracing = "0.1"
+tracing-subscriber = { version = "0.3.17", features = ["env-filter"] }
+tracing-appender-timezone = { git = "https://github.com/skyfffire/tracing-appender-timezone.git" }
+serde = { version = "1.0.188", features = ["derive"] }
+actix-rt = "2.5.0"
+actix-web = "4.0.0-beta.12"
+ctrlc = "3.2.5"
+serde_json = "1.0.105"
+rust_decimal = { version = "1.32.0", features = ["maths"] }
+rust_decimal_macros = "1.32.0"
+

exchanges/src/lib.rs

@@ -23,6 +23,5 @@ pub mod kucoin_spot_ws;
 pub mod kucoin_spot_rest;
 pub mod crypto_spot_ws;
 pub mod bybit_swap_rest;
-pub mod xlsx_utils;
 pub mod bybit_swap_ws;
 

global/src/log_utils.rs

@@ -23,7 +23,7 @@ impl Visit for ErrorMessageVisitor {
 
 // 错误报告发送到指定服务器
 struct ReportingLayer {
-    account_name: String,
+    app_name: String,
 }
 impl<S> Layer<S> for ReportingLayer
     where
@@ -35,6 +35,8 @@ impl<S> Layer<S> for ReportingLayer
                 message: String::new()
             };
             event.record(&mut visitor);
+
+            panic!("Unhandle error: {}", self.app_name)
         }
     }
 }
@@ -51,7 +53,7 @@ pub fn init_log_with_info() {
     let _ = final_init(tracing::Level::INFO.as_str(), 0, "test".to_string());
 }
 
-pub fn final_init(level: &str, port: u32, account_name: String) -> WorkerGuard {
+pub fn final_init(level: &str, port: u32, app_name: String) -> WorkerGuard {
     let mut path = String::new();
     path.push_str("./logs");
     path.push_str(port.to_string().as_str());
@@ -87,7 +89,7 @@ pub fn final_init(level: &str, port: u32, account_name: String) -> WorkerGuard {
         .with_span_events(fmt::format::FmtSpan::FULL);
 
     let reporting_layer = ReportingLayer {
-        account_name
+        app_name
     };
 
     let layer = tracing_subscriber::Registry::default()

src/main.rs

@@ -1,3 +1,16 @@
-fn main() {
-    println!("Hello, world!");
+use tracing::info;
+use tracing_appender_timezone::non_blocking::WorkerGuard;
+
+// 日志级别配置
+fn log_level_init(log_str: String, port: u32, app_name: String) -> WorkerGuard {
+    info!("日志级别读取成功：{}。", log_str);
+    global::log_utils::final_init(log_str.as_str(), port, app_name)
+}
+
+#[tokio::main(flavor = "multi_thread")]
+async fn main() {
+    // 日志级别配置
+    let _guard = log_level_init("info".to_string(), 8888, "data-center".to_string());
+
+    info!("Welcome data center");
 }

standard/src/binance_swap.rs

@@ -8,11 +8,9 @@ use rust_decimal::prelude::FromPrimitive;
 use rust_decimal_macros::dec;
 use serde_json::Value;
 use tokio::sync::mpsc::Sender;
-use tokio::time::Instant;
-use tracing::{error, warn};
-use crate::{Platform, ExchangeEnum, Account, Position, Ticker, Market, Order, OrderCommand, utils, PositionModeEnum};
+use tracing::{error};
+use crate::{Platform, ExchangeEnum, Account, Position, Ticker, Market, Order, utils, PositionModeEnum};
 use exchanges::binance_swap_rest::BinanceSwapRest;
-use global::trace_stack::TraceStack;
 
 #[allow(dead_code)]
 #[derive(Clone)]
@@ -289,8 +287,7 @@ impl Platform for BinanceSwap {
                 deal_amount: Decimal::from_str(res_data_json["executedQty"].as_str().unwrap()).unwrap(),
                 avg_price: Decimal::from_str(res_data_json["avgPrice"].as_str().unwrap()).unwrap(),
                 status: custom_status,
-                order_type: res_data_json["type"].as_str().unwrap().parse().unwrap(),
-                trace_stack: TraceStack::new(0, Instant::now()).on_special("301 binance_swap".to_string()),
+                order_type: res_data_json["type"].as_str().unwrap().parse().unwrap()
             };
             Ok(result)
         } else {
@@ -318,8 +315,7 @@ impl Platform for BinanceSwap {
                     deal_amount: Decimal::from_str(item["executedQty"].as_str().unwrap()).unwrap(),
                     avg_price: Decimal::from_str(item["avgPrice"].as_str().unwrap()).unwrap(),
                     status: custom_status,
-                    order_type: item["type"].as_str().unwrap().parse().unwrap(),
-                    trace_stack: TraceStack::new(0, Instant::now()).on_special("331 binance_swap".to_string()),
+                    order_type: item["type"].as_str().unwrap().parse().unwrap()
                 }
             }).collect();
             Ok(result)
@@ -354,8 +350,6 @@ impl Platform for BinanceSwap {
     async fn set_auto_deposit_status(&mut self, _status: bool) -> Result<String, Error> { Err(Error::new(ErrorKind::NotFound, "binance_swap：该交易所方法未实现".to_string())) }
 
     async fn wallet_transfers(&mut self, _coin: &str, _from: &str, _to: &str, _amount: Decimal) -> Result<String, Error> { Err(Error::new(ErrorKind::NotFound, "binance_swap：该交易所方法未实现".to_string())) }
-
-    async fn command_order(&mut self, _order_command: &mut OrderCommand, _trace_stack: &TraceStack) { warn!("binance_swap：该交易所方法未实现"); }
 }
 
 pub fn format_position_item(position: &serde_json::Value, ct_val: Decimal) -> Position {

standard/src/bitget_swap.rs

@@ -4,18 +4,13 @@ use std::io::{Error, ErrorKind};
 use tokio::sync::mpsc::Sender;
 use std::str::FromStr;
 use async_trait::async_trait;
-use futures::stream::FuturesUnordered;
-use futures::TryStreamExt;
 use rust_decimal::{Decimal, MathematicalOps};
 use rust_decimal::prelude::ToPrimitive;
 use rust_decimal_macros::dec;
 use serde_json::{json, Value};
-use tokio::spawn;
-use tokio::time::Instant;
 use tracing::{error, info};
-use global::trace_stack::TraceStack;
 use crate::exchange::ExchangeEnum;
-use crate::{Account, Market, Order, OrderCommand, Platform, Position, PositionModeEnum, Ticker, utils};
+use crate::{Account, Market, Order, Platform, Position, PositionModeEnum, Ticker, utils};
 
 #[allow(dead_code)]
 #[derive(Clone)]
@@ -374,8 +369,7 @@ impl Platform for BitgetSwap {
             deal_amount: Decimal::ZERO,
             avg_price: Decimal::ZERO,
             status: "NEW".to_string(),
-            order_type: "".to_string(),
-            trace_stack: TraceStack::new(0, Instant::now()).on_special("328 bitget_swap".to_string()),
+            order_type: "".to_string()
         };
         return Ok(result)
     }
@@ -404,8 +398,7 @@ impl Platform for BitgetSwap {
             deal_amount: Decimal::ZERO,
             avg_price: Decimal::ZERO,
             status: "REMOVE".to_string(),
-            order_type: "".to_string(),
-            trace_stack: TraceStack::new(0, Instant::now()).on_special("443 bitget_swap".to_string()),
+            order_type: "".to_string()
         };
         Ok(result)
     }
@@ -442,8 +435,7 @@ impl Platform for BitgetSwap {
                             deal_amount: Decimal::ZERO,
                             avg_price: Decimal::ZERO,
                             status: "REMOVE".to_string(),
-                            order_type: "".to_string(),
-                            trace_stack: TraceStack::new(0, Instant::now()).on_special("457 bitget_swap".to_string()),
+                            order_type: "".to_string()
                         });
                     } else {
                         return Err(Error::new(ErrorKind::Other, cancel_res_data.to_string()));
@@ -515,109 +507,6 @@ impl Platform for BitgetSwap {
         //     Err(Error::new(ErrorKind::Other, res_data.to_string()))
         // }
     }
-
-    async fn command_order(&mut self, order_command: &mut OrderCommand, trace_stack: &TraceStack) {
-        let mut handles = vec![];
-
-        // 下单指令
-        for item in order_command.limits_open.keys() {
-            let mut ts = trace_stack.clone();
-
-            let amount = Decimal::from_str(&*order_command.limits_open[item].get(0).unwrap().clone()).unwrap();
-            let side = order_command.limits_open[item].get(1).unwrap().clone();
-            let price = Decimal::from_str(&*order_command.limits_open[item].get(2).unwrap().clone()).unwrap();
-            let cid = order_command.limits_open[item].get(3).unwrap().clone();
-
-            //  order_name: [数量，方向，价格，c_id]
-            let mut self_clone = self.clone();
-            let handle = spawn(async move {
-                // TraceStack::show_delay(&ts.ins);
-                ts.on_before_send();
-                let result = self_clone.take_order(cid.as_str(), side.as_str(), price, amount).await;
-                ts.on_after_send();
-
-                match result {
-                    Ok(mut result) => {
-                        result.trace_stack = ts;
-
-                        self_clone.order_sender.send(result).await.unwrap();
-                    }
-                    Err(error) => {
-                        info!(?error);
-                        let mut err_order = Order::new();
-                        err_order.custom_id = cid.clone();
-                        err_order.status = "REMOVE".to_string();
-
-                        self_clone.order_sender.send(err_order).await.unwrap();
-                        self_clone.error_sender.send(error).await.unwrap();
-                    }
-                }
-            });
-            handles.push(handle)
-        }
-        let futures = FuturesUnordered::from_iter(handles);
-        // 等待所有任务完成
-        let _: Result<Vec<_>, _> = futures.try_collect().await;
-
-        // 撤销订单
-        let mut cancel_handlers = vec![];
-        for item in order_command.cancel.keys() {
-            let order_id = order_command.cancel[item].get(1).unwrap().clone();
-            let custom_id = order_command.cancel[item].get(0).unwrap().clone();
-
-            let mut self_clone = self.clone();
-            let handle = spawn(async move {
-                let result = self_clone.cancel_order(&order_id, &custom_id).await;
-                match result {
-                    Ok(_) => {
-                        // result_sd.send(result).await.unwrap();
-                    }
-                    Err(error) => {
-                        // 取消失败去查订单。
-                        let query_rst = self_clone.get_order_detail(&order_id, &custom_id).await;
-                        match query_rst {
-                            Ok(order) => {
-                                self_clone.order_sender.send(order).await.unwrap();
-                            }
-                            Err(err) => {
-                                error!("撤单失败，而且查单也失败了，bitget_swap，oid={}, cid={}, err={:?}。", order_id.clone(), custom_id.clone(), err);
-                            }
-                        }
-                        self_clone.error_sender.send(error).await.unwrap();
-                    }
-                }
-            });
-            cancel_handlers.push(handle)
-        }
-        let futures = FuturesUnordered::from_iter(cancel_handlers);
-        // 等待所有任务完成
-        let _: Result<Vec<_>, _> = futures.try_collect().await;
-
-        // 检查订单指令
-        let mut check_handlers = vec![];
-        for item in order_command.check.keys() {
-            let order_id = order_command.check[item].get(1).unwrap().clone();
-            let custom_id = order_command.check[item].get(0).unwrap().clone();
-
-            let mut self_clone = self.clone();
-            let handle = spawn(async move {
-                let result = self_clone.get_order_detail(order_id.as_str(), custom_id.as_str()).await;
-                match result {
-                    Ok(result) => {
-                        self_clone.order_sender.send(result).await.unwrap();
-                    }
-                    Err(error) => {
-                        self_clone.error_sender.send(error).await.unwrap();
-                    }
-                }
-            });
-            check_handlers.push(handle)
-        }
-
-        let futures = FuturesUnordered::from_iter(check_handlers);
-        // 等待所有任务完成
-        let _: Result<Vec<_>, _> = futures.try_collect().await;
-    }
 }
 
 // pub fn format_account_info(balance_data: Value) -> Account {
@@ -665,7 +554,6 @@ pub fn format_order_item(order: Value, ct_val: Decimal) -> Order {
         deal_amount,
         avg_price,
         status: custom_status,
-        order_type: order["orderType"].as_str().unwrap().to_string(),
-        trace_stack: TraceStack::new(0, Instant::now()).on_special("700 bitget_swap".to_string()),
+        order_type: order["orderType"].as_str().unwrap().to_string()
     }
 }

standard/src/bitget_swap_handle.rs

@@ -2,9 +2,7 @@ use std::str::FromStr;
 use rust_decimal::Decimal;
 use rust_decimal::prelude::FromPrimitive;
 use serde_json::Value;
-use tokio::time::Instant;
 use exchanges::response_base::ResponseData;
-use global::trace_stack::TraceStack;
 use crate::{Account, MarketOrder, Order, Position, PositionModeEnum, SpecialOrder};
 
 // 处理账号信息
@@ -82,8 +80,7 @@ pub fn format_order_item(order: Value, ct_val: Decimal) -> Order {
         deal_amount,
         avg_price,
         status: custom_status,
-        order_type: order["orderType"].as_str().unwrap().to_string(),
-        trace_stack: TraceStack::new(0, Instant::now()).on_special("86 bitget_swap_handle".to_string()),
+        order_type: order["orderType"].as_str().unwrap().to_string()
     }
 }
 

standard/src/bybit_swap.rs

@@ -3,17 +3,13 @@ use std::io::{Error, ErrorKind};
 use std::str::FromStr;
 use tokio::sync::mpsc::Sender;
 use async_trait::async_trait;
-use futures::stream::FuturesUnordered;
-use futures::TryStreamExt;
 use rust_decimal::Decimal;
 use serde_json::{from_value, json, Value};
 use rust_decimal::prelude::FromPrimitive;
 use serde::{Deserialize, Serialize};
-use tokio::time::Instant;
 use tracing::{error, trace};
 use exchanges::bybit_swap_rest::BybitSwapRest;
-use crate::{Platform, ExchangeEnum, Account, Position, Ticker, Market, Order, OrderCommand, PositionModeEnum};
-use global::trace_stack::TraceStack;
+use crate::{Platform, ExchangeEnum, Account, Position, Ticker, Market, Order, PositionModeEnum};
 
 #[derive(Debug, Clone, Deserialize, Serialize)]
 #[serde(rename_all = "camelCase")]
@@ -532,108 +528,6 @@ impl Platform for BybitSwap {
         // }
         Err(Error::new(ErrorKind::Other, "暂未实现！"))
     }
-
-    // 指令下单
-    async fn command_order(&mut self, order_command: &mut OrderCommand, trace_stack: &TraceStack) {
-        // 下单指令
-        let mut handles = vec![];
-        for item in order_command.limits_open.keys() {
-            let mut self_clone = self.clone();
-
-            let amount = Decimal::from_str(order_command.limits_open[item].get(0).unwrap_or(&"0".to_string())).unwrap();
-            let side = order_command.limits_open[item].get(1).unwrap().clone();
-            let price = Decimal::from_str(order_command.limits_open[item].get(2).unwrap_or(&"0".to_string())).unwrap();
-            let cid = order_command.limits_open[item].get(3).unwrap().clone();
-
-            let mut ts = trace_stack.clone();
-
-            let handle = tokio::spawn(async move {
-                ts.on_before_send();
-                // TraceStack::show_delay(&ts.ins);
-                let result = self_clone.take_order(cid.as_str(), side.as_str(), price, amount).await;
-                ts.on_after_send();
-                match result {
-                    Ok(mut result) => {
-                        // 记录此订单完成时间
-                        result.trace_stack = ts;
-
-                        self_clone.order_sender.send(result).await.unwrap();
-                    }
-                    Err(error) => {
-                        error!("bybit：下单失败：{:?}", error);
-
-                        let mut err_order = Order::new();
-                        err_order.custom_id = cid.clone();
-                        err_order.status = "REMOVE".to_string();
-
-                        self_clone.order_sender.send(err_order).await.unwrap();
-                        self_clone.error_sender.send(error).await.unwrap();
-                    }
-                }
-            });
-            handles.push(handle)
-        }
-        let futures = FuturesUnordered::from_iter(handles);
-        let _: Result<Vec<_>, _> = futures.try_collect().await;
-
-        // 撤销订单
-        let mut cancel_handles = vec![];
-        for item in order_command.cancel.keys() {
-            let mut self_clone = self.clone();
-
-            let order_id = order_command.cancel[item].get(1).unwrap().clone();
-            let custom_id = order_command.cancel[item].get(0).unwrap().clone();
-
-            let handle = tokio::spawn(async move {
-                let result = self_clone.cancel_order(&order_id, &custom_id).await;
-                match result {
-                    Ok(_) => {
-                        // result_sd.send(result).await.unwrap();
-                    }
-                    Err(error) => {
-                        // 取消失败去查订单。
-                        let query_rst = self_clone.get_order_detail(&order_id, &custom_id).await;
-                        match query_rst {
-                            Ok(order) => {
-                                self_clone.order_sender.send(order).await.unwrap();
-                            }
-                            Err(_err) => {
-                                error!("bybit：撤单失败，而且查单也失败了，oid={}, cid={}。", order_id.clone(), custom_id.clone());
-                            }
-                        }
-                        self_clone.error_sender.send(error).await.unwrap();
-                    }
-                }
-            });
-            cancel_handles.push(handle)
-        }
-        let futures = FuturesUnordered::from_iter(cancel_handles);
-        let _: Result<Vec<_>, _> = futures.try_collect().await;
-
-        // 检查订单指令
-        let mut check_handles = vec![];
-        for item in order_command.check.keys() {
-            let mut self_clone = self.clone();
-
-            let order_id = order_command.check[item].get(1).unwrap().clone();
-            let custom_id = order_command.check[item].get(0).unwrap().clone();
-
-            let handle = tokio::spawn(async move {
-                let result = self_clone.get_order_detail(&order_id, &custom_id).await;
-                match result {
-                    Ok(result) => {
-                        self_clone.order_sender.send(result).await.unwrap();
-                    }
-                    Err(error) => {
-                        self_clone.error_sender.send(error).await.unwrap();
-                    }
-                }
-            });
-            check_handles.push(handle)
-        }
-        let futures = FuturesUnordered::from_iter(check_handles);
-        let _: Result<Vec<_>, _> = futures.try_collect().await;
-    }
 }
 
 pub fn format_position_item(position: &Value, ct_val: Decimal) -> Position {
@@ -687,8 +581,7 @@ fn format_cancel_order_item(order: Value) -> Order {
         deal_amount: Decimal::ZERO,
         avg_price: Decimal::ZERO,
         status: "REMOVE".to_string(),
-        order_type: "limit".to_string(),
-        trace_stack: TraceStack::new(0, Instant::now()).on_special("688 trace_stack".to_string())
+        order_type: "limit".to_string()
     }
 }
 
@@ -701,8 +594,7 @@ fn format_new_order_item(order: Value, price: Decimal, amount: Decimal) -> Order
         deal_amount: Decimal::ZERO,
         avg_price: price,
         status: "NEW".to_string(),
-        order_type: "limit".to_string(),
-        trace_stack: TraceStack::new(0, Instant::now()).on_special("688 trace_stack".to_string())
+        order_type: "limit".to_string()
     }
 }
 
@@ -738,8 +630,7 @@ pub fn format_order_item(order: Value, ct_val: Decimal) -> Order {
         deal_amount,
         avg_price,
         status: custom_status,
-        order_type: "limit".to_string(),
-        trace_stack: TraceStack::new(0, Instant::now()).on_special("688 trace_stack".to_string()),
+        order_type: "limit".to_string()
     };
     return rst_order;
 }

standard/src/bybit_swap_handle.rs

@@ -3,10 +3,8 @@ use rust_decimal::Decimal;
 use rust_decimal::prelude::{FromPrimitive, ToPrimitive};
 use rust_decimal_macros::dec;
 use serde_json::{from_value, Value};
-use tokio::time::Instant;
 use tracing::{error};
 use exchanges::response_base::ResponseData;
-use global::trace_stack::TraceStack;
 use crate::{Account, MarketOrder, Order, Position, PositionModeEnum, SpecialDepth, SpecialOrder, SpecialTicker};
 
 // 处理账号信息
@@ -130,8 +128,7 @@ pub fn format_order_item(order: Value, ct_val: Decimal) -> Order {
         deal_amount,
         avg_price,
         status: custom_status,
-        order_type: "limit".to_string(),
-        trace_stack: TraceStack::new(0, Instant::now()).on_special("120 bybit_handle".to_string()),
+        order_type: "limit".to_string()
     };
 
     return rst_order;

standard/src/gate_swap.rs

@@ -3,17 +3,12 @@ use std::io::{Error, ErrorKind};
 use std::str::FromStr;
 use tokio::sync::mpsc::Sender;
 use async_trait::async_trait;
-use futures::stream::FuturesUnordered;
-use futures::TryStreamExt;
 use rust_decimal::Decimal;
 use rust_decimal::prelude::{FromPrimitive, ToPrimitive};
 use serde_json::{json, Value};
-use tokio::spawn;
-use tokio::time::Instant;
 use tracing::{error, info};
-use crate::{Platform, ExchangeEnum, Account, Position, Ticker, Market, Order, OrderCommand, PositionModeEnum};
+use crate::{Platform, ExchangeEnum, Account, Position, Ticker, Market, Order, PositionModeEnum};
 use exchanges::gate_swap_rest::GateSwapRest;
-use global::trace_stack::TraceStack;
 
 #[allow(dead_code)]
 #[derive(Clone)]
@@ -588,110 +583,6 @@ impl Platform for GateSwap {
             Err(Error::new(ErrorKind::Other, res_data.to_string()))
         }
     }
-
-    // 指令下单
-    async fn command_order(&mut self, order_command: &mut OrderCommand, trace_stack: &TraceStack) {
-        let mut handles = vec![];
-
-        // 下单指令，limits_open里已经包含了limits_close
-        for item in order_command.limits_open.keys() {
-            let mut ts = trace_stack.clone();
-
-            let amount = Decimal::from_str(&*order_command.limits_open[item].get(0).unwrap().clone()).unwrap();
-            let side = order_command.limits_open[item].get(1).unwrap().clone();
-            let price = Decimal::from_str(&*order_command.limits_open[item].get(2).unwrap().clone()).unwrap();
-            let cid = order_command.limits_open[item].get(3).unwrap().clone();
-
-            //  order_name: [数量，方向，价格，c_id]
-            let mut self_clone = self.clone();
-            let handle = spawn(async move {
-                // TraceStack::show_delay(&ts.ins);
-                ts.on_before_send();
-                let result = self_clone.take_order(&cid, &side, price, amount).await;
-                ts.on_after_send();
-
-                match result {
-                    Ok(mut result) => {
-                        result.trace_stack = ts;
-
-                        self_clone.order_sender.send(result).await.unwrap();
-                    }
-                    Err(error) => {
-                        let mut err_order = Order::new();
-                        err_order.custom_id = cid.clone();
-                        err_order.status = "REMOVE".to_string();
-
-                        self_clone.order_sender.send(err_order).await.unwrap();
-                        self_clone.error_sender.send(error).await.unwrap();
-                    }
-                }
-            });
-            handles.push(handle)
-        }
-        let futures = FuturesUnordered::from_iter(handles);
-        // 等待所有任务完成
-        let _: Result<Vec<_>, _> = futures.try_collect().await;
-
-        // 撤销订单
-        let mut cancel_handlers = vec![];
-        for item in order_command.cancel.keys() {
-            let order_id = order_command.cancel[item].get(1).unwrap().clone();
-            let custom_id = order_command.cancel[item].get(0).unwrap().clone();
-
-            let mut self_clone = self.clone();
-            let handle = spawn(async move {
-                let result = self_clone.cancel_order(&order_id, &custom_id).await;
-                match result {
-                    Ok(_) => {
-                        // result_sd.send(result).await.unwrap();
-                    }
-                    Err(error) => {
-                        // 取消失败去查订单。
-                        let query_rst = self_clone.get_order_detail(&order_id, &custom_id).await;
-                        match query_rst {
-                            Ok(order) => {
-                                self_clone.order_sender.send(order).await.unwrap();
-                            }
-                            Err(_err) => {
-                                // error!("撤单失败，而且查单也失败了，gate_io_swap，oid={}, cid={}。", order_id.clone(), custom_id.clone());
-                                // panic!("撤单失败，而且查单也失败了，gate_io_swap，oid={}, cid={}。", order_id.clone(), custom_id.clone());
-                            }
-                        }
-                        self_clone.error_sender.send(error).await.unwrap();
-                    }
-                }
-            });
-            cancel_handlers.push(handle)
-        }
-        let futures = FuturesUnordered::from_iter(cancel_handlers);
-        // 等待所有任务完成
-        let _: Result<Vec<_>, _> = futures.try_collect().await;
-
-        // 检查订单指令
-        let mut check_handlers = vec![];
-        for item in order_command.check.keys() {
-            let order_id = order_command.check[item].get(1).unwrap().clone();
-            let custom_id = order_command.check[item].get(0).unwrap().clone();
-
-            let mut self_clone = self.clone();
-            let handle = spawn(async move {
-                let result = self_clone.get_order_detail(&order_id, &custom_id).await;
-                match result {
-                    Ok(result) => {
-                        self_clone.order_sender.send(result).await.unwrap();
-                    }
-                    Err(error) => {
-                        self_clone.error_sender.send(error).await.unwrap();
-                    }
-                }
-            });
-            check_handlers.push(handle)
-        }
-
-        let futures = FuturesUnordered::from_iter(check_handlers);
-        // 等待所有任务完成
-        let _: Result<Vec<_>, _> = futures.try_collect().await;
-    }
 }
 
 pub fn format_position_item(position: &serde_json::Value, ct_val: Decimal) -> Position {
@@ -748,8 +639,7 @@ pub fn format_order_item(order: serde_json::Value, ct_val: Decimal) -> Order {
         deal_amount,
         avg_price: Decimal::from_str(&order["fill_price"].as_str().unwrap()).unwrap(),
         status: custom_status,
-        order_type: "limit".to_string(),
-        trace_stack: TraceStack::new(0, Instant::now()).on_special("688 trace_stack".to_string()),
+        order_type: "limit".to_string()
     };
     return rst_order;
 }

standard/src/gate_swap_handle.rs

@@ -3,10 +3,8 @@ use rust_decimal::Decimal;
 use rust_decimal::prelude::FromPrimitive;
 use rust_decimal_macros::dec;
 use serde_json::Value;
-use tokio::time::Instant;
 use tracing::{error};
 use exchanges::response_base::ResponseData;
-use global::trace_stack::TraceStack;
 use crate::{Account, MarketOrder, Order, Position, PositionModeEnum, SpecialDepth, SpecialOrder, SpecialTicker};
 
 // 处理账号信息
@@ -113,8 +111,7 @@ pub fn format_order_item(order: serde_json::Value, ct_val: Decimal) -> Order {
         deal_amount,
         avg_price: Decimal::from_f64(order["fill_price"].as_f64().unwrap()).unwrap(),
         status: custom_status,
-        order_type: "limit".to_string(),
-        trace_stack: TraceStack::new(0, Instant::now()).on_special("120 gate_handle".to_string()),
+        order_type: "limit".to_string()
     };
 
     return rst_order;

standard/src/handle_info.rs

@@ -1,14 +1,11 @@
-use std::cmp::Ordering;
 use std::str::FromStr;
 use rust_decimal::{Decimal};
 use rust_decimal::prelude::FromPrimitive;
-use rust_decimal_macros::dec;
 use tracing::{error, info};
 use exchanges::response_base::ResponseData;
-use global::public_params;
 use crate::exchange::ExchangeEnum;
-use crate::{binance_swap_handle, gate_swap_handle, bybit_swap_handle, bitget_swap_handle, kucoin_handle};
-use crate::{Account, MarketOrder, Position, SpecialDepth, SpecialOrder, SpecialTicker};
+use crate::{binance_swap_handle, gate_swap_handle, bybit_swap_handle, bitget_swap_handle, kucoin_handle, Depth};
+use crate::{Account, MarketOrder, Position, SpecialDepth, SpecialOrder};
 
 #[allow(dead_code)]
 pub struct HandleSwapInfo;
@@ -155,88 +152,13 @@ impl HandleSwapInfo {
             }
         }
     }
-
-    // 处理深度信息
-    pub fn handle_special_depth(exchange: ExchangeEnum, res_data: &ResponseData) -> SpecialDepth {
-        let label = res_data.label.clone();
-        // 格式化
-        let mut format_depth = format_depth(exchange, res_data);
-        // 运算、组装
-        make_special_depth(label, &mut format_depth.depth_asks, &mut format_depth.depth_bids, format_depth.t, format_depth.create_at)
-    }
 }
 
-
-pub fn make_special_depth(label: String, depth_asks: &mut Vec<MarketOrder>, depth_bids: &mut Vec<MarketOrder>, t: Decimal, create_at: i64) -> SpecialDepth {
-    depth_asks.sort_by(|a, b| a.price.partial_cmp(&b.price).unwrap_or(Ordering::Equal));
-    depth_bids.sort_by(|a, b| b.price.partial_cmp(&a.price).unwrap_or(Ordering::Equal));
-    // TODO 不排序的话，有4us可以省下来。
-    let mp = (depth_asks[0].price + depth_bids[0].price) * dec!(0.5);
-    let step = (public_params::EFF_RANGE * mp / Decimal::from_usize(public_params::LEVEL).unwrap()).round_dp(mp.scale());
-    let mut ap = Vec::new();
-    let mut bp = Vec::new();
-    let mut av: Vec<Decimal> = Vec::new();
-    let mut bv: Vec<Decimal> = Vec::new();
-    for i in 0..public_params::LEVEL {
-        let price = (depth_asks[0].price + step * Decimal::from_f64(i as f64).unwrap()).round_dp(depth_asks[0].price.scale());
-        ap.push(price);
-    }
-    for i in 0..public_params::LEVEL {
-        let price = (depth_bids[0].price - step * Decimal::from_f64(i as f64).unwrap()).round_dp(depth_bids[0].price.scale());
-        bp.push(price);
-    }
-    let mut ap_price_tag = depth_asks[0].price + step;
-    let mut ap_index = 0;
-    for item in depth_asks.iter() {
-        let price = item.price;
-        let amount = item.amount;
-        if av.get(ap_index).is_none() { av.push(Decimal::ZERO) };
-        if price < ap_price_tag {
-            av[ap_index] += amount;
-        } else {
-            ap_price_tag += step;
-            ap_index += 1;
-            if ap_index == public_params::LEVEL {
-                break;
-            }
-            av[ap_index] += amount
-        }
-    }
-
-    let mut bp_price_tag = depth_bids[0].price - step;
-    let mut bp_index = 0;
-    for item in depth_bids.iter() {
-        let price = item.price;
-        let amount = item.amount;
-        if bv.get(bp_index).is_none() { bv.push(Decimal::ZERO) };
-        if price > bp_price_tag {
-            bv[bp_index] += amount;
-        } else {
-            bp_price_tag -= step;
-            bp_index += 1;
-            if bp_index == public_params::LEVEL {
-                break;
-            }
-            bv[bp_index] += amount
-        }
-    }
-
-    let ticker_info = SpecialTicker { sell: depth_asks[0].price, buy: depth_bids[0].price, mid_price: mp, t, create_at };
-    let depth_info = bp.iter().cloned().chain(bv.iter().cloned()).chain(ap.iter().cloned()).chain(av.iter().cloned()).collect();
-    SpecialDepth {
-        name: label,
-        depth: depth_info,
-        ticker: ticker_info,
-        t,
-        create_at,
-    }
-}
-
-pub fn format_depth(exchange: ExchangeEnum, res_data: &ResponseData) -> DepthParam {
+pub fn format_depth(exchange: ExchangeEnum, res_data: &ResponseData) -> Depth {
     let depth_asks: Vec<MarketOrder>;
     let depth_bids: Vec<MarketOrder>;
     let t: Decimal;
-    let create_at: i64;
+
     match exchange {
         // ExchangeEnum::BinanceSpot => {
         //     depth_asks = binance_swap_handle::format_depth_items(res_data_json["asks"].clone());
@@ -247,21 +169,17 @@ pub fn format_depth(exchange: ExchangeEnum, res_data: &ResponseData) -> DepthPar
         ExchangeEnum::BinanceSwap => {
             depth_asks = binance_swap_handle::format_depth_items(res_data.data["a"].clone());
             depth_bids = binance_swap_handle::format_depth_items(res_data.data["b"].clone());
-            t = Decimal::from_str(&res_data.data["u"].to_string()).unwrap();
-            create_at = res_data.data["E"].as_i64().unwrap() * 1000;
+            t = Decimal::from_str(&res_data.data["E"].to_string()).unwrap();
         }
         ExchangeEnum::GateSwap => {
             depth_asks = gate_swap_handle::format_depth_items(&res_data.data["asks"]);
             depth_bids = gate_swap_handle::format_depth_items(&res_data.data["bids"]);
-            // todo! 有id可以取 保证与py一致
             t = Decimal::from_str(&res_data.data["t"].to_string()).unwrap();
-            create_at = res_data.data["t"].as_i64().unwrap() * 1000;
         }
         ExchangeEnum::KucoinSwap => {
             depth_asks = kucoin_handle::format_depth_items(res_data.data["asks"].clone());
             depth_bids = kucoin_handle::format_depth_items(res_data.data["bids"].clone());
-            t = Decimal::from_str(&res_data.data["sequence"].to_string()).unwrap();
-            create_at = res_data.data["ts"].as_i64().unwrap() * 1000;
+            t = Decimal::from_str(&res_data.data["ts"].to_string()).unwrap();
         }
         // ExchangeEnum::KucoinSpot => {
         //     depth_asks = kucoin_spot_handle::format_depth_items(res_data_json["asks"].clone());
@@ -285,22 +203,17 @@ pub fn format_depth(exchange: ExchangeEnum, res_data: &ResponseData) -> DepthPar
             depth_asks = bitget_swap_handle::format_depth_items(res_data.data[0]["asks"].clone());
             depth_bids = bitget_swap_handle::format_depth_items(res_data.data[0]["bids"].clone());
             t = Decimal::from_str(res_data.data[0]["ts"].as_str().unwrap()).unwrap();
-            create_at = res_data.data[0]["ts"].as_str().unwrap().parse::<i64>().unwrap() * 1000;
         }
         ExchangeEnum::BybitSwap => {
             depth_asks = bybit_swap_handle::format_depth_items(res_data.data["a"].clone());
             depth_bids = bybit_swap_handle::format_depth_items(res_data.data["b"].clone());
             t = Decimal::from_i64(res_data.reach_time).unwrap();
-            create_at = res_data.reach_time * 1000;
         }
     }
 
-    DepthParam {
-        depth_asks,
-        depth_bids,
-        t,
-        create_at
+    Depth {
+        asks: depth_asks,
+        bids: depth_bids,
+        time: t,
     }
 }
-
-

standard/src/kucoin_handle.rs

@@ -3,12 +3,9 @@ use rust_decimal::Decimal;
 use rust_decimal::prelude::FromPrimitive;
 use rust_decimal_macros::dec;
 use serde_json::Value;
-use tokio::time::Instant;
 use exchanges::response_base::ResponseData;
-use global::trace_stack::TraceStack;
 use crate::{Account, MarketOrder, Order, Position, PositionModeEnum, SpecialDepth, SpecialOrder, SpecialTicker, utils};
 use crate::exchange::ExchangeEnum;
-use crate::handle_info::HandleSwapInfo;
 
 // 处理账号信息
 pub fn handle_account_info(res_data: &ResponseData, symbol: &String) -> Account {
@@ -133,16 +130,10 @@ pub fn format_order_item(order: &Value, ct_val: Decimal) -> Order {
         deal_amount,
         avg_price,
         status: custom_status,
-        order_type: order["type"].as_str().unwrap().to_string(),
-        trace_stack: TraceStack::new(0, Instant::now()).on_special("136 kucoin_handle".to_string()),
+        order_type: order["type"].as_str().unwrap().to_string()
     }
 }
 
-// 处理特殊深度数据
-pub fn handle_special_depth(res_data: ResponseData) -> SpecialDepth {
-    HandleSwapInfo::handle_special_depth(ExchangeEnum::KucoinSwap, &res_data)
-}
-
 pub fn format_depth_items(value: serde_json::Value) -> Vec<MarketOrder> {
     let mut depth_items: Vec<MarketOrder> = vec![];
     for value in value.as_array().unwrap() {

standard/src/kucoin_swap.rs

@@ -3,18 +3,14 @@ use std::io::{Error, ErrorKind};
 use std::str::FromStr;
 use tokio::sync::mpsc::Sender;
 use async_trait::async_trait;
-use futures::stream::FuturesUnordered;
-use futures::TryStreamExt;
 use rust_decimal::Decimal;
 use rust_decimal::prelude::{FromPrimitive, ToPrimitive};
 use rust_decimal_macros::dec;
 use serde_json::{json, Value};
-use tokio::time::Instant;
 use tracing::{error, info};
 use exchanges::kucoin_swap_rest::KucoinSwapRest;
-use global::trace_stack::TraceStack;
 use crate::exchange::ExchangeEnum;
-use crate::{Account, kucoin_handle, Market, Order, OrderCommand, Platform, Position, Ticker, utils};
+use crate::{Account, kucoin_handle, Market, Order, Platform, Position, Ticker, utils};
 
 #[allow(dead_code)]
 #[derive(Clone)]
@@ -360,8 +356,7 @@ impl Platform for KucoinSwap {
                 deal_amount: Decimal::ZERO,
                 avg_price: Decimal::ZERO,
                 status: "NEW".to_string(),
-                order_type: "".to_string(),
-                trace_stack: TraceStack::new(0, Instant::now()).on_special("359 kucoin_swap".to_string()),
+                order_type: "".to_string()
             };
             Ok(result)
         } else {
@@ -412,8 +407,7 @@ impl Platform for KucoinSwap {
                 deal_amount: Decimal::ZERO,
                 avg_price: Decimal::ZERO,
                 status: "NEW".to_string(),
-                order_type: "".to_string(),
-                trace_stack: TraceStack::new(0, Instant::now()).on_special("408 kucoin_swap".to_string()),
+                order_type: "".to_string()
             };
             Ok(result)
         } else {
@@ -439,8 +433,7 @@ impl Platform for KucoinSwap {
                 deal_amount: Decimal::ZERO,
                 avg_price: Decimal::ZERO,
                 status: "REMOVE".to_string(),
-                order_type: "".to_string(),
-                trace_stack: TraceStack::new(0, Instant::now()).on_special("436 kucoin_swap".to_string()),
+                order_type: "".to_string()
             };
             Ok(result)
         } else {
@@ -464,8 +457,7 @@ impl Platform for KucoinSwap {
                     deal_amount: Decimal::ZERO,
                     avg_price: Decimal::ZERO,
                     status: "REMOVE".to_string(),
-                    order_type: "".to_string(),
-                    trace_stack: TraceStack::new(0, Instant::now()).on_special("461 kucoin_swap".to_string()),
+                    order_type: "".to_string()
                 }
             ).collect();
             Ok(result)
@@ -488,8 +480,7 @@ impl Platform for KucoinSwap {
                     deal_amount: Decimal::ZERO,
                     avg_price: Decimal::ZERO,
                     status: "REMOVE".to_string(),
-                    order_type: "".to_string(),
-                    trace_stack: TraceStack::new(0, Instant::now()).on_special("486 kucoin_swap".to_string()),
+                    order_type: "".to_string()
                 }
             ).collect();
             Ok(result)
@@ -528,112 +519,6 @@ impl Platform for KucoinSwap {
     async fn wallet_transfers(&mut self, _coin: &str, _from: &str, _to: &str, _amount: Decimal) -> Result<String, Error> {
         Err(Error::new(ErrorKind::NotFound, "kucoin_swap：该交易所方法未实现".to_string()))
     }
-
-    // 指令下单
-    async fn command_order(&mut self, order_command: &mut OrderCommand, trace_stack: &TraceStack) {
-        let mut handles = vec![];
-        // 下单指令，limits_open里已经包含了limits_close，在core里面处理过了
-        for item in order_command.limits_open.keys() {
-            let mut ts = trace_stack.clone();
-
-            let amount = Decimal::from_str(&*order_command.limits_open[item].get(0).unwrap().clone()).unwrap();
-            let side = order_command.limits_open[item].get(1).unwrap().clone();
-            let price = Decimal::from_str(&*order_command.limits_open[item].get(2).unwrap().clone()).unwrap();
-            let cid = order_command.limits_open[item].get(3).unwrap().clone();
-
-            let mut self_clone = self.clone();
-            let handle = tokio::spawn(async move {
-                ts.on_before_send();
-                let result = self_clone.take_order(&cid, &side, price, amount).await;
-                ts.on_after_send();
-
-                match result {
-                    Ok(mut result) => {
-                        // ts.on_after_send();
-                        result.trace_stack = ts.clone();
-
-                        self_clone.order_sender.send(result).await.unwrap();
-                    }
-                    Err(error) => {
-                        let mut err_order = Order::new();
-                        err_order.custom_id = cid.clone();
-                        err_order.status = "REMOVE".to_string();
-
-                        self_clone.order_sender.send(err_order).await.unwrap();
-                        self_clone.error_sender.send(error).await.unwrap();
-                    }
-                }
-            });
-            handles.push(handle)
-        }
-        let futures = FuturesUnordered::from_iter(handles);
-        // 等待所有任务完成
-        let _: Result<Vec<_>, _> = futures.try_collect().await;
-
-        // 撤销订单
-        let mut cancel_handlers = vec![];
-        for item in order_command.cancel.keys() {
-            let order_id = order_command.cancel[item].get(1).unwrap().clone();
-            let custom_id = order_command.cancel[item].get(0).unwrap().clone();
-
-            let mut self_clone = self.clone();
-
-            let handle = tokio::spawn(async move {
-                if order_id != "" {
-                    let result = self_clone.cancel_order(&order_id, &custom_id).await;
-                    match result {
-                        Ok(_) => {
-                            // result_sd.send(result).await.unwrap();
-                        }
-                        Err(error) => {
-                            // 取消失败去查订单。
-                            let query_rst = self_clone.get_order_detail(&order_id, &custom_id).await;
-                            match query_rst {
-                                Ok(order) => {
-                                    self_clone.order_sender.send(order).await.unwrap();
-                                }
-                                Err(_query_err) => {
-                                    // error!(?_query_err);
-                                    // error!("撤单失败，而且查单也失败了，kucoin_swap，oid={}, cid={}。", order_id.clone(), custom_id.clone());
-                                }
-                            }
-                            self_clone.error_sender.send(error).await.unwrap();
-                        }
-                    }
-                }
-            });
-            cancel_handlers.push(handle)
-        }
-        let futures = FuturesUnordered::from_iter(cancel_handlers);
-        // 等待所有任务完成
-        let _: Result<Vec<_>, _> = futures.try_collect().await;
-
-        // 检查订单指令
-        let mut check_handlers = vec![];
-        for item in order_command.check.keys() {
-            let order_id = order_command.check[item].get(1).unwrap().clone();
-            let custom_id = order_command.check[item].get(0).unwrap().clone();
-
-            let mut self_clone = self.clone();
-
-            let handle = tokio::spawn(async move {
-                let result = self_clone.get_order_detail(&order_id, &custom_id).await;
-                match result {
-                    Ok(result) => {
-                        self_clone.order_sender.send(result).await.unwrap();
-                    }
-                    Err(error) => {
-                        self_clone.error_sender.send(error).await.unwrap();
-                    }
-                }
-            });
-            check_handlers.push(handle)
-        }
-
-        let futures = FuturesUnordered::from_iter(check_handlers);
-        // 等待所有任务完成
-        let _: Result<Vec<_>, _> = futures.try_collect().await;
-    }
 }
 
 pub fn format_order_item(order: Value, ct_val: Decimal) -> Order {
@@ -662,7 +547,6 @@ pub fn format_order_item(order: Value, ct_val: Decimal) -> Order {
         deal_amount,
         avg_price,
         status: custom_status,
-        order_type: order["type"].as_str().unwrap().to_string(),
-        trace_stack: TraceStack::new(0, Instant::now()).on_special("655 kucoin_swap".to_string()),
+        order_type: order["type"].as_str().unwrap().to_string()
     }
 }

standard/src/lib.rs

@@ -1,12 +1,8 @@
-use std::collections::{BTreeMap, HashMap};
-use std::fmt;
-use std::fmt::Formatter;
+use std::collections::{BTreeMap};
 use std::io::{Error};
 use async_trait::async_trait;
 use rust_decimal::Decimal;
 use serde_json::Value;
-use tokio::time::Instant;
-use global::trace_stack::TraceStack;
 
 use crate::exchange::ExchangeEnum;
 
@@ -48,49 +44,6 @@ pub enum PositionModeEnum {
     Short,
 }
 
-/// OrderCommand结构体(下单指令)
-/// - `cancel(HashMap<String, Vec<String>)`: 取消订单指令 `{"order_name": [c_id, o_id]}`;
-/// - `check(HashMap<String, Vec<String>>)`: balance挂单的冻结数量 `{"order_name": [数量，方向，价格，c_id]}`;
-/// - `limits_open(HashMap<String, Vec<String>>)`: 总计交易币数量 `{"order_name": [c_id, o_id]}`;
-/// - `limits_close(HashMap<String, Vec<String>>)`: 可用交易币数量 `{"order_name": [c_id, o_id]}`;
-#[derive(Debug, Clone, PartialEq, Eq)]
-pub struct OrderCommand {
-    // 取消订单指令，数据结构例子：
-    pub cancel: HashMap<String, Vec<String>>,
-    // 检验指令，数据结构例子：（暂没找到例子）
-    pub check: HashMap<String, Vec<String>>,
-    // 限开指令，数据结构例子：（暂没找到例子）
-    pub limits_open: HashMap<String, Vec<String>>,
-    // 限收指令，数据结构例子：（暂没找到例子）
-    pub limits_close: HashMap<String, Vec<String>>,
-}
-
-impl OrderCommand {
-    pub fn new() -> OrderCommand {
-        OrderCommand {
-            cancel: Default::default(),
-            check: Default::default(),
-            limits_open: Default::default(),
-            limits_close: Default::default(),
-        }
-    }
-    pub fn is_not_empty(&self) -> bool {
-        let is_empty = self.limits_close.is_empty() && self.limits_open.is_empty() && self.cancel.is_empty() && self.check.is_empty();
-        if is_empty { false } else { true }
-    }
-}
-
-impl fmt::Display for OrderCommand {
-    fn fmt(&self, f: &mut Formatter<'_>) -> fmt::Result {
-        write!(f, "订单触发数据： cancel: {:?}, check: {:?}, limits_open: {:?}, limits_close: {:?}",
-               self.cancel,
-               self.check,
-               self.limits_open,
-               self.limits_close
-        )
-    }
-}
-
 /// Account结构体(账户信息)
 /// - `coin(String)`: 货币;
 /// - `balance(Decimal)`: 总计计价币数量;
@@ -125,12 +78,12 @@ impl Account {
 }
 
 /// Depth结构体(市场深度)
-/// - `time(i64)`: 深度更新时间戳(ms);
+/// - `time(Decimal)`: 深度更新时间戳(ms);
 /// - `asks(Vec<MarketOrder>)`: 卖方深度列表;
 /// - `bids(Vec<MarketOrder>)`: 买方深度列表;
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub struct Depth {
-    pub time: i64,
+    pub time: Decimal,
     pub asks: Vec<MarketOrder>,
     pub bids: Vec<MarketOrder>,
 }
@@ -138,7 +91,7 @@ pub struct Depth {
 impl Depth {
     pub fn new() -> Depth {
         Depth {
-            time: 0,
+            time: Decimal::ZERO,
             asks: vec![],
             bids: vec![],
         }
@@ -284,7 +237,6 @@ pub struct Order {
     pub avg_price: Decimal,
     pub status: String,
     pub order_type: String,
-    pub trace_stack: TraceStack,
 }
 
 impl Order {
@@ -297,8 +249,7 @@ impl Order {
             deal_amount: Default::default(),
             avg_price: Default::default(),
             status: "".to_string(),
-            order_type: "".to_string(),
-            trace_stack: TraceStack::new(0, Instant::now()),
+            order_type: "".to_string()
         }
     }
 }
@@ -570,6 +521,4 @@ pub trait Platform {
     async fn set_auto_deposit_status(&mut self, status: bool) -> Result<String, Error>;
     // 交易账户互转
     async fn wallet_transfers(&mut self, coin: &str, from: &str, to: &str, amount: Decimal) -> Result<String, Error>;
-    // 指令下单
-    async fn command_order(&mut self, order_command: &mut OrderCommand, trace_stack: &TraceStack);
 }