线性合并分钟行情窗口

crates/fidc-core/src/data.rs

@@ -1576,32 +1576,53 @@ impl DataSet {
     }
 
     pub fn add_execution_quotes(&mut self, quotes: Vec<IntradayExecutionQuote>) -> usize {
-        let mut added = 0usize;
+        let mut grouped = HashMap::<NaiveDate, HashMap<String, Vec<IntradayExecutionQuote>>>::new();
-        let mut touched = HashSet::<(NaiveDate, String)>::new();
         for quote in quotes {
-            let key = (quote.date, quote.symbol.clone());
+            grouped
-            let rows = self
-                .execution_quotes_by_date
                 .entry(quote.date)
                 .or_default()
                 .entry(quote.symbol.clone())
-                .or_default();
+                .or_default()
-            if rows.iter().any(|existing| {
+                .push(quote);
-                existing.timestamp == quote.timestamp && existing.symbol == quote.symbol
-            }) {
-                continue;
-            }
-            rows.push(quote);
-            touched.insert(key);
-            added += 1;
         }
-        for (date, symbol) in touched {
+        let mut added = 0usize;
-            if let Some(rows) = self
+        for (date, rows_by_symbol) in grouped {
-                .execution_quotes_by_date
+            let target_by_symbol = self.execution_quotes_by_date.entry(date).or_default();
-                .get_mut(&date)
+            for (symbol, mut incoming) in rows_by_symbol {
-                .and_then(|rows_by_symbol| rows_by_symbol.get_mut(&symbol))
+                incoming.sort_by_key(|quote| quote.timestamp);
-            {
+                incoming.dedup_by(|left, right| left.timestamp == right.timestamp);
-                rows.sort_by_key(|quote| quote.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![
+        assert_eq!(
-            quote("000002.SZ", "09:31:00"),
+            run_data.add_execution_quotes(vec![
-            quote("000001.SZ", "09:31:00"),
+                quote("000002.SZ", "09:31:00"),
-            quote("000002.SZ", "09:30:00"),
+                quote("000001.SZ", "09:31:00"),
-            quote("000001.SZ", "09:30: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);
     }