下推分钟报价订阅过滤

This commit is contained in:
boris
2026-08-26 07:58:55 +08:00
parent d3bacffd8b
commit 82604481b6
2 changed files with 31 additions and 10 deletions
+27 -1
View File
@@ -1,6 +1,6 @@
use std::borrow::Cow; use std::borrow::Cow;
use std::cmp::Reverse; use std::cmp::Reverse;
use std::collections::{BTreeMap, BinaryHeap, HashMap, HashSet}; use std::collections::{BTreeMap, BTreeSet, BinaryHeap, HashMap, HashSet};
use std::sync::{Arc, OnceLock}; use std::sync::{Arc, OnceLock};
use ahash::AHashMap; use ahash::AHashMap;
@@ -1692,11 +1692,24 @@ impl DataSet {
} }
pub fn execution_quotes_on_date(&self, date: NaiveDate) -> Vec<IntradayExecutionQuote> { pub fn execution_quotes_on_date(&self, date: NaiveDate) -> Vec<IntradayExecutionQuote> {
self.execution_quotes_on_date_for_symbols(date, None)
}
pub fn execution_quotes_on_date_for_symbols(
&self,
date: NaiveDate,
symbols: Option<&BTreeSet<String>>,
) -> Vec<IntradayExecutionQuote> {
let Some(rows_by_symbol) = self.execution_quotes_by_date.get(&date) else { let Some(rows_by_symbol) = self.execution_quotes_by_date.get(&date) else {
return Vec::new(); return Vec::new();
}; };
let mut streams = rows_by_symbol let mut streams = rows_by_symbol
.iter() .iter()
.filter(|(symbol, _)| {
symbols
.map(|allowed_symbols| allowed_symbols.contains(*symbol))
.unwrap_or(true)
})
.map(|(symbol, rows)| (symbol.as_str(), rows.as_slice())) .map(|(symbol, rows)| (symbol.as_str(), rows.as_slice()))
.collect::<Vec<_>>(); .collect::<Vec<_>>();
streams.sort_by_key(|(symbol, _)| *symbol); streams.sort_by_key(|(symbol, _)| *symbol);
@@ -3929,6 +3942,19 @@ mod tests {
] ]
); );
assert_eq!(merged[2].last_price, 10.0); assert_eq!(merged[2].last_price, 10.0);
let allowed_symbols = BTreeSet::from(["000001.SZ".to_string()]);
let filtered = run_data.execution_quotes_on_date_for_symbols(date, Some(&allowed_symbols));
assert_eq!(
filtered
.iter()
.map(|row| (row.timestamp.time().to_string(), row.symbol.clone()))
.collect::<Vec<_>>(),
vec![
("09:30:00".to_string(), "000001.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.remove_execution_quotes_on_date(date), 5); assert_eq!(run_data.remove_execution_quotes_on_date(date), 5);
assert_eq!(run_data.execution_quote_count(), 0); assert_eq!(run_data.execution_quote_count(), 0);
} }
+4 -9
View File
@@ -2529,15 +2529,10 @@ where
&mut minute_symbols, &mut minute_symbols,
)?; )?;
} }
let filter_by_subscription = !self.subscriptions.is_empty(); let minute_quotes = self.data.execution_quotes_on_date_for_symbols(
let minute_quotes = self execution_date,
.data (!self.subscriptions.is_empty()).then_some(&self.subscriptions),
.execution_quotes_on_date(execution_date) );
.into_iter()
.filter(|quote| {
!filter_by_subscription || self.subscriptions.contains(&quote.symbol)
})
.collect::<Vec<_>>();
let requires_minute_callbacks = self.strategy.requires_minute_callbacks(); let requires_minute_callbacks = self.strategy.requires_minute_callbacks();
let has_minute_process_listeners = self.process_event_bus.has_listeners_for(&[ let has_minute_process_listeners = self.process_event_bus.has_listeners_for(&[
ProcessEventKind::PreMinute, ProcessEventKind::PreMinute,