保持日内覆盖与按股序列一致

This commit is contained in:
boris
2026-09-07 13:25:07 +08:00
parent d5af51c02b
commit 3f6017d98b
+128 -22
View File
@@ -900,6 +900,37 @@ impl SymbolPriceSeries {
} }
} }
fn apply_intraday_market_overlays(
&mut self,
overlays: &[&IntradayMarketSnapshotOverlay],
) -> Result<(), NaiveDate> {
let mut last_price_changed = false;
for overlay in overlays {
let index = self
.dates
.binary_search(&overlay.date)
.map_err(|_| overlay.date)?;
self.timestamps[index] = overlay.timestamp.clone();
if let Some(last_price) = overlay
.last_price
.filter(|value| value.is_finite() && *value > 0.0)
{
self.last_prices[index] = last_price;
last_price_changed = true;
}
self.bid1s[index] = overlay.bid1;
self.ask1s[index] = overlay.ask1;
self.minute_volumes[index] = overlay.minute_volume;
self.bid1_volumes[index] = overlay.bid1_volume;
self.ask1_volumes[index] = overlay.ask1_volume;
self.trading_phases[index] = overlay.trading_phase.clone();
}
if last_price_changed {
self.last_prefix = prefix_sums(&self.last_prices);
}
Ok(())
}
fn moving_average(&self, date: NaiveDate, lookback: usize, field: PriceField) -> Option<f64> { fn moving_average(&self, date: NaiveDate, lookback: usize, field: PriceField) -> Option<f64> {
if lookback == 0 { if lookback == 0 {
return None; return None;
@@ -2238,7 +2269,7 @@ impl DataSet {
.sum() .sum()
} }
/// Applies sparse intraday fields without rebuilding daily series or indexes. /// Applies sparse intraday fields without rebuilding unaffected daily series or indexes.
/// ///
/// The daily market storage must still be uniquely owned. This is deliberate: /// The daily market storage must still be uniquely owned. This is deliberate:
/// silently using `Arc::make_mut` here would deep-copy the full market panel /// silently using `Arc::make_mut` here would deep-copy the full market panel
@@ -2251,8 +2282,25 @@ impl DataSet {
return Ok(0); return Ok(0);
} }
for (component, strong_count) in [
("daily market panel", Arc::strong_count(&self.market_by_date)),
(
"market series by symbol",
Arc::strong_count(&self.market_series_by_symbol),
),
(
"market series by symbol id",
Arc::strong_count(&self.market_series_by_symbol_id),
),
] {
if strong_count != 1 {
return Err(DataSetError::SharedComponentMutation { component });
}
}
let mut resolved = Vec::with_capacity(overlays.len()); let mut resolved = Vec::with_capacity(overlays.len());
let mut seen = HashSet::<(NaiveDate, u32)>::with_capacity(overlays.len()); let mut seen = HashSet::<(NaiveDate, u32)>::with_capacity(overlays.len());
let mut overlay_indexes_by_symbol_id = BTreeMap::<u32, Vec<usize>>::new();
for overlay in overlays { for overlay in overlays {
let symbol_id = self let symbol_id = self
.symbol_id_by_code .symbol_id_by_code
@@ -2270,33 +2318,71 @@ impl DataSet {
}); });
} }
let row_position = self let row_position = self
.market_symbol_ids_by_date .market_row_positions_by_date
.get(&overlay.date) .as_ref()
.and_then(|symbol_ids| symbol_ids.binary_search(&symbol_id).ok()) .as_ref()
.and_then(|positions_by_date| positions_by_date.get(&overlay.date))
.and_then(|positions| positions.get(symbol_id as usize))
.copied()
.filter(|position| *position != MISSING_ROW_POSITION)
.map(|position| position as usize)
.or_else(|| {
self.market_symbol_ids_by_date
.get(&overlay.date)
.and_then(|symbol_ids| symbol_ids.binary_search(&symbol_id).ok())
})
.ok_or_else(|| DataSetError::MissingSnapshot { .ok_or_else(|| DataSetError::MissingSnapshot {
kind: "intraday_overlay_market", kind: "intraday_overlay_market",
date: overlay.date, date: overlay.date,
symbol: overlay.symbol.clone(), symbol: overlay.symbol.clone(),
})?; })?;
resolved.push((overlay.date, row_position, overlay)); let overlay_index = resolved.len();
resolved.push((overlay.date, row_position, symbol_id, overlay));
overlay_indexes_by_symbol_id
.entry(symbol_id)
.or_default()
.push(overlay_index);
} }
let market_by_date = Arc::get_mut(&mut self.market_by_date).ok_or( let mut series_replacements = Vec::with_capacity(overlay_indexes_by_symbol_id.len());
DataSetError::SharedComponentMutation { for (symbol_id, overlay_indexes) in overlay_indexes_by_symbol_id {
component: "daily market panel", let existing = self
}, .market_series_by_symbol_id
)?; .get(symbol_id as usize)
for (date, row_position, overlay) in resolved { .and_then(Option::as_ref)
.ok_or_else(|| DataSetError::MissingSnapshot {
kind: "intraday_overlay_market_series",
date: resolved[overlay_indexes[0]].0,
symbol: resolved[overlay_indexes[0]].3.symbol.clone(),
})?;
let mut updated = (**existing).clone();
let series_overlays = overlay_indexes
.iter()
.map(|index| &resolved[*index].3)
.collect::<Vec<_>>();
updated
.apply_intraday_market_overlays(&series_overlays)
.map_err(|date| DataSetError::MissingSnapshot {
kind: "intraday_overlay_market_series_date",
date,
symbol: updated.symbol.clone(),
})?;
series_replacements.push((symbol_id, Arc::new(updated)));
}
let market_by_date = Arc::get_mut(&mut self.market_by_date)
.expect("daily market panel uniqueness checked before overlay");
for (date, row_position, _, overlay) in &resolved {
let row = market_by_date let row = market_by_date
.get_mut(&date) .get_mut(date)
.and_then(|rows| rows.get_mut(row_position)) .and_then(|rows| rows.get_mut(*row_position))
.ok_or_else(|| DataSetError::MissingSnapshot { .ok_or_else(|| DataSetError::MissingSnapshot {
kind: "intraday_overlay_market_row", kind: "intraday_overlay_market_row",
date, date: *date,
symbol: overlay.symbol.clone(), symbol: overlay.symbol.clone(),
})?; })?;
debug_assert_eq!(row.symbol, overlay.symbol); debug_assert_eq!(row.symbol, overlay.symbol);
row.timestamp = overlay.timestamp; row.timestamp = overlay.timestamp.clone();
if let Some(last_price) = overlay if let Some(last_price) = overlay
.last_price .last_price
.filter(|value| value.is_finite() && *value > 0.0) .filter(|value| value.is_finite() && *value > 0.0)
@@ -2308,9 +2394,19 @@ impl DataSet {
row.minute_volume = overlay.minute_volume; row.minute_volume = overlay.minute_volume;
row.bid1_volume = overlay.bid1_volume; row.bid1_volume = overlay.bid1_volume;
row.ask1_volume = overlay.ask1_volume; row.ask1_volume = overlay.ask1_volume;
row.trading_phase = overlay.trading_phase; row.trading_phase = overlay.trading_phase.clone();
} }
Ok(seen.len())
let market_series_by_symbol = Arc::get_mut(&mut self.market_series_by_symbol)
.expect("market series map uniqueness checked before overlay");
let market_series_by_symbol_id = Arc::get_mut(&mut self.market_series_by_symbol_id)
.expect("market series id map uniqueness checked before overlay");
for (symbol_id, series) in series_replacements {
let symbol = self.symbol_by_id[symbol_id as usize].to_string();
market_series_by_symbol.insert(symbol, Arc::clone(&series));
market_series_by_symbol_id[symbol_id as usize] = Some(series);
}
Ok(resolved.len())
} }
/// Replaces the run-local execution quote layer without touching the /// Replaces the run-local execution quote layer without touching the
@@ -4966,7 +5062,7 @@ mod tests {
} }
#[test] #[test]
fn unique_dataset_applies_sparse_intraday_overlay_without_rebuilding_daily_series() { fn unique_dataset_applies_sparse_intraday_overlay_to_daily_and_symbol_views() {
let date = NaiveDate::from_ymd_opt(2025, 1, 2).unwrap(); let date = NaiveDate::from_ymd_opt(2025, 1, 2).unwrap();
let mut data = DataSet::from_components( let mut data = DataSet::from_components(
vec![Instrument { vec![Instrument {
@@ -5016,12 +5112,22 @@ mod tests {
assert_eq!(market.ask1_volume, 3_200); assert_eq!(market.ask1_volume, 3_200);
assert_eq!(market.trading_phase.as_deref(), Some("continuous")); assert_eq!(market.trading_phase.as_deref(), Some("continuous"));
assert_eq!(market.close, 10.0); assert_eq!(market.close, 10.0);
assert!(Arc::ptr_eq( let market_series_after = data.market_series_by_symbol_id
[data.symbol_id("000001.SZ").unwrap() as usize]
.as_ref()
.unwrap();
assert!(!Arc::ptr_eq(
&market_series_before, &market_series_before,
data.market_series_by_symbol_id[data.symbol_id("000001.SZ").unwrap() as usize] market_series_after
.as_ref()
.unwrap()
)); ));
assert_eq!(
serde_json::to_value(market_series_after.snapshot_at(0)).unwrap(),
serde_json::to_value(market).unwrap()
);
assert_eq!(
market_series_after.moving_average(date, 1, PriceField::Last),
Some(10.08)
);
} }
#[test] #[test]