From 96d0cc6fc4beddb452f98b687a791f1a11d6ef69 Mon Sep 17 00:00:00 2001 From: boris Date: Sun, 6 Sep 2026 20:55:18 +0800 Subject: [PATCH] perf: build price series by symbol id --- crates/fidc-core/src/data.rs | 137 ++++++++++++++++++----------------- 1 file changed, 70 insertions(+), 67 deletions(-) diff --git a/crates/fidc-core/src/data.rs b/crates/fidc-core/src/data.rs index 65dbcd2..768a985 100644 --- a/crates/fidc-core/src/data.rs +++ b/crates/fidc-core/src/data.rs @@ -1709,44 +1709,81 @@ 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 market_symbol_ids_by_date = + build_group_symbol_ids(&market_by_date, &symbol_id_by_code, |item| { + item.symbol.as_str() + }); + let factor_symbol_ids_by_date = + build_group_symbol_ids(&factor_by_date, &symbol_id_by_code, |item| { + item.symbol.as_str() + }); + let candidate_symbol_ids_by_date = + build_group_symbol_ids(&candidate_by_date, &symbol_id_by_code, |item| { + item.symbol.as_str() + }); + + let mut market_rows_by_symbol_id = (0..symbol_count) + .map(|_| Vec::<&DailyMarketSnapshot>::new()) + .collect::>(); + for (date, rows) in &market_by_date { + let symbol_ids = market_symbol_ids_by_date + .get(date) + .expect("market symbol ids missing while building price series"); + for (row, symbol_id) in rows.iter().zip(symbol_ids) { + market_rows_by_symbol_id[*symbol_id as usize].push(row); } - 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 factor_rows_by_symbol_id = (0..symbol_count) + .map(|_| Vec::<&DailyFactorSnapshot>::new()) + .collect::>(); + for (date, rows) in &factor_by_date { + let symbol_ids = factor_symbol_ids_by_date + .get(date) + .expect("factor symbol ids missing while building adjusted series"); + for (row, symbol_id) in rows.iter().zip(symbol_ids) { + factor_rows_by_symbol_id[*symbol_id as usize].push(row); + } + } + let 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) + .zip(factor_rows_by_symbol_id.into_par_iter()) + .enumerate() + .map(|(symbol_id, (market_rows, factor_rows))| { + if market_rows.is_empty() { + return (None, None); + } + let symbol = symbol_by_id[symbol_id].to_string(); + let market = Arc::new(SymbolPriceSeries::from_sorted_rows(symbol, market_rows)); + let adjusted = AdjustedCloseSeries::new(&market, &factor_rows).map(Arc::new); + (Some(market), adjusted) }) - .collect::>() - .into_iter() - .collect::>(); - let mut factor_rows_by_symbol = AHashMap::<&str, Vec<&DailyFactorSnapshot>>::new(); - for row in factor_by_date.values().flatten() { - factor_rows_by_symbol - .entry(row.symbol.as_str()) - .or_default() - .push(row); + .collect::>(); + let mut market_series_by_symbol = AHashMap::with_capacity(symbol_count); + let mut adjusted_close_series_by_symbol = AHashMap::with_capacity(symbol_count); + let mut market_series_by_symbol_id = Vec::with_capacity(symbol_count); + let mut adjusted_close_series_by_symbol_id = Vec::with_capacity(symbol_count); + for (symbol_id, (market, adjusted)) in series_by_symbol_id.into_iter().enumerate() { + let symbol = &symbol_by_id[symbol_id]; + if let Some(series) = market.as_ref() { + market_series_by_symbol.insert(symbol.to_string(), Arc::clone(series)); + } + if let Some(series) = adjusted.as_ref() { + adjusted_close_series_by_symbol.insert(symbol.to_string(), Arc::clone(series)); + } + market_series_by_symbol_id.push(market); + adjusted_close_series_by_symbol_id.push(adjusted); } - let adjusted_close_series_by_symbol = market_series_by_symbol - .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))) - }) - .collect::>() - .into_iter() - .collect::>(); let factor_texts = factor_texts .into_iter() .filter_map(|mut item| { @@ -1764,36 +1801,14 @@ 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() - }); - let factor_symbol_ids_by_date = - build_group_symbol_ids(&factor_by_date, &symbol_id_by_code, |item| { - item.symbol.as_str() - }); let factor_market_cap_order_by_date = build_factor_market_cap_order(&factor_by_date, &factor_symbol_ids_by_date); - let candidate_symbol_ids_by_date = - build_group_symbol_ids(&candidate_by_date, &symbol_id_by_code, |item| { - item.symbol.as_str() - }); let market_row_positions_by_date = build_dense_row_positions( &market_by_date, &market_symbol_ids_by_date, @@ -1809,18 +1824,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);