优化分钟历史窗口读取并移除滚动双口径

crates/fidc-core/src/data.rs

@@ -1174,6 +1174,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 +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::<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 +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)
+        let removed = Arc::make_mut(&mut self.execution_quotes_by_date).remove(&date);
-            .remove(&date)
+        let Some(rows_by_symbol) = removed else {
-            .map(|rows_by_symbol| rows_by_symbol.into_values().map(|rows| rows.len()).sum())
+            return 0;
-            .unwrap_or_default()
+        };
+        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
+        let end = self
-            .execution_quotes_by_date
+            .execution_quote_dates
-            .values()
+            .partition_point(|quote_date| *quote_date <= date);
-            .filter_map(|rows_by_symbol| rows_by_symbol.get(symbol))
+        let mut quotes = Vec::with_capacity(bar_count);
-            .flat_map(|rows| rows.iter())
+        'dates: for quote_date in self.execution_quote_dates[..end].iter().rev() {
-            .filter(|quote| intraday_quote_visible(quote, date, active_datetime, include_now))
+            let Some(rows) = self
-            .cloned()
+                .execution_quotes_by_date
-            .collect::<Vec<_>>();
+                .get(quote_date)
-        quotes.sort_by_key(|quote| quote.timestamp);
+                .and_then(|rows_by_symbol| rows_by_symbol.get(symbol))
-        take_last(quotes, bar_count)
+            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 +3434,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 +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]

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<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(),
+    );
+}