From 728ed7998d5a5905c631da3633f6a15210f5e771 Mon Sep 17 00:00:00 2001 From: boris Date: Mon, 7 Sep 2026 11:28:33 +0800 Subject: [PATCH] =?UTF-8?q?Revert=20"=E4=BB=A5=E8=BF=90=E8=A1=8C=E8=A6=86?= =?UTF-8?q?=E7=9B=96=E5=B1=82=E9=9A=94=E7=A6=BB=E8=A1=A5=E5=85=85=E8=A1=8C?= =?UTF-8?q?=E6=83=85"?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit This reverts commit 757b5665ca01bceb2bf8afe1ae2fe2dfc9c52c5c. --- crates/fidc-core/src/data.rs | 349 ++++------------------------------- 1 file changed, 41 insertions(+), 308 deletions(-) diff --git a/crates/fidc-core/src/data.rs b/crates/fidc-core/src/data.rs index 9e077a2..167cd77 100644 --- a/crates/fidc-core/src/data.rs +++ b/crates/fidc-core/src/data.rs @@ -340,12 +340,9 @@ pub struct ExecutionQuoteIterator<'a> { heap: BinaryHeap>, } -type ExecutionQuotesBySymbol = HashMap>; -type ExecutionQuotesByDate = HashMap; - impl<'a> ExecutionQuoteIterator<'a> { fn new( - rows_by_symbol: Option<&'a ExecutionQuotesBySymbol>, + rows_by_symbol: Option<&'a HashMap>>, symbols: Option<&BTreeSet>, ) -> Self { let mut streams = rows_by_symbol @@ -368,34 +365,6 @@ 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> { @@ -1345,8 +1314,7 @@ 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_quote_overlays_by_date: Arc, + execution_quotes_by_date: Arc>>>, execution_quote_dates: Arc>, order_book_depth_index: Arc>>, benchmark_by_date: Arc>, @@ -1882,7 +1850,6 @@ 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), @@ -2213,46 +2180,23 @@ impl DataSet { } pub fn execution_quotes_on(&self, date: NaiveDate, symbol: &str) -> &[IntradayExecutionQuote] { - 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 + self.execution_quotes_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 { - 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 + self.execution_quotes_by_date .get(&date) - .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()) + .map(|rows_by_symbol| !rows_by_symbol.is_empty()) + .unwrap_or(false) } 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() @@ -2262,51 +2206,15 @@ impl DataSet { } pub fn execution_quote_count(&self) -> usize { - let mut count = self - .execution_quotes_by_date + self.execution_quotes_by_date .values() .flat_map(|rows_by_symbol| rows_by_symbol.values()) .map(Vec::len) - .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 + .sum() } pub fn add_execution_quotes(&mut self, quotes: Vec) -> usize { - let mut grouped = ExecutionQuotesByDate::new(); + let mut grouped = HashMap::>>::new(); for quote in quotes { grouped .entry(quote.date) @@ -2317,30 +2225,17 @@ impl DataSet { } let mut added = 0usize; let mut new_dates = Vec::new(); - 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); + 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 = !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(); + 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); - 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) - } - }; + let target = target_by_symbol.entry(symbol).or_default(); if target.is_empty() { added = added.saturating_add(incoming.len()); *target = incoming; @@ -2405,13 +2300,6 @@ 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) } @@ -2426,33 +2314,29 @@ impl DataSet { } pub fn remove_execution_quotes_on_date(&mut self, date: NaiveDate) -> usize { - 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 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); } - row_count + rows_by_symbol.into_values().map(|rows| rows.len()).sum() } pub fn release_execution_quotes_on_date(&mut self, date: NaiveDate) -> usize { - 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); - } + 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); // 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 Arc::strong_count(&self.execution_quotes_by_date) == 1 { - Arc::make_mut(&mut self.execution_quotes_by_date).remove(&date); + if row_count == 0 || Arc::strong_count(&self.execution_quotes_by_date) > 1 { + return row_count; } - 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 + self.remove_execution_quotes_on_date(date) } pub fn snapshot_components(&self) -> DataSetSnapshotComponents { @@ -2480,31 +2364,12 @@ impl DataSet { .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 - }; + 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::>(); DataSetSnapshotComponents { instruments, @@ -2625,53 +2490,6 @@ 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); @@ -3187,22 +3005,14 @@ impl DataSet { .map(daily_market_price_bar) .collect(), Some("1m") => { - 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::>() - }; + 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::>(); bars.sort_by(|left, right| { left.date .cmp(&right.date) @@ -5007,10 +4817,6 @@ 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 @@ -5033,14 +4839,10 @@ 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 @@ -5917,75 +5719,6 @@ 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();