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

This reverts commit d0639558b3.

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)
@@ -1174,6 +1225,7 @@ pub struct DataSet {
     candidate_symbol_ids_by_date: Arc<BTreeMap<NaiveDate, Vec<u32>>>,
     corporate_actions_by_date: Arc<BTreeMap<NaiveDate, Vec<CorporateAction>>>,
     execution_quotes_by_date: Arc<HashMap<NaiveDate, HashMap<String, Vec<IntradayExecutionQuote>>>>,
+    execution_quote_dates: Arc<Vec<NaiveDate>>,
     order_book_depth_index: Arc<HashMap<(NaiveDate, String), Vec<IntradayOrderBookDepthLevel>>>,
     benchmark_by_date: Arc<BTreeMap<NaiveDate, BenchmarkSnapshot>>,
     market_series_by_symbol: Arc<AHashMap<String, Arc<SymbolPriceSeries>>>,
@@ -1441,6 +1493,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::<Vec<_>>();
+        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 +1518,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 +1703,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 +1748,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
     }
 
@@ -1706,48 +1774,37 @@ 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 {
-            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::<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
+        self.execution_quotes_iter_on_date_for_symbols(date, symbols)
+            .cloned()
+            .collect()
     }
 
     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 +1958,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::<Vec<_>>();
-        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<NaiveDate> {
@@ -3396,13 +3466,6 @@ fn normalize_history_frequency(frequency: &str) -> Option<String> {
     }
 }
 
-fn take_last<T>(mut rows: Vec<T>, count: usize) -> Vec<T> {
-    if rows.len() <= count {
-        return rows;
-    }
-    rows.split_off(rows.len() - count)
-}
-
 fn group_by_date<T, F>(rows: Vec<T>, mut date_of: F) -> BTreeMap<NaiveDate, Vec<T>>
 where
     F: FnMut(&T) -> NaiveDate,
@@ -3864,6 +3927,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 +3953,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]
@@ -3991,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(
-                    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::<BTreeSet<_>>();
-                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 &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/src/platform_expr_strategy.rs

@@ -9,8 +9,8 @@ use rhai::{AST, Dynamic, Engine, Map, Scope};
 use crate::broker::{MatchingType, RebalanceCashMode, SlippageModel};
 use crate::cost::ChinaAShareCostModel;
 use crate::data::{
-    DailyMarketSnapshot, EligibleUniverseSnapshot, NumericFactorMap, PriceField,
-    decision_free_float_cap_bn, decision_market_cap_bn,
+    DailyMarketSnapshot, EligibleUniverseSnapshot, PriceField, decision_free_float_cap_bn,
+    decision_market_cap_bn,
 };
 use crate::engine::BacktestError;
 use crate::events::OrderSide;
@@ -424,11 +424,6 @@ pub struct PlatformExprStrategyConfig {
     pub matching_type: MatchingType,
     pub quote_quantity_limit: bool,
     pub current_day_precomputed_factors: bool,
-    /// Use audited Strategy Factory Source Lake rolling fields when the
-    /// runtime intentionally loads only the latest decision rows.  This is
-    /// opt-in; ordinary backtests always recompute rolling values from the
-    /// canonical market series.
-    pub prefer_precomputed_rolling_factors: bool,
     pub intraday_execution_time: Option<NaiveTime>,
     pub delayed_limit_open_exit_enabled: bool,
     pub delayed_limit_open_exit_time: Option<NaiveTime>,
@@ -500,7 +495,6 @@ impl PlatformExprStrategyConfig {
             matching_type: MatchingType::CurrentBarClose,
             quote_quantity_limit: true,
             current_day_precomputed_factors: false,
-            prefer_precomputed_rolling_factors: false,
             intraday_execution_time: None,
             delayed_limit_open_exit_enabled: false,
             delayed_limit_open_exit_time: None,
@@ -808,42 +802,6 @@ fn framework_stock_rolling_factor_requirement(key: &str) -> Option<(&'static str
         .map(|window| (field, window))
 }
 
-fn precomputed_stock_rolling_mean<F>(get: F, field: &str, lookback: usize) -> Option<f64>
-where
-    F: Fn(&str) -> Option<f64>,
-{
-    if lookback == 0 {
-        return None;
-    }
-    let value_for = |key: String| get(&key).filter(|value| value.is_finite());
-    match field.trim().to_ascii_lowercase().as_str() {
-        "close" | "prev_close" | "stock_close" | "price" => {
-            value_for(format!("ma{lookback}_prev_close"))
-                .or_else(|| value_for(format!("ma{lookback}")))
-        }
-        "volume" | "stock_volume" => value_for(format!("avg_volume{lookback}"))
-            .or_else(|| value_for(format!("vma{lookback}"))),
-        _ => None,
-    }
-}
-
-fn precomputed_stock_current_rolling_mean<F>(get: F, field: &str, lookback: usize) -> Option<f64>
-where
-    F: Fn(&str) -> Option<f64>,
-{
-    if lookback == 0 {
-        return None;
-    }
-    let value_for = |key: String| get(&key).filter(|value| value.is_finite());
-    match field.trim().to_ascii_lowercase().as_str() {
-        "close" | "prev_close" | "stock_close" | "price" => {
-            value_for(format!("ma{lookback}_current_back_adjusted_close"))
-        }
-        "volume" | "stock_volume" => value_for(format!("avg_volume{lookback}_current")),
-        _ => None,
-    }
-}
-
 struct SelectiveExpressionScope<'a> {
     inner: Scope<'static>,
     required_identifiers: &'a AHashSet<String>,
@@ -3817,24 +3775,13 @@ impl PlatformExprStrategy {
         date: NaiveDate,
         symbol_id: u32,
         symbol: &str,
-        extra_factors: Option<&NumericFactorMap>,
         field: &str,
         lookback: usize,
     ) -> Option<f64> {
-        let computed = || {
-            ctx.data
-                .market_decision_numeric_moving_average_by_symbol_id(
-                    date, symbol_id, symbol, field, lookback,
-                )
-        };
-        let precomputed = extra_factors.and_then(|factors| {
-            precomputed_stock_rolling_mean(|key| factors.get(key).copied(), field, lookback)
-        });
-        if self.config.prefer_precomputed_rolling_factors {
-            precomputed.or_else(computed)
-        } else {
-            computed().or(precomputed)
-        }
+        ctx.data
+            .market_decision_numeric_moving_average_by_symbol_id(
+                date, symbol_id, symbol, field, lookback,
+            )
     }
 
     fn stock_current_rolling_mean(
@@ -3843,23 +3790,12 @@ impl PlatformExprStrategy {
         date: NaiveDate,
         symbol_id: u32,
         symbol: &str,
-        extra_factors: Option<&NumericFactorMap>,
         field: &str,
         lookback: usize,
     ) -> Option<f64> {
-        let computed = || {
-            ctx.data.market_current_numeric_moving_average_by_symbol_id(
-                date, symbol_id, symbol, field, lookback,
-            )
-        };
-        let precomputed = extra_factors.and_then(|factors| {
-            precomputed_stock_current_rolling_mean(|key| factors.get(key).copied(), field, lookback)
-        });
-        if self.config.prefer_precomputed_rolling_factors {
-            precomputed.or_else(computed)
-        } else {
-            computed().or(precomputed)
-        }
+        ctx.data.market_current_numeric_moving_average_by_symbol_id(
+            date, symbol_id, symbol, field, lookback,
+        )
     }
 
     fn stock_state_at_time(
@@ -3984,16 +3920,8 @@ impl PlatformExprStrategy {
             if !self.stock_rolling_requirements.requires(field, lookback) {
                 return f64::NAN;
             }
-            self.stock_decision_rolling_mean(
-                ctx,
-                date,
-                symbol_id,
-                symbol,
-                Some(&factor.extra_factors),
-                field,
-                lookback,
-            )
-            .unwrap_or(f64::NAN)
+            self.stock_decision_rolling_mean(ctx, date, symbol_id, symbol, field, lookback)
+                .unwrap_or(f64::NAN)
         };
         let stock_ma_short = rolling("close", self.config.stock_short_ma_days);
         let stock_ma_mid = rolling("close", self.config.stock_mid_ma_days);
@@ -6109,9 +6037,6 @@ impl PlatformExprStrategy {
                             day.date,
                             stock.symbol_id,
                             &stock.symbol,
-                            ctx.data
-                                .factor_by_symbol_id(day.date, stock.symbol_id)
-                                .map(|factor| &factor.extra_factors),
                             field,
                             lookback,
                         )
@@ -6767,9 +6692,6 @@ impl PlatformExprStrategy {
                     day.date,
                     stock.symbol_id,
                     &stock.symbol,
-                    ctx.data
-                        .factor_by_symbol_id(day.date, stock.symbol_id)
-                        .map(|factor| &factor.extra_factors),
                     other,
                     lookback,
                 )
@@ -6819,9 +6741,6 @@ impl PlatformExprStrategy {
                     day.date,
                     stock.symbol_id,
                     &stock.symbol,
-                    ctx.data
-                        .factor_by_symbol_id(day.date, stock.symbol_id)
-                        .map(|factor| &factor.extra_factors),
                     other,
                     lookback,
                 )
@@ -12173,7 +12092,6 @@ mod tests {
         PlatformRebalanceSchedule, PlatformScheduleFrequency, PlatformStopTakeReferencePriceMode,
         PlatformTradeAction, PlatformUniverseActionKind, RuntimeHelperResolution,
         SelectionRiskDeferral, StockFilterQuoteUsage, framework_stock_rolling_factor_requirement,
-        precomputed_stock_current_rolling_mean, precomputed_stock_rolling_mean,
     };
     use crate::{
         AlgoOrderStyle, BenchmarkSnapshot, CandidateEligibility, CorporateAction,
@@ -12241,39 +12159,6 @@ mod tests {
         assert_eq!(framework_stock_rolling_factor_requirement("alpha001"), None);
     }
 
-    #[test]
-    fn source_lake_precomputed_rolling_aliases_use_only_canonical_fields() {
-        let factors = BTreeMap::from([
-            ("ma5_prev_close".to_string(), 11.0),
-            ("ma5".to_string(), 12.0),
-            ("ma5_current_close".to_string(), 99.0),
-            ("ma5_current_back_adjusted_close".to_string(), 13.0),
-            ("avg_volume5".to_string(), 100.0),
-            ("avg_volume5_current".to_string(), 120.0),
-        ]);
-        let get = |key: &str| factors.get(key).copied();
-        assert_eq!(precomputed_stock_rolling_mean(get, "close", 5), Some(11.0));
-        assert_eq!(
-            precomputed_stock_current_rolling_mean(get, "close", 5),
-            Some(13.0)
-        );
-        assert_eq!(
-            precomputed_stock_rolling_mean(get, "volume", 5),
-            Some(100.0)
-        );
-        assert_eq!(
-            precomputed_stock_current_rolling_mean(get, "volume", 5),
-            Some(120.0)
-        );
-        // A source row that only has a raw current-close alias must not be
-        // mistaken for the adjusted rolling value.
-        let raw_only = BTreeMap::from([("ma5_current_close".to_string(), 99.0)]);
-        assert_eq!(
-            precomputed_stock_current_rolling_mean(|key| raw_only.get(key).copied(), "close", 5,),
-            None
-        );
-    }
-
     #[test]
     fn typed_runtime_numbers_preserve_legacy_rhai_formatting() {
         let RuntimeHelperResolution::Number(value) =

crates/fidc-core/tests/intraday_history_performance.rs

@@ -0,0 +1,222 @@
+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<NaiveDate>) {
+    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::<Vec<_>>();
+    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::<Vec<_>>();
+    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::<Vec<_>>();
+    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<_>>(),
+        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<_>>(),
+        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(),
+    );
+}
+
+#[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}"
+    );
+}