修改binance获取深度接口为WebSocket

src/exchange_middle_ware.rs

@@ -1,7 +1,7 @@
-use std::io::{Error, ErrorKind};
+use std::io::{ Error, ErrorKind};
 use std::collections::{BTreeMap};
 use serde_json::json;
-use crate::exchange_libs::{BinanceExc, OkxExc, SocketTool};
+use crate::exchange_libs::{BinanceExc, OkxExc, ReqData, SocketTool};
 
 // 深度结构体
 #[derive(Debug)]
@@ -119,23 +119,24 @@ 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) -> Result<Depth, Error> {
+    pub async fn get_binance_depth(&self, symbol: &String, limit: i32, callback: fn(Depth)) {
         let real_symbol = self.get_real_symbol(symbol, "".to_string());
-        // SocketTool::binance_run_depth(vec![symbol], limit.to_string(), aa);
-        let req_data = self.binance_exc.binance_depth(&real_symbol, &limit.to_string()).await;
-        if req_data.code == "0" {
-            let req_data_str = req_data.data;
-            let req_data_json: serde_json::Value = serde_json::from_str(&*req_data_str).unwrap();
-            let depth_asks: Vec<DepthItem> = parse_depth_items(&req_data_json["asks"]);
-            let depth_bids: Vec<DepthItem> = parse_depth_items(&req_data_json["bids"]);
-            let result = Depth {
-                asks: depth_asks,
-                bids: depth_bids,
-            };
-            Ok(result)
-        } else {
-            Err(Error::new(ErrorKind::Other, req_data.message))
-        }
+        let get_res_data =|res_data:ReqData|{
+            if res_data.code == "0" {
+                let res_data_str = res_data.data;
+                let res_data_json: serde_json::Value = serde_json::from_str(&*res_data_str).unwrap();
+                let depth_asks: Vec<DepthItem> = parse_depth_items(&res_data_json["asks"]);
+                let depth_bids: Vec<DepthItem> = parse_depth_items(&res_data_json["bids"]);
+                let result = Depth {
+                    asks: depth_asks,
+                    bids: depth_bids,
+                };
+                callback(result)
+            } else {
+                panic!("get_binance_depth: {}",res_data.message);
+            }
+        };
+        SocketTool::binance_run_depth(vec![&real_symbol], limit.to_string(), get_res_data)
     }
 
 
@@ -145,12 +146,12 @@ impl Exchange {
     // limit: 返回条数, 最大 1000
     pub async fn get_binance_klines(&self, symbol: &String, interval: &String, limit: &i32) -> Result<Vec<Record>, Error> {
         let real_symbol = self.get_real_symbol(symbol, "".to_string());
-        let req_data = self.binance_exc.binance_k(&real_symbol, interval, &limit.to_string()).await;
-        if req_data.code == "0" {
-            let req_data_str = req_data.data;
-            let req_data_json: Vec<Vec<serde_json::Value>> = serde_json::from_str(&*req_data_str).unwrap();
+        let res_data = self.binance_exc.binance_k(&real_symbol, interval, &limit.to_string()).await;
+        if res_data.code == "0" {
+            let res_data_str = res_data.data;
+            let res_data_json: Vec<Vec<serde_json::Value>> = serde_json::from_str(&*res_data_str).unwrap();
             let mut result: Vec<Record> = vec![];
-            for item in req_data_json.iter() {
+            for item in res_data_json.iter() {
                 let record = Record {
                     time: item[0].as_i64().unwrap_or(0),
                     open: item[1].as_str().unwrap_or("0").parse().unwrap_or(0.0),
@@ -163,7 +164,7 @@ impl Exchange {
             }
             Ok(result)
         } else {
-            Err(Error::new(ErrorKind::Other, req_data.message))
+            Err(Error::new(ErrorKind::Other, res_data.message))
         }
     }
 
@@ -171,12 +172,12 @@ impl Exchange {
     // symbol: 交易币对, "BTC_USDT"
     pub async fn get_okx_account(&self, symbol: &String) -> Result<Account, Error> {
         let real_symbol = self.get_real_symbol(symbol, ",".to_string());
-        let req_data = self.okx_exc.okx_acc(&real_symbol).await;
-        if req_data.code == "0" {
-            let req_data_str = req_data.data;
-            let req_data_json: Vec<serde_json::Value> = serde_json::from_str(&*req_data_str).unwrap();
+        let res_data = self.okx_exc.okx_acc(&real_symbol).await;
+        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 symbol_array: Vec<&str> = symbol.split("_").collect();
-            let details = req_data_json[0]["details"].as_array().unwrap();
+            let details = res_data_json[0]["details"].as_array().unwrap();
             let default_info = json!({"availBal":"0","fixedBal":"0"});
             let stocks_info = details.iter().find(|item| item["ccy"].as_str().unwrap() == symbol_array[0].to_string()).unwrap_or(&default_info);
             let balance_info = details.iter().find(|item| item["ccy"].as_str().unwrap() == symbol_array[1].to_string()).unwrap_or(&default_info);
@@ -188,7 +189,7 @@ impl Exchange {
             };
             Ok(result)
         } else {
-            Err(Error::new(ErrorKind::Other, req_data.message))
+            Err(Error::new(ErrorKind::Other, res_data.message))
         }
     }
 
@@ -200,14 +201,14 @@ impl Exchange {
     // amount: 委托数量
     pub async fn place_okx_order(&self, symbol: &String, side: &String, order_type: &String, price: &String, amount: &String) -> Result<String, Error> {
         let real_symbol = self.get_real_symbol(symbol, "-".to_string());
-        let req_data = self.okx_exc.okx_order(&real_symbol, "cash", side, order_type, price, amount).await;
-        if req_data.code == "0" {
-            let req_data_str = req_data.data;
-            let req_data_json: Vec<serde_json::Value> = serde_json::from_str(&*req_data_str).unwrap();
-            let result = req_data_json[0]["ordId"].as_str().unwrap().parse().unwrap();
+        let res_data = self.okx_exc.okx_order(&real_symbol, "cash", side, order_type, price, amount).await;
+        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 = res_data_json[0]["ordId"].as_str().unwrap().parse().unwrap();
             Ok(result)
         } else {
-            Err(Error::new(ErrorKind::Other, req_data.message))
+            Err(Error::new(ErrorKind::Other, res_data.message))
         }
     }
 
@@ -216,22 +217,22 @@ impl Exchange {
     // order_id: 订单ID, "590910403358593111"
     pub async fn get_okx_order(&self, symbol: &String, order_id: &String) -> Result<Order, Error> {
         let real_symbol = self.get_real_symbol(symbol, "-".to_string());
-        let req_data = self.okx_exc.okx_get_order(&real_symbol, order_id).await;
-        if req_data.code == "0" {
-            let req_data_str = req_data.data;
-            let req_data_json: Vec<serde_json::Value> = serde_json::from_str(&*req_data_str).unwrap();
+        let res_data = self.okx_exc.okx_get_order(&real_symbol, order_id).await;
+        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: req_data_json[0]["ordId"].as_str().unwrap().parse().unwrap(),
-                price: req_data_json[0]["px"].as_str().unwrap_or("0").parse().unwrap_or(0.0),
-                amount: req_data_json[0]["sz"].as_str().unwrap_or("0").parse().unwrap_or(0.0),
-                deal_amount: req_data_json[0]["accFillSz"].as_str().unwrap_or("0").parse().unwrap_or(0.0),
-                avg_price: req_data_json[0]["avgPx"].as_str().unwrap_or("0").parse().unwrap_or(0.0),
-                status: req_data_json[0]["state"].as_str().unwrap().parse().unwrap(),
-                order_type: req_data_json[0]["instType"].as_str().unwrap().parse().unwrap(),
+                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(),
             };
             Ok(result)
         } else {
-            Err(Error::new(ErrorKind::Other, req_data.message))
+            Err(Error::new(ErrorKind::Other, res_data.message))
         }
     }
 
@@ -240,15 +241,15 @@ impl Exchange {
     // order_id: 订单ID, "590910403358593111"
     pub async fn cancel_okx_order(&self, symbol: &String, order_id: &String) -> Result<bool, Error> {
         let real_symbol = self.get_real_symbol(symbol, "-".to_string());
-        let req_data = self.okx_exc.okx_revocation_order(&real_symbol, order_id).await;
-        if req_data.code == "0" {
-            let req_data_str = req_data.data;
-            let req_data_json: Vec<serde_json::Value> = serde_json::from_str(&*req_data_str).unwrap();
-            let order_info = req_data_json[0]["sCode"].as_str().unwrap();
+        let res_data = self.okx_exc.okx_revocation_order(&real_symbol, order_id).await;
+        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 order_info = res_data_json[0]["sCode"].as_str().unwrap();
             let result = if order_info == "0" { true } else { false };
             Ok(result)
         } else {
-            Err(Error::new(ErrorKind::Other, req_data.message))
+            Err(Error::new(ErrorKind::Other, res_data.message))
         }
     }
 
@@ -261,11 +262,11 @@ impl Exchange {
         btree_map.insert("instType", "SPOT");
         let result = self.okx_exc.get_v("/api/v5/public/instruments".to_string(), btree_map).await;
         match result {
-            Ok(req_data) => {
+            Ok(res_data) => {
                 let symbol_array: Vec<&str> = symbol.split("_").collect();
-                let req_data_str = req_data.data;
-                let req_data_json: serde_json::Value = serde_json::from_str(&*req_data_str).unwrap();
-                let order_info = req_data_json["data"].as_array().unwrap();
+                let res_data_str = res_data.data;
+                let res_data_json: serde_json::Value = serde_json::from_str(&*res_data_str).unwrap();
+                let order_info = res_data_json["data"].as_array().unwrap();
                 let info = order_info.iter().find(|item| item["baseCcy"].as_str().unwrap() == symbol_array[0] && item["quoteCcy"].as_str().unwrap() == symbol_array[1]).unwrap();
                 println!("\n\n{:?}", info);
 
@@ -284,20 +285,20 @@ impl Exchange {
                     ct_val: 0.01,
                 };
                 println!("\n\n{:?}", result);
-                // let order_info = req_data_json;
+                // let order_info = res_data_json;
             }
             Err(err) => {}
         }
         // let real_symbol = self.get_real_symbol(symbol, "-".to_string());
-        // let req_data = self.okx_exc.get_v("/api/v5/public/instruments", order_id).await;
-        // if req_data.code == "0" {
-        //     let req_data_str = req_data.data;
-        //     let req_data_json: Vec<serde_json::Value> = serde_json::from_str(&*req_data_str).unwrap();
-        //     let order_info = req_data_json[0]["sCode"].as_str().unwrap();
+        // let res_data = self.okx_exc.get_v("/api/v5/public/instruments", order_id).await;
+        // 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 order_info = res_data_json[0]["sCode"].as_str().unwrap();
         //     let result = if order_info == "0" { true } else { false };
         //     Ok(result)
         // } else {
-        //     Err(Error::new(ErrorKind::Other, req_data.message))
+        //     Err(Error::new(ErrorKind::Other, res_data.message))
         // }
     }
 }
@@ -318,8 +319,9 @@ fn parse_depth_items(value: &serde_json::Value) -> Vec<DepthItem> {
 // 单元测试集
 #[cfg(test)]
 mod tests {
-    use crate::exchange_middle_ware::{Exchange};
-    use crate::exchange_libs::{is_proxy};
+    use std::io::{self, Write};
+    use crate::exchange_middle_ware::{Depth, Exchange};
+    use crate::exchange_libs::{is_proxy, ReqData};
 
     // new Exchange
     fn new_exchange() -> Exchange {
@@ -346,10 +348,11 @@ mod tests {
     // 测试binance获取深度信息
     #[tokio::test]
     async fn test_get_binance_depth() {
-        is_proxy();
         let exchange = new_exchange();
-        let depth = exchange.get_binance_depth(&"DOGE_USDT".to_string(), 10).await;
-        println!("test_get_binance_depth:{:?}", depth);
+        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;
     }
 
     // 测试binance获取k线