diff --git a/crates/fidc-core/src/data.rs b/crates/fidc-core/src/data.rs index 4725d40..47ff342 100644 --- a/crates/fidc-core/src/data.rs +++ b/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, } +struct GroupedSnapshotComponents { + market_by_date: BTreeMap>, + factor_by_date: BTreeMap>, + candidate_by_date: BTreeMap>, + benchmark_by_date: BTreeMap, + corporate_actions_by_date: BTreeMap>, +} + #[derive(Debug, Clone)] pub struct DataSetSnapshotComponents { pub instruments: Vec, @@ -1093,13 +1112,20 @@ struct BenchmarkPriceSeries { } impl BenchmarkPriceSeries { - fn new(rows: &[BenchmarkSnapshot]) -> Self { - let mut sorted = rows.to_vec(); - sorted.sort_by_key(|row| row.date); - let dates = sorted.iter().map(|row| row.date).collect::>(); - let opens = sorted.iter().map(|row| row.open).collect::>(); - let closes = sorted.iter().map(|row| row.close).collect::>(); - let prev_closes = sorted.iter().map(|row| row.prev_close).collect::>(); + fn from_sorted<'a, I>(rows: I) -> Self + where + I: IntoIterator, + { + let mut dates = Vec::new(); + let mut opens = Vec::new(); + 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, + mut bundles: Vec, + execution_quotes: Vec, + ) -> Result { + 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, market: Vec, @@ -1397,20 +1516,77 @@ impl DataSet { order_book_depth: Vec, factor_texts: Vec, ) -> Result { - let benchmark_code = collect_benchmark_code(&benchmarks)?; - let calendar = TradingCalendar::new(benchmarks.iter().map(|item| item.date).collect()); - let factors = normalize_factor_snapshots(factors); + Self::build_from_components( + 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, + market: Vec, + factors: Vec, + candidates: Vec, + benchmarks: Vec, + corporate_actions: Vec, + execution_quotes: Vec, + futures_params: Vec, + order_book_depth: Vec, + factor_texts: Vec, + grouped: Option, + ) -> Result { + 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::>(); + 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::>(); - - 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::>::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::>(); - 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::>(); execution_quote_dates.sort_unstable(); let order_book_depth_index = build_order_book_depth_index(order_book_depth); - let benchmark_by_date = benchmarks - .into_iter() - .map(|item| (item.date, item)) - .collect::>(); - let benchmark_series_cache = - BenchmarkPriceSeries::new(&benchmark_by_date.values().cloned().collect::>()); + let benchmark_series_cache = BenchmarkPriceSeries::from_sorted(benchmark_by_date.values()); let futures_params_by_symbol = build_futures_params_index(futures_params); Ok(Self { @@ -3530,6 +3698,28 @@ fn normalize_history_frequency(frequency: &str) -> Option { } } +fn validate_daily_bundle_component_dates( + 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(rows: Vec, mut date_of: F) -> BTreeMap> where F: FnMut(&T) -> NaiveDate, @@ -3657,19 +3847,21 @@ where .map(|index| &rows[index]) } -fn collect_benchmark_code(benchmarks: &[BenchmarkSnapshot]) -> Result { - let mut codes = benchmarks - .iter() - .map(|row| row.benchmark.clone()) - .collect::>(); - codes.sort_unstable(); - codes.dedup(); - - if codes.len() == 1 { - Ok(codes.remove(0)) - } else { - Err(DataSetError::MultipleBenchmarks) +fn collect_benchmark_code<'a, I>(benchmarks: I) -> Result +where + I: IntoIterator, +{ + let mut benchmark_code = None; + for benchmark in benchmarks { + match benchmark_code { + None => benchmark_code = Some(benchmark.benchmark.as_str()), + Some(code) if code == benchmark.benchmark => {} + Some(_) => return Err(DataSetError::MultipleBenchmarks), + } } + benchmark_code + .map(str::to_owned) + .ok_or(DataSetError::MultipleBenchmarks) } fn prefix_sums(values: &[f64]) -> Vec { @@ -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::>(); + 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));