添加bitget

exchanges/src/bitget_swap_rest.rs

@@ -0,0 +1,408 @@
+use std::collections::BTreeMap;
+use std::str::FromStr;
+use reqwest::Client;
+use reqwest::header::HeaderMap;
+use rust_decimal::Decimal;
+use tracing::info;
+use crate::http_tool::RestTool;
+use crate::response_base::ResponseData;
+use ring::hmac;
+use rust_decimal::prelude::FromPrimitive;
+use serde_json::Value;
+
+#[derive(Clone, Debug)]
+pub struct BitgetSwapRest {
+    pub tag: String,
+    base_url: String,
+    client: Client,
+    login_param: BTreeMap<String, String>,                  // 登录参数
+    delays: Vec<i64>,
+    max_delay: i64,
+    avg_delay: Decimal,
+}
+
+impl BitgetSwapRest {
+    pub fn new(is_colo: bool, login_param: BTreeMap<String, String>) -> BitgetSwapRest {
+        return BitgetSwapRest::new_with_tag("default-BitgetSwapRest".to_string(), is_colo, login_param)
+    }
+
+    // 构造Bitget，可以自定义tag
+    pub fn new_with_tag(tag: String, is_colo: bool, login_param: BTreeMap<String, String>) -> BitgetSwapRest {
+        let base_url = if is_colo {
+            "https://api.bitget.com".to_string()
+        } else {
+            "https://api.bitget.com".to_string()
+        };
+
+        if is_colo {
+            info!("开启高速(未配置，走普通：{})通道", base_url);
+        } else {
+            info!("走普通通道：{}", base_url);
+        }
+
+        BitgetSwapRest {
+            tag,
+            base_url,
+            client: Client::new(),
+            login_param,
+            delays: vec![],
+            max_delay: 0,
+            avg_delay: Decimal::ZERO,
+        }
+    }
+
+    //获取合约信息
+    pub async fn get_contracts(&mut self, symbol: String) -> ResponseData {
+        let params = serde_json::json!({
+            "symbol": symbol,
+            "productType": "USDT-FUTURES"
+         });
+        let data = self.request("GET".to_string(),
+                                "/api/v2".to_string(),
+                                "/mix/market/contracts".to_string(),
+                                false,
+                                params.to_string(),
+        ).await;
+        data
+    }
+
+    //获取所有合约信息
+    pub async fn get_all_contracts(&mut self) -> ResponseData {
+        let params = serde_json::json!({
+            "productType": "USDT-FUTURES"
+         });
+        let data = self.request("GET".to_string(),
+                                "/api/v2".to_string(),
+                                "/mix/market/contracts".to_string(),
+                                false,
+                                params.to_string(),
+        ).await;
+        data
+    }
+
+    //获取行情信息
+    pub async fn get_tickers(&mut self, symbol: String) -> ResponseData {
+        let params = serde_json::json!({
+            "symbol": symbol,
+            "productType": "USDT-FUTURES"
+         });
+        let data = self.request("GET".to_string(),
+                                "/api/v2".to_string(),
+                                "/mix/market/ticker".to_string(),
+                                true,
+                                params.to_string(),
+        ).await;
+        data
+    }
+
+    // 获取服务器时间
+    pub async fn get_server_time(&mut self) -> ResponseData {
+        let params = serde_json::json!({});
+        self.request("GET".to_string(),
+                     "/api/v2".to_string(),
+                     "/public/time".to_string(),
+                     false,
+                     params.to_string(),
+        ).await
+    }
+
+    // 获取账户信息
+    pub async fn get_account_info(&mut self) -> ResponseData {
+        let params = serde_json::json!({
+            "productType": "USDT-FUTURES"
+        });
+
+        self.request("GET".to_string(),
+                     "/api/v2".to_string(),
+                     "/mix/account/accounts".to_string(),
+                     true,
+                     params.to_string(),
+        ).await
+    }
+
+    // 获取仓位信息（单个）
+    pub async fn get_single_position(&mut self, params: Value) -> ResponseData {
+        self.request("GET".to_string(),
+                     "/api/v2".to_string(),
+                     "/mix/position/single-position".to_string(),
+                     true,
+                     params.to_string(),
+        ).await
+    }
+
+    // 获取所有仓位
+    pub async fn get_all_position(&mut self, params: Value) -> ResponseData {
+        self.request("GET".to_string(),
+                     "/api/v2".to_string(),
+                     "/mix/position/all-position".to_string(),
+                     true,
+                     params.to_string(),
+        ).await
+    }
+
+    // 下单
+    pub async fn swap_order(&mut self, params: Value) -> ResponseData {
+        self.request("POST".to_string(),
+                     "/api/v2".to_string(),
+                     "/mix/order/place-order".to_string(),
+                     true,
+                     params.to_string(),
+        ).await
+    }
+
+    // 撤单
+    pub async fn cancel_order(&mut self, params: Value) -> ResponseData {
+        self.request("POST".to_string(),
+                     "/api/v2".to_string(),
+                     "/mix/order/cancel-order".to_string(),
+                     true,
+                     params.to_string(),
+        ).await
+    }
+
+    // 获取订单详情
+    pub async fn get_order(&mut self, params: Value) -> ResponseData {
+        self.request("GET".to_string(),
+                     "/api/v2".to_string(),
+                     "/mix/order/detail".to_string(),
+                     true,
+                     params.to_string(),
+        ).await
+    }
+
+    // 获取当前未成交订单
+    pub async fn get_pending_orders(&mut self) -> ResponseData {
+        let params = serde_json::json!({
+            "productType": "USDT-FUTURES"
+        });
+
+        self.request("GET".to_string(),
+                     "/api/v2".to_string(),
+                     "/mix/order/orders-pending".to_string(),
+                     true,
+                     params.to_string(),
+        ).await
+    }
+
+    // 调整杠杆
+    pub async fn set_leverage(&mut self, params: Value) -> ResponseData {
+        self.request("POST".to_string(),
+                     "/api/v2".to_string(),
+                     "/mix/account/set-leverage".to_string(),
+                     true,
+                     params.to_string(),
+        ).await
+    }
+
+    // 调整持仓模式（单向、双向）
+    pub async fn set_position_mode(&mut self, params: Value) -> ResponseData {
+        self.request("POST".to_string(),
+                     "/api/v2".to_string(),
+                     "/mix/account/set-position-mode".to_string(),
+                     true,
+                     params.to_string(),
+        ).await
+    }
+
+    // 发送请求
+    pub async fn request(&mut self,
+                         method: 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();
+        let mut passphrase = "".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();
+        }
+        if self.login_param.contains_key("pass_key") {
+            passphrase = self.login_param.get("pass_key").unwrap().to_string();
+        }
+        let mut is_login_param = true;
+        if access_key == "" || secret_key == "" || passphrase == "" {
+            is_login_param = false
+        }
+
+        //请求头配置-如果需要登陆则存在额外配置
+        let mut body = "".to_string();
+        let timestamp = chrono::Utc::now().timestamp_millis().to_string();
+        let mut headers = HeaderMap::new();
+        headers.insert("Content-Type", "application/json".parse().unwrap());
+        headers.insert("locale", "en-US".parse().unwrap());
+        if method == "POST" {
+            body = params.clone();
+        }
+
+        //是否需要登陆-- 组装sing
+        if is_login {
+            if !is_login_param {
+                let e = ResponseData::error(self.tag.clone(), "登录参数错误！".to_string());
+                return e;
+            } else {
+                //需要登陆-且登陆参数齐全
+                // trace!("param:{}", params);
+                // trace!("body:{}", body);
+                //组装sing
+                let sing = Self::sign(secret_key.clone(),
+                                      method.clone(),
+                                      prefix_url.clone(),
+                                      request_url.clone(),
+                                      params.clone(),
+                                      body.clone(),
+                                      timestamp.clone(),
+                );
+                //组装header
+                headers.extend(Self::headers(sing, timestamp, passphrase, access_key));
+            }
+        }
+
+
+        // trace!("headers:{:?}", headers);
+        let base_url = format!("{}{}", prefix_url.clone(), request_url.clone());
+        let start_time = chrono::Utc::now().timestamp_millis();
+        let get_response = self.http_tool(
+            format!("{}{}", prefix_url.clone(), request_url.clone()),
+            method.to_string(),
+            params.clone(),
+            headers,
+        ).await;
+        let time_array = chrono::Utc::now().timestamp_millis() - start_time;
+        self.delays.push(time_array);
+        self.get_delay_info();
+        let res_data = Self::res_data_analysis(get_response, base_url, params);
+        res_data
+    }
+
+    pub fn res_data_analysis(result: Result<ResponseData, reqwest::Error>, base_url: String, params: String) -> ResponseData {
+        match result {
+            Ok(res_data) => {
+                if res_data.code != 200 {
+                    let json_value = res_data.data;
+                    let error = ResponseData::new("".to_string(),
+                                                  res_data.code,
+                                                  format!("请求地址:{}, 请求参数:{}, 响应结果:{}", base_url, params, res_data.message),
+                                                  json_value);
+                    error
+                } else {
+                    let json_value = res_data.data;
+
+
+                    let code = json_value["code"].as_str().unwrap();
+                    if code == "00000" {
+                        let data = serde_json::to_string(&json_value["data"]).unwrap();
+                        let success = ResponseData::new("".to_string(), 200, "success".to_string(), data.parse().unwrap());
+                        success
+                    } else {
+                        let code_num = i16::from_str(json_value["code"].as_str().unwrap()).unwrap();
+                        let error = ResponseData::new("".to_string(), code_num,
+                                                      format!("请求地址:{}, 请求参数:{}, 响应结果:{}", base_url, params, json_value.as_str().unwrap()),
+                                                      json_value);
+                        error
+                    }
+                }
+            }
+            Err(err) => {
+                let error = ResponseData::error("".to_string(), format!("json 解析失败:{}", err));
+                error
+            }
+        }
+    }
+
+    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();
+    }
+
+    // 封装请求头
+    pub fn headers(sign: String, timestamp: String, passphrase: String, access_key: String) -> HeaderMap {
+        let mut headers = HeaderMap::new();
+        headers.insert("ACCESS-KEY", access_key.parse().unwrap());
+        headers.insert("ACCESS-SIGN", sign.parse().unwrap());
+        headers.insert("ACCESS-TIMESTAMP", timestamp.parse().unwrap());
+        headers.insert("ACCESS-PASSPHRASE", passphrase.parse().unwrap());
+        headers
+    }
+
+
+    async fn http_tool(&mut self, request_path: String, request_type: String, params: String, headers: HeaderMap) -> Result<ResponseData, reqwest::Error> {
+        let res_data: ResponseData;
+        /****请求接口与 地址*/
+        let url = format!("{}{}", self.base_url.to_string(), request_path);
+        let request_type = request_type.clone().to_uppercase();
+        let addrs_url: String = if RestTool::parse_params_to_str(params.clone()) == "" {
+            url.clone()
+        } else {
+            format!("{}?{}", url.clone(), RestTool::parse_params_to_str(params.clone()))
+        };
+        // trace!("url:{}", url);
+        // trace!("addrs_url:{}", addrs_url);
+        // let params_json: serde_json::Value = serde_json::from_str(&params).unwrap();
+        // trace!("params_json:{}",params_json);
+        // trace!("headers:{:?}",headers);
+
+        let req = match request_type.as_str() {
+            "GET" => self.client.get(addrs_url.clone()).headers(headers),
+            "POST" => self.client.post(url.clone()).body(params).headers(headers),
+            "DELETE" => self.client.delete(addrs_url.clone()).headers(headers),
+            // "PUT" => self.client.put(url.clone()).json(&params),
+            _ => return Ok(ResponseData::error(self.tag.clone(), format!("错误的请求类型:{}", request_type.clone()))), // 处理未知请求类型
+        };
+
+        let response = req.send().await?;
+        if response.status().is_success() {
+            // 读取响应的内容
+            let body = response.text().await?;
+            // trace!("ok-----{}", body);
+            res_data = ResponseData::new(self.tag.clone(), 200, "success".to_string(), body.parse().unwrap());
+        } else {
+            let body = response.text().await?;
+            res_data = ResponseData::error(self.tag.clone(), body.to_string())
+        }
+
+        Ok(res_data)
+    }
+
+    // 对请求进行签名处理
+    pub fn sign(secret_key: String,
+            method: String, prefix_url: String, request_url: String,
+            params: String, body: String, timestamp: String) -> String {
+        let url_param_str = RestTool::parse_params_to_str(params);
+        let base_url = if method == "GET" && url_param_str.len() > 0 {
+            format!("{}{}?{}", prefix_url, request_url, url_param_str)
+        } else {
+            format!("{}{}", prefix_url, request_url)
+        };
+
+        // 时间戳 + 请求类型+ 请求参数字符串
+        let message = format!("{}{}{}{}", timestamp, method, base_url, body);
+        // trace!("message:{}",message);
+
+        // 做签名
+        let hmac_key = hmac::Key::new(hmac::HMAC_SHA256, secret_key.as_bytes());
+        let result = hmac::sign(&hmac_key, &message.as_bytes());
+        let sign = base64::encode(result);
+        sign
+    }
+}

exchanges/src/bitget_swap_ws.rs

@@ -0,0 +1,323 @@
+use std::sync::Arc;
+use std::sync::atomic::{AtomicBool, Ordering};
+use chrono::{Utc};
+use futures_channel::mpsc::{UnboundedReceiver, UnboundedSender};
+use serde_json::{json, Value};
+use tracing::{info, trace, warn};
+use ring::hmac;
+use tokio::sync::Mutex;
+use tokio_tungstenite::tungstenite::{Error, Message};
+use crate::response_base::ResponseData;
+use crate::socket_tool::{AbstractWsMode, HeartbeatType};
+
+pub enum BitgetSwapWsType {
+    Public,
+    Private,
+}
+
+#[derive(Clone)]
+pub enum BitgetSwapSubscribeType {
+    PuTrade,
+    PuBooks1,
+    PuCandle1m,
+
+    PrAccount,
+    PrPosition,
+    PrOrders,
+}
+
+#[derive(Clone)]
+#[allow(dead_code)]
+pub struct BitgetSwapLogin {
+    pub api_key: String,
+    pub secret_key: String,
+    pub passphrase_key: String,
+}
+
+#[derive(Clone)]
+pub struct BitgetSwapWs {
+    tag: String,                                              // 类型
+    address_url: String,                                        // 地址
+    login_param: Option<BitgetSwapLogin>,                       // 账号
+    symbol_s: Vec<String>,                                      // 币对
+    subscribe_types: Vec<BitgetSwapSubscribeType>,              // 订阅
+    heartbeat_time: u64,                                        // 心跳间隔
+}
+
+impl BitgetSwapWs {
+    pub fn new(is_colo: bool, login_param: Option<BitgetSwapLogin>, ws_type: BitgetSwapWsType) -> BitgetSwapWs {
+        return BitgetSwapWs::new_with_tag("default-BitgetSwapWs".to_string(), is_colo, login_param, ws_type);
+    }
+
+    pub fn new_with_tag(tag: String, is_colo: bool, login_param: Option<BitgetSwapLogin>, ws_type: BitgetSwapWsType) -> BitgetSwapWs {
+        let address_url = match ws_type {
+            BitgetSwapWsType::Public => {
+                "wss://ws.bitget.com/v2/ws/public".to_string()
+            }
+            BitgetSwapWsType::Private => {
+                "wss://ws.bitget.com/v2/ws/private".to_string()
+            }
+        };
+
+        if is_colo {
+            info!("开启高速(未配置，走普通：{})通道",address_url);
+        } else {
+            info!("走普通通道：{}",address_url);
+        }
+
+        BitgetSwapWs {
+            tag,
+            address_url,
+            login_param,
+            symbol_s: vec![],
+            subscribe_types: vec![],
+            heartbeat_time: 1000 * 10
+        }
+    }
+
+    // 添加订阅信息
+    pub fn set_subscribe(&mut self, subscribe_types: Vec<BitgetSwapSubscribeType>) {
+        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_uppercase();
+            // 字符串替换
+            *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 {
+                BitgetSwapSubscribeType::PuTrade => false,
+                BitgetSwapSubscribeType::PuBooks1 => false,
+                BitgetSwapSubscribeType::PuCandle1m => false,
+
+                BitgetSwapSubscribeType::PrAccount => true,
+                BitgetSwapSubscribeType::PrOrders => true,
+                BitgetSwapSubscribeType::PrPosition => true
+            } {
+                return true;
+            }
+        }
+        false
+    }
+
+    /*******************************************************************************************************/
+    /*****************************************工具函数*******************************************************/
+    /*******************************************************************************************************/
+    // 枚举解析成json
+    pub fn enum_to_json(symbol: String, subscribe_type: BitgetSwapSubscribeType) -> Value {
+        match subscribe_type {
+            // 公共订阅
+            BitgetSwapSubscribeType::PuTrade => {
+                json!({
+                    "instType": "USDT-FUTURES",
+                    "channel": "trade",
+                    "instId": symbol,
+                })
+            },
+            BitgetSwapSubscribeType::PuBooks1 => {
+                json!({
+                    "instType": "USDT-FUTURES",
+                    "channel": "books1",
+                    "instId": symbol,
+                })
+            },
+            BitgetSwapSubscribeType::PuCandle1m => {
+                json!({
+                    "instType": "USDT-FUTURES",
+                    "channel": "candle1m",
+                    "instId": symbol,
+                })
+            },
+
+            // 私有订阅
+            BitgetSwapSubscribeType::PrAccount => {
+                json!({
+                    "instType": "USDT-FUTURES",
+                    "channel": "account",
+                    "coin": "default",
+                })
+            },
+            BitgetSwapSubscribeType::PrPosition => {
+                json!({
+                    "instType": "USDT-FUTURES",
+                    "channel": "positions",
+                    "instId": "default"
+                })
+            },
+            BitgetSwapSubscribeType::PrOrders => {
+                json!({
+                    "instType": "USDT-FUTURES",
+                    "channel": "orders",
+                    "instId": "default"
+                })
+            },
+        }
+    }
+
+    // 订阅信息生成
+    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_json(symbol.clone(), subscribe_type.clone());
+                params.push(ty_str);
+            }
+        }
+        let str = json!({
+            "op": "subscribe",
+            "args": params
+        });
+
+        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 address_url = self.address_url.clone();
+        let tag = self.tag.clone();
+        let heartbeat_time = self.heartbeat_time.clone();
+
+        // 设置订阅
+        let subscription = self.get_subscription();
+        let subscribe_array = vec![subscription.to_string()];
+
+        //心跳-- 方法内部线程启动
+        let write_tx_clone1 = Arc::clone(write_tx_am);
+        tokio::spawn(async move {
+            let ping_str = json!("ping");
+            AbstractWsMode::ping_or_pong(write_tx_clone1, HeartbeatType::Custom(ping_str.as_str().unwrap().to_string()), heartbeat_time).await;
+        });
+
+        //链接
+        let login_param = self.login_param.clone();
+        let write_tx_clone2 = Arc::clone(write_tx_am);
+        let t2 = tokio::spawn(async move {
+            let write_to_socket_rx_arc = Arc::new(Mutex::new(write_to_socket_rx));
+
+            loop {
+                if !is_shutdown_arc.load(Ordering::Relaxed) {
+                    break;
+                }
+
+                info!("bitget_usdt_swap socket 连接中……");
+
+                // 登录相关
+                if login_is {
+                    let login_param_c = login_param.clone().unwrap();
+                    let timestamp = Utc::now().timestamp().to_string();
+                    // 时间戳 + 请求类型+ 请求参数字符串
+                    let message = format!("{}GET{}", timestamp, "/user/verify");
+                    let hmac_key = hmac::Key::new(hmac::HMAC_SHA256, login_param_c.secret_key.as_bytes());
+                    let result = hmac::sign(&hmac_key, &message.as_bytes());
+                    let sign = base64::encode(result);
+
+                    let login_json = json!({
+                        "op": "login",
+                        "args": [{
+                            "apiKey": login_param_c.api_key,
+                            "passphrase": login_param_c.passphrase_key,
+                            "timestamp": timestamp,
+                            "sign": sign
+                        }]
+                    });
+                    let login_str = login_json.to_string();
+                    info!("发起ws登录: {}", login_str);
+                    let write_tx_c = Arc::clone(&write_tx_clone2);
+                    AbstractWsMode::send_subscribe(write_tx_c, Message::Text(login_str)).await;
+                }
+
+                // ws层重连
+                AbstractWsMode::ws_connect_async(is_shutdown_arc.clone(), handle_function.clone(), address_url.clone(),
+                                                 login_is, tag.clone(), subscribe_array.clone(), write_to_socket_rx_arc.clone(),
+                                                 Self::message_text, Self::message_ping, Self::message_pong, Self::message_binary).await;
+
+                warn!("bitget_usdt_swap socket 断连，重连中……");
+            }
+        });
+        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::Null));
+    }
+    // 数据解析-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 {
+        let mut res_data = ResponseData::new("".to_string(), 200, text.clone(), Value::Null);
+        match text.as_str() {
+            "pong" => {
+                res_data.code = -301;
+                res_data.channel = "pong".to_string();
+                res_data.message = "success".to_string();
+            },
+            _ => {
+                let json_value: Value = serde_json::from_str(&text).unwrap();
+
+                if json_value.get("event").is_some() && json_value["event"].as_str() == Some("login") {
+                    if json_value.get("code").is_some() && json_value["code"] == 0 {
+                        res_data.message = "登陆成功".to_string();
+                    } else {
+                        res_data.message = format!("登陆失败:({},{})", json_value.get("code").as_ref().unwrap(), json_value.get("msg").as_ref().unwrap());
+                    }
+                    res_data.channel = "login".to_string();
+                    res_data.code = -200;
+                    res_data.data = json_value.clone();
+                } else if json_value.get("event").is_some() && json_value["event"].as_str() == Some("subscribe") {
+                    res_data.code = -201;
+                    res_data.data = json_value.clone();
+                    res_data.channel = format!("{}", json_value["arg"]["channel"].as_str().unwrap());
+                    res_data.message = "success".to_string();
+                } else if json_value.get("action").is_some() {
+                    res_data.data = json_value.clone();
+                    res_data.message = "success".to_string();
+                    res_data.channel = format!("{}", json_value["arg"]["channel"].as_str().unwrap());
+                    res_data.reach_time = json_value["ts"].as_i64().unwrap() * 1000;
+                }
+            }
+        }
+
+        res_data
+    }
+}

exchanges/src/lib.rs

@@ -15,3 +15,5 @@ pub mod mexc_swap_rest;
 pub mod mexc_swap_ws;
 pub mod bybit_swap_rest;
 pub mod bybit_swap_ws;
+pub mod bitget_swap_rest;
+pub mod bitget_swap_ws;

src/bitget_usdt_swap_data_listener.rs

@@ -0,0 +1,117 @@
+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 rust_decimal::Decimal;
+use tokio::sync::{Mutex};
+use tracing::info;
+use exchanges::bitget_swap_rest::BitgetSwapRest;
+use exchanges::bitget_swap_ws::{BitgetSwapSubscribeType, BitgetSwapWs, BitgetSwapWsType};
+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 = "bitget_usdt_swap";
+
+lazy_static! {
+    // 给其他模块使用
+    pub static ref INDICATOR_MAP: Mutex<HashMap<String, Indicators>> = Mutex::new(HashMap::new());
+
+    static ref TRADES_MAP: Mutex<TradeMap> = Mutex::new(HashMap::new());
+    static ref MUL_MAP: Mutex<HashMap<String, Decimal>> = Mutex::new(HashMap::new());
+}
+
+pub async fn run_listener(is_shutdown_arc: Arc<AtomicBool>) {
+    let name = "bitget_usdt_swap_listener";
+    // 订阅所有币种
+    let login = BTreeMap::new();
+    let mut bitget_rest = BitgetSwapRest::new(false, login);
+    let response = bitget_rest.get_all_contracts().await;
+    let mut symbols = vec![];
+    if response.code == 200 {
+        let symbol_infos = response.data.as_array().unwrap();
+        let mut mul_map = MUL_MAP.lock().await;
+        for symbol_info in symbol_infos {
+            let symbol = symbol_info["symbol"].as_str().unwrap().replace("USDT", "_USDT");
+
+            mul_map.insert(symbol.clone(), Decimal::ONE);
+            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 = BitgetSwapWs::new_with_tag(ws_name, false, None, BitgetSwapWsType::Public);
+            ws.set_subscribe(vec![
+                BitgetSwapSubscribeType::PuTrade,
+            ]);
+
+            // 建立链接
+            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 * 2;
+            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 {
+            tokio::time::sleep(Duration::from_secs(1800)).await;
+            delete_db_by_exchange(EXCHANGE_NAME, "trades", 5 * 60).await;
+        }
+    });
+}
+
+// 读取数据
+pub async fn data_listener(response: ResponseData) {
+    if response.code != 200 {
+        return;
+    }
+    match response.channel.as_str() {
+        // 订单流数据
+        "trade" => {
+            let mut trades = ExchangeStructHandler::trades_handle(ExchangeEnum::BitgetSwap, &response);
+            let mul_map = MUL_MAP.lock().await;
+
+            for trade in trades.iter_mut() {
+                // 真实交易量处理，因为gate的量都是张数
+                let mul = mul_map[trade.symbol.as_str()];
+                let mut real_size = trade.size * mul * trade.price;
+                real_size.rescale(2);
+                trade.size = real_size;
+
+                // 更新到本地数据库
+                let trades_map = TRADES_MAP.lock().await;
+                update_trade(trade, trades_map, EXCHANGE_NAME).await;
+            }
+        }
+        _ => {
+            info!("48 未知的数据类型: {:?}", response)
+        }
+    }
+}

src/main.rs

@@ -8,6 +8,7 @@ mod coinex_usdt_swap_data_listener;
 mod phemex_usdt_swap_data_listener;
 mod mexc_usdt_swap_data_listener;
 mod bybit_usdt_swap_data_listener;
+mod bitget_usdt_swap_data_listener;
 mod msv;
 mod rank;
 
@@ -40,6 +41,7 @@ async fn main() {
     phemex_usdt_swap_data_listener::run_listener(running.clone()).await;
     mexc_usdt_swap_data_listener::run_listener(running.clone()).await;
     bybit_usdt_swap_data_listener::run_listener(running.clone()).await;
+    bitget_usdt_swap_data_listener::run_listener(running.clone()).await;
     // panic错误捕获，panic级别的错误直接退出
     // let panic_running = running.clone();
     std::panic::set_hook(Box::new(move |panic_info| {

src/server.rs

@@ -5,7 +5,7 @@ use serde::{Deserialize, Serialize};
 use serde_json::{json, Value};
 use tracing::{info};
 use actix_cors::Cors;
-use crate::{binance_usdt_swap_data_listener, bybit_usdt_swap_data_listener, coinex_usdt_swap_data_listener, gate_usdt_swap_data_listener, mexc_usdt_swap_data_listener, phemex_usdt_swap_data_listener, rank};
+use crate::{binance_usdt_swap_data_listener, bitget_usdt_swap_data_listener, bybit_usdt_swap_data_listener, coinex_usdt_swap_data_listener, gate_usdt_swap_data_listener, mexc_usdt_swap_data_listener, phemex_usdt_swap_data_listener, rank};
 
 // 定义用于反序列化查询参数的结构体
 #[derive(Serialize, Deserialize, Clone)]
@@ -65,6 +65,9 @@ async fn get_rank_list(query: web::Query<RankQuery>) -> impl Responder {
         "bybit_usdt_swap" => {
             bybit_usdt_swap_data_listener::INDICATOR_MAP.lock().await
         },
+        "bitget_usdt_swap" => {
+            bitget_usdt_swap_data_listener::INDICATOR_MAP.lock().await
+        },
         _ => {
             let response = Response {
                 query: serde_json::to_value(&query.into_inner()).unwrap(),
@@ -112,6 +115,7 @@ async fn get_exchanges() -> impl Responder {
         "phemex_usdt_swap",
         "mexc_usdt_swap",
         "bybit_usdt_swap",
+        "bitget_usdt_swap",
     ];
     let response_data = json!(exchanges);
 

standard/src/bitget_swap.rs

@@ -0,0 +1,559 @@
+use std::collections::{BTreeMap};
+use exchanges::bitget_swap_rest::BitgetSwapRest;
+use std::io::{Error, ErrorKind};
+use tokio::sync::mpsc::Sender;
+use std::str::FromStr;
+use async_trait::async_trait;
+use rust_decimal::{Decimal, MathematicalOps};
+use rust_decimal::prelude::ToPrimitive;
+use rust_decimal_macros::dec;
+use serde_json::{json, Value};
+use tracing::{error, info};
+use crate::exchange::ExchangeEnum;
+use crate::{Account, Market, Order, Platform, Position, PositionModeEnum, Ticker, utils};
+
+#[allow(dead_code)]
+#[derive(Clone)]
+pub struct BitgetSwap {
+    exchange: ExchangeEnum,
+    symbol: String,
+    is_colo: bool,
+    params: BTreeMap<String, String>,
+    request: BitgetSwapRest,
+    market: Market,
+    order_sender: Sender<Order>,
+    error_sender: Sender<Error>,
+}
+
+impl BitgetSwap {
+    pub async fn new(symbol: String, is_colo: bool, params: BTreeMap<String, String>, order_sender: Sender<Order>, error_sender: Sender<Error>) -> BitgetSwap {
+        let market = Market::new();
+        let mut bitget_swap = BitgetSwap {
+            exchange: ExchangeEnum::BitgetSwap,
+            symbol: symbol.to_uppercase(),
+            is_colo,
+            params: params.clone(),
+            request: BitgetSwapRest::new(is_colo, params.clone()),
+            market,
+            order_sender,
+            error_sender,
+        };
+        bitget_swap.market = BitgetSwap::get_market(&mut bitget_swap).await.unwrap();
+        // 修改持仓模式
+        let mode_result = bitget_swap.set_dual_mode("", true).await;
+        match mode_result {
+            Ok(ok) => {
+                info!("BitgetSwap:设置持仓模式成功！{:?}", ok);
+            }
+            Err(error) => {
+                error!("BitgetSwap:设置持仓模式失败！{:?}", error)
+            }
+        }
+        // 设置持仓杠杆
+        // let lever_rate_result = bitget_swap.set_dual_leverage("10").await;
+        // match lever_rate_result {
+        //     Ok(ok) => {
+        //         info!("BitgetSwap:设置持仓杠杆成功！{:?}", ok);
+        //     }
+        //     Err(error) => {
+        //         error!("BitgetSwap:设置持仓杠杆失败！{:?}", error)
+        //     }
+        // }
+
+        return bitget_swap;
+    }
+}
+
+#[async_trait]
+impl Platform for BitgetSwap {
+    fn clone_box(&self) -> Box<dyn Platform + Send + Sync> { Box::new(self.clone()) }
+
+    fn get_self_exchange(&self) -> ExchangeEnum { ExchangeEnum::BitgetSwap }
+
+    fn get_self_symbol(&self) -> String { self.symbol.clone() }
+
+    fn get_self_is_colo(&self) -> bool { self.is_colo }
+
+    fn get_self_params(&self) -> BTreeMap<String, String> { self.params.clone() }
+
+    fn get_self_market(&self) -> Market { self.market.clone() }
+
+    fn get_request_delays(&self) -> Vec<i64> {
+        // self.request.get_delays()
+        vec![]
+    }
+
+    fn get_request_avg_delay(&self) -> Decimal {
+        // self.request.get_avg_delay()
+        Decimal::ZERO
+    }
+
+    fn get_request_max_delay(&self) -> i64 { 0 }
+
+    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"].as_str().unwrap().to_string();
+            Ok(result)
+        } else {
+            Err(Error::new(ErrorKind::Other, res_data.to_string()))
+        }
+    }
+
+    async fn get_account(&mut self) -> Result<Account, Error> {
+        let response = self.request.get_account_info().await;
+
+        if response.code == 200 {
+            for data in response.data.as_array().unwrap() {
+                if data["marginCoin"].as_str().unwrap() != "USDT" {
+                    continue
+                }
+
+                // 格式化account信息
+                let mut account = Account {
+                    coin: data["marginCoin"].to_string(),
+                    balance: Decimal::from_str(data["accountEquity"].as_str().unwrap()).unwrap(),
+                    available_balance: Decimal::from_str(data["available"].as_str().unwrap()).unwrap(),
+                    frozen_balance: Default::default(),
+                    stocks: Default::default(),
+                    available_stocks: Default::default(),
+                    frozen_stocks: Default::default(),
+                };
+                account.frozen_balance = account.balance - account.available_balance;
+
+                return Ok(account)
+            }
+
+            Err(Error::new(ErrorKind::NotFound, format!("bitget_usdt_swap 未能找到USDT账户：{}。", response.to_string())))
+        } else {
+            Err(Error::new(ErrorKind::Other, response.to_string()))
+        }
+    }
+
+    async fn get_spot_account(&mut self) -> Result<Vec<Account>, Error> {
+        Err(Error::new(ErrorKind::NotFound, "bitget_swap get_spot_account：该交易所方法未实现".to_string()))
+    }
+
+    async fn get_position(&mut self) -> Result<Vec<Position>, Error> { Err(Error::new(ErrorKind::NotFound, "bitget_swap get_position：该交易所方法未实现".to_string())) }
+
+    async fn get_positions(&mut self) -> Result<Vec<Position>, Error> {
+        let params = json!({
+            "productType": "USDT-FUTURES",
+            "marginCoin": "USDT"
+        });
+        let response = self.request.get_all_position(params).await;
+        info!(?response);
+
+        if response.code != 200 {
+            return Err(Error::new(ErrorKind::NotFound, format!("bitget_swap 获取仓位异常{:?}", response).to_string()))
+        }
+
+        // 正常处理持仓信息
+        let mut positions: Vec<Position> = vec![];
+        if response.data.is_null() {
+            return Ok(positions)
+        }
+
+        let positions_json = response.data.as_array().unwrap();
+        for position_json in positions_json {
+            let symbol = position_json["symbol"].as_str().unwrap().to_string();
+            let margin_level = Decimal::from_str(position_json["leverage"].as_str().unwrap()).unwrap();
+            let amount = Decimal::from_str(position_json["total"].as_str().unwrap()).unwrap();
+            let frozen_amount = Decimal::from_str(position_json["locked"].as_str().unwrap()).unwrap();
+            let price = Decimal::from_str(position_json["openPriceAvg"].as_str().unwrap()).unwrap();
+            let profit = Decimal::from_str(position_json["unrealizedPL"].as_str().unwrap()).unwrap();
+            let position_mode = match position_json["posMode"].as_str().unwrap() {
+                "hedge_mode" => {
+                    match position_json["holdSide"].as_str().unwrap() {
+                        "short" => {
+                            PositionModeEnum::Short
+                        }
+                        "long" => {
+                            PositionModeEnum::Long
+                        },
+                        _ => {
+                            panic!("bitget_usdt_swap: 未知的持仓模式与持仓方向: {}, {}",
+                                   position_json["posMode"].as_str().unwrap(), position_json["holdSide"].as_str().unwrap())
+                        }
+                    }
+                },
+                "one_way_mode" => {
+                    PositionModeEnum::Both
+                },
+                _ => {
+                    panic!("bitget_usdt_swap: 未知的持仓模式: {}", position_json["posMode"].as_str().unwrap())
+                }
+            };
+            let margin = Decimal::from_str(position_json["marginSize"].as_str().unwrap()).unwrap();
+
+            positions.push(Position {
+                symbol,
+                margin_level,
+                amount,
+                frozen_amount,
+                price,
+                profit,
+                position_mode,
+                margin,
+            });
+        }
+
+        Ok(positions)
+    }
+
+    async fn get_ticker(&mut self) -> Result<Ticker, Error> {
+        return self.get_ticker_symbol(self.symbol.clone()).await
+    }
+
+    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_tickers(symbol_format).await;
+        if res_data.code == 200 {
+            let res_data_json = res_data.data;
+            let ticker_info = res_data_json[0].clone();
+            let time = (Decimal::from_str(&*ticker_info["ts"].as_str().unwrap()).unwrap() / dec!(1000)).floor().to_i64().unwrap();
+            let result = Ticker {
+                time,
+                high: Decimal::from_str(ticker_info["high24h"].as_str().unwrap()).unwrap(),
+                low: Decimal::from_str(ticker_info["low24h"].as_str().unwrap()).unwrap(),
+                sell: Decimal::from_str(ticker_info["askPr"].as_str().unwrap()).unwrap(),
+                buy: Decimal::from_str(ticker_info["bidPr"].as_str().unwrap()).unwrap(),
+                last: Decimal::from_str(ticker_info["lastPr"].as_str().unwrap()).unwrap(),
+                volume: Decimal::from_str(ticker_info["quoteVolume"].as_str().unwrap()).unwrap(),
+            };
+            Ok(result)
+        } else {
+            Err(Error::new(ErrorKind::Other, res_data.to_string()))
+        }
+    }
+
+    async fn get_market(&mut self) -> Result<Market, Error> {
+        self.get_market_symbol(self.symbol.clone()).await
+    }
+
+    async fn get_market_symbol(&mut self, symbol: String) -> Result<Market, Error> {
+        let symbol_format = utils::format_symbol(symbol.clone(), "");
+        let response = self.request.get_contracts(symbol_format.clone()).await;
+
+        if response.code == 200 {
+            let res_data_json = response.data.as_array().unwrap();
+            let market_info = res_data_json[0].clone();
+
+            info!(?market_info);
+            if !market_info["symbol"].as_str().unwrap().to_string().eq(&symbol_format) {
+                return Err(Error::new(ErrorKind::NotFound, format!("符号未找到：symbol={}, response={:?}", symbol_format, response))).unwrap();
+            }
+
+            let base_asset = market_info["baseCoin"].as_str().unwrap().to_string();
+            let quote_asset = market_info["quoteCoin"].as_str().unwrap().to_string();
+            let price_precision = Decimal::from_str(market_info["pricePlace"].as_str().unwrap()).unwrap();
+            let tick_size = Decimal::TEN.powd(Decimal::NEGATIVE_ONE * price_precision);
+            let amount_precision = Decimal::from_str(market_info["volumePlace"].as_str().unwrap()).unwrap();
+            let amount_size = Decimal::TEN.powd(Decimal::NEGATIVE_ONE * amount_precision);
+            let min_qty = Decimal::NEGATIVE_ONE;
+            let max_qty = Decimal::NEGATIVE_ONE;
+            // let ct_val = Decimal::from_str(&market_info["sizeMultiplier"].as_str().unwrap()).unwrap();
+            let ct_val = Decimal::ONE;
+
+            let result = Market {
+                symbol: format!("{}_{}", base_asset, quote_asset),
+                base_asset,
+                quote_asset,
+                tick_size,
+                amount_size,
+                price_precision,
+                amount_precision,
+                min_qty,
+                max_qty,
+                min_notional: min_qty,
+                max_notional: max_qty,
+                ct_val,
+            };
+            Ok(result)
+        } else {
+            Err(Error::new(ErrorKind::Other, response.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 params = json!({
+            "symbol": symbol_format,
+            "productType": "USDT-FUTURES",
+            "clientOid": custom_id,
+            "orderId": order_id
+        });
+
+        let ct_val = self.market.ct_val;
+        let response = self.request.get_order(params).await;
+        if response.code == 200 {
+            let res_data_json = response.data;
+            let result = format_order_item(res_data_json, ct_val);
+            Ok(result)
+        } else {
+            Err(Error::new(ErrorKind::Other, response.to_string()))
+        }
+    }
+
+    async fn get_orders_list(&mut self, _status: &str) -> Result<Vec<Order>, Error> {
+        Err(Error::new(ErrorKind::NotFound, "bitget_swap get_orders_list：该交易所方法未实现".to_string()))
+        // let symbol_format = utils::format_symbol(self.symbol.clone(), "");
+        // let ct_val = self.market.ct_val;
+        // let res_data = self.request.get_unfilled_orders(symbol_format.to_string(), "".to_string(), "".to_string(), "".to_string(), "100".to_string(), "".to_string()).await;
+        // if res_data.code == 200 {
+        //     let res_data_str = &res_data.data;
+        //     let res_data_json: Vec<serde_json::Value> = serde_json::from_str(res_data_str).unwrap();
+        //     let result = res_data_json.iter().map(|item| format_order_item(item.clone(), ct_val)).collect();
+        //     Ok(result)
+        // } else {
+        //     Err(Error::new(ErrorKind::Other, res_data.to_string()))
+        // }
+    }
+
+    async fn take_order(&mut self, custom_id: &str, origin_side: &str, price: Decimal, amount: Decimal) -> Result<Order, Error> {
+        let ct_val = self.market.ct_val;
+
+        return self.take_order_symbol(self.symbol.clone(), ct_val, 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, "");
+        let final_size = amount / ct_val;
+        let mut params = json!({
+            "symbol": symbol_format,
+            "clientOid": custom_id,
+            "productType": "USDT-FUTURES",
+            "marginMode": "crossed",
+            "marginCoin": "USDT",
+            "size": final_size.to_string()
+        });
+        if price.eq(&Decimal::ZERO) {
+            params["orderType"] = json!("market");
+            params["force"] = json!("gtc");
+        } else {
+            params["price"] = json!(price.to_string());
+            params["orderType"] = json!("limit");
+            params["force"] = json!("gtc");
+        };
+        match origin_side {
+            "kd" => {
+                params["side"] = json!("buy");
+                params["tradeSide"] = json!("open");
+            }
+            "pd" => {
+                params["side"] = json!("buy");
+                params["tradeSide"] = json!("close");
+            }
+            "kk" => {
+                params["side"] = json!("sell");
+                params["tradeSide"] = json!("open");
+            }
+            "pk" => {
+                params["side"] = json!("sell");
+                params["tradeSide"] = json!("close");
+            }
+            _ => { panic!("bitget_usdt_swap 下单参数错误"); }
+        };
+        let res_data = self.request.swap_order(params).await;
+        if res_data.code != 200 {
+            return Err(Error::new(ErrorKind::Other, res_data.to_string()))
+        }
+
+        let res_data_json = res_data.data;
+        let result = Order {
+            id: res_data_json["orderId"].as_str().unwrap().to_string(),
+            custom_id: res_data_json["clientOid"].as_str().unwrap().to_string(),
+            price: Decimal::ZERO,
+            amount: Decimal::ZERO,
+            deal_amount: Decimal::ZERO,
+            avg_price: Decimal::ZERO,
+            status: "NEW".to_string(),
+            order_type: "".to_string()
+        };
+        return Ok(result)
+    }
+
+    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 params = json!({
+            "symbol": symbol_format,
+            "productType": "USDT-FUTURES",
+            "clientOid": custom_id,
+            "orderId": order_id
+        });
+        let response = self.request.cancel_order(params).await;
+
+        // 取消失败，进行报错
+        if response.code != 200 {
+            return Err(Error::new(ErrorKind::NotFound, response.to_string()));
+        }
+
+        let res_data_json = response.data;
+        let result = Order {
+            id: res_data_json["orderId"].as_str().unwrap().to_string(),
+            custom_id: res_data_json["clientOid"].as_str().unwrap().to_string(),
+            price: Decimal::ZERO,
+            amount: Decimal::ZERO,
+            deal_amount: Decimal::ZERO,
+            avg_price: Decimal::ZERO,
+            status: "REMOVE".to_string(),
+            order_type: "".to_string()
+        };
+        Ok(result)
+    }
+
+    async fn cancel_orders(&mut self) -> Result<Vec<Order>, Error> {
+        Err(Error::new(ErrorKind::NotFound, "bitget_swap cancel_orders：该交易所方法未实现".to_string()))
+    }
+
+    async fn cancel_orders_all(&mut self) -> Result<Vec<Order>, Error> {
+        let response = self.request.get_pending_orders().await;
+        info!(?response);
+        if response.code == 200 {
+            let mut result = vec![];
+            let symbol_format = utils::format_symbol(self.symbol.clone(), "");
+
+            if !response.data["entrustedList"].is_null() {
+                let orders_res_data_json = response.data["entrustedList"].as_array().unwrap();
+                for order in orders_res_data_json {
+                    let order_id = order["orderId"].as_str().unwrap().to_string();
+                    let symbol = symbol_format.clone();
+                    let params = json!({
+                        "symbol": symbol,
+                        "productType": "USDT-FUTURES",
+                        "orderId": order_id,
+                    });
+                    let cancel_res_data = self.request.cancel_order(params).await;
+                    if cancel_res_data.code == 200 {
+                        let cancel_res_data_json = cancel_res_data.data;
+                        result.push(Order {
+                            id: cancel_res_data_json["orderId"].as_str().unwrap().to_string(),
+                            custom_id: cancel_res_data_json["clientOid"].as_str().unwrap().to_string(),
+                            price: Decimal::ZERO,
+                            amount: Decimal::ZERO,
+                            deal_amount: Decimal::ZERO,
+                            avg_price: Decimal::ZERO,
+                            status: "REMOVE".to_string(),
+                            order_type: "".to_string()
+                        });
+                    } else {
+                        return Err(Error::new(ErrorKind::Other, cancel_res_data.to_string()));
+                    }
+                }
+            }
+            Ok(result)
+        } else {
+            Err(Error::new(ErrorKind::Other, response.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, "bitget_swap take_stop_loss_order：该交易所方法未实现".to_string()))
+    }
+
+    async fn cancel_stop_loss_order(&mut self, _order_id: &str) -> Result<Value, Error> {
+        Err(Error::new(ErrorKind::NotFound, "bitget_swap cancel_stop_loss_order：该交易所方法未实现".to_string()))
+    }
+
+    async fn set_dual_mode(&mut self, _coin: &str, is_dual_mode: bool) -> Result<String, Error> {
+        let pos_mode = if is_dual_mode {
+            "hedge_mode"
+        } else {
+            "one_way_mode"
+        };
+        let params = json!({
+            "productType": "USDT-FUTURES",
+            "posMode": pos_mode,
+        });
+        let response = self.request.set_position_mode(params).await;
+
+        if response.code != 200 {
+            return Err(Error::new(ErrorKind::Other, format!("设置持仓模式失败:{:?}", response).to_string()))
+        }
+
+        return Ok(response.data.to_string());
+    }
+
+    async fn set_dual_leverage(&mut self, leverage: &str) -> Result<String, Error> {
+        let params = json!({
+            "symbol": "ETHUSDT",
+            "productType": "USDT-FUTURES",
+            "marginCoin": "USDT",
+            "leverage": leverage
+        });
+        let response = self.request.set_leverage(params).await;
+
+        if response.code != 200 {
+            return Err(Error::new(ErrorKind::Other, format!("设置杠杆失败:{:?}", response).to_string()))
+        }
+
+        return Ok(response.data.to_string());
+    }
+
+    async fn set_auto_deposit_status(&mut self, _status: bool) -> Result<String, Error> {
+        Err(Error::new(ErrorKind::NotFound, "bitget_swap set_auto_deposit_status：该交易所方法未实现".to_string()))
+    }
+
+    async fn wallet_transfers(&mut self, _coin: &str, _from: &str, _to: &str, _amount: Decimal) -> Result<String, Error> {
+        Err(Error::new(ErrorKind::NotFound, "bitget_swap wallet_transfers：该交易所方法未实现".to_string()))
+        // let coin_format = coin.to_string().to_uppercase();
+        // let res_data = self.request.wallet_transfer(from.to_string(), to.to_string(), amount.to_string(), coin_format.clone(), "".to_string(), "".to_string()).await;
+        // if res_data.code == 200 {
+        //     let res_data_str = &res_data.data;
+        //     let result = res_data_str.clone();
+        //     Ok(result)
+        // } else {
+        //     Err(Error::new(ErrorKind::Other, res_data.to_string()))
+        // }
+    }
+}
+
+// pub fn format_account_info(balance_data: Value) -> Account {
+//     let balance_coin = balance_data["coin"].as_str().unwrap().to_string().to_uppercase();
+//     let available_balance = Decimal::from_str(balance_data["available"].as_str().unwrap()).unwrap();
+//     let frozen_balance = Decimal::from_str(balance_data["frozen"].as_str().unwrap()).unwrap();
+//     let balance = available_balance + frozen_balance;
+//
+//     Account {
+//         coin: balance_coin,
+//         balance,
+//         available_balance,
+//         frozen_balance,
+//         stocks: Decimal::ZERO,
+//         available_stocks: Decimal::ZERO,
+//         frozen_stocks: Decimal::ZERO,
+//     }
+// }
+
+pub fn format_order_item(order: Value, ct_val: Decimal) -> Order {
+    let price = Decimal::from_str(order["price"].as_str().unwrap_or(order["priceAvg"].as_str().unwrap())).unwrap();
+    let size = Decimal::from_str(order["size"].as_str().unwrap()).unwrap();
+    let status = order["state"].as_str().unwrap();
+    let base_volume = Decimal::from_str(order["quoteVolume"].as_str().unwrap()).unwrap();
+    let avg_price = if order["priceAvg"].is_null() || order["priceAvg"].as_str().unwrap().is_empty() {
+        Decimal::ZERO
+    } else {
+        Decimal::from_str(order["priceAvg"].as_str().unwrap().to_string().as_str()).unwrap()
+    };
+
+    let amount = size * ct_val;
+    let deal_amount = base_volume * ct_val;
+    let custom_status = if ["filled", "cancelled"].contains(&status) {
+        "REMOVE".to_string()
+    } else if ["init", "live", "new", "partially_filled"].contains(&status) {
+        "NEW".to_string()
+    } else {
+        "NULL".to_string()
+    };
+    Order {
+        id: order["orderId"].as_str().unwrap().to_string(),
+        custom_id: order["clientOid"].as_str().unwrap().to_string(),
+        price,
+        amount,
+        deal_amount,
+        avg_price,
+        status: custom_status,
+        order_type: order["orderType"].as_str().unwrap().to_string()
+    }
+}

standard/src/bitget_swap_handle.rs

@@ -0,0 +1,29 @@
+use std::str::FromStr;
+use rust_decimal::Decimal;
+use exchanges::response_base::ResponseData;
+use crate::{Trade};
+
+pub fn format_trade_items(res_data: &ResponseData) -> Vec<Trade> {
+    let arg = res_data.data["arg"].clone();
+    let symbol = arg["instId"].as_str().unwrap().to_string().replace("USDT", "_USDT");
+    let result = res_data.data["data"].as_array().unwrap();
+    let mut trades = vec![];
+
+    for item in result {
+        let mut trade = Trade {
+            id: item["tradeId"].as_str().unwrap().to_string(),
+            time: Decimal::from_str(item["ts"].as_str().unwrap()).unwrap(),
+            size: Decimal::from_str(item["size"].as_str().unwrap()).unwrap(),
+            price: Decimal::from_str(item["price"].as_str().unwrap().to_string().as_str()).unwrap(),
+            symbol: symbol.to_string(),
+        };
+
+        if item["side"].as_str().unwrap().eq("sell") {
+            trade.size = trade.size * Decimal::NEGATIVE_ONE;
+        }
+
+        trades.push(trade)
+    }
+
+    return trades;
+}

standard/src/exchange.rs

@@ -3,6 +3,7 @@ use std::io::Error;
 use tokio::sync::mpsc::Sender;
 use crate::{Order, Platform};
 use crate::binance_swap::BinanceSwap;
+use crate::bitget_swap::BitgetSwap;
 use crate::bybit_swap::BybitSwap;
 use crate::coinex_swap::CoinexSwap;
 use crate::gate_swap::GateSwap;
@@ -23,6 +24,7 @@ pub enum ExchangeEnum {
     PhemexSwap,
     MexcSwap,
     BybitSwap,
+    BitgetSwap,
 }
 
 /// Exchange结构体
@@ -85,6 +87,9 @@ impl Exchange {
             ExchangeEnum::BybitSwap => {
                 Box::new(BybitSwap::new(symbol, is_colo, params, order_sender, error_sender).await)
             }
+            ExchangeEnum::BitgetSwap => {
+                Box::new(BitgetSwap::new(symbol, is_colo, params, order_sender, error_sender).await)
+            }
         }
     }
 }

standard/src/exchange_struct_handler.rs

@@ -1,7 +1,7 @@
 use exchanges::response_base::ResponseData;
 use crate::exchange::ExchangeEnum;
-use crate::{binance_swap_handle, bybit_swap_handle, coinex_swap_handle, gate_swap_handle, mexc_swap_handle, phemex_swap_handle};
-use crate::{ Trade };
+use crate::{binance_swap_handle, bitget_swap_handle, bybit_swap_handle, coinex_swap_handle, gate_swap_handle, mexc_swap_handle, phemex_swap_handle};
+use crate::{Trade};
 
 #[allow(dead_code)]
 pub struct ExchangeStructHandler;
@@ -29,6 +29,9 @@ impl ExchangeStructHandler {
             ExchangeEnum::BybitSwap => {
                 bybit_swap_handle::format_trade_items(&res_data)
             }
+            ExchangeEnum::BitgetSwap => {
+                bitget_swap_handle::format_trade_items(&res_data)
+            }
         }
     }
 }

standard/src/lib.rs

@@ -25,6 +25,8 @@ pub mod mexc_swap;
 pub mod mexc_swap_handle;
 pub mod bybit_swap;
 pub mod bybit_swap_handle;
+pub mod bitget_swap;
+pub mod bitget_swap_handle;
 
 /// 持仓模式枚举
 /// - `Both`：单持仓方向