as-rust/standard/src/kucoin_spot.rs

use std::collections::{BTreeMap, HashMap};
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;
use serde::{Deserialize, Serialize};
use serde_json::json;
use tracing::error;
use exchanges::kucoin_spot_rest::KucoinSpotRest;
use global::trace_stack::TraceStack;
use crate::exchange::ExchangeEnum;
use crate::{Account, Market, Order, OrderCommand, Platform, Position, Ticker, utils};

/// Kucoin交易所账户信息请求数据结构
/// 接口`"/api/v1/accounts"`;
///
/// struct SpotAccount
/// - `id`: String, accountId
/// - `currency`: String, 币种
/// - `account_type`: String, 账户类型，资金(main)账户，现货交易(trade)账户，现货高频交易(trade_hf)账户，杠杆(margin)账户
/// - `balance`: Decimal, 资金总额
/// - `available`: Decimal, 可用余额
/// - `holds`: Decimal, 冻结金额
#[derive(Debug, Deserialize, Serialize)]
#[serde(rename_all = "camelCase")]
struct SpotAccount {
    id: String,
    currency: String,
    account_type: String,
    balance: Decimal,
    available: Decimal,
    holds: Decimal,
}

impl SpotAccount {
    fn new() -> SpotAccount {
        SpotAccount {
            id: "".to_string(),
            currency: "".to_string(),
            account_type: "".to_string(),
            balance: Default::default(),
            available: Default::default(),
            holds: Default::default(),
        }
    }
}

/// Kucoin交易所Ticker信息请求数据结构
/// 接口`"/api/v1/market/orderbook/level1"`;
///
/// struct SpotTicker
/// - `sequence`: String, 序列号
/// - `price`: Decimal, 最新成交价格
/// - `size`: Decimal, 最新成交数量
/// - `best_ask`: Decimal, 最佳卖一价
/// - `best_ask_size`: Decimal, 最佳卖一数量
/// - `best_bid`: Decimal, 最佳买一价
/// - `best_bid_size`: Decimal, 最佳买一数量
/// - `time`: i64, 时间戳
#[derive(Debug, Deserialize, Serialize)]
#[serde(rename_all = "camelCase")]
struct SpotTicker {
    sequence: String,
    price: Decimal,
    size: Decimal,
    best_ask: Decimal,
    best_ask_size: Decimal,
    best_bid: Decimal,
    best_bid_size: Decimal,
    time: i64,
}

/// Kucoin交易所Market信息请求数据结构
/// 接口`"/api/v2/symbols"`;
///
/// struct SpotTicker
/// - `symbol: String, 交易对唯一标识码
/// - `name: String, 交易对名称
/// - `base_currency: String, 商品货币
/// - `quote_currency: String, 计价币种
/// - `fee_currency: String, 交易计算手续费的币种
/// - `market: String, 交易市场
/// - `base_min_size: Decimal, 下单时size的最小值
/// - `quote_min_size: Decimal, 下市价单，funds的最小值
/// - `base_max_size: Decimal, 下单，size的最大值
/// - `quote_max_size: Decimal, 下市价单，funds的最大值
/// - `base_increment: Decimal, 数量增量，下单的size必须为数量增量的正整数倍
/// - `quote_increment: Decimal, 市价单：资金增量，下单的funds必须为资金增量的正整数倍
/// - `price_increment: Decimal, 限价单：价格增量，下单的price必须为价格增量的正整数倍
/// - `price_limit_rate: Decimal, 价格保护阈值
/// - `min_funds: Option<Decimal>, 最小交易金额
/// - `is_margin_enabled: bool, 是否支持杠杆
/// - `enable_trading: bool, 是否可以用于交易
#[derive(Debug, Deserialize, Serialize)]
#[serde(rename_all = "camelCase")]
struct SpotMarket {
    symbol: String,
    name: String,
    base_currency: String,
    quote_currency: String,
    fee_currency: String,
    market: String,
    base_min_size: Decimal,
    quote_min_size: Decimal,
    base_max_size: Decimal,
    quote_max_size: Decimal,
    base_increment: Decimal,
    quote_increment: Decimal,
    price_increment: Decimal,
    price_limit_rate: Decimal,
    min_funds: Option<Decimal>,
    is_margin_enabled: bool,
    enable_trading: bool,
}

/// Kucoin交易所Order信息请求数据结构
/// 接口`"/api/v1/orders/{orderId}"`;
///
/// struct SpotOrder
/// - `id`: String,
/// - `symbol`: String,
/// - `op_type`: String,
/// - `order_type`: String,
/// - `side`: String,
/// - `price`: Decimal,
/// - `size`: Decimal,
/// - `funds`: Decimal,
/// - `deal_funds`: Decimal,
/// - `deal_size`: Decimal,
/// - `fee`: Decimal,
/// - `fee_currency`: String,
/// - `stp`: String,
/// - `stop`: String,
/// - `stop_triggered`: bool,
/// - `stop_price`: Decimal,
/// - `time_in_force`: String,
/// - `post_only`: bool,
/// - `hidden`: bool,
/// - `iceberg`: bool,
/// - `visible_size`: Decimal,
/// - `cancel_after`: i64,
/// - `channel`: String,
/// - `client_oid`: String,
/// - `remark`: String,
/// - `tags`: String,
/// - `is_active`: bool,
/// - `cancel_exist`: bool,
/// - `created_at`: i64,
/// - `trade_type`: String,
#[derive(Debug, Deserialize, Serialize)]
#[serde(rename_all = "camelCase")]
struct SpotOrder {
    id: String,
    symbol: String,
    op_type: String,
    #[serde(rename = "type")]
    order_type: String,
    side: String,
    price: Decimal,
    size: Decimal,
    funds: Decimal,
    deal_funds: Decimal,
    deal_size: Decimal,
    fee: Decimal,
    fee_currency: String,
    stp: String,
    stop: String,
    stop_triggered: bool,
    stop_price: Decimal,
    time_in_force: String,
    post_only: bool,
    hidden: bool,
    iceberg: bool,
    visible_size: Decimal,
    cancel_after: i64,
    channel: String,
    client_oid: String,
    remark: String,
    tags: String,
    is_active: bool,
    cancel_exist: bool,
    created_at: i64,
    trade_type: String,
}

#[allow(dead_code)]
#[derive(Clone)]
pub struct KucoinSpot {
    exchange: ExchangeEnum,
    symbol: String,
    is_colo: bool,
    params: BTreeMap<String, String>,
    request: KucoinSpotRest,
    market: Market,
    order_sender: Sender<Order>,
    error_sender: Sender<Error>,
}

impl KucoinSpot {
    pub async fn new(symbol: String, is_colo: bool, params: BTreeMap<String, String>, order_sender: Sender<Order>, error_sender: Sender<Error>) -> KucoinSpot {
        let market = Market::new();
        let mut kucoin_spot = KucoinSpot {
            exchange: ExchangeEnum::KucoinSpot,
            symbol: symbol.to_uppercase(),
            is_colo,
            params: params.clone(),
            request: KucoinSpotRest::new(is_colo, params.clone()),
            market,
            order_sender,
            error_sender,
        };
        kucoin_spot.market = KucoinSpot::get_market(&mut kucoin_spot).await.unwrap_or(kucoin_spot.market);

        return kucoin_spot;
    }
}

#[async_trait]
impl Platform for KucoinSpot {
    // 克隆方法
    fn clone_box(&self) -> Box<dyn Platform + Send + Sync> { Box::new(self.clone()) }
    fn get_self_exchange(&self) -> ExchangeEnum { ExchangeEnum::KucoinSpot }
    // 获取交易对
    fn get_self_symbol(&self) -> String { self.symbol.clone() }
    // 获取是否使用高速通道
    fn get_self_is_colo(&self) -> bool {
        self.is_colo
    }
    // 获取params信息
    fn get_self_params(&self) -> BTreeMap<String, String> {
        self.params.clone()
    }
    // 获取market信息
    fn get_self_market(&self) -> Market { self.market.clone() }
    // 获取请求时间
    fn get_request_delays(&self) -> Vec<i64> { self.request.get_delays() }
    // 获取请求平均时间
    fn get_request_avg_delay(&self) -> Decimal { self.request.get_avg_delay() }
    // 获取请求最大时间
    fn get_request_max_delay(&self) -> i64 { self.request.get_max_delay() }
    // 获取服务器时间
    async fn get_server_time(&mut self) -> Result<String, Error> {
        let res_data = self.request.get_server_time().await;
        if res_data.code == "200" {
            let res_data_str = &res_data.data;
            let result = res_data_str.clone();
            Ok(result)
        } else {
            Err(Error::new(ErrorKind::Other, res_data.to_string()))
        }
    }

    async fn get_account(&mut self) -> Result<Account, Error> {
        let coin_array: Vec<&str> = self.symbol.split("_").collect();
        let res_data = self.request.get_accounts(coin_array[1].to_string()).await;
        if res_data.code == "200" {
            let res_data_str = &res_data.data;
            let balance_info_list: Vec<SpotAccount> = serde_json::from_str(res_data_str).unwrap();
            let mut balance_info_default = SpotAccount::new();
            balance_info_default.currency = coin_array[1].to_string();
            let balance_info = balance_info_list.iter().find(|&item| item.currency == coin_array[1].to_string()).unwrap_or(&balance_info_default);
            let mut stocks_info_default = SpotAccount::new();
            stocks_info_default.currency = coin_array[0].to_string();
            let stocks_info = balance_info_list.iter().find(|&item| item.currency == coin_array[0].to_string()).unwrap_or(&balance_info_default);
            let result = Account {
                coin: format!("{}_{}", balance_info.currency, stocks_info.currency),
                balance: balance_info.available + balance_info.holds,
                available_balance: balance_info.available,
                frozen_balance: balance_info.holds,
                stocks: stocks_info.available + stocks_info.holds,
                available_stocks: stocks_info.available,
                frozen_stocks: stocks_info.holds,
            };
            Ok(result)
        } else {
            Err(Error::new(ErrorKind::Other, res_data.to_string()))
        }
    }

    async fn get_spot_account(&mut self) -> Result<Vec<Account>, Error> {
        let res_data = self.request.get_accounts("".to_string()).await;
        if res_data.code == "200" {
            let res_data_str = &res_data.data;
            let balance_info_list: Vec<SpotAccount> = serde_json::from_str(res_data_str).unwrap();
            let result = balance_info_list.iter().map(|item| {
                Account {
                    coin: item.currency.to_string(),
                    balance: item.available + item.holds,
                    available_balance: item.available,
                    frozen_balance: item.holds,
                    stocks: Decimal::ZERO,
                    available_stocks: Decimal::ZERO,
                    frozen_stocks: Decimal::ZERO,
                }
            }).collect();
            Ok(result)
        } else {
            Err(Error::new(ErrorKind::Other, res_data.to_string()))
        }
    }

    async fn get_position(&mut self) -> Result<Vec<Position>, Error> {
        Err(Error::new(ErrorKind::NotFound, "kucoin_spot：该交易所方法未实现".to_string()))
    }

    async fn get_positions(&mut self) -> Result<Vec<Position>, Error> {
        Err(Error::new(ErrorKind::NotFound, "kucoin_spot：该交易所方法未实现".to_string()))
    }

    async fn get_ticker(&mut self) -> Result<Ticker, Error> {
        let symbol_format = utils::format_symbol(self.symbol.clone(), "-");
        let res_data = self.request.get_level1(symbol_format).await;
        if res_data.code == "200" {
            let res_data_str = &res_data.data;
            let ticker_info: SpotTicker = serde_json::from_str(res_data_str).unwrap();
            let result = Ticker {
                time: ticker_info.time,
                high: ticker_info.best_ask,
                low: ticker_info.best_bid,
                sell: ticker_info.best_ask,
                buy: ticker_info.best_bid,
                last: ticker_info.price,
                volume: ticker_info.size,
            };
            Ok(result)
        } else {
            Err(Error::new(ErrorKind::Other, res_data.to_string()))
        }
    }

    async fn get_ticker_symbol(&mut self, symbol: String) -> Result<Ticker, Error> {
        let symbol_format = utils::format_symbol(symbol.clone(), "-");
        let res_data = self.request.get_level1(symbol_format).await;
        if res_data.code == "200" {
            let res_data_str = &res_data.data;
            let ticker_info: SpotTicker = serde_json::from_str(res_data_str).unwrap();
            let result = Ticker {
                time: ticker_info.time,
                high: ticker_info.best_ask,
                low: ticker_info.best_bid,
                sell: ticker_info.best_ask,
                buy: ticker_info.best_bid,
                last: ticker_info.price,
                volume: ticker_info.size,
            };
            Ok(result)
        } else {
            Err(Error::new(ErrorKind::Other, res_data.to_string()))
        }
    }

    async fn get_market(&mut self) -> Result<Market, Error> {
        let symbol_format = utils::format_symbol(self.symbol.clone(), "-");
        let res_data = self.request.get_symbols().await;
        if res_data.code == "200" {
            let res_data_str = &res_data.data;
            let market_info_list: Vec<SpotMarket> = serde_json::from_str(res_data_str).unwrap();
            let market_info = market_info_list.iter().find(|&item| item.symbol == symbol_format).unwrap();
            let result = Market {
                symbol: market_info.symbol.replace("-", "_"),
                base_asset: market_info.base_currency.clone(),
                quote_asset: market_info.quote_currency.clone(),
                tick_size: market_info.price_increment,
                amount_size: market_info.base_increment,
                price_precision: Decimal::from_u32(market_info.price_increment.scale()).unwrap(),
                amount_precision: Decimal::from_u32(market_info.base_increment.scale()).unwrap(),
                min_qty: market_info.base_min_size,
                max_qty: market_info.base_max_size,
                min_notional: market_info.price_increment * market_info.base_min_size,
                max_notional: market_info.price_increment * market_info.base_max_size,
                ct_val: Decimal::ONE,
            };
            Ok(result)
        } else {
            Err(Error::new(ErrorKind::Other, res_data.to_string()))
        }
    }

    async fn get_market_symbol(&mut self, symbol: String) -> Result<Market, Error> {
        let symbol_format = utils::format_symbol(symbol.clone(), "-");
        let res_data = self.request.get_symbols().await;
        if res_data.code == "200" {
            let res_data_str = &res_data.data;
            let market_info_list: Vec<SpotMarket> = serde_json::from_str(res_data_str).unwrap();
            let market_info = market_info_list.iter().find(|&item| item.symbol == symbol_format).unwrap();
            let result = Market {
                symbol: market_info.symbol.replace("-", "_"),
                base_asset: market_info.base_currency.clone(),
                quote_asset: market_info.quote_currency.clone(),
                tick_size: market_info.price_increment,
                amount_size: market_info.base_increment,
                price_precision: Decimal::from_u32(market_info.price_increment.scale()).unwrap(),
                amount_precision: Decimal::from_u32(market_info.base_increment.scale()).unwrap(),
                min_qty: market_info.base_min_size,
                max_qty: market_info.base_max_size,
                min_notional: market_info.price_increment * market_info.base_min_size,
                max_notional: market_info.price_increment * market_info.base_max_size,
                ct_val: Decimal::ONE,
            };
            Ok(result)
        } else {
            Err(Error::new(ErrorKind::Other, res_data.to_string()))
        }
    }

    async fn get_order_detail(&mut self, order_id: &str, custom_id: &str) -> Result<Order, Error> {
        let res_data = if order_id != "" { self.request.get_order_by_order_id(order_id.to_string()).await } else { self.request.get_order_by_client_id(custom_id.to_string()).await };
        if res_data.code == "200" {
            let res_data_str = &res_data.data;
            let result = format_order_item(res_data_str.clone());
            Ok(result)
        } else {
            Err(Error::new(ErrorKind::Other, res_data.to_string()))
        }
    }

    async fn get_orders_list(&mut self, _status: &str) -> Result<Vec<Order>, Error> {
        let res_data = self.request.get_order().await;
        if res_data.code == "200" {
            let res_data_str = &res_data.data;
            let order_info_list: Vec<String> = serde_json::from_str(res_data_str).unwrap();
            let result = order_info_list.iter().map(|item| format_order_item(item.clone())).collect();
            Ok(result)
        } else {
            Err(Error::new(ErrorKind::Other, res_data.to_string()))
        }
    }

    async fn take_order(&mut self, custom_id: &str, origin_side: &str, price: Decimal, amount: Decimal) -> Result<Order, Error> {
        let symbol_format = utils::format_symbol(self.symbol.clone(), "-");
        let mut params = json!({
            "symbol": symbol_format.to_string(),
            "clientOid": custom_id,
            "price": price.to_string(),
            "size": amount.to_string()
        });
        if price.eq(&Decimal::ZERO) {
            params["type"] = json!("market");
        } else {
            params["type"] = json!("limit");
        };
        match origin_side {
            "kd" => {
                params["side"] = json!("buy");
            }
            "pd" => {
                params["side"] = json!("sell");
            }
            "kk" => {
                params["side"] = json!("sell");
            }
            "pk" => {
                params["side"] = json!("buy");
            }
            _ => { error!("下单参数错误"); }
        };
        let res_data = self.request.spot_order(params).await;
        if res_data.code == "200" {
            let res_data_str = &res_data.data;
            let res_data_json: serde_json::Value = serde_json::from_str(res_data_str).unwrap();
            let result = Order {
                id: res_data_json["orderId"].as_str().unwrap().to_string(),
                custom_id: custom_id.to_string(),
                price: Decimal::ZERO,
                amount: Decimal::ZERO,
                deal_amount: Decimal::ZERO,
                avg_price: Decimal::ZERO,
                status: "NEW".to_string(),
                order_type: "".to_string(),
                trace_stack: TraceStack::default().on_special("550 kucoin_spot".to_string()),
            };
            Ok(result)
        } else {
            Err(Error::new(ErrorKind::Other, res_data.to_string()))
        }
    }

    async fn take_order_symbol(&mut self, symbol: String, ct_val: Decimal, custom_id: &str, origin_side: &str, price: Decimal, amount: Decimal) -> Result<Order, Error> {
        let symbol_format = utils::format_symbol(symbol.clone(), "-");
        let mut params = json!({
            "symbol": symbol_format.to_string(),
            "clientOid": custom_id,
            "price": price.to_string(),
            "size": amount * ct_val,
        });
        if price.eq(&Decimal::ZERO) {
            params["type"] = json!("market");
        } else {
            params["type"] = json!("limit");
        };
        match origin_side {
            "kd" => {
                params["side"] = json!("buy");
            }
            "pd" => {
                params["side"] = json!("sell");
            }
            "kk" => {
                params["side"] = json!("sell");
            }
            "pk" => {
                params["side"] = json!("buy");
            }
            _ => { error!("下单参数错误"); }
        };
        let res_data = self.request.spot_order(params).await;
        if res_data.code == "200" {
            let res_data_str = &res_data.data;
            let res_data_json: serde_json::Value = serde_json::from_str(res_data_str).unwrap();
            let result = Order {
                id: res_data_json["orderId"].as_str().unwrap().to_string(),
                custom_id: custom_id.to_string(),
                price: Decimal::ZERO,
                amount: Decimal::ZERO,
                deal_amount: Decimal::ZERO,
                avg_price: Decimal::ZERO,
                status: "NEW".to_string(),
                order_type: "".to_string(),
                trace_stack: TraceStack::default().on_special("599 kucoin_spot".to_string()),
            };
            Ok(result)
        } else {
            Err(Error::new(ErrorKind::Other, res_data.to_string()))
        }
    }

    async fn cancel_order(&mut self, order_id: &str, custom_id: &str) -> Result<Order, Error> {
        let res_data = if order_id != "" { self.request.cancel_order_by_order_id(order_id.to_string()).await } else { self.request.cancel_order_by_client_id(custom_id.to_string()).await };
        if res_data.code == "200" {
            let res_data_str = &res_data.data;
            let order_info: serde_json::Value = serde_json::from_str(res_data_str).unwrap();
            let id = if order_id != "" { order_info["cancelledOrderIds"][0].as_str().unwrap().to_string() } else { order_info["cancelledOrderId"].as_str().unwrap().to_string() };
            let custom_id = if order_id != "" { "".to_string() } else { order_info["clientOid"].as_str().unwrap().to_string() };
            let result = Order {
                id,
                custom_id,
                price: Decimal::ZERO,
                amount: Decimal::ZERO,
                deal_amount: Decimal::ZERO,
                avg_price: Decimal::ZERO,
                status: "REMOVE".to_string(),
                order_type: "".to_string(),
                trace_stack: TraceStack::default().on_special("623 kucoin_spot".to_string()),
            };
            Ok(result)
        } else {
            Err(Error::new(ErrorKind::Other, res_data.to_string()))
        }
    }

    async fn cancel_orders(&mut self) -> Result<Vec<Order>, Error> {
        let symbol_format = utils::format_symbol(self.symbol.clone(), "-");
        let res_data = self.request.cancel_order_all(symbol_format).await;
        if res_data.code == "200" {
            let res_data_str = &res_data.data;
            let order_info: serde_json::Value = serde_json::from_str(res_data_str).unwrap();
            let id_list = order_info["cancelledOrderIds"].as_array().unwrap();
            let result = id_list.iter().map(|item| Order {
                id: item["id"].as_str().unwrap().to_string(),
                custom_id: "".to_string(),
                price: Decimal::ZERO,
                amount: Decimal::ZERO,
                deal_amount: Decimal::ZERO,
                avg_price: Decimal::ZERO,
                status: "REMOVE".to_string(),
                order_type: "".to_string(),
                trace_stack: TraceStack::default().on_special("647 kucoin_spot".to_string()),
            }).collect();
            Ok(result)
        } else {
            Err(Error::new(ErrorKind::Other, res_data.to_string()))
        }
    }

    async fn cancel_orders_all(&mut self) -> Result<Vec<Order>, Error> {
        let res_data = self.request.cancel_order_all("".to_string()).await;
        if res_data.code == "200" {
            let res_data_str = &res_data.data;
            let order_info: serde_json::Value = serde_json::from_str(res_data_str).unwrap();
            let id_list = order_info["cancelledOrderIds"].as_array().unwrap();
            let result = id_list.iter().map(|item| Order {
                id: item["id"].as_str().unwrap().to_string(),
                custom_id: "".to_string(),
                price: Decimal::ZERO,
                amount: Decimal::ZERO,
                deal_amount: Decimal::ZERO,
                avg_price: Decimal::ZERO,
                status: "REMOVE".to_string(),
                order_type: "".to_string(),
                trace_stack: TraceStack::default().on_special("670 kucoin_spot".to_string()),
            }).collect();
            Ok(result)
        } else {
            Err(Error::new(ErrorKind::Other, res_data.to_string()))
        }
    }

    async fn set_dual_mode(&mut self, _coin: &str, _is_dual_mode: bool) -> Result<String, Error> {
        Err(Error::new(ErrorKind::NotFound, "kucoin_spot：该交易所方法未实现".to_string()))
    }

    async fn set_dual_leverage(&mut self, _leverage: &str) -> Result<String, Error> {
        Err(Error::new(ErrorKind::NotFound, "kucoin_spot：该交易所方法未实现".to_string()))
    }

    async fn set_auto_deposit_status(&mut self, _status: bool) -> Result<String, Error> {
        Err(Error::new(ErrorKind::NotFound, "kucoin_spot：该交易所方法未实现".to_string()))
    }

    async fn wallet_transfers(&mut self, _coin: &str, _from: &str, _to: &str, _amount: Decimal) -> Result<String, Error> {
        Err(Error::new(ErrorKind::NotFound, "kucoin_spot：该交易所方法未实现".to_string()))
    }

    async fn command_order(&mut self, order_command: OrderCommand, trace_stack: TraceStack) {
        let mut handles = vec![];
        // 撤销订单
        let cancel = order_command.cancel;
        for item in cancel.keys() {
            let mut self_clone = self.clone();
            let cancel_clone = cancel.clone();
            let item_clone = item.clone();
            let order_id = cancel_clone[&item_clone].get(1).unwrap_or(&"".to_string()).clone();
            let custom_id = cancel_clone[&item_clone].get(0).unwrap_or(&"".to_string()).clone();
            let result_sd = self.order_sender.clone();
            let err_sd = self.error_sender.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) => {
                                result_sd.send(order).await.unwrap();
                            }
                            Err(_query_err) => {
                                // error!(?_query_err);
                                // error!("撤单失败，而且查单也失败了，bitget_spot，oid={}, cid={}。", order_id.clone(), custom_id.clone());
                            }
                        }
                        err_sd.send(error).await.unwrap();
                    }
                }
            });
            handles.push(handle)
        }
        // 下单指令
        let mut limits = HashMap::new();
        limits.extend(order_command.limits_open);
        limits.extend(order_command.limits_close);
        for item in limits.keys() {
            let mut self_clone = self.clone();
            let limits_clone = limits.clone();
            let item_clone = item.clone();
            let result_sd = self.order_sender.clone();
            let err_sd = self.error_sender.clone();
            let mut ts = trace_stack.clone();

            let handle = tokio::spawn(async move {
                let value = limits_clone[&item_clone].clone();
                let amount = Decimal::from_str(value.get(0).unwrap_or(&"0".to_string())).unwrap();
                let side = value.get(1).unwrap();
                let price = Decimal::from_str(value.get(2).unwrap_or(&"0".to_string())).unwrap();
                let cid = value.get(3).unwrap();

                //  order_name: [数量，方向，价格，c_id]
                let result = self_clone.take_order(cid, side, price, amount).await;
                match result {
                    Ok(mut result) => {
                        // 记录此订单完成时间
                        ts.on_after_send();
                        result.trace_stack = ts;

                        result_sd.send(result).await.unwrap();
                    }
                    Err(error) => {
                        let mut err_order = Order::new();
                        err_order.custom_id = cid.clone();
                        err_order.status = "REMOVE".to_string();

                        result_sd.send(err_order).await.unwrap();
                        err_sd.send(error).await.unwrap();
                    }
                }
            });
            handles.push(handle)
        }
        // 检查订单指令
        let check = order_command.check;
        for item in check.keys() {
            let mut self_clone = self.clone();
            let check_clone = check.clone();
            let item_clone = item.clone();
            let order_id = check_clone[&item_clone].get(1).unwrap_or(&"".to_string()).clone();
            let custom_id = check_clone[&item_clone].get(0).unwrap_or(&"".to_string()).clone();
            let result_sd = self.order_sender.clone();
            let err_sd = self.error_sender.clone();
            let handle = tokio::spawn(async move {
                let result = self_clone.get_order_detail(&order_id, &custom_id).await;
                match result {
                    Ok(result) => {
                        result_sd.send(result).await.unwrap();
                    }
                    Err(error) => {
                        err_sd.send(error).await.unwrap();
                    }
                }
            });
            handles.push(handle)
        }

        let futures = FuturesUnordered::from_iter(handles);
        let _: Result<Vec<_>, _> = futures.try_collect().await;
    }
}

pub fn format_order_item(data: String) -> Order {
    let order_info: SpotOrder = serde_json::from_str(&data).unwrap();
    Order {
        id: order_info.id,
        custom_id: order_info.client_oid,
        price: order_info.price,
        amount: order_info.size,
        deal_amount: order_info.deal_size,
        avg_price: order_info.deal_funds / order_info.deal_size,
        status: if order_info.is_active { "NEW".to_string() } else { "REMOVE".to_string() },
        order_type: order_info.order_type,
        trace_stack: TraceStack::default().on_special("811 kucoin_spot".to_string()),
    }
}