1.添加binance交易所数据

2.排名接口返回币安是否存在字段

exchanges/src/binance_swap_rest.rs

@@ -0,0 +1,471 @@
+use std::collections::BTreeMap;
+use reqwest::header::HeaderMap;
+use hex;
+use reqwest::Client;
+use rust_decimal::Decimal;
+use rust_decimal::prelude::{FromPrimitive, ToPrimitive};
+use rust_decimal_macros::dec;
+use crate::http_tool::RestTool;
+use crate::response_base::ResponseData;
+use tracing::{error, info, trace};
+use ring::hmac;
+use serde_json::{json, Value};
+
+#[derive(Clone)]
+pub struct BinanceSwapRest {
+    tag: String,
+    base_url: String,
+    client: reqwest::Client,
+    /*******参数*/
+    //登录所需参数
+    login_param: BTreeMap<String, String>,
+    delays: Vec<i64>,
+    max_delay: i64,
+    avg_delay: Decimal,
+}
+
+impl BinanceSwapRest {
+    /*******************************************************************************************************/
+    /*****************************************获取一个对象****************************************************/
+    /*******************************************************************************************************/
+    pub fn new(is_colo: bool, login_param: BTreeMap<String, String>) -> BinanceSwapRest
+    {
+        return BinanceSwapRest::new_with_tag("default-BinanceSwapRest".to_string(), is_colo, login_param);
+    }
+    pub fn new_with_tag(tag: String, is_colo: bool, login_param: BTreeMap<String, String>) -> BinanceSwapRest
+    {
+        let base_url = if is_colo {
+            "https://fapi.binance.com".to_string()
+        } else {
+            "https://fapi.binance.com".to_string()
+        };
+
+        if is_colo {
+            info!("开启高速(未配置，走普通：{})通道",base_url);
+        } else {
+            info!("走普通通道：{}",base_url);
+        }
+        /*****返回结构体*******/
+        BinanceSwapRest {
+            tag,
+            base_url: base_url.to_string(),
+            client: Client::new(),
+            login_param,
+            delays: vec![],
+            max_delay: 0,
+            avg_delay: dec!(0.0),
+        }
+    }
+
+    /*******************************************************************************************************/
+    /*****************************************rest请求函数********************************************************/
+    /*******************************************************************************************************/
+    //获取系统时间
+    pub async fn get_server_time(&mut self) -> ResponseData {
+        let params = json!({
+         });
+
+        let data = self.request("GET".to_string(),
+                                "".to_string(),
+                                "/fapi/v1/time".to_string(),
+                                true,
+                                params.to_string(),
+        ).await;
+        data
+    }
+    //获取交易规则和交易对
+    pub async fn get_exchange_info(&mut self) -> ResponseData {
+        let params = json!({});
+
+        let data = self.request("GET".to_string(),
+                                "".to_string(),
+                                "/fapi/v1/exchangeInfo".to_string(),
+                                false,
+                                params.to_string(),
+        ).await;
+        data
+    }
+    // 获取深度信息
+    pub async fn get_order_book(&mut self, symbol: &str, limit: i64) -> ResponseData {
+        let params = json!({
+            "symbol": symbol,
+            "limit": limit
+        });
+
+        let data = self.request("GET".to_string(),
+                                "".to_string(),
+                                "/fapi/v1/depth".to_string(),
+                                false,
+                                params.to_string(),
+        ).await;
+        data
+    }
+
+
+    //账户信息
+    pub async fn get_account(&mut self) -> ResponseData {
+        let params = json!({
+            "recvWindow":"2000"
+         });
+
+        let data = self.request("GET".to_string(),
+                                "".to_string(),
+                                "/fapi/v2/balance".to_string(),
+                                true,
+                                params.to_string(),
+        ).await;
+        data
+    }
+
+    //查询订单
+    pub async fn get_order(&mut self, symbol: String, order_id: i64, orig_client_order_id: String) -> ResponseData {
+        let mut params = json!({
+            "symbol":symbol,
+            "recvWindow":"2000"
+         });
+        if order_id > 0 {
+            params["orderId"] = json!(order_id);
+        }
+        if orig_client_order_id.len() > 0 {
+            params["origClientOrderId"] = json!(orig_client_order_id);
+        }
+
+        let data = self.request("GET".to_string(),
+                                "".to_string(),
+                                "/fapi/v1/order".to_string(),
+                                true,
+                                params.to_string(),
+        ).await;
+        data
+    }
+    //查看当前全部挂单
+    pub async fn get_open_orders(&mut self, symbol: String) -> ResponseData {
+        let params = json!({
+            "symbol":symbol,
+            "recvWindow":"2000"
+         });
+
+        let data = self.request("GET".to_string(),
+                                "".to_string(),
+                                "/fapi/v1/openOrders".to_string(),
+                                true,
+                                params.to_string(),
+        ).await;
+        data
+    }
+    //查询所有订单(包括历史订单)
+    pub async fn get_all_orders(&mut self, symbol: String, limit: u64, start_time: i64, end_time: i64) -> ResponseData {
+        let params = json!({
+            "symbol":symbol,
+            "limit":limit,
+            "startTime":start_time,
+            "endTime":end_time,
+            "recvWindow":"2000"
+         });
+
+        let data = self.request("GET".to_string(),
+                                "".to_string(),
+                                "/fapi/v1/allOrders".to_string(),
+                                true,
+                                params.to_string(),
+        ).await;
+        data
+    }
+    //当前最优挂单
+    pub async fn get_book_ticker(&mut self, symbol: String) -> ResponseData {
+        let params = json!({
+            "symbol":symbol
+         });
+
+        let data = self.request("GET".to_string(),
+                                "".to_string(),
+                                "/fapi/v1/ticker/bookTicker".to_string(),
+                                true,
+                                params.to_string(),
+        ).await;
+        data
+    }
+    //用户持仓风险V2
+    pub async fn get_position_risk(&mut self, symbol: String) -> ResponseData {
+        let params = json!({
+            "symbol":symbol,
+            "recvWindow":"2000"
+         });
+
+        let data = self.request("GET".to_string(),
+                                "".to_string(),
+                                "/fapi/v2/positionRisk".to_string(),
+                                true,
+                                params.to_string(),
+        ).await;
+        data
+    }
+
+    //下单
+    pub async fn swap_order(&mut self, params: serde_json::Value) -> ResponseData {
+        let data = self.request("POST".to_string(),
+                                "".to_string(),
+                                "/fapi/v1/order".to_string(),
+                                true,
+                                params.to_string(),
+        ).await;
+        data
+    }
+    //更改持仓模式
+    pub async fn change_pos_side(&mut self, dual_side_position: bool) -> ResponseData {
+        let dual_side_position_str = format!("{}", dual_side_position);
+        let params = json!({
+            "dualSidePosition":dual_side_position_str,
+            "recvWindow":"2000"
+         });
+        let data = self.request("POST".to_string(),
+                                "".to_string(),
+                                "/fapi/v1/positionSide/dual".to_string(),
+                                true,
+                                params.to_string(),
+        ).await;
+        data
+    }
+    //撤销订单
+    pub async fn cancel_order(&mut self, symbol: String, order_id: i64, orig_client_order_id: String) -> ResponseData {
+        let mut params = json!({
+            "symbol":symbol,
+            "recvWindow":"2000"
+         });
+        if order_id > 0 {
+            params["orderId"] = json!(order_id);
+        }
+        if orig_client_order_id.len() > 0 {
+            params["origClientOrderId"] = json!(orig_client_order_id);
+        }
+        let data = self.request("DELETE".to_string(),
+                                "".to_string(),
+                                "/fapi/v1/order".to_string(),
+                                true,
+                                params.to_string(),
+        ).await;
+        data
+    }
+    //根据币对 撤销全部订单
+    pub async fn cancel_order_all(&mut self, symbol: String) -> ResponseData {
+        let params = json!({
+            "symbol":symbol,
+            "recvWindow":"2000",
+         });
+        let data = self.request("DELETE".to_string(),
+                                "".to_string(),
+                                "/fapi/v1/allOpenOrders".to_string(),
+                                true,
+                                params.to_string(),
+        ).await;
+        data
+    }
+    //账户成交历史
+    pub async fn get_user_trades(&mut self, symbol: String, start_time: i64, end_time: i64, limit: i64) -> ResponseData {
+        let mut params = json!({
+            "symbol":symbol,
+            "limit":1000,
+            "recvWindow":"1000",
+         });
+        if start_time > 0 {
+            params["startTime"] = json!(start_time);
+        }
+        if end_time > 0 {
+            params["endTime"] = json!(end_time);
+        }
+        if limit > 0 {
+            params["limit"] = json!(limit);
+        }
+        let data = self.request("GET".to_string(),
+                                "".to_string(),
+                                format!("/fapi/v1/userTrades"),
+                                true,
+                                params.to_string(),
+        ).await;
+        data
+    }
+
+
+    /*******************************************************************************************************/
+    /*****************************************工具函数********************************************************/
+    /*******************************************************************************************************/
+    pub fn get_delays(&self) -> Vec<i64> {
+        self.delays.clone()
+    }
+    pub fn get_avg_delay(&self) -> Decimal {
+        self.avg_delay.clone()
+    }
+    pub fn get_max_delay(&self) -> i64 {
+        self.max_delay.clone()
+    }
+    fn get_delay_info(&mut self) {
+        let last_100 = if self.delays.len() > 100 {
+            self.delays[self.delays.len() - 100..].to_vec()
+        } else {
+            self.delays.clone()
+        };
+
+        let max_value = last_100.iter().max().unwrap();
+        if max_value.clone().to_owned() > self.max_delay {
+            self.max_delay = max_value.clone().to_owned();
+        }
+
+        let sum: i64 = last_100.iter().sum();
+        let sum_v = Decimal::from_i64(sum).unwrap();
+        let len_v = Decimal::from_u64(last_100.len() as u64).unwrap();
+        self.avg_delay = (sum_v / len_v).round_dp(1);
+        self.delays = last_100.clone().into_iter().collect();
+    }
+
+    //调用请求
+    async fn request(&mut self,
+                     request_type: String,
+                     prefix_url: String,
+                     request_url: String,
+                     is_login: bool,
+                     params: String) -> ResponseData
+    {
+        // trace!("login_param:{:?}", self.login_param);
+        //解析账号信息
+        let mut access_key = "".to_string();
+        let mut secret_key = "".to_string();
+        if self.login_param.contains_key("access_key") {
+            access_key = self.login_param.get("access_key").unwrap().to_string();
+        }
+        if self.login_param.contains_key("secret_key") {
+            secret_key = self.login_param.get("secret_key").unwrap().to_string();
+        }
+        let mut is_login_param = true;
+        if access_key == "" || secret_key == "" {
+            is_login_param = false
+        }
+
+        //请求头配置-如果需要登录则存在额外配置
+        let mut params_json: Value = serde_json::from_str(params.clone().as_str()).unwrap();
+        // let mut body = "".to_string();
+        let timestamp = chrono::Utc::now().timestamp_millis().to_string();
+        params_json.as_object_mut().unwrap().insert("timestamp".to_string(), Value::from(timestamp));
+
+
+        let mut headers = HeaderMap::new();
+        if request_type == "GET" {
+            headers.insert("Content-Type", "application/json; charset=UTF-8".parse().unwrap());
+        } else if request_type == "POST" || request_type == "PUT" || request_type == "DELETE" {
+            headers.insert("Content-Type", "application/x-www-form-urlencoded".parse().unwrap());
+        }
+
+
+        //是否需要登录-- 组装sing
+        if is_login {
+            if !is_login_param {
+                let e = ResponseData::error(self.tag.clone(), "登录参数错误！".to_string());
+                return e;
+            } else {//需要登录-且登录参数齐全
+                //组装sing
+                let sing = Self::sign(secret_key.clone(),
+                                      params_json.to_string(),
+                );
+                params_json.as_object_mut().unwrap().insert("signature".to_string(), Value::from(sing.clone()));
+                // trace!("sing:{}", sing);
+                //组装header
+                headers.extend(Self::headers(access_key));
+            }
+        }
+
+        trace!("headers:{:?}", headers);
+        let start_time = chrono::Utc::now().timestamp_millis();
+        let response = self.http_tool(
+            format!("{}{}", prefix_url.clone(), request_url.clone()),
+            request_type.to_string(),
+            params_json.to_string(),
+            headers,
+        ).await;
+
+        let time_array = chrono::Utc::now().timestamp_millis() - start_time;
+        self.delays.push(time_array);
+        self.get_delay_info();
+
+        response
+    }
+
+    pub fn headers(access_key: String) -> HeaderMap {
+        let mut headers = HeaderMap::new();
+        headers.insert("X-MBX-APIKEY", access_key.parse().unwrap());
+        headers
+    }
+    pub fn sign(secret_key: String, params_str: String) -> String
+    {
+        /*签名生成*/
+        let params_str_v = RestTool::parse_params_to_str(params_str.to_string());
+        let message = format!("{}", params_str_v);
+
+        // let secret_key2 = "2b5eb11e18796d12d88f13dc27dbbd02c2cc51ff7059765ed9821957d82bb4d9";
+        // let message2 = "symbol=BTCUSDT&side=BUY&type=LIMIT&quantity=1&price=9000&timeInForce=GTC&recvWindow=5000&timestamp=1591702613943";
+        let signed_key = hmac::Key::new(hmac::HMAC_SHA256, secret_key.as_ref());
+        let sign = hex::encode(hmac::sign(&signed_key, message.as_bytes()).as_ref());
+
+        sign
+    }
+
+
+    async fn http_tool(&mut self, request_path: String, request_type: String, params: String, headers: HeaderMap) -> ResponseData {
+        /****请求接口与 地址*/
+        let url = format!("{}{}", self.base_url.to_string(), request_path);
+        let request_type = request_type.clone().to_uppercase();
+        let url_param = RestTool::parse_params_to_str(params.clone());
+        let addrs_url = format!("{}?{}", url.clone(), url_param);
+
+        let params_json: Value = serde_json::from_str(&params).unwrap();
+        trace!("url:{}",url);
+        trace!("addrs_url:{}",addrs_url);
+        trace!("params_json:{}",params_json);
+        trace!("headers:{:?}",headers);
+        trace!("body:{:?}",params);
+
+
+        let request_builder = match request_type.as_str() {
+            "GET" => self.client.get(url.clone()).query(&params_json).headers(headers),
+            "POST" => self.client.post(url.clone()).query(&params_json).headers(headers),
+            "DELETE" => self.client.delete(url.clone()).query(&params_json).headers(headers),
+            // "PUT" => self.client.put(url.clone()).json(&params),
+            _ => {
+                panic!("{}", format!("错误的请求类型:{}", request_type.clone()))
+            }
+        };
+
+        let response = request_builder.send().await.unwrap();
+        let is_success = response.status().is_success(); // 先检查状态码
+        let text = response.text().await.unwrap();
+        return if is_success {
+            self.on_success_data(text)
+        } else {
+            self.on_error_data(text, addrs_url, params)
+        }
+    }
+
+    pub fn on_success_data(&mut self, text: String) -> ResponseData {
+        let data = serde_json::from_str(text.as_str()).unwrap();
+
+        ResponseData::new(self.tag.clone(), 200, "success".to_string(), data)
+    }
+
+    pub fn on_error_data(&mut self, text: String, base_url: String, params: String) -> ResponseData {
+        let json_value = serde_json::from_str::<Value>(&text);
+
+        match json_value {
+            Ok(data) => {
+                let message = data["msg"].as_str().unwrap();
+
+                let mut error = ResponseData::error(self.tag.clone(), message.to_string());
+                error.code = data["code"].as_i64().unwrap().to_i16().unwrap();
+                error.message = format!("请求地址：{}, 请求参数：{}", base_url, params);
+                error
+            }
+            Err(e) => {
+                error!("解析错误：{:?}", e);
+                let error = ResponseData::error("".to_string(), format!("json 解析失败:{}，相关参数:{}", e, text));
+                error
+            }
+        }
+    }
+}

exchanges/src/binance_swap_ws.rs

@@ -0,0 +1,275 @@
+use std::str::FromStr;
+use std::sync::Arc;
+use std::sync::atomic::AtomicBool;
+use std::time::Duration;
+
+use futures_channel::mpsc::{UnboundedReceiver, UnboundedSender};
+use serde_json::{json, Value};
+use tokio::sync::Mutex;
+use tokio_tungstenite::tungstenite::{Error, Message};
+use tracing::{error, info, trace};
+
+use crate::response_base::ResponseData;
+use crate::socket_tool::AbstractWsMode;
+
+//类型
+pub enum BinanceSwapWsType {
+    PublicAndPrivate,
+}
+
+//订阅频道
+#[derive(Clone)]
+pub enum BinanceSwapSubscribeType {
+    PuBookTicker,
+    PuAggTrade,
+    PuDepth20levels100ms,
+    PuDepthUpdate,
+    PuKline,
+}
+
+//账号信息
+#[derive(Clone)]
+#[allow(dead_code)]
+pub struct BinanceSwapLogin {
+    pub api_key: String,
+    pub api_secret: String,
+}
+
+#[derive(Clone)]
+#[allow(dead_code)]
+pub struct BinanceSwapWs {
+    //类型
+    tag: String,
+    //地址
+    address_url: String,
+    //账号
+    login_param: Option<BinanceSwapLogin>,
+    //币对
+    symbol_s: Vec<String>,
+    //订阅
+    subscribe_types: Vec<BinanceSwapSubscribeType>,
+    //心跳间隔
+    heartbeat_time: u64,
+}
+
+impl BinanceSwapWs {
+    /*******************************************************************************************************/
+    /*****************************************获取一个对象****************************************************/
+    /*******************************************************************************************************/
+    pub fn new(is_colo: bool, login_param: Option<BinanceSwapLogin>, ws_type: BinanceSwapWsType) -> BinanceSwapWs {
+        return BinanceSwapWs::new_with_tag("default-BinanceSwapWs".to_string(), is_colo, login_param, ws_type);
+    }
+    pub fn new_with_tag(tag: String, is_colo: bool, login_param: Option<BinanceSwapLogin>, ws_type: BinanceSwapWsType) -> BinanceSwapWs {
+        /*******公共频道-私有频道数据组装*/
+        let address_url = match ws_type {
+            BinanceSwapWsType::PublicAndPrivate => {
+                // "wss://fstream.binance.com/stream?streams=btcusdt@depth20@100ms".to_string(),
+                "wss://fstream.binance.com/stream".to_string()
+            }
+        };
+
+        if is_colo {
+            info!("开启高速(未配置，走普通：{})通道",address_url);
+        } else {
+            info!("走普通通道：{}",address_url);
+        }
+
+        BinanceSwapWs {
+            tag,
+            address_url,
+            login_param,
+            symbol_s: vec![],
+            subscribe_types: vec![],
+            heartbeat_time: 1000 * 20,
+        }
+    }
+
+    /*******************************************************************************************************/
+    /*****************************************订阅函数********************************************************/
+    /*******************************************************************************************************/
+    //手动添加订阅信息
+    pub fn set_subscribe(&mut self, subscribe_types: Vec<BinanceSwapSubscribeType>) {
+        self.subscribe_types.extend(subscribe_types);
+    }
+    //手动添加币对
+    pub fn set_symbols(&mut self, mut b_array: Vec<String>) {
+        for symbol in b_array.iter_mut() {
+            // 小写
+            *symbol = symbol.to_lowercase();
+            // 字符串替换
+            *symbol = symbol.replace("_", "");
+            *symbol = symbol.replace("-", "");
+        }
+        self.symbol_s = b_array;
+    }
+    //频道是否需要登录
+    fn contains_pr(&self) -> bool {
+        for t in self.subscribe_types.clone() {
+            if match t {
+                BinanceSwapSubscribeType::PuBookTicker => false,
+                BinanceSwapSubscribeType::PuAggTrade => false,
+                BinanceSwapSubscribeType::PuDepth20levels100ms => false,
+                BinanceSwapSubscribeType::PuDepthUpdate => false,
+                BinanceSwapSubscribeType::PuKline => false,
+            } {
+                return true;
+            }
+        }
+        false
+    }
+    /*******************************************************************************************************/
+    /*****************************************工具函数********************************************************/
+    /*******************************************************************************************************/
+    //订阅枚举解析
+    pub fn enum_to_string(symbol: String, subscribe_type: BinanceSwapSubscribeType) -> String {
+        match subscribe_type {
+            BinanceSwapSubscribeType::PuAggTrade => {
+                format!("{}@aggTrade", symbol)
+            }
+            BinanceSwapSubscribeType::PuDepth20levels100ms => {
+                format!("{}@depth20@100ms", symbol)
+            }
+            BinanceSwapSubscribeType::PuDepthUpdate => {
+                format!("{}@depth@100ms", symbol)
+            }
+            BinanceSwapSubscribeType::PuBookTicker => {
+                format!("{}@bookTicker", symbol)
+            }
+            BinanceSwapSubscribeType::PuKline => {
+                format!("{}@kline_1m", symbol)
+            }
+        }
+    }
+    //订阅信息生成
+    pub fn get_subscription(&self) -> String {
+        let mut params = vec![];
+        for symbol in &self.symbol_s {
+            for subscribe_type in &self.subscribe_types {
+                let ty_str = Self::enum_to_string(symbol.clone(), subscribe_type.clone());
+                params.push(ty_str);
+            }
+        }
+
+        let str = json!({
+            "method": "SUBSCRIBE",
+            "params":  params,
+            "id": 1
+        });
+        str.to_string()
+    }
+    /*******************************************************************************************************/
+    /*****************************************socket基本*****************************************************/
+    /*******************************************************************************************************/
+    //链接
+    pub async fn ws_connect_async<F, Future>(&mut self,
+                                             is_shutdown_arc: Arc<AtomicBool>,
+                                             handle_function: F,
+                                             _write_tx_am: &Arc<Mutex<UnboundedSender<Message>>>,
+                                             write_to_socket_rx: UnboundedReceiver<Message>) -> Result<(), Error>
+    where
+        F: Fn(ResponseData) -> Future + Clone + Send + 'static + Sync,
+        Future: std::future::Future<Output=()> + Send + 'static, // 确保 Fut 是一个 Future，且输出类型为 ()
+    {
+        let login_is = self.contains_pr();
+        let subscription = self.get_subscription();
+        let address_url = self.address_url.clone();
+        let tag = self.tag.clone();
+        // let heartbeat_time = self.heartbeat_time.clone();
+
+
+        //心跳-- 方法内部线程启动
+        // let write_tx_clone1 = Arc::clone(write_tx_am);
+        // tokio::spawn(async move {
+        //     trace!("线程-异步心跳-开始");
+        //     AbstractWsMode::ping_or_pong(write_tx_clone1, HeartbeatType::Pong, heartbeat_time).await;
+        //     trace!("线程-异步心跳-结束");
+        // });
+
+        //设置订阅
+        let mut subscribe_array = vec![];
+        if login_is {
+            //登录相关
+        }
+        subscribe_array.push(subscription.to_string());
+
+        //链接
+        let t2 = tokio::spawn(async move {
+            let write_to_socket_rx_arc = Arc::new(Mutex::new(write_to_socket_rx));
+
+            loop {
+                info!("binance_usdt_swap socket 连接中……");
+                // ws层重连
+                AbstractWsMode::ws_connect_async(is_shutdown_arc.clone(), handle_function.clone(), address_url.clone(),
+                                                 false, tag.clone(), subscribe_array.clone(), write_to_socket_rx_arc.clone(),
+                                                 Self::message_text, Self::message_ping, Self::message_pong, Self::message_binary).await;
+
+                error!("binance_usdt_swap socket 断连，1s以后重连……");
+                tokio::time::sleep(Duration::from_secs(1)).await;
+            }
+        });
+
+        tokio::try_join!(t2).unwrap();
+        trace!("线程-心跳与链接-结束");
+
+        Ok(())
+    }
+    /*******************************************************************************************************/
+    /*****************************************数据解析*****************************************************/
+    /*******************************************************************************************************/
+    //数据解析-Text
+    pub fn message_text(text: String) -> Option<ResponseData> {
+        let response_data = Self::ok_text(text);
+        Option::from(response_data)
+    }
+    //数据解析-ping
+    pub fn message_ping(_pi: Vec<u8>) -> Option<ResponseData> {
+        return Option::from(ResponseData::new("".to_string(), -300, "success".to_string(), Value::String("pong".to_string())));
+    }
+    //数据解析-pong
+    pub fn message_pong(_po: Vec<u8>) -> Option<ResponseData> {
+        return Option::from(ResponseData::new("".to_string(), -301, "success".to_string(), Value::Null));
+    }
+    //数据解析-二进制
+    pub fn message_binary(_po: Vec<u8>) -> Option<ResponseData> {
+        //二进制WebSocket消息
+        let message_str = format!("Binary:{:?}", _po);
+        Option::from(ResponseData::new("".to_string(), 2, message_str, Value::Null))
+    }
+    //数据解析
+    pub fn ok_text(text: String) -> ResponseData {
+        // trace!("原始数据");
+        // trace!(?text);
+        let mut res_data = ResponseData::new("".to_string(), 200, "success".to_string(), Value::Null);
+        let json_value: Value = serde_json::from_str(&text).unwrap();
+
+        if json_value.get("result").is_some() && json_value.get("id").is_some() &&
+            json_value.get("id").unwrap() == 1
+        {
+            res_data.code = -201;
+            res_data.message = "订阅成功".to_string();
+        } else if json_value.get("error").is_some() {//订阅返回
+            res_data.code = i16::from_str(json_value["error"]["code"].as_str().unwrap()).unwrap();
+            res_data.message = json_value["error"]["msg"].to_string();
+        } else if json_value.get("stream").is_some() {//订阅返回
+            res_data.data = json_value["data"].clone();
+            res_data.code = 200;
+
+            let channel = json_value["stream"].as_str().unwrap();
+            if channel.contains("@aggTrade") {
+                res_data.channel = "aggTrade".to_string();
+            } else if channel.contains("@depth20@100ms") {
+                res_data.channel = "depth".to_string();
+            } else if channel.contains("@bookTicker") {
+                res_data.channel = "bookTicker".to_string();
+            } else if channel.contains("@kline") {
+                res_data.channel = "kline".to_string();
+            } else {
+                res_data.channel = channel.to_string();
+            }
+        } else {
+            res_data.code = -1;
+            res_data.channel = "未知的频道2".to_string();
+        }
+        res_data
+    }
+}

exchanges/src/lib.rs

@@ -7,3 +7,5 @@ pub mod gate_swap_ws;
 pub mod gate_swap_rest;
 pub mod coinex_swap_rest;
 pub mod coinex_swap_ws;
+pub mod binance_swap_ws;
+pub mod binance_swap_rest;

src/binance_usdt_swap_data_listener.rs

@@ -0,0 +1,110 @@
+use std::collections::{BTreeMap, HashMap};
+use std::sync::{Arc};
+use std::sync::atomic::AtomicBool;
+use std::time::Duration;
+use chrono::Utc;
+use lazy_static::lazy_static;
+use tokio::sync::{Mutex};
+use tracing::info;
+use exchanges::binance_swap_rest::BinanceSwapRest;
+use exchanges::binance_swap_ws::{BinanceSwapSubscribeType, BinanceSwapWs, BinanceSwapWsType};
+use exchanges::response_base::ResponseData;
+use rust_decimal_macros::dec;
+use standard::exchange::ExchangeEnum;
+use standard::exchange_struct_handler::ExchangeStructHandler;
+use crate::json_db_utils::{collect_special_trades_json, delete_db_by_exchange};
+use crate::listener_tools::{TradeMap, update_trade};
+use crate::msv::{generate_msv_by_trades, Indicators, parse_json_to_trades};
+
+const EXCHANGE_NAME: &str = "binance_usdt_swap";
+
+lazy_static! {
+    // 给其他模块使用
+    pub static ref INDICATOR_MAP: Mutex<HashMap<String, Indicators>> = Mutex::new(HashMap::new());
+    // static ref LOCAL_DEPTH: Mutex<HashMap<String, Depth>> = Mutex::new(HashMap::new());             // 本地缓存的订单簿
+
+    static ref TRADES_MAP: Mutex<TradeMap> = Mutex::new(HashMap::new());
+}
+
+pub async fn run_listener(is_shutdown_arc: Arc<AtomicBool>) {
+    let name = "binance_usdt_swap_listener";
+    // 订阅所有币种
+    let login = BTreeMap::new();
+    let mut binance_rest = BinanceSwapRest::new(false, login);
+    let response = binance_rest.get_exchange_info().await;
+    let mut symbols = vec![];
+    if response.code == 200 {
+        let data = response.data["symbols"].as_array().unwrap();
+        for info in data {
+            let s = info["symbol"].as_str().unwrap().to_string();
+            if !s.ends_with("USDT") {
+                continue;
+            }
+            let symbol = s.replace("USDT", "_USDT");
+            symbols.push(symbol)
+        }
+    }
+
+    for chunk in symbols.chunks(20) {
+        let ws_name = name.to_string();
+        let (write_tx, write_rx) = futures_channel::mpsc::unbounded();
+        let write_tx_am = Arc::new(Mutex::new(write_tx));
+        let symbols_chunk = chunk.iter().cloned().collect::<Vec<String>>();
+        let is_shutdown_clone = Arc::clone(&is_shutdown_arc);
+
+        tokio::spawn(async move {
+            let mut ws = BinanceSwapWs::new_with_tag(ws_name, false, None, BinanceSwapWsType::PublicAndPrivate);
+            ws.set_subscribe(vec![
+                BinanceSwapSubscribeType::PuAggTrade,
+            ]);
+
+            // 建立链接
+            ws.set_symbols(symbols_chunk);
+            ws.ws_connect_async(is_shutdown_clone, data_listener, &write_tx_am, write_rx).await.expect("链接失败(内部一个心跳线程应该已经关闭了)");
+        });
+    }
+    // 每分钟计算msv
+    tokio::spawn(async move {
+        loop {
+            let end_timestamp = Utc::now().timestamp_millis();
+            let start_timestamp = end_timestamp - 60 * 1000 * 60;
+            for symbol in symbols.clone() {
+                let trades_value = collect_special_trades_json(start_timestamp, end_timestamp, EXCHANGE_NAME, &symbol).await;
+                let trades = parse_json_to_trades(trades_value);
+                let msv = generate_msv_by_trades(trades, dec!(50), vec![], start_timestamp, end_timestamp);
+                let mut indicator_map = INDICATOR_MAP.lock().await;
+                indicator_map.insert(symbol, msv);
+            }
+            tokio::time::sleep(Duration::from_secs(60)).await;
+        }
+    });
+    // 定时删除数据
+    tokio::spawn(async move {
+        loop {
+            delete_db_by_exchange(EXCHANGE_NAME, "trades", 5 * 60).await;
+            tokio::time::sleep(Duration::from_secs(3600)).await;
+        }
+    });
+}
+
+// 读取数据
+pub async fn data_listener(response: ResponseData) {
+    if response.code != 200 {
+        return;
+    }
+    match response.channel.as_str() {
+        // 订单流数据
+        "aggTrade" => {
+            let trades = ExchangeStructHandler::trades_handle(ExchangeEnum::BinanceSwap, &response);
+
+            for trade in trades.iter() {
+                let trades_map = TRADES_MAP.lock().await;
+
+                update_trade(trade, trades_map, EXCHANGE_NAME).await;
+            }
+        }
+        _ => {
+            info!("48 未知的数据类型: {:?}", response)
+        }
+    }
+}

src/main.rs

@@ -2,6 +2,7 @@ mod json_db_utils;
 mod control_c;
 mod server;
 mod listener_tools;
+mod binance_usdt_swap_data_listener;
 mod gate_usdt_swap_data_listener;
 mod coinex_usdt_swap_data_listener;
 mod msv;
@@ -30,6 +31,7 @@ async fn main() {
     // ctrl c退出检查程序
     control_c::exit_handler(running.clone());
     // 启动各交易所的数据监听器
+    binance_usdt_swap_data_listener::run_listener(running.clone()).await;
     gate_usdt_swap_data_listener::run_listener(running.clone()).await;
     coinex_usdt_swap_data_listener::run_listener(running.clone()).await;
     // panic错误捕获，panic级别的错误直接退出

src/msv.rs

@@ -70,8 +70,9 @@ pub fn generate_msv_by_trades(mut trades: Vec<Trade>, mills_back: Decimal, simpl
         }
 
         // 获取到range毫秒以后的预定价格，计算回去的幅度
-        let mut future_ref_price = ref_price;
-        let mut future_range_index = index;
+        let mut future_ref_price_sum = trade.price;
+        let mut future_ref_count = Decimal::ONE;
+        let mut future_range_index = index + 1;
         loop {
             // 下标合法性判断
             if future_range_index >= trades.len() {
@@ -82,13 +83,15 @@ pub fn generate_msv_by_trades(mut trades: Vec<Trade>, mills_back: Decimal, simpl
             let range_time = flag_trade.time - trade.time;
 
             // 判断该ticker是否是range ms以外
-            if range_time > mills_back {
+            if range_time > mills_back && future_ref_count > Decimal::TEN {
                 break;
             }
-            future_ref_price = future_ref_price * GAMMA + flag_trade.price * (Decimal::ONE - GAMMA);
 
             future_range_index += 1;
+            future_ref_price_sum += flag_trade.price;
+            future_ref_count += Decimal::ONE;
         }
+        let future_ref_price = future_ref_price_sum / future_ref_count;
 
         // 计算过去至多100条数据的sigma值 sigma^2 = (1 / (tn-t0))*sum((S(tk) - S(tk-1)) ^ 2)
         let mut sigma_index = index - 1;

src/rank.rs

@@ -3,8 +3,9 @@ use std::collections::HashMap;
 use rust_decimal::{Decimal, MathematicalOps};
 use rust_decimal_macros::dec;
 use serde::{Deserialize, Serialize};
-use serde_json::Value;
+use serde_json::{Value};
 use tokio::sync::MutexGuard;
+use crate::json_db_utils::get_symbols_by_exchange;
 use crate::msv::Indicators;
 
 #[derive(Serialize, Deserialize, Clone)]
@@ -21,6 +22,7 @@ pub struct Rank {
     epr_total: Decimal,
     effective_count: Decimal,
     liquidity_avg: Decimal,
+    is_binance: bool,
 }
 
 const TWENTY: Decimal = dec!(20);
@@ -31,10 +33,10 @@ const TWO_HUNDRED: Decimal = dec!(200);
 // 根据最终的msv计算排行榜
 pub fn generate_rank_by_indicator_map(indicator_map: &MutexGuard<HashMap<String, Indicators>>) -> Value {
     let mut rank_list: Vec<Rank> = vec![];
-
+    let binance_symbols = get_symbols_by_exchange("binance_usdt_swap").as_array().unwrap_or(&vec![]).clone();
     for (key, indicators) in indicator_map.iter() {
         let symbol = key.clone();
-
+        let is_binance = binance_symbols.iter().find(|item| item.as_str().unwrap().to_string() == symbol).is_some();
         // ============== msv相关数据的计算 =================
         let mut msv_abs_total = Decimal::ZERO;
         let mut msv_abs_max = Decimal::ZERO;
@@ -43,7 +45,7 @@ pub fn generate_rank_by_indicator_map(indicator_map: &MutexGuard<HashMap<String,
             let msv_abs_value = value[1].abs();
 
             if msv_abs_value <= Decimal::ZERO {
-                continue
+                continue;
             }
 
             effective_count += Decimal::ONE;
@@ -123,6 +125,7 @@ pub fn generate_rank_by_indicator_map(indicator_map: &MutexGuard<HashMap<String,
             epr_total,
             effective_count,
             liquidity_avg,
+            is_binance,
         };
 
         rank_list.push(rank);

src/server.rs

@@ -95,6 +95,7 @@ async fn get_exchanges() -> impl Responder {
         // "binance_usdt_swap",
         // "htx_usdt_swap",
         // "phemex_usdt_swap",
+        "binance_usdt_swap",
         "gate_usdt_swap",
         "coinex_usdt_swap",
     ];

standard/src/binance_swap.rs

@@ -0,0 +1,387 @@
+use std::collections::BTreeMap;
+use std::io::{Error, ErrorKind};
+use std::result::Result;
+use std::str::FromStr;
+use async_trait::async_trait;
+use rust_decimal::Decimal;
+use rust_decimal::prelude::FromPrimitive;
+use rust_decimal_macros::dec;
+use serde_json::Value;
+use tokio::sync::mpsc::Sender;
+use tracing::{error};
+use crate::{Platform, ExchangeEnum, Account, Position, Ticker, Market, Order, utils, PositionModeEnum};
+use exchanges::binance_swap_rest::BinanceSwapRest;
+
+#[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,
+        };
+        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.ct_val;
+        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: res_data_json["time"].as_i64().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),
+            };
+            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_book_ticker(symbol_format).await;
+        if res_data.code == 200 {
+            let res_data_json: serde_json::Value = res_data.data;
+            let result = Ticker {
+                time: res_data_json["time"].as_i64().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),
+            };
+            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 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(),
+                        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_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(),
+                        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 symbol_format = utils::format_symbol(self.symbol.clone(), "");
+        let res_data = self.request.get_order(symbol_format, order_id.parse().unwrap_or(-1), custom_id.to_string()).await;
+        if res_data.code == 200 {
+            let res_data_json: serde_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()
+            };
+            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 res_data = self.request.get_open_orders(symbol_format).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()
+                }
+            }).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> { Err(Error::new(ErrorKind::NotFound, "binance_swap：该交易所方法未实现".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> { Err(Error::new(ErrorKind::NotFound, "binance_swap：该交易所方法未实现".to_string())) }
+
+    async fn cancel_order(&mut self, _order_id: &str, _custom_id: &str) -> Result<Order, Error> { Err(Error::new(ErrorKind::NotFound, "binance_swap：该交易所方法未实现".to_string())) }
+
+    async fn cancel_orders(&mut self) -> Result<Vec<Order>, Error> { Err(Error::new(ErrorKind::NotFound, "binance_swap：该交易所方法未实现".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> { Err(Error::new(ErrorKind::NotFound, "binance_swap：该交易所方法未实现".to_string())) }
+
+    async fn set_dual_leverage(&mut self, _leverage: &str) -> Result<String, Error> { Err(Error::new(ErrorKind::NotFound, "binance_swap：该交易所方法未实现".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())) }
+}
+
+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(),
+    }
+}

standard/src/binance_swap_handle.rs

@@ -0,0 +1,28 @@
+use std::str::FromStr;
+use rust_decimal::Decimal;
+use rust_decimal::prelude::FromPrimitive;
+use exchanges::response_base::ResponseData;
+use crate::{Trade};
+
+
+pub fn format_trade_items(response: &ResponseData) -> Vec<Trade> {
+    let data = response.data.clone();
+
+    let id = data["a"].as_i64().unwrap().to_string();
+    let time = Decimal::from_i64(data["T"].as_i64().unwrap()).unwrap();
+    let is_sell = data["m"].as_bool().unwrap(); // 买方是否是做市方。如true，则此次成交是一个主动卖出单，否则是一个主动买入单。
+    let mut size = Decimal::from_str(data["q"].as_str().unwrap().to_string().as_str()).unwrap();
+    if is_sell {
+        size = -size
+    }
+    let price = Decimal::from_str(data["p"].as_str().unwrap().to_string().as_str()).unwrap();
+    let symbol = data["s"].as_str().unwrap().to_string().replace("USDT", "_USDT");
+
+    vec![Trade {
+        id,
+        time,
+        size,
+        price,
+        symbol,
+    }]
+}

standard/src/exchange.rs

@@ -2,6 +2,7 @@ use std::collections::{BTreeMap};
 use std::io::Error;
 use tokio::sync::mpsc::Sender;
 use crate::{Order, Platform};
+use crate::binance_swap::BinanceSwap;
 use crate::coinex_swap::CoinexSwap;
 use crate::gate_swap::GateSwap;
 
@@ -14,6 +15,7 @@ use crate::gate_swap::GateSwap;
 /// - `KucoinSwap`: kucoin交易所期货;
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub enum ExchangeEnum {
+    BinanceSwap,
     GateSwap,
     CoinexSwap,
 }
@@ -60,6 +62,9 @@ pub struct Exchange;
 impl Exchange {
     pub async fn new(exchange: ExchangeEnum, symbol: String, is_colo: bool, params: BTreeMap<String, String>, order_sender: Sender<Order>, error_sender: Sender<Error>) -> Box<dyn Platform + Send + Sync> {
         match exchange {
+            ExchangeEnum::BinanceSwap => {
+                Box::new(BinanceSwap::new(symbol, is_colo, params, order_sender, error_sender).await)
+            }
             ExchangeEnum::GateSwap => {
                 Box::new(GateSwap::new(symbol, is_colo, params, order_sender, error_sender).await)
             }

standard/src/exchange_struct_handler.rs

@@ -1,6 +1,6 @@
 use exchanges::response_base::ResponseData;
 use crate::exchange::ExchangeEnum;
-use crate::{coinex_swap_handle, gate_swap_handle};
+use crate::{binance_swap_handle, coinex_swap_handle, gate_swap_handle};
 use crate::{ Trade };
 
 #[allow(dead_code)]
@@ -11,6 +11,9 @@ impl ExchangeStructHandler {
     // 处理成交信息
     pub fn trades_handle(exchange: ExchangeEnum, res_data: &ResponseData) -> Vec<Trade> {
         match exchange {
+            ExchangeEnum::BinanceSwap => {
+                binance_swap_handle::format_trade_items(&res_data)
+            }
             ExchangeEnum::GateSwap => {
                 gate_swap_handle::format_trade_items(&res_data)
             }

standard/src/lib.rs

@@ -17,6 +17,8 @@ mod gate_swap;
 pub mod gate_swap_handle;
 pub mod coinex_swap;
 pub mod coinex_swap_handle;
+pub mod binance_swap;
+pub mod binance_swap_handle;
 
 /// 持仓模式枚举
 /// - `Both`：单持仓方向