zsb/fidc-backtest-engine
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

以运行覆盖层隔离补充行情

Browse Source
This commit is contained in:
boris
2026-09-07 11:22:20 +08:00
parent c280bbc1c3
commit 757b5665ca
1 changed files with 308 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/fidc-core/src/data.rs
+308 -41
View File
@@ -340,9 +340,12 @@ pub struct ExecutionQuoteIterator<'a> {
heap: BinaryHeap<Reverse<(NaiveDateTime, usize, usize)>>,
}
type ExecutionQuotesBySymbol = HashMap<String, Vec<IntradayExecutionQuote>>;
type ExecutionQuotesByDate = HashMap<NaiveDate, ExecutionQuotesBySymbol>;
impl<'a> ExecutionQuoteIterator<'a> {
fn new(
rows_by_symbol: Option<&'a HashMap<String, Vec<IntradayExecutionQuote>>>,
rows_by_symbol: Option<&'a ExecutionQuotesBySymbol>,
symbols: Option<&BTreeSet<String>>,
) -> Self {
let mut streams = rows_by_symbol
@@ -365,6 +368,34 @@ 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<String>>,
) -> 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::<Vec<_>>();
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> {
@@ -1314,7 +1345,8 @@ pub struct DataSet {
candidate_symbol_ids_by_date: Arc<BTreeMap<NaiveDate, Vec<u32>>>,
candidate_row_positions_by_date: Arc<Option<DenseRowPositionIndex>>,
corporate_actions_by_date: Arc<BTreeMap<NaiveDate, Vec<CorporateAction>>>,
execution_quotes_by_date: Arc<HashMap<NaiveDate, HashMap<String, Vec<IntradayExecutionQuote>>>>,
execution_quotes_by_date: Arc<ExecutionQuotesByDate>,
execution_quote_overlays_by_date: Arc<ExecutionQuotesByDate>,
execution_quote_dates: Arc<Vec<NaiveDate>>,
order_book_depth_index: Arc<HashMap<(NaiveDate, String), Vec<IntradayOrderBookDepthLevel>>>,
benchmark_by_date: Arc<BTreeMap<NaiveDate, BenchmarkSnapshot>>,
@@ -1850,6 +1882,7 @@ 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),
@@ -2180,23 +2213,46 @@ impl DataSet {
}
pub fn execution_quotes_on(&self, date: NaiveDate, symbol: &str) -> &[IntradayExecutionQuote] {
self.execution_quotes_by_date
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
.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 {
self.execution_quotes_by_date
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
.get(&date)
.map(|rows_by_symbol| !rows_by_symbol.is_empty())
.unwrap_or(false)
.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())
}
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()
@@ -2206,15 +2262,51 @@ impl DataSet {
}
pub fn execution_quote_count(&self) -> usize {
self.execution_quotes_by_date
let mut count = self
.execution_quotes_by_date
.values()
.flat_map(|rows_by_symbol| rows_by_symbol.values())
.map(Vec::len)
.sum()
.sum::<usize>();
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::<usize>();
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
}
pub fn add_execution_quotes(&mut self, quotes: Vec<IntradayExecutionQuote>) -> usize {
let mut grouped = HashMap::<NaiveDate, HashMap<String, Vec<IntradayExecutionQuote>>>::new();
let mut grouped = ExecutionQuotesByDate::new();
for quote in quotes {
grouped
.entry(quote.date)
@@ -2225,17 +2317,30 @@ impl DataSet {
}
let mut added = 0usize;
let mut new_dates = Vec::new();
let execution_quotes_by_date = Arc::make_mut(&mut self.execution_quotes_by_date);
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);
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 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();
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 = 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)
}
};
if target.is_empty() {
added = added.saturating_add(incoming.len());
*target = incoming;
@@ -2300,6 +2405,13 @@ impl DataSet {
date: NaiveDate,
symbols: Option<&BTreeSet<String>>,
) -> 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)
}
@@ -2314,29 +2426,33 @@ impl DataSet {
}
pub fn remove_execution_quotes_on_date(&mut self, date: NaiveDate) -> usize {
let removed = Arc::make_mut(&mut self.execution_quotes_by_date).remove(&date);
let Some(rows_by_symbol) = removed else {
return 0;
};
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 dates = Arc::make_mut(&mut self.execution_quote_dates);
if let Ok(index) = dates.binary_search(&date) {
dates.remove(index);
}
rows_by_symbol.into_values().map(|rows| rows.len()).sum()
row_count
}
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())
.unwrap_or(0);
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);
}
// 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 row_count == 0 || Arc::strong_count(&self.execution_quotes_by_date) > 1 {
return row_count;
if Arc::strong_count(&self.execution_quotes_by_date) == 1 {
Arc::make_mut(&mut self.execution_quotes_by_date).remove(&date);
}
self.remove_execution_quotes_on_date(date)
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
}
pub fn snapshot_components(&self) -> DataSetSnapshotComponents {
@@ -2364,12 +2480,31 @@ impl DataSet {
.values()
.flat_map(|rows| rows.iter().cloned())
.collect::<Vec<_>>();
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::<Vec<_>>();
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::<Vec<_>>()
} 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
};
DataSetSnapshotComponents {
instruments,
@@ -2490,6 +2625,53 @@ 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<NaiveDateTime>,
symbol: &str,
bar_count: usize,
include_now: bool,
) -> Vec<IntradayExecutionQuote> {
let end = self
.execution_quote_dates
.partition_point(|quote_date| *quote_date <= date);
@@ -3005,14 +3187,22 @@ impl DataSet {
.map(daily_market_price_bar)
.collect(),
Some("1m") => {
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::<Vec<_>>();
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::<Vec<_>>()
} 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::<Vec<_>>()
};
bars.sort_by(|left, right| {
left.date
.cmp(&right.date)
@@ -4817,6 +5007,10 @@ 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
@@ -4839,10 +5033,14 @@ 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
@@ -5719,6 +5917,75 @@ 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::<Vec<_>>();
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();