From 3f6017d98bce6fc234c69f638c34481d1a37e56f Mon Sep 17 00:00:00 2001 From: boris Date: Mon, 7 Sep 2026 13:25:07 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BF=9D=E6=8C=81=E6=97=A5=E5=86=85=E8=A6=86?= =?UTF-8?q?=E7=9B=96=E4=B8=8E=E6=8C=89=E8=82=A1=E5=BA=8F=E5=88=97=E4=B8=80?= =?UTF-8?q?=E8=87=B4?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- crates/fidc-core/src/data.rs | 150 ++++++++++++++++++++++++++++++----- 1 file changed, 128 insertions(+), 22 deletions(-) diff --git a/crates/fidc-core/src/data.rs b/crates/fidc-core/src/data.rs index d903539..9878022 100644 --- a/crates/fidc-core/src/data.rs +++ b/crates/fidc-core/src/data.rs @@ -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 { if lookback == 0 { return None; @@ -2238,7 +2269,7 @@ impl DataSet { .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: /// silently using `Arc::make_mut` here would deep-copy the full market panel @@ -2251,8 +2282,25 @@ impl DataSet { 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 seen = HashSet::<(NaiveDate, u32)>::with_capacity(overlays.len()); + let mut overlay_indexes_by_symbol_id = BTreeMap::>::new(); for overlay in overlays { let symbol_id = self .symbol_id_by_code @@ -2270,33 +2318,71 @@ impl DataSet { }); } let row_position = self - .market_symbol_ids_by_date - .get(&overlay.date) - .and_then(|symbol_ids| symbol_ids.binary_search(&symbol_id).ok()) + .market_row_positions_by_date + .as_ref() + .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 { kind: "intraday_overlay_market", date: overlay.date, 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( - DataSetError::SharedComponentMutation { - component: "daily market panel", - }, - )?; - for (date, row_position, overlay) in resolved { + let mut series_replacements = Vec::with_capacity(overlay_indexes_by_symbol_id.len()); + for (symbol_id, overlay_indexes) in overlay_indexes_by_symbol_id { + let existing = self + .market_series_by_symbol_id + .get(symbol_id as usize) + .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::>(); + 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 - .get_mut(&date) - .and_then(|rows| rows.get_mut(row_position)) + .get_mut(date) + .and_then(|rows| rows.get_mut(*row_position)) .ok_or_else(|| DataSetError::MissingSnapshot { kind: "intraday_overlay_market_row", - date, + date: *date, symbol: overlay.symbol.clone(), })?; debug_assert_eq!(row.symbol, overlay.symbol); - row.timestamp = overlay.timestamp; + row.timestamp = overlay.timestamp.clone(); if let Some(last_price) = overlay .last_price .filter(|value| value.is_finite() && *value > 0.0) @@ -2308,9 +2394,19 @@ impl DataSet { row.minute_volume = overlay.minute_volume; row.bid1_volume = overlay.bid1_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 @@ -4966,7 +5062,7 @@ mod tests { } #[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 mut data = DataSet::from_components( vec![Instrument { @@ -5016,12 +5112,22 @@ mod tests { assert_eq!(market.ask1_volume, 3_200); assert_eq!(market.trading_phase.as_deref(), Some("continuous")); 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, - data.market_series_by_symbol_id[data.symbol_id("000001.SZ").unwrap() as usize] - .as_ref() - .unwrap() + market_series_after )); + 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]