让分钟行情按流式迭代器处理

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<Reverse<(NaiveDateTime, usize, usize)>>,
+}
+
+impl<'a> ExecutionQuoteIterator<'a> {
+    fn new(
+        rows_by_symbol: Option<&'a HashMap<String, Vec<IntradayExecutionQuote>>>,
+        symbols: Option<&BTreeSet<String>>,
+    ) -> 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::<Vec<_>>();
+        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<Self::Item> {
+        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<String>>,
+    ) -> 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<String>>,
     ) -> Vec<IntradayExecutionQuote> {
-        let Some(rows_by_symbol) = self.execution_quotes_by_date.get(&date) else {
+        self.execution_quotes_iter_on_date_for_symbols(date, symbols)
-            return Vec::new();
+            .cloned()
-        };
+            .collect()
-        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::<Vec<_>>();
-        streams.sort_by_key(|(symbol, _)| *symbol);
-        let total_rows = streams.iter().map(|(_, rows)| rows.len()).sum();
-        let mut heap = BinaryHeap::<Reverse<(NaiveDateTime, usize, usize)>>::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
     }
 
     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::<Vec<_>>();
+        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));

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(
+                // 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::<BTreeSet<_>>();
-                let mut minute_cursor = 0usize;
+                let mut minute_group = Vec::new();
-                while minute_cursor < minute_quotes.len() {
+                while let Some(first_quote) = minute_quotes.next() {
-                    let minute_timestamp = minute_quotes[minute_cursor].timestamp;
+                    let minute_timestamp = first_quote.timestamp;
                     let minute_time = minute_timestamp.time();
-                    let mut minute_end = minute_cursor + 1;
+                    minute_group.clear();
-                    while minute_end < minute_quotes.len()
+                    minute_group.push(first_quote);
-                        && minute_quotes[minute_end].timestamp == minute_timestamp
+                    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 &quote 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);
             }
 

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::<i64>();
+        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::<i64>();
+        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}"
+    );
+}