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

perf(data): retain exact repeated and reference-matched series columns

Browse Source
This commit is contained in:
boris
2026-09-14 03:29:05 +08:00
committed by boris
parent c62ae1206f
commit 996b909608
2 changed files with 319 additions and 45 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/fidc-core/src/data.rs
+128 -45
View File
@@ -313,8 +313,8 @@ pub enum QuoteObservationKind {
/// Sparse same-day fields layered onto an already-built immutable daily panel.
///
/// These fields do not participate in daily price series, adjustment series,
/// symbol indexes, or rolling windows. Applying them in place lets the runner
/// These fields leave daily OHLC, adjustment series and symbol indexes intact,
/// but update quote history and Last-price rolling windows. Applying them lets the runner
/// reuse the candidate-planning `DataSet` as the final execution `DataSet`
/// without rebuilding the full market panel.
#[derive(Debug, Clone, Serialize, Deserialize)]
@@ -597,17 +597,21 @@ pub fn decision_free_float_cap_bn(factor: &DailyFactorSnapshot) -> f64 {
#[derive(Debug, Clone)]
struct SymbolPriceSeries {
base: Arc<SymbolDailySeriesBase>,
timestamps: Vec<Option<String>>,
last_prices: Vec<f64>,
bid1s: Vec<f64>,
ask1s: Vec<f64>,
minute_volumes: Vec<u64>,
bid1_volumes: Vec<u64>,
ask1_volumes: Vec<u64>,
trading_phases: Vec<Option<String>>,
last_prefix: Vec<f64>,
timestamps: RepeatedValues<Option<String>>,
last_prices: ReferenceMatchedValues,
bid1s: ReferenceMatchedValues,
ask1s: ReferenceMatchedValues,
minute_volumes: RepeatedValues<u64>,
bid1_volumes: RepeatedValues<u64>,
ask1_volumes: RepeatedValues<u64>,
trading_phases: RepeatedValues<Option<String>>,
last_prefix: ReferenceMatchedValues,
}
#[path = "series_columns.rs"]
mod series_columns;
use series_columns::{ReferenceMatchedValues, RepeatedValues};
#[derive(Debug)]
struct SymbolDailySeriesBase {
symbol: String,
@@ -840,42 +844,42 @@ impl SymbolPriceSeries {
);
let row_count = rows.len();
let mut dates = Vec::with_capacity(row_count);
let mut timestamps = Vec::with_capacity(row_count);
let mut timestamps = RepeatedValues::new();
let mut day_opens = Vec::with_capacity(row_count);
let mut opens = Vec::with_capacity(row_count);
let mut highs = Vec::with_capacity(row_count);
let mut lows = Vec::with_capacity(row_count);
let mut closes = Vec::with_capacity(row_count);
let mut prev_closes = Vec::with_capacity(row_count);
let mut last_prices = Vec::with_capacity(row_count);
let mut bid1s = Vec::with_capacity(row_count);
let mut ask1s = Vec::with_capacity(row_count);
let mut last_prices = ReferenceMatchedValues::Identical;
let mut bid1s = ReferenceMatchedValues::Identical;
let mut ask1s = ReferenceMatchedValues::Identical;
let mut volumes = Vec::with_capacity(row_count);
let mut minute_volumes = Vec::with_capacity(row_count);
let mut bid1_volumes = Vec::with_capacity(row_count);
let mut ask1_volumes = Vec::with_capacity(row_count);
let mut trading_phases = Vec::with_capacity(row_count);
let mut minute_volumes = RepeatedValues::new();
let mut bid1_volumes = RepeatedValues::new();
let mut ask1_volumes = RepeatedValues::new();
let mut trading_phases = RepeatedValues::new();
let mut paused = Vec::with_capacity(row_count);
let mut upper_limits = Vec::with_capacity(row_count);
let mut lower_limits = Vec::with_capacity(row_count);
let mut price_ticks = Vec::with_capacity(row_count);
for row in rows {
dates.push(row.date);
timestamps.push(row.timestamp.clone());
timestamps.push(&row.timestamp, row_count);
day_opens.push(row.day_open);
opens.push(row.open);
highs.push(row.high);
lows.push(row.low);
closes.push(row.close);
prev_closes.push(row.prev_close);
last_prices.push(row.last_price);
bid1s.push(row.bid1);
ask1s.push(row.ask1);
last_prices.push(row.last_price, &closes, row_count);
bid1s.push(row.bid1, &closes, row_count);
ask1s.push(row.ask1, &closes, row_count);
volumes.push(row.volume);
minute_volumes.push(row.minute_volume);
bid1_volumes.push(row.bid1_volume);
ask1_volumes.push(row.ask1_volume);
trading_phases.push(row.trading_phase.clone());
minute_volumes.push(&row.minute_volume, row_count);
bid1_volumes.push(&row.bid1_volume, row_count);
ask1_volumes.push(&row.ask1_volume, row_count);
trading_phases.push(&row.trading_phase, row_count);
paused.push(row.paused);
upper_limits.push(row.upper_limit);
lower_limits.push(row.lower_limit);
@@ -885,7 +889,7 @@ impl SymbolPriceSeries {
let open_prefix = prefix_sums(&opens);
let close_prefix = prefix_sums(&closes);
let prev_close_prefix = prefix_sums(&prev_closes);
let last_prefix = prefix_sums(&last_prices);
let last_prefix = last_prices.prefix();
let mut valid_volume_sum_prefix = Vec::with_capacity(volumes.len() + 1);
let mut valid_volume_count_prefix = Vec::with_capacity(volumes.len() + 1);
valid_volume_sum_prefix.push(0.0);
@@ -958,23 +962,23 @@ impl SymbolPriceSeries {
.dates
.binary_search(&overlay.date)
.map_err(|_| overlay.date)?;
self.timestamps[index] = overlay.timestamp.clone();
self.timestamps.set(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;
self.last_prices.set(index, last_price, &self.base.closes);
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();
self.bid1s.set(index, overlay.bid1, &self.base.closes);
self.ask1s.set(index, overlay.ask1, &self.base.closes);
self.minute_volumes.set(index, overlay.minute_volume);
self.bid1_volumes.set(index, overlay.bid1_volume);
self.ask1_volumes.set(index, overlay.ask1_volume);
self.trading_phases.set(index, overlay.trading_phase.clone());
}
if last_price_changed {
self.last_prefix = prefix_sums(&self.last_prices);
self.last_prefix = self.last_prices.prefix();
}
Ok(())
}
@@ -1194,7 +1198,7 @@ impl SymbolPriceSeries {
PriceField::DayOpen => &self.day_opens,
PriceField::Open => &self.opens,
PriceField::Close => &self.closes,
PriceField::Last => &self.last_prices,
PriceField::Last => self.last_prices.values(&self.closes),
}
}
@@ -1211,7 +1215,7 @@ impl SymbolPriceSeries {
PriceField::DayOpen => &self.day_open_prefix,
PriceField::Open => &self.open_prefix,
PriceField::Close => &self.close_prefix,
PriceField::Last => &self.last_prefix,
PriceField::Last => self.last_prefix.values(&self.close_prefix),
}
}
@@ -1225,9 +1229,9 @@ impl SymbolPriceSeries {
high: self.highs[index],
low: self.lows[index],
close: self.closes[index],
last_price: self.last_prices[index],
bid1: self.bid1s[index],
ask1: self.ask1s[index],
last_price: self.last_prices.values(&self.closes)[index],
bid1: self.bid1s.values(&self.closes)[index],
ask1: self.ask1s.values(&self.closes)[index],
prev_close: self.prev_closes[index],
volume: self.volumes[index],
minute_volume: self.minute_volumes[index],
@@ -1248,12 +1252,12 @@ impl SymbolPriceSeries {
"high" => Some(self.highs[index]),
"low" => Some(self.lows[index]),
"close" | "price" => Some(self.closes[index]),
"last" | "last_price" => Some(self.last_prices[index]),
"last" | "last_price" => Some(self.last_prices.values(&self.closes)[index]),
"prev_close" | "pre_close" => Some(self.prev_closes[index]),
"volume" => Some(self.volumes[index] as f64),
"minute_volume" => Some(self.minute_volumes[index] as f64),
"bid1" => Some(self.bid1s[index]),
"ask1" => Some(self.ask1s[index]),
"bid1" => Some(self.bid1s.values(&self.closes)[index]),
"ask1" => Some(self.ask1s.values(&self.closes)[index]),
"bid1_volume" => Some(self.bid1_volumes[index] as f64),
"ask1_volume" => Some(self.ask1_volumes[index] as f64),
"upper_limit" => Some(self.upper_limits[index]),
@@ -6556,6 +6560,85 @@ mod tests {
}
}
#[test]
fn series_columns_preserve_full_snapshots_and_distinct_price_bits() {
for mixed in [false, true] {
let mut rows = (0..6).map(|index| {
let date = NaiveDate::from_ymd_opt(2025, 1, 2 + index).unwrap();
let mut row = market_row(&date.to_string(), 10. + index as f64, 1_000);
row.minute_volume = 7;
row.trading_phase = Some("continuous".to_string());
row
}).collect::<Vec<_>>();
if mixed {
rows[2].last_price = 0.;
rows[3].bid1 = -0.;
rows[4].ask1 = f64::from_bits(0x7ff8_0000_0000_0042);
rows[4].timestamp = Some("2025-01-06 10:21:00".to_string());
rows[4].trading_phase = None;
rows[4].minute_volume = 10_000;
}
let series = SymbolPriceSeries::new("000001.SZ".to_string(), &rows);
for (index, expected) in rows.iter().enumerate() {
let actual = series.snapshot_at(index);
assert_eq!(serde_json::to_value(&actual).unwrap(), serde_json::to_value(expected).unwrap());
assert_eq!(actual.last_price.to_bits(), expected.last_price.to_bits());
assert_eq!(actual.bid1.to_bits(), expected.bid1.to_bits());
assert_eq!(actual.ask1.to_bits(), expected.ask1.to_bits());
}
let expected_prefix = prefix_sums(&rows.iter().map(|row| row.last_price).collect::<Vec<_>>());
let bits = |values: &[f64]| values.iter().map(|value| value.to_bits()).collect::<Vec<_>>();
assert_eq!(bits(series.prefix_for(PriceField::Last)), bits(&expected_prefix));
if !mixed {
assert!(matches!(series.last_prices, ReferenceMatchedValues::Identical));
assert!(matches!(series.bid1s, ReferenceMatchedValues::Identical));
assert!(matches!(series.ask1s, ReferenceMatchedValues::Identical));
assert_eq!(series.price_values_for(PriceField::Last).as_ptr(), series.closes.as_ptr());
}
}
}
#[test]
fn series_overlay_materializes_only_changed_values_and_preserves_history_cutoff() {
let rows = [
market_row("2025-01-02", 10., 1_000),
market_row("2025-01-03", 12., 2_000),
market_row("2025-01-06", 14., 3_000),
];
let original = SymbolPriceSeries::new("000001.SZ".to_string(), &rows);
let mut changed = original.clone();
let overlay = IntradayMarketSnapshotOverlay {
date: rows[2].date, symbol: "000001.SZ".to_string(),
timestamp: Some("2025-01-06 13:20:00".to_string()), last_price: Some(15.),
bid1: 14., ask1: 15.01, minute_volume: 30, bid1_volume: 20, ask1_volume: 10,
trading_phase: Some("continuous".to_string()),
};
changed.apply_intraday_market_overlays(&[&overlay]).unwrap();
assert!(Arc::ptr_eq(&original.base, &changed.base));
assert!(matches!(original.last_prices, ReferenceMatchedValues::Identical));
assert!(matches!(changed.last_prices, ReferenceMatchedValues::Owned(_)));
assert!(matches!(changed.bid1s, ReferenceMatchedValues::Identical));
let mut expected = rows[2].clone();
expected.timestamp = overlay.timestamp.clone();
expected.last_price = 15.;
expected.bid1 = overlay.bid1;
expected.ask1 = overlay.ask1;
expected.minute_volume = overlay.minute_volume;
expected.bid1_volume = overlay.bid1_volume;
expected.ask1_volume = overlay.ask1_volume;
expected.trading_phase = overlay.trading_phase.clone();
assert_eq!(serde_json::to_value(changed.snapshot_at(2)).unwrap(), serde_json::to_value(expected).unwrap());
assert_eq!(original.snapshot_at(2).last_price, 14.);
assert_eq!(changed.moving_average(rows[1].date, 2, PriceField::Last), Some(11.));
assert_eq!(changed.trailing_values(rows[1].date, 2, PriceField::Last), vec![10., 12.]);
assert_eq!(changed.trailing_snapshots(rows[2].date, 2, false).len(), 2);
assert_eq!(changed.trailing_numeric_values(rows[2].date, 2, "last", false), vec![10., 12.]);
assert_eq!(changed.moving_average(rows[2].date, 2, PriceField::Last), Some(13.5));
let mut unknown = overlay;
unknown.date = NaiveDate::from_ymd_opt(2025, 2, 1).unwrap();
assert_eq!(changed.apply_intraday_market_overlays(&[&unknown]), Err(unknown.date));
}
#[test]
fn day_open_moving_average_uses_its_own_historical_column() {
let mut first = market_row("2025-01-02", 10.0, 100);
crates/fidc-core/src/series_columns.rs
+191
View File
@@ -0,0 +1,191 @@
use std::ops::Index;
use super::prefix_sums;
#[derive(Debug, Clone)]
pub(super) enum ReferenceMatchedValues {
Identical,
Owned(Vec<f64>),
}
impl ReferenceMatchedValues {
pub(super) fn push(&mut self, value: f64, reference: &[f64], capacity: usize) {
let previous_len = reference.len().checked_sub(1).expect("reference row is missing");
match self {
Self::Identical if value.to_bits() == reference[previous_len].to_bits() => {}
Self::Identical => {
let mut values = Vec::with_capacity(capacity);
values.extend_from_slice(&reference[..previous_len]);
values.push(value);
*self = Self::Owned(values);
}
Self::Owned(values) => {
debug_assert_eq!(values.len(), previous_len);
values.push(value);
}
}
}
pub(super) fn values<'a>(&'a self, reference: &'a [f64]) -> &'a [f64] {
match self {
Self::Identical => reference,
Self::Owned(values) => {
debug_assert_eq!(values.len(), reference.len());
values
}
}
}
pub(super) fn set(&mut self, index: usize, value: f64, reference: &[f64]) {
assert!(index < reference.len(), "series index out of bounds");
match self {
Self::Owned(values) => values[index] = value,
Self::Identical if value.to_bits() == reference[index].to_bits() => {}
Self::Identical => {
let mut values = reference.to_vec();
values[index] = value;
*self = Self::Owned(values);
}
}
}
pub(super) fn prefix(&self) -> Self {
match self {
Self::Identical => Self::Identical,
Self::Owned(values) => Self::Owned(prefix_sums(values)),
}
}
}
#[derive(Debug, Clone)]
pub(super) struct RepeatedValues<T> {
repeated: T,
values: Option<Vec<T>>,
len: usize,
}
impl<T: Default + Clone + Eq> RepeatedValues<T> {
pub(super) fn new() -> Self {
Self { repeated: T::default(), values: None, len: 0 }
}
pub(super) fn push(&mut self, value: &T, capacity: usize) {
if let Some(values) = &mut self.values {
values.push(value.clone());
} else if self.len == 0 {
self.repeated = value.clone();
} else if *value != self.repeated {
let mut values = Vec::with_capacity(capacity);
values.resize(self.len, std::mem::take(&mut self.repeated));
values.push(value.clone());
self.values = Some(values);
}
self.len += 1;
}
pub(super) fn set(&mut self, index: usize, value: T) {
assert!(index < self.len, "series index out of bounds");
if let Some(values) = &mut self.values {
values[index] = value;
} else if value != self.repeated {
let mut values = vec![std::mem::take(&mut self.repeated); self.len];
values[index] = value;
self.values = Some(values);
}
}
}
impl<T> Index<usize> for RepeatedValues<T> {
type Output = T;
fn index(&self, index: usize) -> &T {
assert!(index < self.len, "series index out of bounds");
match &self.values {
Some(values) => &values[index],
None => &self.repeated,
}
}
}
#[cfg(test)]
mod tests {
use super::*;
fn bits(values: &[f64]) -> Vec<u64> {
values.iter().map(|value| value.to_bits()).collect()
}
#[test]
fn identical_prices_share_only_after_exact_bit_comparison() {
let reference = [10., -0., f64::from_bits(0x7ff8_0000_0000_0042), f64::INFINITY];
let mut column = ReferenceMatchedValues::Identical;
for (index, value) in reference.iter().copied().enumerate() {
column.push(value, &reference[..=index], reference.len());
}
assert!(matches!(column, ReferenceMatchedValues::Identical));
assert_eq!(column.values(&reference).as_ptr(), reference.as_ptr());
let prefix = prefix_sums(&reference);
assert_eq!(bits(column.prefix().values(&prefix)), bits(&prefix));
let original = column.clone();
column.set(1, 0., &reference);
assert!(matches!(column, ReferenceMatchedValues::Owned(_)));
assert_eq!(column.values(&reference)[1].to_bits(), 0_f64.to_bits());
assert_eq!(bits(original.values(&reference)), bits(&reference));
assert_eq!(bits(column.prefix().values(&prefix)), bits(&prefix_sums(column.values(&reference))));
}
#[test]
fn differing_prices_preserve_zero_nan_payloads_and_prior_rows() {
let reference = [10., 11., f64::from_bits(0x7ff8_0000_0000_0042), 13.];
for actual in [
[10., 0., reference[2], 13.],
[10., 11., f64::from_bits(0x7ff8_0000_0000_0043), 13.],
] {
let mut column = ReferenceMatchedValues::Identical;
for (index, value) in actual.iter().copied().enumerate() {
column.push(value, &reference[..=index], actual.len());
}
assert!(matches!(column, ReferenceMatchedValues::Owned(_)));
assert_eq!(bits(column.values(&reference)), bits(&actual));
}
}
#[test]
fn repeated_values_preserve_nonzero_values_and_copy_on_change() {
let mut column = RepeatedValues::new();
for _ in 0..128 { column.push(&7_u64, 128); }
assert!(column.values.is_none());
assert_eq!(column[127], 7);
column.set(0, 7);
assert!(column.values.is_none());
let mut changed = column.clone();
changed.set(64, 9);
assert_eq!(changed[64], 9);
assert_eq!(changed[63], 7);
assert_eq!(column[64], 7);
}
#[test]
fn optional_values_keep_none_distinct_from_empty_and_repeated_text() {
for repeated in [None, Some(String::new()), Some("continuous".to_string())] {
let mut column = RepeatedValues::new();
for _ in 0..12 { column.push(&repeated, 16); }
assert!(column.values.is_none());
assert_eq!(column[0], repeated);
column.push(&Some("closing".to_string()), 16);
assert_eq!(column[11], repeated);
assert_eq!(column[12].as_deref(), Some("closing"));
column.set(5, None);
assert_eq!(column[5], None);
assert_eq!(column[4], repeated);
}
}
#[test]
#[should_panic(expected = "series index out of bounds")]
fn repeated_values_reject_out_of_range_access() {
let column = RepeatedValues::<u64>::new();
let _ = column[0];
}
}