diff --git a/crates/fidc-core/src/data.rs b/crates/fidc-core/src/data.rs index 167cd77..37d30b0 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 @@ -1314,7 +1317,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_quotes_by_date: Arc, execution_quote_dates: Arc>, order_book_depth_index: Arc>>, benchmark_by_date: Arc>, @@ -2183,7 +2186,7 @@ impl DataSet { self.execution_quotes_by_date .get(&date) .and_then(|rows_by_symbol| rows_by_symbol.get(symbol)) - .map(Vec::as_slice) + .map(|rows| rows.as_slice()) .unwrap_or(&[]) } @@ -2209,7 +2212,7 @@ impl DataSet { self.execution_quotes_by_date .values() .flat_map(|rows_by_symbol| rows_by_symbol.values()) - .map(Vec::len) + .map(|rows| rows.len()) .sum() } @@ -2228,14 +2231,22 @@ impl DataSet { 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(); + let target_by_symbol = Arc::make_mut( + execution_quotes_by_date + .entry(date) + .or_insert_with(|| Arc::new(HashMap::new())), + ); 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 = Arc::make_mut( + target_by_symbol + .entry(symbol) + .or_insert_with(|| Arc::new(Vec::new())), + ); if target.is_empty() { added = added.saturating_add(incoming.len()); *target = incoming; @@ -2300,7 +2311,10 @@ impl DataSet { date: NaiveDate, symbols: Option<&BTreeSet>, ) -> ExecutionQuoteIterator<'_> { - ExecutionQuoteIterator::new(self.execution_quotes_by_date.get(&date), symbols) + ExecutionQuoteIterator::new( + self.execution_quotes_by_date.get(&date).map(Arc::as_ref), + symbols, + ) } pub fn execution_quotes_on_date_for_symbols( @@ -2322,14 +2336,14 @@ impl DataSet { if let Ok(index) = dates.binary_search(&date) { dates.remove(index); } - rows_by_symbol.into_values().map(|rows| rows.len()).sum() + rows_by_symbol.values().map(|rows| rows.len()).sum() } 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()) + .map(|rows_by_symbol| rows_by_symbol.values().map(|rows| rows.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. @@ -4529,7 +4543,7 @@ fn build_futures_params_index( fn build_execution_quote_index( execution_quotes: Vec, -) -> HashMap>> { +) -> ExecutionQuotesByDate { let mut grouped = HashMap::>>::new(); for quote in execution_quotes { grouped @@ -4547,6 +4561,15 @@ fn build_execution_quote_index( } grouped + .into_iter() + .map(|(date, rows_by_symbol)| { + let rows_by_symbol = rows_by_symbol + .into_iter() + .map(|(symbol, rows)| (symbol, Arc::new(rows))) + .collect(); + (date, Arc::new(rows_by_symbol)) + }) + .collect() } fn build_order_book_depth_index( @@ -4849,6 +4872,68 @@ mod tests { )); } + #[test] + fn execution_quote_updates_copy_only_the_modified_date_and_symbol_path() { + let first_date = NaiveDate::parse_from_str("2025-01-02", "%Y-%m-%d").unwrap(); + let second_date = NaiveDate::parse_from_str("2025-01-03", "%Y-%m-%d").unwrap(); + let quote = |date: NaiveDate, symbol: &str, time: &str| IntradayExecutionQuote { + date, + timestamp: NaiveDateTime::parse_from_str( + &format!("{date} {time}"), + "%Y-%m-%d %H:%M:%S", + ) + .unwrap(), + symbol: symbol.to_string(), + last_price: 10.0, + bid1: 9.99, + ask1: 10.01, + bid1_volume: 10_000, + ask1_volume: 10_000, + volume_delta: 10_000, + amount_delta: 100_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), benchmark_row("2025-01-03", 12.1)], + Vec::new(), + vec![ + quote(first_date, "000001.SZ", "10:18:00"), + quote(first_date, "000002.SZ", "10:18:00"), + quote(second_date, "000001.SZ", "10:18:00"), + ], + ) + .unwrap(); + let mut run_data = data.clone(); + + run_data.add_execution_quotes(vec![quote(first_date, "000001.SZ", "10:19:00")]); + + assert!(!Arc::ptr_eq( + &data.execution_quotes_by_date, + &run_data.execution_quotes_by_date + )); + let base_first = data.execution_quotes_by_date.get(&first_date).unwrap(); + let run_first = run_data.execution_quotes_by_date.get(&first_date).unwrap(); + assert!(!Arc::ptr_eq(base_first, run_first)); + assert!(!Arc::ptr_eq( + base_first.get("000001.SZ").unwrap(), + run_first.get("000001.SZ").unwrap() + )); + assert!(Arc::ptr_eq( + base_first.get("000002.SZ").unwrap(), + run_first.get("000002.SZ").unwrap() + )); + assert!(Arc::ptr_eq( + data.execution_quotes_by_date.get(&second_date).unwrap(), + run_data.execution_quotes_by_date.get(&second_date).unwrap() + )); + assert_eq!(data.execution_quotes_on(first_date, "000001.SZ").len(), 1); + assert_eq!(run_data.execution_quotes_on(first_date, "000001.SZ").len(), 2); + } + #[test] fn daily_bundle_constructor_matches_flat_component_constructor() { let dates = [