perf(data): normalize owned daily bundles in parallel with stable errors

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::<Vec<_>>();
+        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<String> {
     }
 }
 
+fn normalize_daily_snapshot_bundle(
+    mut bundle: DailySnapshotBundle,
+) -> Result<DailySnapshotBundle, DataSetError> {
+    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<T, D, S>(
     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::<Vec<_>>();
+        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();