From ed126a36300e2c47d018b978fa8ac92490defb38 Mon Sep 17 00:00:00 2001 From: boris Date: Thu, 27 Aug 2026 13:14:41 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BC=98=E5=8C=96=E5=88=86=E9=92=9F=E5=8E=86?= =?UTF-8?q?=E5=8F=B2=E7=AA=97=E5=8F=A3=E8=AF=BB=E5=8F=96=E5=B9=B6=E7=A7=BB?= =?UTF-8?q?=E9=99=A4=E6=BB=9A=E5=8A=A8=E5=8F=8C=E5=8F=A3=E5=BE=84?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- crates/fidc-core/src/data.rs | 81 ++++++-- .../tests/intraday_history_performance.rs | 179 ++++++++++++++++++ 2 files changed, 239 insertions(+), 21 deletions(-) create mode 100644 crates/fidc-core/tests/intraday_history_performance.rs diff --git a/crates/fidc-core/src/data.rs b/crates/fidc-core/src/data.rs index 0545d4d..2bac616 100644 --- a/crates/fidc-core/src/data.rs +++ b/crates/fidc-core/src/data.rs @@ -1174,6 +1174,7 @@ pub struct DataSet { candidate_symbol_ids_by_date: Arc>>, corporate_actions_by_date: Arc>>, execution_quotes_by_date: Arc>>>, + execution_quote_dates: Arc>, order_book_depth_index: Arc>>, benchmark_by_date: Arc>, market_series_by_symbol: Arc>>, @@ -1441,6 +1442,8 @@ impl DataSet { } 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 @@ -1464,6 +1467,7 @@ impl DataSet { candidate_symbol_ids_by_date: Arc::new(candidate_symbol_ids_by_date), corporate_actions_by_date: Arc::new(corporate_actions_by_date), execution_quotes_by_date: Arc::new(execution_quotes_by_date), + execution_quote_dates: Arc::new(execution_quote_dates), order_book_depth_index: Arc::new(order_book_depth_index), benchmark_by_date: Arc::new(benchmark_by_date), market_series_by_symbol: Arc::new(market_series_by_symbol), @@ -1648,9 +1652,14 @@ impl DataSet { .push(quote); } let mut added = 0usize; + let mut new_dates = Vec::new(); let execution_quotes_by_date = Arc::make_mut(&mut self.execution_quotes_by_date); for (date, rows_by_symbol) in grouped { + let date_is_new = !execution_quotes_by_date.contains_key(&date); let target_by_symbol = execution_quotes_by_date.entry(date).or_default(); + if date_is_new { + new_dates.push(date); + } for (symbol, mut incoming) in rows_by_symbol { incoming.sort_by_key(|quote| quote.timestamp); incoming.dedup_by(|left, right| left.timestamp == right.timestamp); @@ -1688,6 +1697,14 @@ impl DataSet { *target = merged; } } + if !new_dates.is_empty() { + let dates = Arc::make_mut(&mut self.execution_quote_dates); + for date in new_dates { + if let Err(index) = dates.binary_search(&date) { + dates.insert(index, date); + } + } + } added } @@ -1744,10 +1761,18 @@ impl DataSet { } pub fn remove_execution_quotes_on_date(&mut self, date: NaiveDate) -> usize { - Arc::make_mut(&mut self.execution_quotes_by_date) - .remove(&date) - .map(|rows_by_symbol| rows_by_symbol.into_values().map(|rows| rows.len()).sum()) - .unwrap_or_default() + let removed = Arc::make_mut(&mut self.execution_quotes_by_date).remove(&date); + let Some(rows_by_symbol) = removed else { + return 0; + }; + let dates = Arc::make_mut(&mut self.execution_quote_dates); + if let Ok(index) = dates.binary_search(&date) { + dates.remove(index); + } + rows_by_symbol + .into_values() + .map(|rows| rows.len()) + .sum() } pub fn snapshot_components(&self) -> DataSetSnapshotComponents { @@ -1901,16 +1926,29 @@ impl DataSet { if bar_count == 0 { return Vec::new(); } - let mut quotes = self - .execution_quotes_by_date - .values() - .filter_map(|rows_by_symbol| rows_by_symbol.get(symbol)) - .flat_map(|rows| rows.iter()) - .filter(|quote| intraday_quote_visible(quote, date, active_datetime, include_now)) - .cloned() - .collect::>(); - quotes.sort_by_key(|quote| quote.timestamp); - take_last(quotes, bar_count) + let end = self + .execution_quote_dates + .partition_point(|quote_date| *quote_date <= date); + let mut quotes = Vec::with_capacity(bar_count); + 'dates: for quote_date in self.execution_quote_dates[..end].iter().rev() { + let Some(rows) = self + .execution_quotes_by_date + .get(quote_date) + .and_then(|rows_by_symbol| rows_by_symbol.get(symbol)) + else { + continue; + }; + for quote in rows.iter().rev() { + if intraday_quote_visible(quote, date, active_datetime, include_now) { + quotes.push(quote.clone()); + if quotes.len() == bar_count { + break 'dates; + } + } + } + } + quotes.reverse(); + quotes } pub fn trading_dates(&self, start: NaiveDate, end: NaiveDate) -> Vec { @@ -3396,13 +3434,6 @@ fn normalize_history_frequency(frequency: &str) -> Option { } } -fn take_last(mut rows: Vec, count: usize) -> Vec { - if rows.len() <= count { - return rows; - } - rows.split_off(rows.len() - count) -} - fn group_by_date(rows: Vec, mut date_of: F) -> BTreeMap> where F: FnMut(&T) -> NaiveDate, @@ -3864,6 +3895,10 @@ mod tests { &data.execution_quotes_by_date, &run_data.execution_quotes_by_date )); + assert!(Arc::ptr_eq( + &data.execution_quote_dates, + &run_data.execution_quote_dates + )); run_data.add_execution_quotes(vec![IntradayExecutionQuote { date, @@ -3886,6 +3921,10 @@ mod tests { &data.execution_quotes_by_date, &run_data.execution_quotes_by_date )); + assert!(!Arc::ptr_eq( + &data.execution_quote_dates, + &run_data.execution_quote_dates + )); } #[test] diff --git a/crates/fidc-core/tests/intraday_history_performance.rs b/crates/fidc-core/tests/intraday_history_performance.rs new file mode 100644 index 0000000..6fd6208 --- /dev/null +++ b/crates/fidc-core/tests/intraday_history_performance.rs @@ -0,0 +1,179 @@ +use std::hint::black_box; +use std::time::Instant; + +use chrono::{Duration, NaiveDate, NaiveDateTime, NaiveTime}; +use fidc_core::{ + BenchmarkSnapshot, DailyMarketSnapshot, DataSet, Instrument, IntradayExecutionQuote, +}; + +const SYMBOL: &str = "000001.SZ"; + +fn dataset(day_count: usize, bars_per_day: usize) -> (DataSet, Vec) { + let start = NaiveDate::from_ymd_opt(2025, 1, 1).expect("valid start date"); + let dates = (0..day_count) + .map(|offset| start + Duration::days(offset as i64)) + .collect::>(); + let markets = dates + .iter() + .map(|date| DailyMarketSnapshot { + date: *date, + symbol: SYMBOL.to_string(), + timestamp: None, + day_open: 10.0, + open: 10.0, + high: 10.5, + low: 9.5, + close: 10.0, + last_price: 10.0, + bid1: 9.99, + ask1: 10.01, + prev_close: 10.0, + volume: 1_000_000, + minute_volume: 1_000, + bid1_volume: 10_000, + ask1_volume: 10_000, + trading_phase: Some("continuous".to_string()), + paused: false, + upper_limit: 11.0, + lower_limit: 9.0, + price_tick: 0.01, + }) + .collect::>(); + let benchmarks = dates + .iter() + .map(|date| BenchmarkSnapshot { + date: *date, + benchmark: "000852.SH".to_string(), + open: 1_000.0, + close: 1_000.0, + prev_close: 1_000.0, + volume: 10_000_000, + }) + .collect::>(); + let mut quotes = Vec::with_capacity(day_count * bars_per_day); + for date in &dates { + let session_start = date + .and_hms_opt(9, 30, 0) + .expect("valid session start"); + for offset in 0..bars_per_day { + let timestamp = session_start + Duration::minutes(offset as i64); + quotes.push(IntradayExecutionQuote { + date: *date, + symbol: SYMBOL.to_string(), + timestamp, + last_price: 10.0 + offset as f64 / 10_000.0, + bid1: 9.99, + ask1: 10.01, + bid1_volume: 10_000, + ask1_volume: 10_000, + volume_delta: 1_000, + amount_delta: 10_000.0, + trading_phase: Some("continuous".to_string()), + }); + } + } + let data = DataSet::from_components_with_actions_and_quotes( + vec![Instrument { + symbol: SYMBOL.to_string(), + name: "平安银行".to_string(), + board: "SZ".to_string(), + round_lot: 100, + listed_at: Some(start - Duration::days(1_000)), + delisted_at: None, + status: "active".to_string(), + }], + markets, + Vec::new(), + Vec::new(), + benchmarks, + Vec::new(), + quotes, + ) + .expect("build intraday history dataset"); + (data, dates) +} + +fn timestamp(date: NaiveDate, time: &str) -> NaiveDateTime { + let time = NaiveTime::parse_from_str(time, "%H:%M:%S").expect("valid time"); + date.and_time(time) +} + +#[test] +fn intraday_history_is_bounded_by_visibility_and_preserves_order() { + let (data, dates) = dataset(3, 4); + let rows = data.history_intraday_quotes_at( + dates[1], + Some(timestamp(dates[1], "09:32:00")), + SYMBOL, + 3, + false, + ); + assert_eq!( + rows.iter().map(|row| row.timestamp).collect::>(), + vec![ + timestamp(dates[0], "09:33:00"), + timestamp(dates[1], "09:30:00"), + timestamp(dates[1], "09:31:00"), + ] + ); + + let including_now = data.history_intraday_quotes_at( + dates[1], + Some(timestamp(dates[1], "09:32:00")), + SYMBOL, + 3, + true, + ); + assert_eq!( + including_now + .iter() + .map(|row| row.timestamp) + .collect::>(), + vec![ + timestamp(dates[1], "09:30:00"), + timestamp(dates[1], "09:31:00"), + timestamp(dates[1], "09:32:00"), + ] + ); +} + +#[test] +#[ignore = "manual release-mode intraday history benchmark"] +fn benchmark_bounded_intraday_history() { + let (data, dates) = dataset(250, 240); + let active_datetime = timestamp(*dates.last().expect("last date"), "13:29:00"); + + for _ in 0..5 { + black_box(data.history_intraday_quotes_at( + active_datetime.date(), + Some(active_datetime), + SYMBOL, + 30, + true, + )); + } + + let started = Instant::now(); + let mut checksum = 0_i64; + for _ in 0..200 { + let rows = data.history_intraday_quotes_at( + active_datetime.date(), + Some(active_datetime), + SYMBOL, + 30, + true, + ); + checksum += rows + .last() + .expect("history row") + .timestamp + .and_utc() + .timestamp(); + black_box(&rows); + } + let elapsed = started.elapsed(); + eprintln!( + "intraday_history_benchmark iterations=200 rows_per_dataset=60000 elapsed_seconds={:.6} checksum={checksum}", + elapsed.as_secs_f64(), + ); +}