From c86a0e2339d291f7eee2ff5173a85884e7746986 Mon Sep 17 00:00:00 2001 From: boris Date: Thu, 27 Aug 2026 13:25:28 +0800 Subject: [PATCH] =?UTF-8?q?=E8=AE=A9=E5=88=86=E9=92=9F=E8=A1=8C=E6=83=85?= =?UTF-8?q?=E6=8C=89=E6=B5=81=E5=BC=8F=E8=BF=AD=E4=BB=A3=E5=99=A8=E5=A4=84?= =?UTF-8?q?=E7=90=86?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- crates/fidc-core/src/data.rs | 97 +++++++++++++------ crates/fidc-core/src/engine.rs | 42 +++++--- .../tests/intraday_history_performance.rs | 43 ++++++++ 3 files changed, 137 insertions(+), 45 deletions(-) diff --git a/crates/fidc-core/src/data.rs b/crates/fidc-core/src/data.rs index 2bac616..e502d3a 100644 --- a/crates/fidc-core/src/data.rs +++ b/crates/fidc-core/src/data.rs @@ -321,6 +321,57 @@ impl IntradayExecutionQuote { } } +/// A borrowed, timestamp-ordered merge of the execution-quote streams for one +/// trading day. The iterator keeps only stream cursors and never clones quote +/// payloads; callers decide how much of the day they need to retain. +pub struct ExecutionQuoteIterator<'a> { + streams: Vec<(&'a str, &'a [IntradayExecutionQuote])>, + heap: BinaryHeap>, +} + +impl<'a> ExecutionQuoteIterator<'a> { + fn new( + rows_by_symbol: Option<&'a HashMap>>, + symbols: Option<&BTreeSet>, + ) -> Self { + let mut streams = rows_by_symbol + .into_iter() + .flat_map(|rows_by_symbol| rows_by_symbol.iter()) + .filter(|(symbol, _)| { + symbols + .map(|allowed_symbols| allowed_symbols.contains(*symbol)) + .unwrap_or(true) + }) + .map(|(symbol, rows)| (symbol.as_str(), rows.as_slice())) + .collect::>(); + streams.sort_by_key(|(symbol, _)| *symbol); + + let mut heap = BinaryHeap::with_capacity(streams.len()); + for (stream_index, (_, rows)) in streams.iter().enumerate() { + if let Some(first) = rows.first() { + heap.push(Reverse((first.timestamp, stream_index, 0))); + } + } + Self { streams, heap } + } +} + +impl<'a> Iterator for ExecutionQuoteIterator<'a> { + type Item = &'a IntradayExecutionQuote; + + fn next(&mut self) -> Option { + let Reverse((_timestamp, stream_index, row_index)) = self.heap.pop()?; + let rows = self.streams.get(stream_index)?.1; + let quote = rows.get(row_index)?; + let next_index = row_index + 1; + if let Some(next) = rows.get(next_index) { + self.heap + .push(Reverse((next.timestamp, stream_index, next_index))); + } + Some(quote) + } +} + impl CorporateAction { pub fn split_ratio(&self) -> f64 { 1.0 + self.share_bonus.max(0.0) + self.share_gift.max(0.0) @@ -1723,41 +1774,22 @@ impl DataSet { self.execution_quotes_on_date_for_symbols(date, None) } + pub fn execution_quotes_iter_on_date_for_symbols( + &self, + date: NaiveDate, + symbols: Option<&BTreeSet>, + ) -> ExecutionQuoteIterator<'_> { + ExecutionQuoteIterator::new(self.execution_quotes_by_date.get(&date), symbols) + } + pub fn execution_quotes_on_date_for_symbols( &self, date: NaiveDate, symbols: Option<&BTreeSet>, ) -> Vec { - let Some(rows_by_symbol) = self.execution_quotes_by_date.get(&date) else { - return Vec::new(); - }; - let mut streams = rows_by_symbol - .iter() - .filter(|(symbol, _)| { - symbols - .map(|allowed_symbols| allowed_symbols.contains(*symbol)) - .unwrap_or(true) - }) - .map(|(symbol, rows)| (symbol.as_str(), rows.as_slice())) - .collect::>(); - streams.sort_by_key(|(symbol, _)| *symbol); - let total_rows = streams.iter().map(|(_, rows)| rows.len()).sum(); - let mut heap = BinaryHeap::>::new(); - for (stream_index, (_, rows)) in streams.iter().enumerate() { - if let Some(first) = rows.first() { - heap.push(Reverse((first.timestamp, stream_index, 0))); - } - } - let mut merged = Vec::with_capacity(total_rows); - while let Some(Reverse((_timestamp, stream_index, row_index))) = heap.pop() { - let (_, rows) = streams[stream_index]; - merged.push(rows[row_index].clone()); - let next_index = row_index + 1; - if let Some(next) = rows.get(next_index) { - heap.push(Reverse((next.timestamp, stream_index, next_index))); - } - } - merged + self.execution_quotes_iter_on_date_for_symbols(date, symbols) + .cloned() + .collect() } pub fn remove_execution_quotes_on_date(&mut self, date: NaiveDate) -> usize { @@ -4030,6 +4062,11 @@ mod tests { ("09:32:00".to_string(), "000001.SZ".to_string()), ] ); + let streamed_keys = run_data + .execution_quotes_iter_on_date_for_symbols(date, None) + .map(|row| (row.timestamp.time().to_string(), row.symbol.clone())) + .collect::>(); + assert_eq!(streamed_keys, keys); assert_eq!(merged[2].last_price, 10.0); let allowed_symbols = BTreeSet::from(["000001.SZ".to_string()]); let filtered = run_data.execution_quotes_on_date_for_symbols(date, Some(&allowed_symbols)); diff --git a/crates/fidc-core/src/engine.rs b/crates/fidc-core/src/engine.rs index 935df29..3ee1506 100644 --- a/crates/fidc-core/src/engine.rs +++ b/crates/fidc-core/src/engine.rs @@ -2723,10 +2723,16 @@ where &mut minute_symbols, )?; } - let minute_quotes = self.data.execution_quotes_on_date_for_symbols( - execution_date, - (!self.subscriptions.is_empty()).then_some(&self.subscriptions), - ); + // Keep the iterator attached to an O(1) DataSet clone. This + // preserves the immutable quote snapshot for the day while + // allowing lazy quote loads and broker state updates on self. + let quote_data = self.data.clone(); + let mut minute_quotes = quote_data + .execution_quotes_iter_on_date_for_symbols( + execution_date, + (!self.subscriptions.is_empty()).then_some(&self.subscriptions), + ) + .peekable(); let requires_minute_callbacks = self.strategy.requires_minute_callbacks(); let has_minute_process_listeners = self.process_event_bus.has_listeners_for(&[ ProcessEventKind::PreMinute, @@ -2741,17 +2747,22 @@ where .filter(|rule| rule.stage == ScheduleStage::Minute) .filter_map(|rule| rule.time_rule.as_ref()?.minute_of_day()) .collect::>(); - let mut minute_cursor = 0usize; - while minute_cursor < minute_quotes.len() { - let minute_timestamp = minute_quotes[minute_cursor].timestamp; + let mut minute_group = Vec::new(); + while let Some(first_quote) = minute_quotes.next() { + let minute_timestamp = first_quote.timestamp; let minute_time = minute_timestamp.time(); - let mut minute_end = minute_cursor + 1; - while minute_end < minute_quotes.len() - && minute_quotes[minute_end].timestamp == minute_timestamp + minute_group.clear(); + minute_group.push(first_quote); + while minute_quotes + .peek() + .is_some_and(|quote| quote.timestamp == minute_timestamp) { - minute_end += 1; + minute_group.push( + minute_quotes + .next() + .expect("peeked minute quote must be available"), + ); } - let minute_group = &minute_quotes[minute_cursor..minute_end]; let schedule_candidate = minute_schedule_all_times || minute_schedule_minutes .contains(&(minute_time.hour() * 60 + minute_time.minute())); @@ -2760,7 +2771,6 @@ where && !schedule_candidate && !self.has_open_orders() { - minute_cursor = minute_end; continue; } let minute_open_orders = self.open_order_views(); @@ -2802,7 +2812,7 @@ where result.fills.as_slice(), )?; if requires_minute_callbacks { - for quote in minute_group { + for "e in &minute_group { minute_decision.merge_from(self.strategy.on_minute( &StrategyContext { execution_date, @@ -2907,8 +2917,10 @@ where ProcessEventKind::PostMinute, format!("minute:{minute_timestamp}:post"), )?; - minute_cursor = minute_end; } + drop(minute_group); + drop(minute_quotes); + drop(quote_data); self.data.remove_execution_quotes_on_date(execution_date); } diff --git a/crates/fidc-core/tests/intraday_history_performance.rs b/crates/fidc-core/tests/intraday_history_performance.rs index 6fd6208..9d4722a 100644 --- a/crates/fidc-core/tests/intraday_history_performance.rs +++ b/crates/fidc-core/tests/intraday_history_performance.rs @@ -177,3 +177,46 @@ fn benchmark_bounded_intraday_history() { elapsed.as_secs_f64(), ); } + +#[test] +#[ignore = "manual release-mode quote-stream benchmark"] +fn benchmark_borrowed_execution_quote_stream() { + let (data, dates) = dataset(250, 240); + let date = *dates.last().expect("last date"); + let symbols = std::collections::BTreeSet::from([SYMBOL.to_string()]); + + for _ in 0..5 { + black_box(data.execution_quotes_on_date_for_symbols(date, Some(&symbols))); + black_box( + data.execution_quotes_iter_on_date_for_symbols(date, Some(&symbols)) + .count(), + ); + } + + let materialized_started = Instant::now(); + let mut materialized_checksum = 0_i64; + for _ in 0..5_000 { + let rows = data.execution_quotes_on_date_for_symbols(date, Some(&symbols)); + materialized_checksum += rows + .iter() + .map(|quote| quote.timestamp.and_utc().timestamp()) + .sum::(); + black_box(rows); + } + let materialized_seconds = materialized_started.elapsed().as_secs_f64(); + + let streamed_started = Instant::now(); + let mut streamed_checksum = 0_i64; + for _ in 0..5_000 { + let count = data + .execution_quotes_iter_on_date_for_symbols(date, Some(&symbols)) + .map(|quote| quote.timestamp.and_utc().timestamp()) + .sum::(); + streamed_checksum += count; + black_box(count); + } + let streamed_seconds = streamed_started.elapsed().as_secs_f64(); + eprintln!( + "quote_stream_benchmark iterations=5000 rows_per_day=240 materialized_seconds={materialized_seconds:.6} streamed_seconds={streamed_seconds:.6} materialized_checksum={materialized_checksum} streamed_checksum={streamed_checksum}" + ); +}