diff --git a/crates/fidc-core/src/data.rs b/crates/fidc-core/src/data.rs index a6e957a..9534b6c 100644 --- a/crates/fidc-core/src/data.rs +++ b/crates/fidc-core/src/data.rs @@ -1576,32 +1576,53 @@ impl DataSet { } pub fn add_execution_quotes(&mut self, quotes: Vec) -> usize { - let mut added = 0usize; - let mut touched = HashSet::<(NaiveDate, String)>::new(); + let mut grouped = HashMap::>>::new(); for quote in quotes { - let key = (quote.date, quote.symbol.clone()); - let rows = self - .execution_quotes_by_date + grouped .entry(quote.date) .or_default() .entry(quote.symbol.clone()) - .or_default(); - if rows.iter().any(|existing| { - existing.timestamp == quote.timestamp && existing.symbol == quote.symbol - }) { - continue; - } - rows.push(quote); - touched.insert(key); - added += 1; + .or_default() + .push(quote); } - for (date, symbol) in touched { - if let Some(rows) = self - .execution_quotes_by_date - .get_mut(&date) - .and_then(|rows_by_symbol| rows_by_symbol.get_mut(&symbol)) - { - rows.sort_by_key(|quote| quote.timestamp); + let mut added = 0usize; + for (date, rows_by_symbol) in grouped { + let target_by_symbol = self.execution_quotes_by_date.entry(date).or_default(); + 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(); + if target.is_empty() { + added = added.saturating_add(incoming.len()); + *target = incoming; + continue; + } + let mut existing = std::mem::take(target).into_iter().peekable(); + let mut incoming = incoming.into_iter().peekable(); + let mut merged = Vec::with_capacity(existing.len() + incoming.len()); + while let (Some(existing_quote), Some(incoming_quote)) = + (existing.peek(), incoming.peek()) + { + match existing_quote.timestamp.cmp(&incoming_quote.timestamp) { + std::cmp::Ordering::Less => { + merged.push(existing.next().expect("peeked existing quote")); + } + std::cmp::Ordering::Greater => { + merged.push(incoming.next().expect("peeked incoming quote")); + added = added.saturating_add(1); + } + std::cmp::Ordering::Equal => { + merged.push(existing.next().expect("peeked existing quote")); + incoming.next(); + } + } + } + merged.extend(existing); + for quote in incoming { + merged.push(quote); + added = added.saturating_add(1); + } + *target = merged; } } added @@ -3785,12 +3806,25 @@ mod tests { trading_phase: Some("continuous".to_string()), }; let mut run_data = data.clone(); - run_data.add_execution_quotes(vec![ - quote("000002.SZ", "09:31:00"), - quote("000001.SZ", "09:31:00"), - quote("000002.SZ", "09:30:00"), - quote("000001.SZ", "09:30:00"), - ]); + assert_eq!( + run_data.add_execution_quotes(vec![ + quote("000002.SZ", "09:31:00"), + quote("000001.SZ", "09:31:00"), + quote("000002.SZ", "09:30:00"), + quote("000001.SZ", "09:30:00"), + ]), + 4 + ); + let mut conflicting = quote("000001.SZ", "09:31:00"); + conflicting.last_price = 99.0; + assert_eq!( + run_data.add_execution_quotes(vec![ + conflicting, + quote("000001.SZ", "09:32:00"), + quote("000001.SZ", "09:32:00"), + ]), + 1 + ); let merged = run_data.execution_quotes_on_date(date); let keys = merged @@ -3804,9 +3838,11 @@ mod tests { ("09:30:00".to_string(), "000002.SZ".to_string()), ("09:31:00".to_string(), "000001.SZ".to_string()), ("09:31:00".to_string(), "000002.SZ".to_string()), + ("09:32:00".to_string(), "000001.SZ".to_string()), ] ); - assert_eq!(run_data.remove_execution_quotes_on_date(date), 4); + assert_eq!(merged[2].last_price, 10.0); + assert_eq!(run_data.remove_execution_quotes_on_date(date), 5); assert_eq!(run_data.execution_quote_count(), 0); }