添加订阅okx订单

src/exchange_middle_ware.rs

@@ -1,13 +1,10 @@
+use std::env;
 use std::io::{BufRead, Error, ErrorKind};
-use std::collections::{BTreeMap};
 use std::future::Future;
-use std::io;
-use std::pin::Pin;
 use std::sync::Arc;
-use std::time::Duration;
 use serde_json::json;
 use tokio::sync::Mutex;
-use crate::Bot;
+use crate::{Bot};
 use crate::exchange_libs::{BinanceExc, OkxExc, ResponseData, SocketTool};
 
 // 深度结构体
@@ -126,7 +123,7 @@ impl Exchange {
     // 获取币安深度信息
     // symbol: 交易币对, "BTC_USDT"
     // limit: 返回条数, 最大 5000. 可选值:[5, 10, 20, 50, 100, 500, 1000, 5000]
-    pub async fn get_binance_depth(&self, symbol: &String, limit: i32, mut bot_arc: Arc<Mutex<Bot>>) {
+    pub async fn subscribe_binance_depth(&self, symbol: &String, limit: i32, mut bot_arc: Arc<Mutex<Bot>>) {
         let real_symbol = self.get_real_symbol(symbol, "".to_string());
 
         let get_res_data = move |res_data: ResponseData| {
@@ -144,7 +141,7 @@ impl Exchange {
                     };
 
                     {
-                        let bot = bot_arc_clone.lock().await;
+                        let mut bot = bot_arc_clone.lock().await;
                         bot.depth_handler(result)
                     }
                 } else {
@@ -238,21 +235,44 @@ impl Exchange {
         if res_data.code == "0" {
             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 = Order {
-                id: res_data_json[0]["ordId"].as_str().unwrap().parse().unwrap(),
-                price: res_data_json[0]["px"].as_str().unwrap_or("0").parse().unwrap_or(0.0),
-                amount: res_data_json[0]["sz"].as_str().unwrap_or("0").parse().unwrap_or(0.0),
-                deal_amount: res_data_json[0]["accFillSz"].as_str().unwrap_or("0").parse().unwrap_or(0.0),
-                avg_price: res_data_json[0]["avgPx"].as_str().unwrap_or("0").parse().unwrap_or(0.0),
-                status: res_data_json[0]["state"].as_str().unwrap().parse().unwrap(),
-                order_type: res_data_json[0]["instType"].as_str().unwrap().parse().unwrap(),
-            };
+            let result = parse_order_info(res_data_json);
             Ok(result)
         } else {
             Err(Error::new(ErrorKind::Other, res_data.message))
         }
     }
 
+
+    // OKX查询订单
+    // symbol: 交易币对, "BTC_USDT"
+    // order_id: 订单ID, "590910403358593111"
+    pub async fn subscribe_okx_order(&self, symbol: &String, limit: &String, mut bot_arc: Arc<Mutex<Bot>>) {
+        let real_symbol = self.get_real_symbol(symbol, "-".to_string());
+
+        let get_res_data = move |res_data: ResponseData| {
+            let bot_arc_clone = Arc::clone(&bot_arc);
+
+            async move {
+                if res_data.code == "0" {
+                    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 = parse_order_info(res_data_json);
+                    {
+                        let mut bot = bot_arc_clone.lock().await;
+                        bot.order_change_response(result).await;
+                    }
+                } else {
+                    panic!("get_binance_depth: {}", res_data.message);
+                }
+            }
+        };
+
+        let okx_access_key = env::var("okx_access_key").unwrap();
+        let okx_secret_key = env::var("okx_secret_key").unwrap();
+        let okx_passphrase = env::var("okx_passphrase").unwrap();
+        SocketTool::okx_pr_run_orders(vec![&real_symbol], okx_access_key, okx_secret_key, okx_passphrase, get_res_data);
+    }
+
     // OKX撤销订单
     // symbol: 交易币对, "BTC_USDT"
     // order_id: 订单ID, "590910403358593111"
@@ -322,6 +342,19 @@ impl Exchange {
     }
 }
 
+fn parse_order_info(res_data_json: Vec<serde_json::Value>) -> Order {
+    let order = Order {
+        id: res_data_json[0]["ordId"].as_str().unwrap().parse().unwrap(),
+        price: res_data_json[0]["px"].as_str().unwrap_or("0").parse().unwrap_or(0.0),
+        amount: res_data_json[0]["sz"].as_str().unwrap_or("0").parse().unwrap_or(0.0),
+        deal_amount: res_data_json[0]["accFillSz"].as_str().unwrap_or("0").parse().unwrap_or(0.0),
+        avg_price: res_data_json[0]["avgPx"].as_str().unwrap_or("0").parse().unwrap_or(0.0),
+        status: res_data_json[0]["state"].as_str().unwrap().parse().unwrap(),
+        order_type: res_data_json[0]["instType"].as_str().unwrap().parse().unwrap(),
+    };
+    return order;
+}
+
 // 深度信息买单/卖单处理
 fn parse_depth_items(value: &serde_json::Value) -> Vec<DepthItem> {
     let mut depth_items: Vec<DepthItem> = vec![];
@@ -338,19 +371,23 @@ fn parse_depth_items(value: &serde_json::Value) -> Vec<DepthItem> {
 // 单元测试集
 #[cfg(test)]
 mod tests {
+    use std::env;
     use std::io::{self, Write};
-    use crate::exchange_middle_ware::{Depth, Exchange};
-    use crate::exchange_libs::{is_proxy, ResponseData};
+    use crate::exchange_middle_ware::{Exchange};
+    use crate::exchange_libs::{ResponseData, SocketTool, http_enable_proxy};
 
     // new Exchange
     fn new_exchange() -> Exchange {
-        Exchange::new("a4cf4f54-f4d3-447d-a57c-166fd1ead2e0".to_string(), "556DAB6773CA26DDAAA114F7044138CA".to_string(), "rust_Test123".to_string())
+        let okx_access_key = env::var("okx_access_key").unwrap();
+        let okx_secret_key = env::var("okx_secret_key").unwrap();
+        let okx_passphrase = env::var("okx_passphrase").unwrap();
+        Exchange::new(okx_access_key, okx_secret_key, okx_passphrase)
     }
 
     // 测试new Exchange
     #[tokio::test]
     async fn test_new_exchange() {
-        is_proxy();
+        http_enable_proxy();
         let exchange = new_exchange();
         println!("test_new_exchange:okx_access_key:{},okx_secret_key:{},okx_passphrase:{}", exchange.okx_access_key, exchange.okx_secret_key, exchange.okx_passphrase);
     }
@@ -358,26 +395,31 @@ mod tests {
     // 测试交易对处理
     #[tokio::test]
     async fn test_get_real_symbol() {
-        is_proxy();
+        http_enable_proxy();
         let exchange = new_exchange();
         let real_symbol = exchange.get_real_symbol(&"BTC_USDT".to_string(), "".to_string());
         println!("test_get_real_symbol:{}", real_symbol);
     }
 
     // 测试binance获取深度信息
-    // #[tokio::test]
-    // async fn test_get_binance_depth() {
-    //     let exchange = new_exchange();
-    //     let get_depth_fn = |depth:Depth|{
-    //         writeln!(io::stdout(), "test_get_binance_depth：{:?}", depth).unwrap();
-    //     };
-    //     exchange.get_binance_depth(&"DOGE_USDT".to_string(), 10,get_depth_fn).await;
-    // }
+    #[tokio::test]
+    async fn test_get_binance_depth() {
+        let exchange = new_exchange();
+        let get_res_data = move |res_data: ResponseData| {
+
+            // writeln!(stdout, "111111111111111111111111111").expect("TODO: panic message");
+            async move {
+                let mut stdout = io::stdout();
+                writeln!(stdout, "test_get_binance_depth: {:?}", res_data).expect("TODO: panic message");
+            }
+        };
+        SocketTool::binance_run_depth(vec![&"BTCUSDT"], "10".to_string(), get_res_data)
+    }
 
     // 测试binance获取k线
     #[tokio::test]
     async fn test_get_binance_klines() {
-        is_proxy();
+        http_enable_proxy();
         let exchange = new_exchange();
         let klines = exchange.get_binance_klines(&"DOGE_USDT".to_string(), &"5m".to_string(), &10).await;
         println!("test_get_binance_klines:{:?}", klines);
@@ -386,7 +428,7 @@ mod tests {
     // 测试okx查询账户信息
     #[tokio::test]
     async fn test_get_okx_account() {
-        is_proxy();
+        http_enable_proxy();
         let exchange = new_exchange();
         let account = exchange.get_okx_account(&"BTC_USDT".to_string()).await;
         println!("test_get_okx_account:{:?}", account);
@@ -395,7 +437,7 @@ mod tests {
     // 测试okx下订单
     #[tokio::test]
     async fn test_place_okx_order() {
-        is_proxy();
+        http_enable_proxy();
         let exchange = new_exchange();
         let order_id = exchange.place_okx_order(&"BTC_USDT".to_string(), &"buy".to_string(), &"limit".to_string(), &"20000".to_string(), &"0.0001".to_string()).await;
         println!("test_place_okx_order:{:?}", order_id);
@@ -404,16 +446,19 @@ mod tests {
     // 测试okx查询订单
     #[tokio::test]
     async fn test_get_okx_order() {
-        is_proxy();
+        http_enable_proxy();
         let exchange = new_exchange();
-        let order = exchange.get_okx_order(&"BTC_USDT".to_string(), &"612034971737800726".to_string()).await;
-        println!("test_get_okx_order:{:?}", order);
+        let get_res_data = move |res_data: ResponseData| {
+            writeln!(io::stdout(), "Current number: {:?}", res_data).expect("TODO: panic message");
+            async move {}
+        };
+        SocketTool::okx_pr_run_orders(vec![&"BTC-USDT"], exchange.okx_access_key, exchange.okx_secret_key, exchange.okx_passphrase, get_res_data);
     }
 
     // 测试okx撤单
     #[tokio::test]
     async fn test_cancel_okx_order() {
-        is_proxy();
+        http_enable_proxy();
         let exchange = new_exchange();
         let is_success = exchange.cancel_okx_order(&"BTC_USDT".to_string(), &"612034971737800726".to_string()).await;
         println!("test_cancel_okx_order:{:?}", is_success);
@@ -422,7 +467,7 @@ mod tests {
     // 测试okx撤单
     #[tokio::test]
     async fn test_get_okx_instruments() {
-        is_proxy();
+        http_enable_proxy();
         let exchange = new_exchange();
         let is_success = exchange.get_okx_instruments(&"BTC_USDT".to_string()).await;
         println!("test_cancel_okx_order:{:?}", is_success);