as-rust/standard/src/binance_swap.rs

use std::collections::BTreeMap;
use std::io::{Error, ErrorKind};
use std::result::Result;
use std::str::FromStr;
use async_trait::async_trait;
use futures::stream::FuturesUnordered;
use futures::TryStreamExt;
use rust_decimal::Decimal;
use rust_decimal::prelude::FromPrimitive;
use rust_decimal_macros::dec;
use serde_json::{json, Value};
use tokio::spawn;
use tokio::sync::mpsc::Sender;
use tokio::time::Instant;
use tracing::{error, info};
use crate::{Platform, ExchangeEnum, Account, Position, Ticker, Market, Order, OrderCommand, utils, PositionModeEnum, Record};
use exchanges::binance_swap_rest::BinanceSwapRest;
use global::trace_stack::TraceStack;

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

impl BinanceSwap {
    pub async fn new(symbol: String, is_colo: bool, params: BTreeMap<String, String>, order_sender: Sender<Order>, error_sender: Sender<Error>) -> BinanceSwap {
        let market = Market::new();
        let mut binance_swap = BinanceSwap {
            exchange: ExchangeEnum::BinanceSwap,
            symbol: symbol.to_uppercase(),
            is_colo,
            params: params.clone(),
            request: BinanceSwapRest::new(is_colo, params.clone()),
            market,
            order_sender,
            error_sender,
        };

        // 修改持仓模式
        let mode_result = binance_swap.set_dual_mode("", false).await;
        match mode_result {
            Ok(ok) => {
                info!("Binance:设置持仓模式成功！{:?}", ok);
            }
            Err(error) => {
                error!("Binance:设置持仓模式失败！mode_result={}", error)
            }
        }
        // 设置持仓杠杆
        let lever_rate_result = binance_swap.set_dual_leverage("1").await;
        match lever_rate_result {
            Ok(ok) => {
                info!("Binance:设置持仓杠杆成功！{:?}", ok);
            }
            Err(error) => {
                error!("Binance:设置持仓杠杆失败！{:?}", error)
            }
        }
        // 获取市场信息
        binance_swap.market = BinanceSwap::get_market(&mut binance_swap).await.unwrap_or(binance_swap.market);
        return binance_swap;
    }
}

#[async_trait]
impl Platform for BinanceSwap {
    // 克隆方法
    fn clone_box(&self) -> Box<dyn Platform + Send + Sync> { Box::new(self.clone()) }
    // 获取交易所模式
    fn get_self_exchange(&self) -> ExchangeEnum {
        ExchangeEnum::BinanceSwap
    }
    // 获取交易对
    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_json = res_data.data;
            let result = res_data_json["serverTime"].to_string();
            Ok(result)
        } else {
            Err(Error::new(ErrorKind::Other, res_data.to_string()))
        }
    }
    // 获取账号信息
    async fn get_account(&mut self) -> Result<Account, Error> {
        let symbol_array: Vec<&str> = self.symbol.split("_").collect();
        let res_data = self.request.get_account().await;
        if res_data.code == 200 {
            let res_data_json = res_data.data.as_array().unwrap();
            let balance_info = res_data_json.iter().find(|item| item["asset"].as_str().unwrap().to_string() == symbol_array[1].to_string());
            match balance_info {
                None => {
                    error!("binance_swap：格式化账号信息错误!\nget_account: res_data={:?}", res_data);
                    Err(Error::new(ErrorKind::Other, res_data.to_string()))
                }
                Some(value) => {
                    let balance = Decimal::from_str(value["balance"].as_str().unwrap()).unwrap();
                    let available_balance = Decimal::from_str(value["availableBalance"].as_str().unwrap()).unwrap();
                    let frozen_balance = balance - available_balance;
                    let result = Account {
                        coin: value["asset"].as_str().unwrap().to_string(),
                        balance,
                        available_balance,
                        frozen_balance,
                        stocks: Decimal::ZERO,
                        available_stocks: Decimal::ZERO,
                        frozen_stocks: Decimal::ZERO,
                    };
                    Ok(result)
                }
            }
        } else {
            Err(Error::new(ErrorKind::Other, res_data.to_string()))
        }
    }

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

    // 获取仓位信息
    async fn get_position(&mut self) -> Result<Vec<Position>, Error> {
        let symbol_format = utils::format_symbol(self.symbol.clone(), "");
        let ct_val = self.market.multiplier;
        let res_data = self.request.get_position_risk(symbol_format).await;
        if res_data.code == 200 {
            let res_data_json = res_data.data.as_array().unwrap();
            let result = res_data_json.iter().map(|item| { format_position_item(item, ct_val) }).collect();
            Ok(result)
        } else {
            Err(Error::new(ErrorKind::Other, res_data.to_string()))
        }
    }

    async fn get_positions(&mut self) -> Result<Vec<Position>, Error> {
        let res_data = self.request.get_position_risk("".to_string()).await;
        if res_data.code == 200 {
            let res_data_json = res_data.data.as_array().unwrap();
            let result = res_data_json.iter().map(|item| { format_position_item(item, Decimal::ONE) }).collect();
            Ok(result)
        } else {
            Err(Error::new(ErrorKind::Other, res_data.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_book_ticker(symbol_format).await;
        if res_data.code == 200 {
            let res_data_json: serde_json::Value = res_data.data;
            let result = Ticker {
                time: Decimal::from_i64(res_data_json["time"].as_i64().unwrap()).unwrap(),
                high: Decimal::from_str(res_data_json["askPrice"].as_str().unwrap()).unwrap(),
                low: Decimal::from_str(res_data_json["bidPrice"].as_str().unwrap()).unwrap(),
                sell: Decimal::from_str(res_data_json["askPrice"].as_str().unwrap()).unwrap(),
                buy: Decimal::from_str(res_data_json["bidPrice"].as_str().unwrap()).unwrap(),
                last: dec!(-1),
                volume: dec!(-1),
                open_interest: Default::default(),
            };
            Ok(result)
        } else {
            Err(Error::new(ErrorKind::Other, res_data.to_string()))
        }
    }

    async fn get_record(&mut self, _interval: String) -> Result<Vec<Record>, Error> {
        Err(Error::new(ErrorKind::NotFound, "binance_usdt_swap：该交易所方法未实现".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_book_ticker(symbol_format).await;
        if res_data.code == 200 {
            let res_data_json: serde_json::Value = res_data.data;
            let result = Ticker {
                time: Decimal::from_i64(res_data_json["time"].as_i64().unwrap()).unwrap(),
                high: Decimal::from_str(res_data_json["askPrice"].as_str().unwrap()).unwrap(),
                low: Decimal::from_str(res_data_json["bidPrice"].as_str().unwrap()).unwrap(),
                sell: Decimal::from_str(res_data_json["askPrice"].as_str().unwrap()).unwrap(),
                buy: Decimal::from_str(res_data_json["bidPrice"].as_str().unwrap()).unwrap(),
                last: dec!(-1),
                volume: dec!(-1),
                open_interest: Default::default(),
            };
            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_exchange_info().await;
        if res_data.code == 200 {
            let res_data_json = res_data.data;
            let symbols = res_data_json["symbols"].as_array().unwrap();
            let market_info = symbols.iter().find(|&item| item["symbol"].as_str().unwrap() == symbol_format);
            match market_info {
                None => {
                    error!("binance_swap：获取Market信息错误!\nget_market:res_data={:?}", res_data_json);
                    Err(Error::new(ErrorKind::Other, res_data_json.to_string()))
                }
                Some(value) => {
                    let base_asset = value["baseAsset"].as_str().unwrap_or("").to_string();
                    let quote_asset = value["quoteAsset"].as_str().unwrap_or("").to_string();

                    let filter_array = value["filters"].as_array().unwrap().clone();
                    let price_filter = filter_array.iter().find(|&item| item["filterType"].as_str().unwrap() == "PRICE_FILTER").unwrap();
                    let lot_size_filter = filter_array.iter().find(|&item| item["filterType"].as_str().unwrap() == "LOT_SIZE").unwrap();
                    let min_notional_filter = filter_array.iter().find(|&item| item["filterType"].as_str().unwrap() == "MIN_NOTIONAL").unwrap();

                    let result = Market {
                        symbol: format!("{}_{}", base_asset, quote_asset),
                        base_asset,
                        quote_asset,
                        tick_size: Decimal::from_str(&price_filter["tickSize"].as_str().unwrap()).unwrap(),
                        amount_size: Decimal::from_str(lot_size_filter["stepSize"].as_str().unwrap()).unwrap(),
                        price_precision: Decimal::from_f64(value["pricePrecision"].as_f64().unwrap()).unwrap(),
                        amount_precision: Decimal::from_f64(value["quantityPrecision"].as_f64().unwrap()).unwrap(),
                        min_qty: Decimal::from_str(lot_size_filter["minQty"].as_str().unwrap()).unwrap(),
                        max_qty: Decimal::from_str(lot_size_filter["maxQty"].as_str().unwrap()).unwrap(),
                        min_notional: Decimal::from_str(min_notional_filter["notional"].as_str().unwrap()).unwrap(),
                        max_notional: Decimal::from_str(price_filter["maxPrice"].as_str().unwrap()).unwrap(),
                        multiplier: 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_exchange_info().await;
        if res_data.code == 200 {
            let res_data_json: serde_json::Value = res_data.data;
            let symbols = res_data_json["symbols"].as_array().unwrap();
            let market_info = symbols.iter().find(|&item| item["symbol"].as_str().unwrap() == symbol_format);
            match market_info {
                None => {
                    error!("binance_swap：获取Market信息错误!\nget_market:res_data={:?}", res_data_json);
                    Err(Error::new(ErrorKind::Other, res_data_json.to_string()))
                }
                Some(value) => {
                    let base_asset = value["baseAsset"].as_str().unwrap_or("").to_string();
                    let quote_asset = value["quoteAsset"].as_str().unwrap_or("").to_string();

                    let filter_array = value["filters"].as_array().unwrap().clone();
                    let price_filter = filter_array.iter().find(|&item| item["filterType"].as_str().unwrap() == "PRICE_FILTER").unwrap();
                    let lot_size_filter = filter_array.iter().find(|&item| item["filterType"].as_str().unwrap() == "LOT_SIZE").unwrap();

                    let result = Market {
                        symbol: format!("{}_{}", base_asset, quote_asset),
                        base_asset,
                        quote_asset,
                        tick_size: Decimal::from_str(&price_filter["tickSize"].as_str().unwrap()).unwrap(),
                        amount_size: Decimal::from_str(lot_size_filter["stepSize"].as_str().unwrap()).unwrap(),
                        price_precision: Decimal::from_f64(value["pricePrecision"].as_f64().unwrap()).unwrap(),
                        amount_precision: Decimal::from_f64(value["quantityPrecision"].as_f64().unwrap()).unwrap(),
                        min_qty: Decimal::from_str(lot_size_filter["minQty"].as_str().unwrap()).unwrap(),
                        max_qty: Decimal::from_str(lot_size_filter["maxQty"].as_str().unwrap()).unwrap(),
                        min_notional: Decimal::from_str(price_filter["minPrice"].as_str().unwrap()).unwrap(),
                        max_notional: Decimal::from_str(price_filter["maxPrice"].as_str().unwrap()).unwrap(),
                        multiplier: 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 symbol_format = utils::format_symbol(self.symbol.clone(), "");
        let mut params = json!({"symbol": symbol_format.clone()});

        if order_id != "" { params["orderId"] = json!(order_id) } else { params["origClientOrderId"] = json!(custom_id) }
        let res_data = self.request.get_order(params).await;
        if res_data.code == 200 {
            let res_data_json: Value = res_data.data;

            let status = res_data_json["status"].as_str().unwrap();
            let custom_status = if ["CANCELED", "EXPIRED", "FILLED"].contains(&status) { "REMOVE".to_string() } else if status == "NEW" { "NEW".to_string() } else {
                error!("binance_swap：格式化订单状态错误！\nget_order_detail:res_data={:?}", res_data_json);
                panic!("binance_swap：格式化订单状态错误！\nget_order_detail:res_data={:?}", res_data_json)
            };
            let result = Order {
                id: res_data_json["orderId"].to_string(),
                custom_id: res_data_json["clientOrderId"].as_str().unwrap().parse().unwrap(),
                price: Decimal::from_str(res_data_json["price"].as_str().unwrap()).unwrap(),
                amount: Decimal::from_str(res_data_json["origQty"].as_str().unwrap()).unwrap(),
                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()),
            };
            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 symbol_format = utils::format_symbol(self.symbol.clone(), "");
        let params = json!({
            "symbol":symbol_format.clone()
        });
        let res_data = self.request.get_open_orders(params).await;
        if res_data.code == 200 {
            let res_data_json = res_data.data.as_array().unwrap();
            let order_info: Vec<_> = res_data_json.iter().filter(|item| item["contract"].as_str().unwrap_or("") == self.symbol).collect();
            let result = order_info.iter().map(|&item| {
                let status = item["status"].as_str().unwrap();
                let custom_status = if ["CANCELED", "EXPIRED", "FILLED"].contains(&status) { "REMOVE".to_string() } else if status == "NEW" { "NEW".to_string() } else {
                    error!("binance_swap：格式化订单状态错误！\nget_order_detail:res_data={:?}", res_data);
                    panic!("binance_swap：格式化订单状态错误！\nget_order_detail:res_data={:?}", res_data)
                };
                Order {
                    id: item["orderId"].to_string(),
                    custom_id: item["clientOrderId"].as_str().unwrap().parse().unwrap(),
                    price: Decimal::from_str(item["price"].as_str().unwrap()).unwrap(),
                    amount: Decimal::from_str(item["origQty"].as_str().unwrap()).unwrap(),
                    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()),
                }
            }).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> {
        self.take_order_symbol(self.symbol.clone(), self.market.multiplier, custom_id, origin_side, price, amount).await
    }

    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!({
            "newClientOrderId": custom_id.to_string(),
            "symbol": symbol_format,
        });
        let size = amount / ct_val;
        params["quantity"] = json!(size);
        if price.eq(&Decimal::ZERO) {
            params["type"] = json!("MARKET");
        } else {
            params["price"] = json!(price.to_string());
            params["type"] = json!("LIMIT");
            params["timeInForce"] = json!("GTC");
        };
        match origin_side {
            "kd" => {
                params["side"] = json!("BUY");
                // params["positionSide"] = json!("LONG");
            }
            "pd" => {
                params["side"] = json!("SELL");
                params["reduceOnly"] = json!(true);
                // params["positionSide"] = json!("LONG");
            }
            "kk" => {
                params["side"] = json!("SELL");
                // params["positionSide"] = json!("SHORT");
            }
            "pk" => {
                params["side"] = json!("BUY");
                params["reduceOnly"] = json!(true);
                // params["positionSide"] = json!("SHORT");
            }
            _ => { error!("下单参数错误"); }
        };
        let res_data = self.request.swap_order(params).await;
        if res_data.code == 200 {
            let res_data_json: Value = res_data.data;
            let id = res_data_json["orderId"].to_string();
            let custom_id = res_data_json["clientOrderId"].as_str().unwrap().to_string();
            let price = Decimal::from_str(res_data_json["price"].as_str().unwrap()).unwrap();
            let amount = Decimal::from_str(res_data_json["origQty"].as_str().unwrap()).unwrap();
            let deal_amount = Decimal::from_str(res_data_json["executedQty"].as_str().unwrap()).unwrap();
            let avg_price = Decimal::from_str(res_data_json["avgPrice"].as_str().unwrap()).unwrap();
            let result = Order {
                id,
                custom_id,
                price,
                amount,
                deal_amount,
                avg_price,
                status: "NEW".to_string(),
                order_type: "".to_string(),
                trace_stack: TraceStack::new(0, Instant::now()).on_special("387 binance_swap".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 symbol_format = utils::format_symbol(self.symbol.clone(), "");
        let mut params = json!({
            "symbol": symbol_format.clone(),
        });
        if order_id != "" { params["orderId"] = json!(order_id) }
        if custom_id != "" { params["origClientOrderId"] = json!(custom_id) }
        // println!("{}", params);
        let res_data = self.request.cancel_order(params).await;
        if res_data.code == 200 {
            let res_data_json = &res_data.data;
            let cum_quote = Decimal::from_str(res_data_json["cumQuote"].as_str().unwrap()).unwrap();
            let id = res_data_json["orderId"].to_string();
            let custom_id = res_data_json["clientOrderId"].as_str().unwrap().to_string();
            let price = Decimal::from_str(res_data_json["price"].as_str().unwrap()).unwrap();
            let amount = Decimal::from_str(res_data_json["origQty"].as_str().unwrap()).unwrap();
            let deal_amount = Decimal::from_str(res_data_json["executedQty"].as_str().unwrap()).unwrap();
            let avg_price = if cum_quote > Decimal::ZERO { deal_amount / cum_quote } else { Decimal::ZERO };
            let result = Order {
                id,
                custom_id,
                price,
                amount,
                deal_amount,
                avg_price,
                status: "REMOVE".to_string(),
                order_type: "".to_string(),
                trace_stack: TraceStack::new(0, Instant::now()).on_special("419 binance_swap".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 params = json!({
            "symbol": symbol_format.clone(),
        });
        let res_data = self.request.cancel_order_all(params).await;
        println!("{:?}", res_data);
        if res_data.code == 200 {
            let result = vec![Order {
                id: "".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: "limit".to_string(),
                trace_stack: TraceStack::new(0, Instant::now()).on_special("484 trace_stack".to_string()),
            }];
            Ok(result)
        } else {
            Err(Error::new(ErrorKind::Other, res_data.to_string()))
        }
    }

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


    async fn take_stop_loss_order(&mut self, _stop_price: Decimal, _price: Decimal, _side: &str) -> Result<Value, Error> {
        Err(Error::new(ErrorKind::NotFound, "binance_swap：该交易所方法未实现".to_string()))
    }

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

    async fn set_dual_mode(&mut self, _coin: &str, is_dual_mode: bool) -> Result<String, Error> {
        let res_data = self.request.change_pos_side(is_dual_mode).await;
        if res_data.code == 200 {
            let res_data_str = &res_data.data;
            let result = res_data_str.clone();
            Ok(result.to_string())
        } else {
            Err(Error::new(ErrorKind::Other, res_data.to_string()))
        }
    }

    async fn set_dual_leverage(&mut self, leverage: &str) -> Result<String, Error> {
        let symbol_format = utils::format_symbol(self.symbol.clone(), "");
        let params = json!({
            "symbol": symbol_format,
            "leverage" : leverage,
        });
        let res_data = self.request.setting_dual_leverage(params).await;
        if res_data.code == 200 {
            let res_data_str = res_data.data;
            let result = res_data_str.clone();
            Ok(result.to_string())
        } else {
            Err(Error::new(ErrorKind::Other, res_data.to_string()))
        }
    }

    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) {
        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 {
    let mut position_mode = match position["positionSide"].as_str().unwrap_or("") {
        "BOTH" => PositionModeEnum::Both,
        "LONG" => PositionModeEnum::Long,
        "SHORT" => PositionModeEnum::Short,
        _ => {
            error!("binance_swap：格式化持仓模式错误！\nformat_position_item:position={:?}", position);
            panic!("binance_swap：格式化持仓模式错误！\nformat_position_item:position={:?}", position)
        }
    };
    let size = Decimal::from_str(position["positionAmt"].as_str().unwrap()).unwrap();
    let amount = size * ct_val;
    match position_mode {
        PositionModeEnum::Both => {
            position_mode = match amount {
                amount if amount > Decimal::ZERO => PositionModeEnum::Long,
                amount if amount < Decimal::ZERO => PositionModeEnum::Short,
                _ => { PositionModeEnum::Both }
            };
        }
        _ => {}
    }
    Position {
        symbol: position["symbol"].as_str().unwrap_or("").parse().unwrap(),
        margin_level: Decimal::from_str(position["leverage"].as_str().unwrap()).unwrap(),
        amount,
        frozen_amount: Decimal::ZERO,
        price: Decimal::from_str(position["entryPrice"].as_str().unwrap()).unwrap(),
        profit: Decimal::from_str(position["unRealizedProfit"].as_str().unwrap()).unwrap(),
        position_mode,
        margin: Decimal::from_str(position["isolatedMargin"].as_str().unwrap()).unwrap(),
    }
}