From fe7243bbc3deaf59565382b3cd58968f52008be9 Mon Sep 17 00:00:00 2001 From: boris Date: Sat, 12 Sep 2026 19:00:02 +0800 Subject: [PATCH] perf(data): normalize owned daily bundles in parallel with stable errors --- crates/fidc-core/src/data.rs | 114 ++++++++++++++++++++++------------- 1 file changed, 73 insertions(+), 41 deletions(-) diff --git a/crates/fidc-core/src/data.rs b/crates/fidc-core/src/data.rs index 97702e8..e988575 100644 --- a/crates/fidc-core/src/data.rs +++ b/crates/fidc-core/src/data.rs @@ -1586,48 +1586,15 @@ impl DataSet { benchmark_by_date: BTreeMap::new(), corporate_actions_by_date: BTreeMap::new(), }; - for mut bundle in bundles { + // Indexed collection retains chronological error precedence while each + // worker validates and normalizes only its owned day buffers. + let prepared = bundles + .into_par_iter() + .map(normalize_daily_snapshot_bundle) + .collect::>(); + for bundle in prepared { + let bundle = bundle?; let date = bundle.date; - if bundle.benchmark.date != date { - return Err(DataSetError::InvalidDailyBundleComponentDate { - kind: "benchmark", - bundle_date: date, - row_date: bundle.benchmark.date, - symbol: bundle.benchmark.benchmark.clone(), - }); - } - validate_daily_bundle_component_dates( - &bundle.market, - date, - "market", - |row| row.date, - |row| row.symbol.as_str(), - )?; - validate_daily_bundle_component_dates( - &bundle.factors, - date, - "factor", - |row| row.date, - |row| row.symbol.as_str(), - )?; - validate_daily_bundle_component_dates( - &bundle.candidates, - date, - "candidate", - |row| row.date, - |row| row.symbol.as_str(), - )?; - validate_daily_bundle_component_dates( - &bundle.corporate_actions, - date, - "corporate_action", - |row| row.date, - |row| row.symbol.as_str(), - )?; - sort_rows_by_symbol_if_needed(&mut bundle.market, |row| row.symbol.as_str()); - bundle.factors = normalize_factor_snapshots(bundle.factors)?; - sort_rows_by_symbol_if_needed(&mut bundle.factors, |row| row.symbol.as_str()); - sort_rows_by_symbol_if_needed(&mut bundle.candidates, |row| row.symbol.as_str()); if !bundle.market.is_empty() { grouped.market_by_date.insert(date, bundle.market); } @@ -4472,6 +4439,38 @@ fn normalize_history_frequency(frequency: &str) -> Option { } } +fn normalize_daily_snapshot_bundle( + mut bundle: DailySnapshotBundle, +) -> Result { + let date = bundle.date; + if bundle.benchmark.date != date { + return Err(DataSetError::InvalidDailyBundleComponentDate { + kind: "benchmark", + bundle_date: date, + row_date: bundle.benchmark.date, + symbol: bundle.benchmark.benchmark.clone(), + }); + } + validate_daily_bundle_component_dates( + &bundle.market, date, "market", |row| row.date, |row| row.symbol.as_str(), + )?; + validate_daily_bundle_component_dates( + &bundle.factors, date, "factor", |row| row.date, |row| row.symbol.as_str(), + )?; + validate_daily_bundle_component_dates( + &bundle.candidates, date, "candidate", |row| row.date, |row| row.symbol.as_str(), + )?; + validate_daily_bundle_component_dates( + &bundle.corporate_actions, date, "corporate_action", |row| row.date, + |row| row.symbol.as_str(), + )?; + sort_rows_by_symbol_if_needed(&mut bundle.market, |row| row.symbol.as_str()); + bundle.factors = normalize_factor_snapshots(bundle.factors)?; + sort_rows_by_symbol_if_needed(&mut bundle.factors, |row| row.symbol.as_str()); + sort_rows_by_symbol_if_needed(&mut bundle.candidates, |row| row.symbol.as_str()); + Ok(bundle) +} + fn validate_daily_bundle_component_dates( rows: &[T], bundle_date: NaiveDate, @@ -5542,6 +5541,39 @@ mod tests { )); } + #[test] + fn parallel_daily_bundle_validation_keeps_earliest_error_and_component_order() { + let bundles = || (2..30).rev().map(|day| { + let date = NaiveDate::from_ymd_opt(2025, 1, day).unwrap(); + let mut benchmark = benchmark_row("2025-01-01", 20.0); + benchmark.date = date; + DailySnapshotBundle { + date, benchmark, + market: vec![market_row("2025-01-01", 10.0, 100)], + factors: Vec::new(), candidates: Vec::new(), corporate_actions: Vec::new(), + } + }).collect::>(); + for threads in [1, 2, 8] { + let pool = rayon::ThreadPoolBuilder::new().num_threads(threads).build().unwrap(); + for _ in 0..4 { + let result = pool.install(|| DataSet::from_daily_bundles_with_execution_quotes( + Vec::new(), bundles(), Vec::new(), + )); + assert!(matches!(result, Err(DataSetError::InvalidDailyBundleComponentDate { + kind: "market", bundle_date, .. + }) if bundle_date == NaiveDate::from_ymd_opt(2025, 1, 2).unwrap())); + } + let mut values = bundles(); + values.last_mut().unwrap().benchmark.date = NaiveDate::from_ymd_opt(2025, 1, 1).unwrap(); + let result = pool.install(|| DataSet::from_daily_bundles_with_execution_quotes( + Vec::new(), values, Vec::new(), + )); + assert!(matches!(result, Err(DataSetError::InvalidDailyBundleComponentDate { + kind: "benchmark", bundle_date, .. + }) if bundle_date == NaiveDate::from_ymd_opt(2025, 1, 2).unwrap())); + } + } + #[test] fn direct_symbol_id_snapshot_lookups_preserve_alignment_for_sparse_rows() { let date = NaiveDate::parse_from_str("2025-01-02", "%Y-%m-%d").unwrap();