From 5a7c49a4543b584503ff0d2c2ae513c68d25de8a Mon Sep 17 00:00:00 2001 From: boris Date: Sun, 6 Sep 2026 06:18:21 +0800 Subject: [PATCH] perf: build market series by symbol id --- crates/fidc-core/src/data.rs | 122 ++++++++++++++++------------------- 1 file changed, 56 insertions(+), 66 deletions(-) diff --git a/crates/fidc-core/src/data.rs b/crates/fidc-core/src/data.rs index 65dbcd2..a242321 100644 --- a/crates/fidc-core/src/data.rs +++ b/crates/fidc-core/src/data.rs @@ -1318,8 +1318,6 @@ pub struct DataSet { execution_quote_dates: Arc>, order_book_depth_index: Arc>>, benchmark_by_date: Arc>, - market_series_by_symbol: Arc>>, - adjusted_close_series_by_symbol: Arc>>, market_series_by_symbol_id: Arc>>>, adjusted_close_series_by_symbol_id: Arc>>>, market_series_end_positions_by_calendar_index: Arc>, @@ -1709,44 +1707,68 @@ impl DataSet { .into_iter() .map(|instrument| (instrument.symbol.clone(), instrument)) .collect::>(); - let mut market_rows_by_symbol = AHashMap::>::new(); - for row in market_by_date.values().flatten() { - if let Some(rows) = market_rows_by_symbol.get_mut(row.symbol.as_str()) { - rows.push(row); - continue; + let symbol_id_by_code = build_symbol_id_index( + &instruments, + &market_by_date, + &factor_by_date, + &candidate_by_date, + ); + let symbol_count = symbol_id_by_code.len(); + let mut symbol_by_id = vec![Arc::::from(""); symbol_count]; + for (symbol, symbol_id) in &symbol_id_by_code { + symbol_by_id[*symbol_id as usize] = Arc::::from(symbol.as_str()); + } + let mut instruments_by_symbol_id = vec![None; symbol_count]; + for (symbol, instrument) in &instruments { + if let Some(symbol_id) = symbol_id_by_code.get(symbol).copied() { + instruments_by_symbol_id[symbol_id as usize] = Some(instrument.clone()); } - market_rows_by_symbol.insert(row.symbol.clone(), vec![row]); } - let market_rows_by_symbol = market_rows_by_symbol.into_iter().collect::>(); - let market_series_by_symbol = market_rows_by_symbol + + let mut market_rows_by_symbol_id = (0..symbol_count) + .map(|_| Vec::<&DailyMarketSnapshot>::new()) + .collect::>(); + for row in market_by_date.values().flatten() { + let symbol_id = *symbol_id_by_code + .get(row.symbol.as_str()) + .expect("market symbol missing from FIDC symbol index"); + market_rows_by_symbol_id[symbol_id as usize].push(row); + } + let market_series_by_symbol_id = market_rows_by_symbol_id .into_par_iter() - .map(|(symbol, rows)| { - let series = Arc::new(SymbolPriceSeries::from_sorted_rows(symbol.clone(), rows)); - (symbol, series) + .enumerate() + .map(|(symbol_id, rows)| { + if rows.is_empty() { + return None; + } + Some(Arc::new(SymbolPriceSeries::from_sorted_rows( + symbol_by_id[symbol_id].to_string(), + rows, + ))) }) - .collect::>() - .into_iter() - .collect::>(); - let mut factor_rows_by_symbol = AHashMap::<&str, Vec<&DailyFactorSnapshot>>::new(); + .collect::>(); + let mut factor_rows_by_symbol_id = (0..symbol_count) + .map(|_| Vec::<&DailyFactorSnapshot>::new()) + .collect::>(); for row in factor_by_date.values().flatten() { - factor_rows_by_symbol - .entry(row.symbol.as_str()) - .or_default() - .push(row); + let symbol_id = *symbol_id_by_code + .get(row.symbol.as_str()) + .expect("factor symbol missing from FIDC symbol index"); + factor_rows_by_symbol_id[symbol_id as usize].push(row); } - let adjusted_close_series_by_symbol = market_series_by_symbol + let adjusted_close_series_by_symbol_id = market_series_by_symbol_id .par_iter() - .filter_map(|(symbol, market)| { - let factor_rows = factor_rows_by_symbol - .get(symbol.as_str()) - .map(Vec::as_slice) - .unwrap_or_default(); - AdjustedCloseSeries::new(market, factor_rows) - .map(|series| (symbol.clone(), Arc::new(series))) + .enumerate() + .map(|(symbol_id, market)| { + market.as_ref().and_then(|market| { + AdjustedCloseSeries::new( + market, + factor_rows_by_symbol_id[symbol_id].as_slice(), + ) + .map(Arc::new) + }) }) - .collect::>() - .into_iter() - .collect::>(); + .collect::>(); let factor_texts = factor_texts .into_iter() .filter_map(|mut item| { @@ -1764,22 +1786,6 @@ impl DataSet { .map(|item| ((item.date, item.symbol.clone(), item.field.clone()), item)) .collect::>(); - let symbol_id_by_code = build_symbol_id_index( - &instruments, - &market_by_date, - &factor_by_date, - &candidate_by_date, - ); - let mut symbol_by_id = vec![Arc::::from(""); symbol_id_by_code.len()]; - for (symbol, symbol_id) in &symbol_id_by_code { - symbol_by_id[*symbol_id as usize] = Arc::::from(symbol.as_str()); - } - let mut instruments_by_symbol_id = vec![None; symbol_id_by_code.len()]; - for (symbol, instrument) in &instruments { - if let Some(symbol_id) = symbol_id_by_code.get(symbol).copied() { - instruments_by_symbol_id[symbol_id as usize] = Some(instrument.clone()); - } - } let market_symbol_ids_by_date = build_group_symbol_ids(&market_by_date, &symbol_id_by_code, |item| { item.symbol.as_str() @@ -1809,18 +1815,6 @@ impl DataSet { &candidate_symbol_ids_by_date, symbol_id_by_code.len(), ); - let mut market_series_by_symbol_id = vec![None; symbol_id_by_code.len()]; - for (symbol, series) in &market_series_by_symbol { - if let Some(symbol_id) = symbol_id_by_code.get(symbol).copied() { - market_series_by_symbol_id[symbol_id as usize] = Some(Arc::clone(series)); - } - } - let mut adjusted_close_series_by_symbol_id = vec![None; symbol_id_by_code.len()]; - for (symbol, series) in &adjusted_close_series_by_symbol { - if let Some(symbol_id) = symbol_id_by_code.get(symbol).copied() { - adjusted_close_series_by_symbol_id[symbol_id as usize] = Some(Arc::clone(series)); - } - } let market_series_end_positions_by_calendar_index = build_calendar_series_end_positions(&market_series_by_symbol_id, &calendar); let execution_quotes_by_date = build_execution_quote_index(execution_quotes); @@ -1853,8 +1847,6 @@ impl DataSet { execution_quote_dates: Arc::new(execution_quote_dates), order_book_depth_index: Arc::new(order_book_depth_index), benchmark_by_date: Arc::new(benchmark_by_date), - market_series_by_symbol: Arc::new(market_series_by_symbol), - adjusted_close_series_by_symbol: Arc::new(adjusted_close_series_by_symbol), market_series_by_symbol_id: Arc::new(market_series_by_symbol_id), adjusted_close_series_by_symbol_id: Arc::new(adjusted_close_series_by_symbol_id), market_series_end_positions_by_calendar_index: Arc::new( @@ -1993,7 +1985,7 @@ impl DataSet { } fn market_series(&self, symbol: &str) -> Option<&SymbolPriceSeries> { - self.market_series_by_symbol.get(symbol).map(Arc::as_ref) + self.market_series_by_symbol_id(self.symbol_id(symbol)?) } fn market_series_by_symbol_id(&self, symbol_id: u32) -> Option<&SymbolPriceSeries> { @@ -2003,9 +1995,7 @@ impl DataSet { } fn adjusted_close_series(&self, symbol: &str) -> Option<&AdjustedCloseSeries> { - self.adjusted_close_series_by_symbol - .get(symbol) - .map(Arc::as_ref) + self.adjusted_close_series_by_symbol_id(self.symbol_id(symbol)?) } fn adjusted_close_series_by_symbol_id(&self, symbol_id: u32) -> Option<&AdjustedCloseSeries> {