diff --git a/crates/fidc-core/src/data.rs b/crates/fidc-core/src/data.rs
index 1b3ccf4..7b6446c 100644
--- a/crates/fidc-core/src/data.rs
+++ b/crates/fidc-core/src/data.rs
@@ -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);
diff --git a/crates/fidc-core/src/series_columns.rs b/crates/fidc-core/src/series_columns.rs
new file mode 100644
index 0000000..3cbd42c
--- /dev/null
+++ b/crates/fidc-core/src/series_columns.rs
@@ -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];
+    }
+}