添加订单流数据

Cargo.lock

@@ -1420,6 +1420,7 @@ dependencies = [
  "tracing",
  "tracing-appender",
  "tracing-subscriber",
+ "uuid",
 ]
 
 [[package]]
@@ -1697,6 +1698,9 @@ name = "uuid"
 version = "1.6.1"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "5e395fcf16a7a3d8127ec99782007af141946b4795001f876d54fb0d55978560"
+dependencies = [
+ "getrandom",
+]
 
 [[package]]
 name = "valuable"

Cargo.toml

@@ -17,4 +17,5 @@ serde = { version = "1.0.193", features = ["derive"] }
 chrono = "0.4"
 rust_decimal = { version = "1.32.0", features = ["maths"] }
 rust_decimal_macros = "1.32.0"
-toml = "0.8.8"
+toml = "0.8.8"
+uuid = { version = "1.6.1", features = ["v4"] }

src/binance_swap/binance_swap_standard.rs

@@ -66,6 +66,7 @@ pub async fn standard_trades(symbol: &str, limit: &str, trade_id: &str) -> Resul
         let result = trades_list.iter().map(|item| {
             Trades {
                 id: item.id.to_string(),
+                data_type: "ticker".to_string(),
                 symbol: symbol.to_string(),
                 create_time: item.time.to_string(),
                 size: if item.is_buyer_maker { item.qty.to_string() } else { format!("-{}", item.qty.to_string()) },

src/export/html.rs

@@ -5,15 +5,19 @@ use chrono::NaiveDateTime;
 use handlebars::Handlebars;
 use rust_decimal::Decimal;
 use rust_decimal::prelude::ToPrimitive;
+use uuid::Uuid;
 use rust_decimal_macros::dec;
 use serde::{Deserialize, Serialize};
 use tracing::info;
 use crate::struct_standard::Trades;
+use crate::utils::utils::ConfigInfo;
 
 #[derive(Debug, Clone, Deserialize, Serialize)]
 pub struct ExportExchangeTickerInfo {
     pub name: String,
     pub ticker_info: Vec<Trades>,
+    pub recall_ticker_info: Vec<Trades>,
+    pub max_price: String,
 }
 
 #[derive(Debug, Clone, Deserialize, Serialize)]
@@ -22,34 +26,81 @@ pub struct SeriesInfo {
     pub data: Vec<Trades>,
 }
 
-pub fn export_html(export_info: Vec<ExportExchangeTickerInfo>, symbol: &str, start_at: &str, end_at: &str, export_path: Option<&str>, export_name: Option<&str>) {
+pub fn export_html(export_info: Vec<ExportExchangeTickerInfo>, start_at: &str, end_at: &str, config: ConfigInfo) {
     info!("正在生成网页,请稍后!");
-    let path = format!("{}/{}.html", export_path.unwrap_or("./"), export_name.unwrap_or("export")).replace("//", "/");
+    let export_path = if config.export_path == "" { "./" } else { config.export_path.as_str() };
+    let export_name = if config.export_name == "" { "export" } else { config.export_name.as_str() };
+    let path = format!("{}/{}.html", export_path, export_name).replace("//", "/");
     // 创建 Handlebars 实例
     let mut handlebars = Handlebars::new();
 
     let mut name_list: Vec<String> = vec![];
-    for exchange_info in export_info.clone().into_iter() {
-        name_list.push(format!("'{}'", exchange_info.name));
-    }
+    let series_info: Vec<SeriesInfo> = export_info.iter().map(|export_exchange_ticker_info| {
+        name_list.push(format!("'{}'", export_exchange_ticker_info.name));
+        let mut ticker_info = vec![];
+        for trades in export_exchange_ticker_info.ticker_info.clone() {
+            // 赋值原ticker数据
+            ticker_info.push(trades.clone());
+
+            // 计算交易量差
+            let mut volume = Decimal::ONE;
+            let mut short_volume = Decimal::ONE;
+            let mut long_volume = Decimal::ONE;
+            let mut sum_volume = Decimal::ONE;
+            for recall_trades in export_exchange_ticker_info.recall_ticker_info.clone() {
+                if recall_trades.create_time <= trades.create_time {
+                    let size = Decimal::from_str(&recall_trades.size).unwrap();
+                    volume += size;
+                    sum_volume += size.abs();
+                    if size > dec!(0) { long_volume += size } else { short_volume += size.abs() }
+                } else {
+                    break;
+                }
+            };
+            if long_volume / sum_volume >= dec!(0.7) || short_volume / sum_volume >= dec!(0.7) {
+                // 新增订单流主动性数据
+                let max_price = Decimal::from_str(&export_exchange_ticker_info.max_price.clone()).unwrap() * dec!(1.005);
+                ticker_info.push(Trades {
+                    id: Uuid::new_v4().to_string()[0..8].to_string(),
+                    data_type: "recall".to_string(),
+                    symbol: trades.symbol,
+                    create_time: trades.create_time,
+                    size: volume.to_string(),
+                    price: max_price.to_string(),
+                });
+            }
+        }
+        // 对订单流主动性数据(recall)去重
+        let mut ticker_set = std::collections::HashSet::new();
+        let mut end_ticker_info = vec![];
+        for trades in ticker_info {
+            if ticker_set.insert((trades.data_type.clone(), trades.create_time.clone())) {
+                end_ticker_info.push(trades.clone());
+            }
+        }
+        SeriesInfo {
+            name: export_exchange_ticker_info.name.clone(),
+            data: end_ticker_info.clone(),
+        }
+    }).collect();
 
     let start_time_d = Decimal::from_str(start_at).unwrap() * dec!(1000) + dec!(8) * dec!(3600000);
     let end_time_d = Decimal::from_str(end_at).unwrap() * dec!(1000) + dec!(8) * dec!(3600000);
     let start_time = NaiveDateTime::from_timestamp_millis(start_time_d.to_i64().unwrap()).unwrap().format("%Y-%m-%d %H:%M:%S%.3f").to_string();
     let end_time = NaiveDateTime::from_timestamp_millis(end_time_d.to_i64().unwrap()).unwrap().format("%Y-%m-%d %H:%M:%S%.3f").to_string();
 
-    let series_info: Vec<_> = export_info.iter().map(|item| {
-        SeriesInfo {
-            name: item.name.clone(),
-            data: item.ticker_info.clone(),
-        }
-    }).collect();
+    // let series_info: Vec<_> = export_info.iter().map(|item| {
+    //     SeriesInfo {
+    //         name: item.name.clone(),
+    //         data: item.ticker_info.clone(),
+    //     }
+    // }).collect();
 
     let data = serde_json::json!({
         "chart_title": format!("{} Ticker数据", name_list.join("、").replace("'","")),
         "legend_data": format!("[{}]", name_list.join(", ")),
         "series_info": series_info.clone(),
-        "symbol": symbol.to_uppercase(),
+        "symbol": config.symbol.to_uppercase(),
         "start_at": start_time,
         "end_at": end_time,
     });

src/gate_swap/gate_swap_standard.rs

@@ -27,6 +27,7 @@ pub(crate) async fn standard_trades(symbol: &str, start_at: &str, end_at: &str,
         let result = trades_list.iter().map(|item| {
             Trades {
                 id: item.id.to_string(),
+                data_type: "ticker".to_string(),
                 symbol: item.contract.clone(),
                 create_time: (item.create_time_ms * 1000.0).to_string(),
                 size: (item.size * ct_val.unwrap_or(Decimal::ONE)).to_string(),

src/handle_ticker.rs

@@ -72,6 +72,7 @@ pub async fn get_binance_ticker_info(symbol: &str, start_at: &str, end_at: &str)
     for agg_trades in agg_ticker_info_list.clone() {
         ticker_info_list.push(Trades {
             id: agg_trades.start_id,
+            data_type: "ticker".to_string(),
             symbol: agg_trades.symbol,
             create_time: agg_trades.create_time,
             size: agg_trades.size,

src/main.rs

@@ -1,56 +1,84 @@
+use std::str::FromStr;
 use chrono::{NaiveDateTime};
+use rust_decimal::Decimal;
 use tracing::{error, info};
 use crate::export::html::ExportExchangeTickerInfo;
+use crate::struct_standard::Trades;
 use crate::utils::logs;
 
 pub mod binance_swap;
 pub mod gate_swap;
+pub mod okx_swap;
+pub mod robot_data;
 pub mod utils;
 pub mod http;
 pub mod struct_standard;
 pub mod export;
 pub mod handle_ticker;
-pub mod okx_swap;
 
 #[tokio::main]
 async fn main() {
     logs::init_log_with_info();
 
     let config = utils::utils::get_config_info("./config.toml");
-    if http::proxy::ParsingDetail::http_enable_proxy(config.proxy_address) {
+    let clone_config = config.clone();
+    if http::proxy::ParsingDetail::http_enable_proxy(clone_config.proxy_address) {
         info!("检测有代理配置，配置走代理");
     }
+    // robot_data::robot_data_standard::standard_robot_info("localhost_test_account_fresh", "1702010000", "1702031000").await;
+    let symbol = clone_config.symbol;
+    let start_at = NaiveDateTime::parse_from_str(&config.range_time[0].clone(), "%Y-%m-%d %H:%M:%S").unwrap().timestamp() - 8 * 3600;
+    let end_at = NaiveDateTime::parse_from_str(&config.range_time[1].clone(), "%Y-%m-%d %H:%M:%S").unwrap().timestamp() - 8 * 3600;
 
-    let symbol = config.symbol;
-    let start_at = (NaiveDateTime::parse_from_str(&config.range_time[0].clone(), "%Y-%m-%d %H:%M:%S").unwrap().timestamp() - 8 * 3600).to_string();
-    let end_at = (NaiveDateTime::parse_from_str(&config.range_time[1].clone(), "%Y-%m-%d %H:%M:%S").unwrap().timestamp() - 8 * 3600).to_string();
-    let export_path = if config.export_path == "" { None } else { Option::from(config.export_path.as_str()) };
-    let export_name = if config.export_name == "" { None } else { Option::from(config.export_name.as_str()) };
+    let recall_start_at = start_at - config.recall_time;
 
-    println!("{} {}", start_at, end_at);
     let mut exchange_list = vec![];
-    for exchange in config.exchanges {
+    for exchange in config.exchanges.clone() {
         let exchange_up = exchange.to_uppercase();
         let exchange_result = match exchange_up.as_str() {
             "BINANCE" => {
-                let ticker_info = handle_ticker::get_binance_ticker_info(&symbol, &start_at, &end_at).await;
+                let recall_ticker_info = handle_ticker::get_binance_ticker_info(&symbol, &recall_start_at.to_string(), &end_at.to_string()).await;
+                let ticker_info: Vec<Trades> = recall_ticker_info.iter().filter(|item| item.create_time.parse::<i64>().unwrap() >= start_at).cloned().collect();
+                let mut max_price = Decimal::ZERO;
+                for trades in ticker_info.clone() {
+                    let trades_price = Decimal::from_str(&trades.price).unwrap();
+                    max_price = if trades_price > max_price { trades_price } else { max_price };
+                };
                 ExportExchangeTickerInfo {
                     name: exchange.to_string(),
                     ticker_info,
+                    recall_ticker_info,
+                    max_price: max_price.to_string(),
                 }
             }
             "GATE" => {
-                let ticker_info = handle_ticker::get_gate_ticker_info(&symbol, &start_at, &end_at).await;
+                let recall_ticker_info = handle_ticker::get_gate_ticker_info(&symbol, &recall_start_at.to_string(), &end_at.to_string()).await;
+                let ticker_info: Vec<Trades> = recall_ticker_info.iter().filter(|item| item.create_time.parse::<i64>().unwrap() >= start_at).cloned().collect();
+                let mut max_price = Decimal::ZERO;
+                for trades in ticker_info.clone() {
+                    let trades_price = Decimal::from_str(&trades.price).unwrap();
+                    max_price = if trades_price > max_price { trades_price } else { max_price };
+                };
                 ExportExchangeTickerInfo {
                     name: exchange.to_string(),
                     ticker_info,
+                    recall_ticker_info,
+                    max_price: max_price.to_string(),
                 }
             }
             "OKX" => {
-                let ticker_info = handle_ticker::get_okx_ticker_info(&symbol, &start_at, &end_at).await;
+                let recall_ticker_info = handle_ticker::get_okx_ticker_info(&symbol, &recall_start_at.to_string(), &end_at.to_string()).await;
+                let ticker_info: Vec<Trades> = recall_ticker_info.iter().filter(|item| item.create_time.parse::<i64>().unwrap() >= start_at).cloned().collect();
+                let mut max_price = Decimal::ZERO;
+                for trades in ticker_info.clone() {
+                    let trades_price = Decimal::from_str(&trades.price).unwrap();
+                    max_price = if trades_price > max_price { trades_price } else { max_price };
+                };
                 ExportExchangeTickerInfo {
                     name: exchange.to_string(),
                     ticker_info,
+                    recall_ticker_info,
+                    max_price: max_price.to_string(),
                 }
             }
             _ => {
@@ -61,5 +89,5 @@ async fn main() {
         exchange_list.push(exchange_result);
     }
 
-    export::html::export_html(exchange_list, &symbol, &start_at, &end_at, export_path, export_name);
+    export::html::export_html(exchange_list, &start_at.to_string(), &end_at.to_string(), config.clone());
 }

src/okx_swap/okx_swap_standard.rs

@@ -47,6 +47,7 @@ pub(crate) async fn standard_history_candles(symbol: &str, start_at: &str, end_a
 
                 Trades {
                     id: item.trade_id.to_string(),
+                    data_type: "ticker".to_string(),
                     symbol: symbol.to_string(),
                     create_time: item.ts.to_string(),
                     size: (size_i * ct_val).to_string(),

src/robot_data/mod.rs

@@ -0,0 +1,2 @@
+pub mod robot_data_rest;
+pub mod robot_data_standard;

src/robot_data/robot_data_rest.rs

@@ -0,0 +1,17 @@
+use reqwest::header::HeaderMap;
+use crate::http::request::{post, Response};
+
+pub async fn get_robot_info(robot_name: &str, start_at: &str, end_at: &str) -> Response {
+    let url = "http://as.skyfffire.com:8848/basic/getDealRecords";
+
+    let params = serde_json::json!({
+        "robotName":robot_name,
+        "beginTime":start_at,
+        "endTime":end_at,
+    });
+    let mut header = HeaderMap::new();
+    header.insert("auth", "43626546liangjiang".parse().unwrap());
+    header.insert("Content-Type", "application/json".parse().unwrap());
+    post(url, params.to_string(), Option::from(header)).await
+}
+

src/robot_data/robot_data_standard.rs

@@ -0,0 +1,23 @@
+use rust_decimal::Decimal;
+use serde::{Deserialize, Serialize};
+use crate::robot_data::robot_data_rest::get_robot_info;
+
+#[derive(Debug, Deserialize, Serialize)]
+#[serde(rename_all = "camelCase")]
+struct RobotInfo {
+    id: i64,
+    ref_price: Decimal,
+    reg_price: Decimal,
+    num: Decimal,
+    trigger_time: Decimal,
+    robot_name: String,
+    side: String,
+}
+
+pub async fn standard_robot_info(robot_name: &str, start_at: &str, end_at: &str) {
+    let res_data = get_robot_info(robot_name, start_at, end_at).await;
+    if res_data.code == "200" {
+        let robot_info: Vec<RobotInfo> = serde_json::from_str(&res_data.data).unwrap();
+        println!("{:?}", robot_info);
+    }
+}

src/struct_standard.rs

@@ -4,6 +4,7 @@ use serde::{Deserialize, Serialize};
 #[derive(Debug, Clone, PartialEq, Eq, Hash, Deserialize, Serialize)]
 pub struct Trades {
     pub id: String,
+    pub data_type: String,
     pub symbol: String,
     pub create_time: String,
     pub size: String,

src/utils/utils.rs

@@ -17,6 +17,7 @@ pub struct ConfigInfo {
     pub exchanges: Vec<String>,
     pub symbol: String,
     pub range_time: Vec<String>,
+    pub recall_time: i64,
     pub export_path: String,
     pub export_name: String,
 }
@@ -28,6 +29,7 @@ impl ConfigInfo {
             exchanges: vec![],
             symbol: "".to_string(),
             range_time: vec![],
+            recall_time: 0,
             export_path: "".to_string(),
             export_name: "".to_string(),
         }