From c83526a6a439e589e2769b09c644a85b98631353 Mon Sep 17 00:00:00 2001 From: boris Date: Sun, 21 Jun 2026 03:57:07 +0800 Subject: [PATCH] =?UTF-8?q?=E6=87=92=E5=8A=A0=E8=BD=BD=E6=97=A5=E7=BA=BF?= =?UTF-8?q?=E5=BA=8F=E5=88=97=E7=BC=93=E5=AD=98=E9=99=8D=E4=BD=8E=E5=9B=9E?= =?UTF-8?q?=E6=B5=8B=E5=86=85=E5=AD=98?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- crates/fidc-core/src/data.rs | 112 +++++++++++++++++------------------ 1 file changed, 53 insertions(+), 59 deletions(-) diff --git a/crates/fidc-core/src/data.rs b/crates/fidc-core/src/data.rs index a65002c..e12afd2 100644 --- a/crates/fidc-core/src/data.rs +++ b/crates/fidc-core/src/data.rs @@ -1,7 +1,7 @@ use std::collections::{BTreeMap, HashMap, HashSet}; use std::fs; use std::path::Path; -use std::sync::{Arc, OnceLock}; +use std::sync::{Arc, OnceLock, RwLock}; use chrono::{NaiveDate, NaiveDateTime}; use serde::{Deserialize, Serialize}; @@ -992,7 +992,7 @@ pub struct DataSet { execution_quotes_by_date: HashMap>>, order_book_depth_index: HashMap<(NaiveDate, String), Vec>, benchmark_by_date: BTreeMap, - market_series_by_symbol: HashMap, + market_series_by_symbol: Arc>>>, benchmark_series_cache: BenchmarkPriceSeries, eligible_universe_by_date: Arc>>>, benchmark_code: String, @@ -1276,7 +1276,6 @@ impl DataSet { .into_iter() .map(|item| (item.date, item)) .collect::>(); - let market_series_by_symbol = build_market_series(&market_by_date); let benchmark_series_cache = BenchmarkPriceSeries::new(&benchmark_by_date.values().cloned().collect::>()); let futures_params_by_symbol = build_futures_params_index(futures_params); @@ -1293,7 +1292,7 @@ impl DataSet { execution_quotes_by_date, order_book_depth_index, benchmark_by_date, - market_series_by_symbol, + market_series_by_symbol: Arc::new(RwLock::new(HashMap::new())), benchmark_series_cache, eligible_universe_by_date: Arc::new(OnceLock::new()), benchmark_code, @@ -1344,6 +1343,39 @@ impl DataSet { .and_then(|rows| find_arc_by_symbol(rows, symbol, |row| row.symbol.as_str())) } + fn market_series(&self, symbol: &str) -> Option> { + if let Some(series) = self + .market_series_by_symbol + .read() + .expect("market series cache lock poisoned") + .get(symbol) + .cloned() + { + return Some(series); + } + + let rows = self + .market_by_date + .values() + .filter_map(|day_rows| find_arc_by_symbol(day_rows, symbol, |row| row.symbol.as_str())) + .collect::>(); + if rows.is_empty() { + return None; + } + + let series = Arc::new(SymbolPriceSeries::new(symbol.to_string(), rows)); + let mut cache = self + .market_series_by_symbol + .write() + .expect("market series cache lock poisoned"); + Some( + cache + .entry(symbol.to_string()) + .or_insert_with(|| Arc::clone(&series)) + .clone(), + ) + } + pub fn factor(&self, date: NaiveDate, symbol: &str) -> Option<&DailyFactorSnapshot> { self.factor_by_date .get(&date) @@ -1585,8 +1617,7 @@ impl DataSet { bar_count: usize, include_now: bool, ) -> Vec { - self.market_series_by_symbol - .get(symbol) + self.market_series(symbol) .map(|series| series.trailing_snapshots(date, bar_count, include_now)) .unwrap_or_default() } @@ -2143,13 +2174,12 @@ impl DataSet { symbol: &str, field: PriceField, ) -> Option { - self.market_series_by_symbol - .get(symbol) + self.market_series(symbol) .and_then(|series| series.price_on_or_before(date, field)) } pub fn market_before(&self, date: NaiveDate, symbol: &str) -> Option<&DailyMarketSnapshot> { - let series = self.market_series_by_symbol.get(symbol)?; + let series = self.market_series(symbol)?; let end = series.previous_completed_end_index(date)?; if end == 0 { return None; @@ -2222,8 +2252,7 @@ impl DataSet { } pub fn market_closes_up_to(&self, date: NaiveDate, symbol: &str, lookback: usize) -> Vec { - self.market_series_by_symbol - .get(symbol) + self.market_series(symbol) .map(|series| series.trailing_values(date, lookback, PriceField::Close)) .unwrap_or_default() } @@ -2236,8 +2265,7 @@ impl DataSet { field: &str, include_now: bool, ) -> Vec { - self.market_series_by_symbol - .get(symbol) + self.market_series(symbol) .map(|series| series.trailing_numeric_values(date, bar_count, field, include_now)) .unwrap_or_default() } @@ -2283,8 +2311,7 @@ impl DataSet { } pub fn market_decision_close(&self, date: NaiveDate, symbol: &str) -> Option { - self.market_series_by_symbol - .get(symbol) + self.market_series(symbol) .and_then(|series| series.decision_price_on_or_before(date)) } @@ -2294,8 +2321,7 @@ impl DataSet { symbol: &str, lookback: usize, ) -> Option { - self.market_series_by_symbol - .get(symbol) + self.market_series(symbol) .and_then(|series| series.decision_close_moving_average(date, lookback)) } @@ -2305,8 +2331,7 @@ impl DataSet { symbol: &str, lookback: usize, ) -> Option { - self.market_series_by_symbol - .get(symbol) + self.market_series(symbol) .and_then(|series| series.decision_volume_moving_average(date, lookback)) } @@ -2393,12 +2418,10 @@ impl DataSet { let field = normalize_field(field); match field.as_str() { "close" | "prev_close" | "stock_close" | "price" => self - .market_series_by_symbol - .get(symbol) + .market_series(symbol) .and_then(|series| series.decision_close_moving_average(date, lookback)), "volume" | "stock_volume" => self - .market_series_by_symbol - .get(symbol) + .market_series(symbol) .and_then(|series| series.decision_volume_moving_average(date, lookback)), "day_open" | "dayopen" => { self.market_moving_average(date, symbol, lookback, PriceField::DayOpen) @@ -2424,8 +2447,7 @@ impl DataSet { self.market_moving_average(date, symbol, lookback, PriceField::Close) } "volume" | "stock_volume" => self - .market_series_by_symbol - .get(symbol) + .market_series(symbol) .and_then(|series| series.current_volume_moving_average(date, lookback)) .or_else(|| self.factor_moving_average(date, symbol, "daily_volume", lookback)), "day_open" | "dayopen" => { @@ -2452,28 +2474,23 @@ impl DataSet { let field = normalize_field(field); match field.as_str() { "close" | "prev_close" | "stock_close" | "price" => self - .market_series_by_symbol - .get(symbol) + .market_series(symbol) .and_then(|series| series.decision_prev_close_values(date, lookback)) .unwrap_or_default(), "volume" | "stock_volume" => self - .market_series_by_symbol - .get(symbol) + .market_series(symbol) .and_then(|series| series.decision_volume_values(date, lookback)) .unwrap_or_default(), "day_open" | "dayopen" => self - .market_series_by_symbol - .get(symbol) + .market_series(symbol) .map(|series| series.trailing_values(date, lookback, PriceField::DayOpen)) .unwrap_or_default(), "open" => self - .market_series_by_symbol - .get(symbol) + .market_series(symbol) .map(|series| series.trailing_values(date, lookback, PriceField::Open)) .unwrap_or_default(), "last" | "last_price" => self - .market_series_by_symbol - .get(symbol) + .market_series(symbol) .map(|series| series.trailing_values(date, lookback, PriceField::Last)) .unwrap_or_default(), other => self.factor_numeric_values(date, symbol, other, lookback), @@ -2505,8 +2522,7 @@ impl DataSet { lookback: usize, field: PriceField, ) -> Option { - self.market_series_by_symbol - .get(symbol) + self.market_series(symbol) .and_then(|series| series.moving_average(date, lookback, field)) } @@ -3565,28 +3581,6 @@ mod optional_date_format { } } -fn build_market_series( - market_by_date: &BTreeMap>>, -) -> HashMap { - let mut grouped = HashMap::>::new(); - for rows in market_by_date.values() { - for row in rows { - grouped - .entry(row.symbol.clone()) - .or_default() - .push(row.as_ref()); - } - } - - grouped - .into_iter() - .map(|(symbol, rows)| { - let series = SymbolPriceSeries::new(symbol.clone(), rows); - (symbol, series) - }) - .collect() -} - fn build_futures_params_index( rows: Vec, ) -> HashMap> {