优化按日快照数据集构造

crates/fidc-core/src/data.rs

@@ -72,6 +72,17 @@ pub enum DataSetError {
     },
     #[error("benchmark snapshot missing for {date}")]
     MissingBenchmark { date: NaiveDate },
+    #[error("duplicate daily snapshot bundle for {date}")]
+    DuplicateDailyBundle { date: NaiveDate },
+    #[error(
+        "{kind} snapshot date {row_date} does not match daily bundle {bundle_date} for {symbol}"
+    )]
+    InvalidDailyBundleComponentDate {
+        kind: &'static str,
+        bundle_date: NaiveDate,
+        row_date: NaiveDate,
+        symbol: String,
+    },
 }
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
@@ -415,6 +426,14 @@ pub struct DailySnapshotBundle {
     pub corporate_actions: Vec<CorporateAction>,
 }
 
+struct GroupedSnapshotComponents {
+    market_by_date: BTreeMap<NaiveDate, Vec<DailyMarketSnapshot>>,
+    factor_by_date: BTreeMap<NaiveDate, Vec<DailyFactorSnapshot>>,
+    candidate_by_date: BTreeMap<NaiveDate, Vec<CandidateEligibility>>,
+    benchmark_by_date: BTreeMap<NaiveDate, BenchmarkSnapshot>,
+    corporate_actions_by_date: BTreeMap<NaiveDate, Vec<CorporateAction>>,
+}
+
 #[derive(Debug, Clone)]
 pub struct DataSetSnapshotComponents {
     pub instruments: Vec<Instrument>,
@@ -1093,13 +1112,20 @@ struct BenchmarkPriceSeries {
 }
 
 impl BenchmarkPriceSeries {
-    fn new(rows: &[BenchmarkSnapshot]) -> Self {
+    fn from_sorted<'a, I>(rows: I) -> Self
-        let mut sorted = rows.to_vec();
+    where
-        sorted.sort_by_key(|row| row.date);
+        I: IntoIterator<Item = &'a BenchmarkSnapshot>,
-        let dates = sorted.iter().map(|row| row.date).collect::<Vec<_>>();
+    {
-        let opens = sorted.iter().map(|row| row.open).collect::<Vec<_>>();
+        let mut dates = Vec::new();
-        let closes = sorted.iter().map(|row| row.close).collect::<Vec<_>>();
+        let mut opens = Vec::new();
-        let prev_closes = sorted.iter().map(|row| row.prev_close).collect::<Vec<_>>();
+        let mut closes = Vec::new();
+        let mut prev_closes = Vec::new();
+        for row in rows {
+            dates.push(row.date);
+            opens.push(row.open);
+            closes.push(row.close);
+            prev_closes.push(row.prev_close);
+        }
         let open_prefix = prefix_sums(&opens);
         let close_prefix = prefix_sums(&closes);
         Self {
@@ -1315,6 +1341,99 @@ impl DataSet {
         )
     }
 
+    pub fn from_daily_bundles_with_execution_quotes(
+        instruments: Vec<Instrument>,
+        mut bundles: Vec<DailySnapshotBundle>,
+        execution_quotes: Vec<IntradayExecutionQuote>,
+    ) -> Result<Self, DataSetError> {
+        bundles.sort_by_key(|bundle| bundle.date);
+        if let Some(pair) = bundles.windows(2).find(|pair| pair[0].date == pair[1].date) {
+            return Err(DataSetError::DuplicateDailyBundle { date: pair[1].date });
+        }
+        let mut grouped = GroupedSnapshotComponents {
+            market_by_date: BTreeMap::new(),
+            factor_by_date: BTreeMap::new(),
+            candidate_by_date: BTreeMap::new(),
+            benchmark_by_date: BTreeMap::new(),
+            corporate_actions_by_date: BTreeMap::new(),
+        };
+        for mut bundle in bundles {
+            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(),
+            )?;
+            bundle.market.sort_by(|left, right| left.symbol.cmp(&right.symbol));
+            bundle.factors = normalize_factor_snapshots(bundle.factors);
+            bundle
+                .factors
+                .sort_by(|left, right| left.symbol.cmp(&right.symbol));
+            bundle
+                .candidates
+                .sort_by(|left, right| left.symbol.cmp(&right.symbol));
+            if !bundle.market.is_empty() {
+                grouped.market_by_date.insert(date, bundle.market);
+            }
+            if !bundle.factors.is_empty() {
+                grouped.factor_by_date.insert(date, bundle.factors);
+            }
+            if !bundle.candidates.is_empty() {
+                grouped.candidate_by_date.insert(date, bundle.candidates);
+            }
+            if !bundle.corporate_actions.is_empty() {
+                grouped
+                    .corporate_actions_by_date
+                    .insert(date, bundle.corporate_actions);
+            }
+            grouped.benchmark_by_date.insert(date, bundle.benchmark);
+        }
+        Self::build_from_components(
+            instruments,
+            Vec::new(),
+            Vec::new(),
+            Vec::new(),
+            Vec::new(),
+            Vec::new(),
+            execution_quotes,
+            Vec::new(),
+            Vec::new(),
+            Vec::new(),
+            Some(grouped),
+        )
+    }
+
     pub fn from_components_with_actions_quotes_and_futures(
         instruments: Vec<Instrument>,
         market: Vec<DailyMarketSnapshot>,
@@ -1397,20 +1516,77 @@ impl DataSet {
         order_book_depth: Vec<IntradayOrderBookDepthLevel>,
         factor_texts: Vec<FactorTextValue>,
     ) -> Result<Self, DataSetError> {
-        let benchmark_code = collect_benchmark_code(&benchmarks)?;
+        Self::build_from_components(
-        let calendar = TradingCalendar::new(benchmarks.iter().map(|item| item.date).collect());
+            instruments,
+            market,
+            factors,
+            candidates,
+            benchmarks,
+            corporate_actions,
+            execution_quotes,
+            futures_params,
+            order_book_depth,
+            factor_texts,
+            None,
+        )
+    }
+
+    #[allow(clippy::too_many_arguments)]
+    fn build_from_components(
+        instruments: Vec<Instrument>,
+        market: Vec<DailyMarketSnapshot>,
+        factors: Vec<DailyFactorSnapshot>,
+        candidates: Vec<CandidateEligibility>,
+        benchmarks: Vec<BenchmarkSnapshot>,
+        corporate_actions: Vec<CorporateAction>,
+        execution_quotes: Vec<IntradayExecutionQuote>,
+        futures_params: Vec<FuturesTradingParameter>,
+        order_book_depth: Vec<IntradayOrderBookDepthLevel>,
+        factor_texts: Vec<FactorTextValue>,
+        grouped: Option<GroupedSnapshotComponents>,
+    ) -> Result<Self, DataSetError> {
+        let (
+            market_by_date,
+            factor_by_date,
+            candidate_by_date,
+            benchmark_by_date,
+            corporate_actions_by_date,
+        ) = if let Some(grouped) = grouped {
+            (
+                grouped.market_by_date,
+                grouped.factor_by_date,
+                grouped.candidate_by_date,
+                grouped.benchmark_by_date,
+                grouped.corporate_actions_by_date,
+            )
+        } else {
+            let mut market_by_date = group_by_date(market, |item| item.date);
+            sort_groups_by_symbol(&mut market_by_date, |item| item.symbol.as_str());
             let factors = normalize_factor_snapshots(factors);
+            let mut factor_by_date = group_by_date(factors, |item| item.date);
+            sort_groups_by_symbol(&mut factor_by_date, |item| item.symbol.as_str());
+            let mut candidate_by_date = group_by_date(candidates, |item| item.date);
+            sort_groups_by_symbol(&mut candidate_by_date, |item| item.symbol.as_str());
+            let benchmark_by_date = benchmarks
+                .into_iter()
+                .map(|item| (item.date, item))
+                .collect::<BTreeMap<_, _>>();
+            let corporate_actions_by_date = group_by_date(corporate_actions, |item| item.date);
+            (
+                market_by_date,
+                factor_by_date,
+                candidate_by_date,
+                benchmark_by_date,
+                corporate_actions_by_date,
+            )
+        };
+        let benchmark_code = collect_benchmark_code(benchmark_by_date.values())?;
+        let calendar = TradingCalendar::new(benchmark_by_date.keys().copied().collect());
 
         let instruments = instruments
             .into_iter()
             .map(|instrument| (instrument.symbol.clone(), instrument))
             .collect::<HashMap<_, _>>();
-
-        let mut market_by_date = group_by_date(market, |item| item.date);
-        sort_groups_by_symbol(&mut market_by_date, |item| item.symbol.as_str());
-
-        let mut factor_by_date = group_by_date(factors, |item| item.date);
-        sort_groups_by_symbol(&mut factor_by_date, |item| item.symbol.as_str());
         let mut market_rows_by_symbol = AHashMap::<String, Vec<&DailyMarketSnapshot>>::new();
         for row in market_by_date.values().flatten() {
             if let Some(rows) = market_rows_by_symbol.get_mut(row.symbol.as_str()) {
@@ -1466,8 +1642,6 @@ impl DataSet {
             .map(|item| ((item.date, item.symbol.clone(), item.field.clone()), item))
             .collect::<HashMap<_, _>>();
 
-        let mut candidate_by_date = group_by_date(candidates, |item| item.date);
-        sort_groups_by_symbol(&mut candidate_by_date, |item| item.symbol.as_str());
         let symbol_id_by_code = build_symbol_id_index(
             &instruments,
             &market_by_date,
@@ -1498,18 +1672,12 @@ impl DataSet {
                 adjusted_close_series_by_symbol_id[symbol_id as usize] = Some(Arc::clone(series));
             }
         }
-        let corporate_actions_by_date = group_by_date(corporate_actions, |item| item.date);
         let execution_quotes_by_date = build_execution_quote_index(execution_quotes);
         let mut execution_quote_dates = execution_quotes_by_date.keys().copied().collect::<Vec<_>>();
         execution_quote_dates.sort_unstable();
         let order_book_depth_index = build_order_book_depth_index(order_book_depth);
 
-        let benchmark_by_date = benchmarks
+        let benchmark_series_cache = BenchmarkPriceSeries::from_sorted(benchmark_by_date.values());
-            .into_iter()
-            .map(|item| (item.date, item))
-            .collect::<BTreeMap<_, _>>();
-        let benchmark_series_cache =
-            BenchmarkPriceSeries::new(&benchmark_by_date.values().cloned().collect::<Vec<_>>());
         let futures_params_by_symbol = build_futures_params_index(futures_params);
 
         Ok(Self {
@@ -3530,6 +3698,28 @@ fn normalize_history_frequency(frequency: &str) -> Option<String> {
     }
 }
 
+fn validate_daily_bundle_component_dates<T, D, S>(
+    rows: &[T],
+    bundle_date: NaiveDate,
+    kind: &'static str,
+    date_of: D,
+    symbol_of: S,
+) -> Result<(), DataSetError>
+where
+    D: Fn(&T) -> NaiveDate,
+    S: Fn(&T) -> &str,
+{
+    if let Some(row) = rows.iter().find(|row| date_of(row) != bundle_date) {
+        return Err(DataSetError::InvalidDailyBundleComponentDate {
+            kind,
+            bundle_date,
+            row_date: date_of(row),
+            symbol: symbol_of(row).to_string(),
+        });
+    }
+    Ok(())
+}
+
 fn group_by_date<T, F>(rows: Vec<T>, mut date_of: F) -> BTreeMap<NaiveDate, Vec<T>>
 where
     F: FnMut(&T) -> NaiveDate,
@@ -3657,19 +3847,21 @@ where
         .map(|index| &rows[index])
 }
 
-fn collect_benchmark_code(benchmarks: &[BenchmarkSnapshot]) -> Result<String, DataSetError> {
+fn collect_benchmark_code<'a, I>(benchmarks: I) -> Result<String, DataSetError>
-    let mut codes = benchmarks
+where
-        .iter()
+    I: IntoIterator<Item = &'a BenchmarkSnapshot>,
-        .map(|row| row.benchmark.clone())
+{
-        .collect::<Vec<_>>();
+    let mut benchmark_code = None;
-    codes.sort_unstable();
+    for benchmark in benchmarks {
-    codes.dedup();
+        match benchmark_code {
-
+            None => benchmark_code = Some(benchmark.benchmark.as_str()),
-    if codes.len() == 1 {
+            Some(code) if code == benchmark.benchmark => {}
-        Ok(codes.remove(0))
+            Some(_) => return Err(DataSetError::MultipleBenchmarks),
-    } else {
-        Err(DataSetError::MultipleBenchmarks)
         }
+    }
+    benchmark_code
+        .map(str::to_owned)
+        .ok_or(DataSetError::MultipleBenchmarks)
 }
 
 fn prefix_sums(values: &[f64]) -> Vec<f64> {
@@ -4044,6 +4236,199 @@ mod tests {
         ));
     }
 
+    #[test]
+    fn daily_bundle_constructor_matches_flat_component_constructor() {
+        let dates = [
+            NaiveDate::from_ymd_opt(2025, 1, 2).unwrap(),
+            NaiveDate::from_ymd_opt(2025, 1, 3).unwrap(),
+        ];
+        let symbols = ["000001.SZ", "600000.SH"];
+        let instruments = symbols
+            .iter()
+            .map(|symbol| Instrument {
+                symbol: (*symbol).to_string(),
+                name: (*symbol).to_string(),
+                board: symbol.rsplit_once('.').unwrap().1.to_string(),
+                round_lot: 100,
+                listed_at: None,
+                delisted_at: None,
+                status: "active".to_string(),
+            })
+            .collect::<Vec<_>>();
+        let mut market = Vec::new();
+        let mut factors = Vec::new();
+        let mut candidates = Vec::new();
+        let mut benchmarks = Vec::new();
+        let mut corporate_actions = Vec::new();
+        let mut execution_quotes = Vec::new();
+        let mut bundles = Vec::new();
+        for (date_index, date) in dates.into_iter().enumerate() {
+            let date_text = date.format("%Y-%m-%d").to_string();
+            let mut day_market = Vec::new();
+            let mut day_factors = Vec::new();
+            let mut day_candidates = Vec::new();
+            for (symbol_index, symbol) in symbols.into_iter().enumerate().rev() {
+                let close = 10.0 + date_index as f64 + symbol_index as f64;
+                let mut market_row = market_row(&date_text, close, 1_000_000);
+                market_row.symbol = symbol.to_string();
+                let factor_row = DailyFactorSnapshot {
+                    date,
+                    symbol: symbol.to_string(),
+                    market_cap_bn: 100.0 + close,
+                    free_float_cap_bn: 80.0 + close,
+                    pe_ttm: 0.0,
+                    turnover_ratio: Some(0.02),
+                    effective_turnover_ratio: Some(0.01),
+                    extra_factors: NumericFactorMap::from([(Cow::Borrowed("quality"), close)]),
+                };
+                let candidate_row = CandidateEligibility {
+                    date,
+                    symbol: symbol.to_string(),
+                    is_st: false,
+                    is_star_st: false,
+                    is_new_listing: false,
+                    is_paused: false,
+                    allow_buy: true,
+                    allow_sell: true,
+                    is_kcb: false,
+                    is_one_yuan: false,
+                    risk_level_code: None,
+                };
+                market.push(market_row.clone());
+                factors.push(factor_row.clone());
+                candidates.push(candidate_row.clone());
+                day_market.push(market_row);
+                day_factors.push(factor_row);
+                day_candidates.push(candidate_row);
+            }
+            let benchmark = benchmark_row(&date_text, 20.0 + date_index as f64);
+            benchmarks.push(benchmark.clone());
+            let corporate_action = CorporateAction {
+                date,
+                symbol: symbols[0].to_string(),
+                payable_date: Some(date),
+                share_cash: 0.1,
+                share_bonus: 0.02,
+                share_gift: 0.03,
+                issue_quantity: 0.0,
+                issue_price: 0.0,
+                reform: false,
+                adjust_factor: Some(1.05),
+                successor_symbol: None,
+                successor_ratio: None,
+                successor_cash: None,
+            };
+            corporate_actions.push(corporate_action.clone());
+            execution_quotes.push(IntradayExecutionQuote {
+                date,
+                symbol: symbols[0].to_string(),
+                timestamp: date.and_hms_opt(10, 18, 0).unwrap(),
+                last_price: 12.3 + date_index as f64,
+                bid1: 12.2 + date_index as f64,
+                ask1: 12.4 + date_index as f64,
+                bid1_volume: 1000,
+                ask1_volume: 1200,
+                volume_delta: 500,
+                amount_delta: 6150.0,
+                trading_phase: Some("continuous_auction".to_string()),
+            });
+            bundles.push(DailySnapshotBundle {
+                date,
+                benchmark,
+                market: day_market,
+                factors: day_factors,
+                candidates: day_candidates,
+                corporate_actions: vec![corporate_action],
+            });
+        }
+
+        let flat = DataSet::from_components_with_actions_and_quotes(
+            instruments.clone(),
+            market,
+            factors,
+            candidates,
+            benchmarks,
+            corporate_actions,
+            execution_quotes.clone(),
+        )
+        .expect("flat dataset");
+        let grouped = DataSet::from_daily_bundles_with_execution_quotes(instruments, bundles, execution_quotes)
+            .expect("daily bundle dataset");
+
+        assert_eq!(flat.calendar().days(), grouped.calendar().days());
+        assert_eq!(flat.benchmark_code(), grouped.benchmark_code());
+        for date in dates {
+            for symbol in symbols {
+                assert_eq!(
+                    flat.market(date, symbol).map(|row| row.close),
+                    grouped.market(date, symbol).map(|row| row.close)
+                );
+                assert_eq!(
+                    flat.factor(date, symbol).map(|row| row.market_cap_bn),
+                    grouped.factor(date, symbol).map(|row| row.market_cap_bn)
+                );
+                assert_eq!(
+                    flat.candidate(date, symbol).map(|row| row.allow_buy),
+                    grouped.candidate(date, symbol).map(|row| row.allow_buy)
+                );
+            }
+            assert_eq!(
+                flat.corporate_actions_on(date).len(),
+                grouped.corporate_actions_on(date).len()
+            );
+            assert_eq!(
+                flat.corporate_actions_on(date)[0].adjust_factor,
+                grouped.corporate_actions_on(date)[0].adjust_factor
+            );
+            let flat_quote = flat.execution_quotes_on(date, symbols[0]);
+            let grouped_quote = grouped.execution_quotes_on(date, symbols[0]);
+            assert_eq!(flat_quote.len(), grouped_quote.len());
+            assert_eq!(flat_quote[0].timestamp, grouped_quote[0].timestamp);
+            assert_eq!(flat_quote[0].last_price, grouped_quote[0].last_price);
+            assert_eq!(flat_quote[0].volume_delta, grouped_quote[0].volume_delta);
+        }
+    }
+
+    #[test]
+    fn daily_bundle_constructor_rejects_duplicate_or_mismatched_dates() {
+        let date = NaiveDate::from_ymd_opt(2025, 1, 2).unwrap();
+        let benchmark = benchmark_row("2025-01-02", 20.0);
+        let empty_bundle = || DailySnapshotBundle {
+            date,
+            benchmark: benchmark.clone(),
+            market: Vec::new(),
+            factors: Vec::new(),
+            candidates: Vec::new(),
+            corporate_actions: Vec::new(),
+        };
+        let duplicate = DataSet::from_daily_bundles_with_execution_quotes(
+            Vec::new(),
+            vec![empty_bundle(), empty_bundle()],
+            Vec::new(),
+        );
+        assert!(matches!(
+            duplicate,
+            Err(DataSetError::DuplicateDailyBundle { date: value }) if value == date
+        ));
+
+        let mut mismatched = empty_bundle();
+        mismatched.market.push(market_row("2025-01-03", 10.0, 1));
+        let mismatch = DataSet::from_daily_bundles_with_execution_quotes(
+            Vec::new(),
+            vec![mismatched],
+            Vec::new(),
+        );
+        assert!(matches!(
+            mismatch,
+            Err(DataSetError::InvalidDailyBundleComponentDate {
+                kind: "market",
+                bundle_date,
+                row_date,
+                ..
+            }) if bundle_date == date && row_date == NaiveDate::from_ymd_opt(2025, 1, 3).unwrap()
+        ));
+    }
+
     #[test]
     fn combined_symbol_snapshot_lookup_uses_alignment_and_falls_back_for_sparse_rows() {
         let date = NaiveDate::parse_from_str("2025-01-02", "%Y-%m-%d").unwrap();
@@ -5055,11 +5440,12 @@ mod tests {
 
     #[test]
     fn benchmark_decision_close_windows_exclude_current_close() {
-        let series = BenchmarkPriceSeries::new(&[
+        let rows = [
             benchmark_row("2025-01-02", 100.0),
             benchmark_row("2025-01-03", 200.0),
             benchmark_row("2025-01-06", 9_999.0),
-        ]);
+        ];
+        let series = BenchmarkPriceSeries::from_sorted(rows.iter());
         let decision_date = NaiveDate::parse_from_str("2025-01-06", "%Y-%m-%d").unwrap();
 
         assert_eq!(series.decision_close(decision_date), Some(9_998.0));