交易所接口串联

strategy/Cargo.toml

@@ -18,4 +18,5 @@ chrono = "0.4.26"
 tracing = "0.1"
 tracing-subscriber = "0.3.17"
 standard = { path = "../standard" }
-global = { path = "../global" }
+global = { path = "../global" }
+exchanges = { path = "../exchanges" }

strategy/src/model.rs

@@ -1,6 +1,8 @@
 use std::collections::{HashMap};
 use rust_decimal::Decimal;
 use rust_decimal_macros::dec;
+use serde_derive::{Deserialize, Serialize};
+
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub struct LocalPosition{
     // 做多仓位
@@ -82,6 +84,27 @@ pub struct OrderInfo{
     pub fee: Decimal
 }
 
+#[derive(Serialize, Deserialize)]
+pub struct OriginalTrade {
+    // 成交价格
+    pub p: Decimal,
+    // 买方是否是做市方。如true，则此次成交是一个主动卖出单，否则是一个主动买入单。
+    pub m: bool
+}
+
+#[derive(Serialize, Deserialize)]
+pub struct OriginalTicker {
+    // 更新ID
+    pub u: i64,
+    // 买单最优挂单价格
+    pub b: Decimal,
+    // 买单最优挂单数量
+    pub B: Decimal,
+    // 卖单最优挂单价格
+    pub a: Decimal,
+    // 卖单最优挂单数量
+    pub A: Decimal
+}
 
 
 #[derive(Debug)]

strategy/src/quant.rs

@@ -1,18 +1,20 @@
 use std::collections::{BTreeMap, HashMap};
 use std::ops::Div;
 use std::str::FromStr;
+use std::sync::mpsc;
 use std::thread;
 use std::thread::{sleep};
 use std::time::Duration;
 use chrono::{Timelike, Utc};
 use rust_decimal::Decimal;
 use rust_decimal_macros::dec;
+use exchanges::binance_usdt_swap_ws::BinanceUsdtSwapWs;
 use global::public_params::{ASK_PRICE_INDEX, BID_PRICE_INDEX, LENGTH};
 use standard::{Market, Platform, Position, PositionModeEnum, SpecialDepth, SpecialTicker, Ticker};
 use standard::exchange::Exchange;
 use standard::exchange::ExchangeEnum::GateSwap;
 
-use crate::model::{LocalPosition, OrderCommand, OrderInfo, TraderMsg};
+use crate::model::{LocalPosition, OrderCommand, OrderInfo, OriginalTicker, OriginalTrade, TraderMsg};
 use crate::params::Params;
 use crate::predictor::Predictor;
 use crate::strategy::Strategy;
@@ -224,7 +226,7 @@ impl Quant {
         return quant_obj;
     }
 
-    pub fn update_order(&mut self, data: OrderInfo){
+    pub fn update_local_order(&mut self, data: OrderInfo){
         /*
          更新订单
             首先直接复写本地订单
@@ -435,18 +437,6 @@ impl Quant {
         }
     }
 
-    pub fn update_equity(&mut self, data: Decimal){
-        /*
-            更新保证金信息
-            合约一直更新
-            现货只有当出现异常时更新
-        */
-        if self.exchange.contains("spot"){
-            return;
-        }
-        self.local_cash = data * self.used_pct
-    }
-
     // 检测初始数据是否齐全
     pub fn check_ready(&mut self){
         // 检查 ticker 行情
@@ -550,31 +540,6 @@ impl Quant {
         }
     }
 
-    pub fn update_exit(&mut self, data: String){
-        // 底层触发停机
-        self.exit_msg = data;
-
-    }
-    pub fn stop(&mut self){
-        self.mode_signal = 80;
-        // 等strategy onExit 彻底执行完毕 进入沉默状态之后 再进入exit 否则可能导致多处同时操作订单
-        // 尽量减少大仓位直接take平
-        // TODO:发起停机 self.loop.create_task(self.exit(delay=10))
-    }
-
-    pub fn exit(self, delay: i8){
-        println!("预约退出操作 delay: {}", delay);
-        if delay > 0{
-            sleep(Duration::from_secs(delay as u64));
-        }
-        println!("开始退出操作");
-        println!("为避免api失效导致遗漏仓位 建议人工复查");
-        // TODO: 检查仓位 await self.rest.check_position(hold_coin=self.hold_coin)
-        // stop flag
-
-
-    }
-
     pub fn update_position(&mut self, data: Vec<Position>){
         if data.is_empty(){
             return;
@@ -727,10 +692,18 @@ impl Quant {
         }
     }
 
-    pub async fn get_equity(&mut self, platform: &Box<dyn Platform + Send + Sync>){
+    pub async fn update_equity(&mut self, platform: &Box<dyn Platform + Send + Sync>){
         match platform.get_account().await {
             Ok(val)=> {
-                self.update_equity(val.balance);
+                /*
+                   更新保证金信息
+                   合约一直更新
+                   现货只有当出现异常时更新
+               */
+                if self.exchange.contains("spot"){
+                    return;
+                }
+                self.local_cash = val.balance * self.used_pct
             },
             Err(e) => {
                 println!("获取账户信息错误: {:?}", e);
@@ -747,10 +720,57 @@ impl Quant {
         exchange_params.insert("access_key".to_string(), "your_secret_key".to_string());
         let platform:Box<dyn Platform+Send+Sync> = Exchange::new(GateSwap, self.symbol.clone(), false, exchange_params);
 
-        for i in &self.ref_name{
-            // TODO: 启动参考ws 参考盘口使用fast行情性能消耗更大 使用普通行情可以节省性能
-            // self.loop.create_task(self.ws_ref[i].run(is_auth=0, sub_trade=0, sub_fast=_sub_fast))
-        }
+        // TODO: 启动参考ws 参考盘口使用fast行情性能消耗更大 使用普通行情可以节省性能
+        let mut btree_map: BTreeMap<String, String> = BTreeMap::new();
+        btree_map.insert("lable".parse().unwrap(), "binance".parse().unwrap());//交易行名称
+        let (tx, rx) = mpsc::channel();
+        let symbol = self.symbol.clone();
+        thread::spawn(move || {
+            let ba_exc = BinanceUsdtSwapWs::new(false, btree_map, tx);
+            ba_exc.custom_subscribe(vec![symbol.as_str()], 1, 0);
+        });
+        thread::spawn(move || {
+            // trade
+            let mut decimal = 99u32;
+            let mut max_buy = Decimal::ZERO;
+            let mut min_sell = Decimal::ZERO;
+            // ticker
+            let mut update_flag_u = 0i64;
+
+            loop {
+                match rx.recv() {
+                    Ok(data) => {
+                        if data.channel == "aggTrade" {
+                            let trade:OriginalTrade = serde_json::from_str(data.data.as_str()).unwrap();
+                            if decimal == 99{
+                                decimal = trade.p.scale();
+                            }
+                            if trade.p > max_buy {
+                                max_buy = trade.p
+                            }
+                            if trade.p < min_sell {
+                                min_sell = trade.p
+                            }
+                        } else if data.channel == "bookTicker" {
+                            let ticker:OriginalTicker = serde_json::from_str(data.data.as_str()).unwrap();
+                            if ticker.u > update_flag_u {
+
+                            } else {
+                                update_flag_u = ticker.u;
+                                
+                            }
+                        } else if data.channel == "depth"{
+
+                        }
+                    },
+                    Err(_) => {
+                        println!("Channel has been closed!");
+                        break;
+                    }
+                }
+            }
+        });
+
         sleep(Duration::from_secs(1));
         // 买入平台币操作 await self.rest.buy_token()
         // 获取市场信息
@@ -759,7 +779,7 @@ impl Quant {
         let ticker = platform.get_ticker().await.expect("获取价格信息异常!");
         let mp = (ticker.buy + ticker.sell) / Decimal::TWO;
         // 获取账户信息
-        self.get_equity(&platform).await;
+        self.update_equity(&platform).await;
         // 初始资金
         let start_cash = self.local_cash.clone();
         let start_coin = self.local_cash.clone();