添加条数限制

添加过滤内部行为

config.toml.sample

@@ -1,17 +1,19 @@
 # 配置代理
 proxy_address = "127.0.0.1:7890"
 # 配置币对
-symbol = "doge_usdt"
+symbol = "xrp_usdt"
 # 配置交易所 目前支持["binance", "gate", "okx"]
-exchanges = ["binance", "gate", "okx", ]
+exchanges = ["binance", "gate"]
 # 配置查询时间(格式：2023-12-6 15:00:00)
-range_time = ["2023-12-5 15:00:00", "2023-12-5 15:01:00"]
-# 前置时间(单位ms)
-recall_time = 30000
+range_time = ["2023-12-11 10:12:00", "2023-12-11 10:12:59"]
+# 回溯时间(单位ms)
+recall_time = 1000
+# 最大回溯条数
+recall_max_count = 1000
 # 主动性跌比率(0.01代表1%)
-short_volume_rate = 0.7
+short_volume_rate = 0.55
 # 主动性涨比率(0.01代表1%)
-long_volume_rate = 0.7
+long_volume_rate = 0.55
 # 导出路径(不填则为默认路径"./")
 export_path = ""
 # 导出文件名字(不填则为默认名字"export")

src/binance_swap/binance_swap_standard.rs

@@ -71,6 +71,7 @@ pub async fn standard_trades(symbol: &str, limit: &str, trade_id: &str) -> Resul
                 create_time: item.time.to_string(),
                 size: if item.is_buyer_maker { item.qty.to_string() } else { format!("-{}", item.qty.to_string()) },
                 price: item.price.clone(),
+                is_effect: true,
             }
         }).collect();
         Ok(result)

src/export/html.rs

@@ -89,6 +89,7 @@ pub fn export_html(export_info: Vec<ExportExchangeTickerInfo>, start_at: &str, e
                 create_time: trades.create_time,
                 size: volume.to_string(),
                 price: max_price.to_string(),
+                is_effect: true,
             });
         }
     }

src/gate_swap/gate_swap_standard.rs

@@ -15,6 +15,7 @@ struct SwapTrades {
     create_time_ms: f64,
     size: Decimal,
     price: String,
+    is_internal: Option<bool>,
 }
 
 pub(crate) async fn standard_trades(symbol: &str, start_at: &str, end_at: &str, ct_val: Option<Decimal>) -> Result<Vec<Trades>, Error> {
@@ -32,6 +33,7 @@ pub(crate) async fn standard_trades(symbol: &str, start_at: &str, end_at: &str,
                 create_time: (item.create_time_ms * 1000.0).to_string(),
                 size: (item.size * ct_val.unwrap_or(Decimal::ONE)).to_string(),
                 price: item.price.clone(),
+                is_effect: if item.is_internal.is_none() { true } else { false },
             }
         }).collect();
         Ok(result)

src/handle_ticker.rs

@@ -33,7 +33,7 @@ pub async fn get_gate_ticker_info(symbol: &str, start_at: &str, end_at: &str) ->
         };
     };
     let mut set = std::collections::HashSet::new();
-    ticker_info_list.clone().into_iter().filter(|e| set.insert(e.clone())).collect()
+    ticker_info_list.into_iter().filter(|trades| trades.is_effect && set.insert(trades.clone())).collect()
 }
 
 pub async fn get_binance_ticker_info(symbol: &str, start_at: &str, end_at: &str) -> Vec<Trades> {
@@ -77,6 +77,7 @@ pub async fn get_binance_ticker_info(symbol: &str, start_at: &str, end_at: &str)
             create_time: agg_trades.create_time,
             size: agg_trades.size,
             price: agg_trades.price,
+            is_effect: true,
         })
         // let id_diff = Decimal::from_str(&agg_trades.end_id).unwrap() - Decimal::from_str(&agg_trades.start_id).unwrap();
         // if id_diff > Decimal::ONE {
@@ -93,13 +94,13 @@ pub async fn get_binance_ticker_info(symbol: &str, start_at: &str, end_at: &str)
         //     })
         // }
     }
-    ticker_info_list
+    ticker_info_list.iter().filter(|trades| trades.is_effect).cloned().collect()
 }
 
 pub(crate) async fn get_okx_ticker_info(symbol: &str, start_at: &str, end_at: &str) -> Vec<Trades> {
     let ct_val = okx_swap_standard::standard_ct_val(symbol).await;
 
-    let  start_at_d = Decimal::from_str(start_at).unwrap();
+    let start_at_d = Decimal::from_str(start_at).unwrap();
     let end_at_d = Decimal::from_str(end_at).unwrap();
 
     // let end_time = if end_at_d - start_at_d > dec!(3600) { (start_at_d + dec!(3600)).to_string() } else { end_at_d.to_string() };
@@ -148,8 +149,8 @@ pub(crate) async fn get_okx_ticker_info(symbol: &str, start_at: &str, end_at: &s
     }
 
     let mut set = std::collections::HashSet::new();
-    list_array.clone().into_iter().filter(|e| {
-        e.create_time > start_at_d_str && e.create_time < end_at_d_str && set.insert(e.clone())
+    list_array.clone().into_iter().filter(|trades| {
+        trades.create_time > start_at_d_str && trades.create_time < end_at_d_str && trades.is_effect && set.insert(trades.clone())
     }).collect()
     //
     // list_array = list_array.iter().filter(|item| item.create_time <= end_at.to_string()).cloned().collect();

src/main.rs

@@ -1,6 +1,7 @@
 use std::str::FromStr;
 use chrono::{NaiveDateTime};
 use rust_decimal::Decimal;
+use rust_decimal::prelude::ToPrimitive;
 use tracing::{error, info};
 use crate::export::html::ExportExchangeTickerInfo;
 use crate::struct_standard::Trades;
@@ -21,12 +22,12 @@ async fn main() {
     logs::init_log_with_info();
 
     let config = utils::utils::get_config_info("./config.toml");
-    let clone_config = config.clone();
-    if http::proxy::ParsingDetail::http_enable_proxy(clone_config.proxy_address) {
+    let config_clone = config.clone();
+    if http::proxy::ParsingDetail::http_enable_proxy(config_clone.proxy_address) {
         info!("检测有代理配置，配置走代理");
     }
     // robot_data::robot_data_standard::standard_robot_info("localhost_test_account_fresh", "1702010000", "1702031000").await;
-    let symbol = clone_config.symbol;
+    let symbol = config_clone.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;
 
@@ -38,46 +39,49 @@ async fn main() {
         let exchange_result = match exchange_up.as_str() {
             "BINANCE" => {
                 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 ticker_info: Vec<Trades> = recall_ticker_info.iter().filter(|item| item.create_time.parse::<i64>().unwrap() >= start_at * 1000).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 };
                 };
+                let start_index = recall_ticker_info.len() - ticker_info.len() - config_clone.recall_max_count.to_usize().unwrap();
                 ExportExchangeTickerInfo {
                     name: exchange.to_string(),
                     ticker_info,
-                    recall_ticker_info,
+                    recall_ticker_info: recall_ticker_info[start_index..].to_vec(),
                     max_price: max_price.to_string(),
                 }
             }
             "GATE" => {
                 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 ticker_info: Vec<Trades> = recall_ticker_info.iter().filter(|item| item.create_time.parse::<i64>().unwrap() >= start_at * 1000).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 };
                 };
+                let start_index = recall_ticker_info.len() - ticker_info.len() - config_clone.recall_max_count.to_usize().unwrap();
                 ExportExchangeTickerInfo {
                     name: exchange.to_string(),
                     ticker_info,
-                    recall_ticker_info,
+                    recall_ticker_info: recall_ticker_info[start_index..].to_vec(),
                     max_price: max_price.to_string(),
                 }
             }
             "OKX" => {
                 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 ticker_info: Vec<Trades> = recall_ticker_info.iter().filter(|item| item.create_time.parse::<i64>().unwrap() >= start_at * 1000).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 };
                 };
+                let start_index = recall_ticker_info.len() - ticker_info.len() - config_clone.recall_max_count.to_usize().unwrap();
                 ExportExchangeTickerInfo {
                     name: exchange.to_string(),
                     ticker_info,
-                    recall_ticker_info,
+                    recall_ticker_info: recall_ticker_info[start_index..].to_vec(),
                     max_price: max_price.to_string(),
                 }
             }

src/okx_swap/okx_swap_standard.rs

@@ -52,6 +52,7 @@ pub(crate) async fn standard_history_candles(symbol: &str, start_at: &str, end_a
                     create_time: item.ts.to_string(),
                     size: (size_i * ct_val).to_string(),
                     price: item.px.clone(),
+                    is_effect: true,
                 }
             }).collect()
         } else {

src/robot_data/robot_data_standard.rs

@@ -46,6 +46,7 @@ pub async fn standard_robot_info(robot_name: &str, start_at: &str, end_at: &str)
                 create_time: item.trigger_time.clone(),
                 size: item.num.clone(),
                 price: item.ref_price.clone(),
+                is_effect: true,
             }
         }).collect();
         Ok(result)

src/struct_standard.rs

@@ -9,6 +9,7 @@ pub struct Trades {
     pub create_time: String,
     pub size: String,
     pub price: String,
+    pub is_effect: bool,
 }
 
 #[derive(Debug, Clone, PartialEq, Eq, Hash, Deserialize, Serialize)]

src/utils/utils.rs

@@ -19,6 +19,7 @@ pub struct ConfigInfo {
     pub symbol: String,
     pub range_time: Vec<String>,
     pub recall_time: i64,
+    pub recall_max_count: Decimal,
     pub short_volume_rate: Decimal,
     pub long_volume_rate: Decimal,
     pub export_path: String,
@@ -33,6 +34,7 @@ impl ConfigInfo {
             symbol: "".to_string(),
             range_time: vec![],
             recall_time: 0,
+            recall_max_count: Decimal::ZERO,
             short_volume_rate: Decimal::ZERO,
             long_volume_rate: Decimal::ZERO,
             export_path: "".to_string(),