配置方式修改为json 统一参数类型

config.json

@@ -0,0 +1,32 @@
+{
+  "broker_id": "hl_3_7",
+  "account_name": "localhost_test_account",
+  "access_key": "4181c882718a95e72122ac1d52c88533",
+  "secret_key": "de82d1507b843ff08d81a0e9b878b721359f274937216b307834b570b676fa3c",
+  "pass_key": "",
+  "exchange": "gate_usdt_swap",
+  "pair": "bch_usdt",
+  "open": 0.9,
+  "close": 0.00017,
+  "lever_rate": 0.5,
+  "hold_coin": 0,
+  "interval": 100,
+  "ref_exchange": [
+    "binance_usdt_swap"
+  ],
+  "ref_pair": [
+    "sc_usdt"
+  ],
+  "used_pct": 0.9,
+  "stop_loss": 0.01,
+  "gamma": 0.999,
+  "grid": 1,
+  "colo": 0,
+  "log_level": "info",
+  "port": 5555,
+  "max_spread": 0.001,
+  "min_spread": 0.001,
+  "rl_num": 30,
+  "max_position_value": 20,
+  "ira": 20
+}

exchanges/tests/binance_swap_test.rs

@@ -1,73 +1,68 @@
-use std::cmp::max;
 use std::collections::BTreeMap;
 use std::sync::Arc;
 use std::sync::atomic::AtomicBool;
+use std::time::Duration;
+use chrono::Utc;
 
+use futures_util::StreamExt;
+use serde_json::json;
 use tokio::sync::Mutex;
-use tracing::{info, trace};
+use tokio_tungstenite::tungstenite::Message;
+use tracing::trace;
 
 use exchanges::binance_swap_rest::BinanceSwapRest;
 use exchanges::binance_swap_ws::{BinanceSwapLogin, BinanceSwapSubscribeType, BinanceSwapWs, BinanceSwapWsType};
-use exchanges::response_base::ResponseData;
-use global::trace_stack::TraceStack;
+use exchanges::socket_tool::AbstractWsMode;
 
 const ACCESS_KEY: &str = "";
 const SECRET_KEY: &str = "";
 
 
 //ws-订阅公共频道信息
-#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+#[tokio::test(flavor = "multi_thread", worker_threads = 5)]
 async fn ws_custom_subscribe() {
     global::log_utils::init_log_with_trace();
 
 
     let (write_tx, write_rx) = futures_channel::mpsc::unbounded();
-    let (read_tx, mut read_rx) = futures_channel::mpsc::unbounded::<ResponseData>();
+    let (read_tx, mut read_rx) = futures_channel::mpsc::unbounded();
 
     // let (write_tx, write_rx) = tokio::sync::broadcast::channel::<Message>(10);
     // let (read_tx, mut read_rx) = tokio::sync::broadcast::channel::<ResponseData>(10);
+    let login_param = BinanceSwapLogin {
+        api_key: ACCESS_KEY.to_string(),
+        api_secret: SECRET_KEY.to_string(),
+    };
 
-
-    let mut ws = get_ws(None);
+    let mut ws = get_ws(Option::from(login_param));
     ws.set_symbols(vec!["BTC_USDT".to_string()]);
     ws.set_subscribe(vec![
-        BinanceSwapSubscribeType::PuBookTicker,
+        // BinanceSwapSubscribeType::PuBookTicker,
         // BinanceSwapSubscribeType::PuAggTrade,
+        BinanceSwapSubscribeType::PuMarkPrice,
         // BinanceSwapSubscribeType::PuDepth20levels100ms,
+
+        BinanceSwapSubscribeType::PrAccount,
     ]);
 
 
     let write_tx_am = Arc::new(Mutex::new(write_tx));
-    let is_shutdown_arc = Arc::new(AtomicBool::new(true));
+    let bool_v1 = Arc::new(AtomicBool::new(true));
 
     //读取
-    let _is_shutdown_arc_clone = Arc::clone(&is_shutdown_arc);
+    let _bool_v1_clone = Arc::clone(&bool_v1);
     let _tr = tokio::spawn(async move {
         trace!("线程-数据读取-开启");
-        let mut max_delay = 0i64;
         loop {
-            // 从通道中接收并丢弃所有的消息，直到通道为空
-            while let Ok(Some(data)) = read_rx.try_next() {
-                // 消息被忽略
-                let mut trace_stack = TraceStack::new(0, Instant::now());
-                trace_stack.on_before_unlock_core();
-                trace_stack.on_after_network(data.time);
-
-                let delay = trace_stack.before_unlock_core - trace_stack.after_network;
-                max_delay = max(max_delay, delay);
-                info!("{}us, max={}us", delay, max_delay);
-
-                // 从通道中接收并丢弃所有的消息，直到通道为空
-                while let Ok(Some(_)) = read_rx.try_next() {
-                    // 消息被忽略
-                }
+            if let Some(data) = read_rx.next().await {
+                trace!("读取数据data:{:?}",data)
             }
         }
         // trace!("线程-数据读取-结束");
     });
 
     //写数据
-    // let bool_v2_clone = Arc::clone(&is_shutdown_arc);
+    // let bool_v2_clone = Arc::clone(&bool_v1);
     // let write_tx_clone = Arc::clone(&write_tx_am);
     // let su = ws.get_subscription();
     // let tw = tokio::spawn(async move {
@@ -90,7 +85,7 @@ async fn ws_custom_subscribe() {
 
     let t1 = tokio::spawn(async move {
         //链接
-        let bool_v3_clone = Arc::clone(&is_shutdown_arc);
+        let bool_v3_clone = Arc::clone(&bool_v1);
         ws.ws_connect_async(bool_v3_clone, &write_tx_am, write_rx, read_tx).await.expect("链接失败(内部一个心跳线程应该已经关闭了)");
         trace!("test 唯一线程结束--");
     });
@@ -99,76 +94,6 @@ async fn ws_custom_subscribe() {
     trace!("重启！");
     trace!("参考交易所关闭");
     return;
-
-    //************************************
-    //************************************
-    //************************************
-    //************************************
-    //************************************
-    //************************************
-    //************************************
-    //11 点31 分
-
-    // let mut is_shutdown_arc = Arc::new(AtomicBool::new(true));
-    // //创建读写通道
-    // let (write_tx, write_rx) = futures_channel::mpsc::unbounded();
-    // let (read_tx, mut read_rx) = futures_channel::mpsc::unbounded();
-    // // 封装 write_tx 到 Arc 和 Mutex
-    // let write_tx_am = Arc::new(Mutex::new(write_tx));
-    //
-    // //对象
-    // let mut ws = get_ws(None);
-    // // 币对
-    // ws.set_symbols(vec!["BTC_USDT".to_string()]);
-    // //订阅
-    // ws.set_subscribe(vec![
-    //     BinanceSwapSubscribeType::PuBookTicker,
-    //     BinanceSwapSubscribeType::PuAggTrade,
-    //     BinanceSwapSubscribeType::PuDepth20levels100ms,
-    // ]);
-    //
-    //
-    // //模拟业务场景 开启链接
-    // let is_shutdown_arc_clone = Arc::clone(&is_shutdown_arc);
-    // let write_tx_clone1 = Arc::clone(&write_tx_am);
-    // let t1 = tokio::spawn(async move {
-    //     ws.ws_connect_async(is_shutdown_arc_clone, write_tx_clone1, write_rx, &read_tx).await.unwrap();
-    //     trace!("ws_connect_async 完成");
-    // });
-    //
-    // //模拟业务场景 一直监听数据
-    // let t2 = tokio::spawn(async move {
-    //     loop {
-    //         if let Some(data) = read_rx.next().await {
-    //             trace!("读取数据data:{:?}",data)
-    //         }
-    //     }
-    //     trace!("数据读取退出 完成");
-    // });
-    //
-    //
-    // //模拟用户主动写入数据
-    // // let write_tx_clone2 = Arc::clone(&write_tx_am);
-    // // let t3 = tokio::spawn(async move {
-    // //     //模拟心跳
-    // //     loop {
-    // //         tokio::time::sleep(Duration::from_millis(5000)).await;
-    // //         let mut write_tx_clone = write_tx_clone2.lock().unwrap();
-    // //         match write_tx_clone.unbounded_send(Message::Pong(Vec::from("pong"))) {
-    // //             Ok(_) => {
-    // //                 trace!("发送心跳");
-    // //                 continue;
-    // //             }
-    // //             Err(_) => {
-    // //                 break;
-    // //             }
-    // //         }
-    // //     }
-    // //     trace!("主动推出 完成");
-    // // });
-    // // tokio::try_join!(y,y1,y2).unwrap();
-    // tokio::try_join!(t1,t2).unwrap();
-    // trace!("323123213");
 }
 
 //rest-获取服务器时间
@@ -201,6 +126,66 @@ async fn rest_get_account_test() {
     trace!(?rep_data)
 }
 
+//rest-调整开仓杠杆
+#[tokio::test]
+async fn rest_get_leverage_test() {
+    global::log_utils::init_log_with_trace();
+
+    let mut rest = get_rest();
+    let rep_data = rest.get_leverage("DOGEUSDT".to_string(),1).await;
+    trace!(?rep_data)
+}
+//rest-查询订单
+#[tokio::test]
+async fn rest_get_order_test() {
+    global::log_utils::init_log_with_trace();
+
+    let mut rest = get_rest();
+    let rep_data = rest.get_order("DOGEUSDT".to_string(),-1,"".to_string()).await;
+    trace!(?rep_data)
+}
+
+//rest-查看当前全部挂单
+#[tokio::test]
+async fn rest_get_open_orders_test() {
+    global::log_utils::init_log_with_trace();
+
+    let mut rest = get_rest();
+    let rep_data = rest.get_open_orders("DOGEUSDT".to_string()).await;
+    trace!(?rep_data)
+}
+
+//rest-查询所有订单(包括历史订单)
+#[tokio::test]
+async fn rest_get_all_orders_test() {
+    global::log_utils::init_log_with_trace();
+
+    let mut rest = get_rest();
+    let rep_data = rest.get_all_orders("DOGEUSDT".to_string(),100,0,0).await;
+    trace!(?rep_data)
+}
+
+//rest-当前最优挂单
+#[tokio::test]
+async fn rest_get_book_ticker_test() {
+    global::log_utils::init_log_with_trace();
+
+    let mut rest = get_rest();
+    let rep_data = rest.get_book_ticker("DOGEUSDT".to_string()).await;
+    trace!(?rep_data)
+}
+
+//rest-用户持仓风险V2
+#[tokio::test]
+async fn rest_get_position_risk_test() {
+    global::log_utils::init_log_with_trace();
+
+    let mut rest = get_rest();
+    let rep_data = rest.get_position_risk("DOGEUSDT".to_string()).await;
+    trace!(?rep_data)
+}
+
+
 
 //rest-根据币对 撤销全部订单
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
@@ -228,6 +213,48 @@ async fn rest_get_user_trades_test() {
     trace!(?rep_data)
 }
 
+//rest-生成 listenKey
+#[tokio::test]
+async fn rest_get_listenKey_test() {
+    global::log_utils::init_log_with_trace();
+
+    let mut rest = get_rest();
+    let rep_data = rest.get_listen_key().await;
+    trace!(?rep_data)
+}
+
+//杠杆设置
+#[tokio::test]
+async fn rest_change_pos_side_test() {
+    global::log_utils::init_log_with_trace();
+
+    let mut rest = get_rest();
+    let rep_data = rest.change_pos_side(true).await;
+    trace!(?rep_data)
+}
+
+//下单
+#[tokio::test]
+async fn rest_swap_order_test() {
+    global::log_utils::init_log_with_trace();
+
+    let mut rest = get_rest();
+    let timestamp = Utc::now().timestamp_millis() /  1000;
+    let timestamp_600 = (timestamp + 700);
+    trace!("{:?}-==={:?}---{:?}--",timestamp,timestamp_600,(timestamp_600 > (timestamp + 600)));
+    let json_str = json!({
+        "symbol":"XRPUSDT",
+        "side":"BUY",
+        "type":"LIMIT",
+        "quantity":"10",
+        "price":"0.59",
+        "timeInForce":"GTC",
+        "goodTillDate":timestamp_600,
+    });
+    let rep_data = rest.swap_order(json_str).await;
+    trace!(?rep_data)
+}
+
 
 fn get_ws(btree_map: Option<BinanceSwapLogin>) -> BinanceSwapWs {
     let binance_ws = BinanceSwapWs::new(false,

global/src/params.rs

@@ -61,6 +61,22 @@ impl Params {
         // 解析 TOML 数据到 Params 结构体
         let params: Params = from_str(&contents)?;
 
+        Ok(params)
+    }
+    pub fn new_json(file_path: &str) -> Result<Params, Box<dyn std::error::Error>> {
+        // 打开文件并读取内容
+        // let mut file = File::open(file_path)?;
+        // let mut contents = String::new();
+        // file.read_to_string(&mut contents)?;
+
+        let json_contents = std::fs::read_to_string(file_path)?;
+        // 使用serde_json库来解析JSON文件内容，并将其转换为Ship结构体
+        let params: Params = serde_json::from_str(&json_contents)?;
+
+
+        // 解析 TOML 数据到 Params 结构体
+        // let params: Params = from_str(&contents)?;
+
         Ok(params)
     }
 }

src/main.rs

@@ -2,6 +2,7 @@ mod server;
 mod control_c;
 mod core_libs;
 
+use std::fs;
 use std::sync::Arc;
 use std::sync::atomic::{AtomicBool, Ordering};
 use std::time::Duration;
@@ -28,7 +29,6 @@ fn read_params() -> Params {
         if !arg.contains("--config") {
             continue;
         }
-
         let p: Vec<&str> = arg.split("=").collect();
         path = p[1];
     }
@@ -39,10 +39,31 @@ fn read_params() -> Params {
     return params;
 }
 
+// 获取本地配置- json 文件解析
+fn read_params_json() -> Params {
+    let mut path = "config.json";
+
+    let args: Vec<String> = std::env::args().collect();
+
+    for arg in &args {
+        if !arg.contains("--config") {
+            continue;
+        }
+        let p: Vec<&str> = arg.split("=").collect();
+        path = p[1];
+    }
+
+    println!("配置文件路径：{}", path);
+    let params = Params::new_json(path).unwrap();
+    return params;
+}
+
 #[tokio::main(flavor = "multi_thread", worker_threads = 2)]
 async fn main() {
     // 日志级别配置
-    let params = read_params();
+    // let params = read_params();
+    let params = read_params_json();
+
     // 日志级别配置
     let _guard = log_level_init(params.log_level.clone(), params.port.clone(), params.account_name.clone());
     info!("配置读取成功：{:?}。", params);