From 757b5665ca01bceb2bf8afe1ae2fe2dfc9c52c5c Mon Sep 17 00:00:00 2001 From: boris Date: Mon, 7 Sep 2026 11:22:20 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BB=A5=E8=BF=90=E8=A1=8C=E8=A6=86=E7=9B=96?= =?UTF-8?q?=E5=B1=82=E9=9A=94=E7=A6=BB=E8=A1=A5=E5=85=85=E8=A1=8C=E6=83=85?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- crates/fidc-core/src/data.rs | 349 +++++++++++++++++++++++++++++++---- 1 file changed, 308 insertions(+), 41 deletions(-) diff --git a/crates/fidc-core/src/data.rs b/crates/fidc-core/src/data.rs index 167cd77..9e077a2 100644 --- a/crates/fidc-core/src/data.rs +++ b/crates/fidc-core/src/data.rs @@ -340,9 +340,12 @@ pub struct ExecutionQuoteIterator<'a> { heap: BinaryHeap>, } +type ExecutionQuotesBySymbol = HashMap>; +type ExecutionQuotesByDate = HashMap; + impl<'a> ExecutionQuoteIterator<'a> { fn new( - rows_by_symbol: Option<&'a HashMap>>, + rows_by_symbol: Option<&'a ExecutionQuotesBySymbol>, symbols: Option<&BTreeSet>, ) -> Self { let mut streams = rows_by_symbol @@ -365,6 +368,34 @@ impl<'a> ExecutionQuoteIterator<'a> { } Self { streams, heap } } + + fn new_with_overlay( + base_rows_by_symbol: Option<&'a ExecutionQuotesBySymbol>, + overlay_rows_by_symbol: &'a ExecutionQuotesBySymbol, + symbols: Option<&BTreeSet>, + ) -> Self { + let mut streams = base_rows_by_symbol + .into_iter() + .flat_map(|rows_by_symbol| rows_by_symbol.iter()) + .filter(|(symbol, _)| !overlay_rows_by_symbol.contains_key(*symbol)) + .chain(overlay_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> { @@ -1314,7 +1345,8 @@ pub struct DataSet { candidate_symbol_ids_by_date: Arc>>, candidate_row_positions_by_date: Arc>, corporate_actions_by_date: Arc>>, - execution_quotes_by_date: Arc>>>, + execution_quotes_by_date: Arc, + execution_quote_overlays_by_date: Arc, execution_quote_dates: Arc>, order_book_depth_index: Arc>>, benchmark_by_date: Arc>, @@ -1850,6 +1882,7 @@ impl DataSet { candidate_row_positions_by_date: Arc::new(candidate_row_positions_by_date), corporate_actions_by_date: Arc::new(corporate_actions_by_date), execution_quotes_by_date: Arc::new(execution_quotes_by_date), + execution_quote_overlays_by_date: Arc::new(HashMap::new()), 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), @@ -2180,23 +2213,46 @@ impl DataSet { } pub fn execution_quotes_on(&self, date: NaiveDate, symbol: &str) -> &[IntradayExecutionQuote] { - self.execution_quotes_by_date + if self.execution_quote_overlays_by_date.is_empty() { + return self + .execution_quotes_by_date + .get(&date) + .and_then(|rows_by_symbol| rows_by_symbol.get(symbol)) + .map(Vec::as_slice) + .unwrap_or(&[]); + } + self.execution_quote_overlays_by_date .get(&date) .and_then(|rows_by_symbol| rows_by_symbol.get(symbol)) + .or_else(|| { + self.execution_quotes_by_date + .get(&date) + .and_then(|rows_by_symbol| rows_by_symbol.get(symbol)) + }) .map(Vec::as_slice) .unwrap_or(&[]) } pub fn has_execution_quotes_on_date(&self, date: NaiveDate) -> bool { - self.execution_quotes_by_date + if self.execution_quote_overlays_by_date.is_empty() { + return self + .execution_quotes_by_date + .get(&date) + .is_some_and(|rows_by_symbol| !rows_by_symbol.is_empty()); + } + self.execution_quote_overlays_by_date .get(&date) - .map(|rows_by_symbol| !rows_by_symbol.is_empty()) - .unwrap_or(false) + .is_some_and(|rows_by_symbol| !rows_by_symbol.is_empty()) + || self + .execution_quotes_by_date + .get(&date) + .is_some_and(|rows_by_symbol| !rows_by_symbol.is_empty()) } pub fn execution_quote_key_set(&self) -> HashSet<(NaiveDate, String)> { self.execution_quotes_by_date .iter() + .chain(self.execution_quote_overlays_by_date.iter()) .flat_map(|(date, rows_by_symbol)| { rows_by_symbol .keys() @@ -2206,15 +2262,51 @@ impl DataSet { } pub fn execution_quote_count(&self) -> usize { - self.execution_quotes_by_date + let mut count = self + .execution_quotes_by_date .values() .flat_map(|rows_by_symbol| rows_by_symbol.values()) .map(Vec::len) - .sum() + .sum::(); + for (date, rows_by_symbol) in self.execution_quote_overlays_by_date.iter() { + for (symbol, rows) in rows_by_symbol { + count = count + .saturating_sub( + self.execution_quotes_by_date + .get(date) + .and_then(|base| base.get(symbol)) + .map(Vec::len) + .unwrap_or(0), + ) + .saturating_add(rows.len()); + } + } + count + } + + fn execution_quote_count_on_date(&self, date: NaiveDate) -> usize { + let base = self.execution_quotes_by_date.get(&date); + let mut count = base + .into_iter() + .flat_map(|rows_by_symbol| rows_by_symbol.values()) + .map(Vec::len) + .sum::(); + if let Some(overlay) = self.execution_quote_overlays_by_date.get(&date) { + for (symbol, rows) in overlay { + count = count + .saturating_sub( + base.and_then(|rows_by_symbol| rows_by_symbol.get(symbol)) + .map(Vec::len) + .unwrap_or(0), + ) + .saturating_add(rows.len()); + } + } + count } pub fn add_execution_quotes(&mut self, quotes: Vec) -> usize { - let mut grouped = HashMap::>>::new(); + let mut grouped = ExecutionQuotesByDate::new(); for quote in quotes { grouped .entry(quote.date) @@ -2225,17 +2317,30 @@ impl DataSet { } let mut added = 0usize; let mut new_dates = Vec::new(); - let execution_quotes_by_date = Arc::make_mut(&mut self.execution_quotes_by_date); + let base_quotes = Arc::clone(&self.execution_quotes_by_date); + let execution_quote_overlays_by_date = + Arc::make_mut(&mut self.execution_quote_overlays_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(); + let date_is_new = !base_quotes.contains_key(&date) + && !execution_quote_overlays_by_date.contains_key(&date); + let target_by_symbol = execution_quote_overlays_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); - let target = target_by_symbol.entry(symbol).or_default(); + let target = match target_by_symbol.entry(symbol) { + std::collections::hash_map::Entry::Occupied(entry) => entry.into_mut(), + std::collections::hash_map::Entry::Vacant(entry) => { + let base_rows = base_quotes + .get(&date) + .and_then(|rows_by_symbol| rows_by_symbol.get(entry.key())) + .cloned() + .unwrap_or_default(); + entry.insert(base_rows) + } + }; if target.is_empty() { added = added.saturating_add(incoming.len()); *target = incoming; @@ -2300,6 +2405,13 @@ impl DataSet { date: NaiveDate, symbols: Option<&BTreeSet>, ) -> ExecutionQuoteIterator<'_> { + if let Some(overlay) = self.execution_quote_overlays_by_date.get(&date) { + return ExecutionQuoteIterator::new_with_overlay( + self.execution_quotes_by_date.get(&date), + overlay, + symbols, + ); + } ExecutionQuoteIterator::new(self.execution_quotes_by_date.get(&date), symbols) } @@ -2314,29 +2426,33 @@ impl DataSet { } pub fn remove_execution_quotes_on_date(&mut self, date: NaiveDate) -> usize { - let removed = Arc::make_mut(&mut self.execution_quotes_by_date).remove(&date); - let Some(rows_by_symbol) = removed else { - return 0; - }; + let row_count = self.execution_quote_count_on_date(date); + Arc::make_mut(&mut self.execution_quote_overlays_by_date).remove(&date); + Arc::make_mut(&mut self.execution_quotes_by_date).remove(&date); 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() + row_count } pub fn release_execution_quotes_on_date(&mut self, date: NaiveDate) -> usize { - let row_count = self - .execution_quotes_by_date - .get(&date) - .map(|rows_by_symbol| rows_by_symbol.values().map(Vec::len).sum()) - .unwrap_or(0); + let row_count = self.execution_quote_count_on_date(date); + if self.execution_quote_overlays_by_date.contains_key(&date) { + Arc::make_mut(&mut self.execution_quote_overlays_by_date).remove(&date); + } // Run data shares this immutable map with the prepared-data cache. Arc::make_mut here // would clone every date just to remove one entry and would not release the cached base. - if row_count == 0 || Arc::strong_count(&self.execution_quotes_by_date) > 1 { - return row_count; + if Arc::strong_count(&self.execution_quotes_by_date) == 1 { + Arc::make_mut(&mut self.execution_quotes_by_date).remove(&date); } - self.remove_execution_quotes_on_date(date) + if !self.execution_quotes_by_date.contains_key(&date) + && !self.execution_quote_overlays_by_date.contains_key(&date) + && let Ok(index) = self.execution_quote_dates.binary_search(&date) + { + Arc::make_mut(&mut self.execution_quote_dates).remove(index); + } + row_count } pub fn snapshot_components(&self) -> DataSetSnapshotComponents { @@ -2364,12 +2480,31 @@ impl DataSet { .values() .flat_map(|rows| rows.iter().cloned()) .collect::>(); - let execution_quotes = self - .execution_quotes_by_date - .values() - .flat_map(|rows_by_symbol| rows_by_symbol.values()) - .flat_map(|rows| rows.iter().cloned()) - .collect::>(); + let execution_quotes = if self.execution_quote_overlays_by_date.is_empty() { + self.execution_quotes_by_date + .values() + .flat_map(|rows_by_symbol| rows_by_symbol.values()) + .flat_map(|rows| rows.iter().cloned()) + .collect::>() + } else { + let mut quotes = Vec::with_capacity(self.execution_quote_count()); + for (date, rows_by_symbol) in self.execution_quotes_by_date.iter() { + let overlay = self.execution_quote_overlays_by_date.get(date); + for (symbol, rows) in rows_by_symbol { + if overlay.is_some_and(|overlay| overlay.contains_key(symbol)) { + continue; + } + quotes.extend(rows.iter().cloned()); + } + } + quotes.extend( + self.execution_quote_overlays_by_date + .values() + .flat_map(|rows_by_symbol| rows_by_symbol.values()) + .flat_map(|rows| rows.iter().cloned()), + ); + quotes + }; DataSetSnapshotComponents { instruments, @@ -2490,6 +2625,53 @@ impl DataSet { if bar_count == 0 { return Vec::new(); } + if self.execution_quote_overlays_by_date.is_empty() { + return self.history_intraday_quotes_from_base( + date, + active_datetime, + symbol, + bar_count, + include_now, + ); + } + 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_quote_overlays_by_date + .get(quote_date) + .and_then(|rows_by_symbol| rows_by_symbol.get(symbol)) + .or_else(|| { + 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 + } + + fn history_intraday_quotes_from_base( + &self, + date: NaiveDate, + active_datetime: Option, + symbol: &str, + bar_count: usize, + include_now: bool, + ) -> Vec { let end = self .execution_quote_dates .partition_point(|quote_date| *quote_date <= date); @@ -3005,14 +3187,22 @@ impl DataSet { .map(daily_market_price_bar) .collect(), Some("1m") => { - let mut bars = self - .execution_quotes_by_date - .iter() - .filter(|(date, _)| **date >= start && **date <= end) - .filter_map(|(_, rows_by_symbol)| rows_by_symbol.get(symbol)) - .flat_map(|rows| rows.iter()) - .map(intraday_quote_price_bar) - .collect::>(); + let mut bars = if self.execution_quote_overlays_by_date.is_empty() { + self.execution_quotes_by_date + .iter() + .filter(|(date, _)| **date >= start && **date <= end) + .filter_map(|(_, rows_by_symbol)| rows_by_symbol.get(symbol)) + .flat_map(|rows| rows.iter()) + .map(intraday_quote_price_bar) + .collect::>() + } else { + self.execution_quote_dates + .iter() + .filter(|date| **date >= start && **date <= end) + .flat_map(|date| self.execution_quotes_on(*date, symbol).iter()) + .map(intraday_quote_price_bar) + .collect::>() + }; bars.sort_by(|left, right| { left.date .cmp(&right.date) @@ -4817,6 +5007,10 @@ mod tests { &data.execution_quotes_by_date, &run_data.execution_quotes_by_date )); + assert!(Arc::ptr_eq( + &data.execution_quote_overlays_by_date, + &run_data.execution_quote_overlays_by_date + )); assert!(Arc::ptr_eq( &data.execution_quote_dates, &run_data.execution_quote_dates @@ -4839,10 +5033,14 @@ mod tests { assert_eq!(data.execution_quote_count(), 0); assert_eq!(run_data.execution_quote_count(), 1); - assert!(!Arc::ptr_eq( + assert!(Arc::ptr_eq( &data.execution_quotes_by_date, &run_data.execution_quotes_by_date )); + assert!(!Arc::ptr_eq( + &data.execution_quote_overlays_by_date, + &run_data.execution_quote_overlays_by_date + )); assert!(!Arc::ptr_eq( &data.execution_quote_dates, &run_data.execution_quote_dates @@ -5719,6 +5917,75 @@ mod tests { assert_eq!(run_data.execution_quote_count(), 0); } + #[test] + fn execution_quote_overlay_preserves_base_precedence_and_merged_iteration() { + let date = NaiveDate::parse_from_str("2025-01-02", "%Y-%m-%d").unwrap(); + let quote = |symbol: &str, time: &str, last_price: f64| IntradayExecutionQuote { + date, + timestamp: NaiveDateTime::parse_from_str( + &format!("2025-01-02 {time}"), + "%Y-%m-%d %H:%M:%S", + ) + .unwrap(), + symbol: symbol.to_string(), + last_price, + bid1: last_price, + ask1: last_price, + bid1_volume: 10_000, + ask1_volume: 10_000, + volume_delta: 10_000, + amount_delta: last_price * 10_000.0, + trading_phase: Some("continuous".to_string()), + }; + let data = DataSet::from_components_with_actions_and_quotes( + Vec::new(), + Vec::new(), + Vec::new(), + Vec::new(), + vec![benchmark_row("2025-01-02", 12.0)], + Vec::new(), + vec![ + quote("000001.SZ", "09:30:00", 10.0), + quote("000001.SZ", "09:31:00", 10.1), + quote("000002.SZ", "09:30:00", 20.0), + ], + ) + .unwrap(); + let mut run_data = data.clone(); + + assert_eq!( + run_data.add_execution_quotes(vec![ + quote("000001.SZ", "09:31:00", 99.0), + quote("000001.SZ", "09:32:00", 10.2), + ]), + 1 + ); + let symbol_rows = run_data.execution_quotes_on(date, "000001.SZ"); + assert_eq!(symbol_rows.len(), 3); + assert_eq!(symbol_rows[1].last_price, 10.1); + assert_eq!(data.execution_quotes_on(date, "000001.SZ").len(), 2); + + let merged = run_data + .execution_quotes_iter_on_date_for_symbols(date, None) + .map(|row| (row.timestamp.time().to_string(), row.symbol.clone())) + .collect::>(); + assert_eq!( + merged, + vec![ + ("09:30:00".to_string(), "000001.SZ".to_string()), + ("09:30:00".to_string(), "000002.SZ".to_string()), + ("09:31:00".to_string(), "000001.SZ".to_string()), + ("09:32:00".to_string(), "000001.SZ".to_string()), + ] + ); + assert_eq!(run_data.execution_quote_count(), 4); + assert_eq!(run_data.snapshot_components().execution_quotes.len(), 4); + + assert_eq!(run_data.release_execution_quotes_on_date(date), 4); + assert_eq!(run_data.execution_quotes_on(date, "000001.SZ").len(), 2); + assert_eq!(run_data.execution_quote_count(), 3); + } + #[test] fn shared_execution_quote_release_does_not_clone_the_base_map() { let date = NaiveDate::parse_from_str("2025-01-02", "%Y-%m-%d").unwrap();