fix(data): DBN spread-instrument filter — root cause of Bug 2 contamination

Direct inspection of ES.FUT_2024-Q1.dbn.zst (databento python client, offline
kubectl-cp from PVC) pinpoints the source of the 8,799 corrupt bars that
sanitize_bars (Fix 25) caught at the bar-level gate.

Q1 file content:
  Outrights (correct): ESH4/ESM4/ESU4/ESZ4/ESH5  — 43,353 records (76.87%)
                       price range $5,063–$5,478
  Spreads  (poison):   ESH4-ESM4/ESM4-ESU4/...   — 13,048 records (23.13%)
                       price range $47.30–$219.70

Calendar spreads trade at the price DIFFERENCE between adjacent contracts
(~$60-150 cost-of-carry roll basis). Databento's stype_in=parent resolution
for ES.FUT returns BOTH outrights AND every spread combination in the same
DBN stream. The legacy decoder keyed only by ts_event + dedup-by-volume;
during low-volume overnight windows + active rollover periods, spread bars
beat the outright on volume and survived the dedup. 780 spread bars
survived in Q1 alone; ~8,800 across 2024-2026.

Fix: build dbn::TsSymbolMap from metadata once, resolve each record's
instrument_id → symbol, skip any symbol containing `-` (spread separator).
Same-ts dedup-by-volume continues to handle legitimate front/back-month
overlap among outrights.

Adds `time = "0.3"` to ml/Cargo.toml (dbn::TsSymbolMap uses time::Date).

Why complementary to the Fix 25 sanitize_bars gate:
  - Spread filter (this commit) catches the cause precisely by symbol pattern,
    but only for instruments where parent-symbol resolution is the source.
  - sanitize_bars catches the symptom universally by close-ratio bound, but
    can't distinguish a low-basis spread from a fast-moving outright in
    extreme cases.
  Both layers active: spread filter dispatches at parser level, sanitize as
  defense-in-depth backstop for unknown future contamination shapes.

Validation: `cargo check -p ml --example train_baseline_rl` clean.

Refs: Bug 2 chain (#191, #194, label_scale=5443 leaks), today's
sanitize_bars find (Fix 25). Closes the contamination-source investigation.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Cargo.lock

@@ -6012,6 +6012,7 @@ dependencies = [
  "tempfile",
  "test-case",
  "thiserror 1.0.69",
+ "time",
  "tokio",
  "tokio-test",
  "toml",

crates/ml/Cargo.toml

@@ -87,6 +87,7 @@ thiserror.workspace = true
 anyhow.workspace = true
 chrono.workspace = true
 chrono-tz = "0.10"  # Timezone support for market hours calculations (Wave C)
+time = "0.3"  # Required by dbn::TsSymbolMap for instrument_id → symbol resolution
 csv = "1.3"  # CSV serialization for action export (Wave 3 Task 3.1)
 rand.workspace = true
 rand_chacha = "0.3"  # ChaCha RNG for reproducible Monte Carlo permutation tests

crates/ml/examples/baseline_common/mod.rs

@@ -69,20 +69,46 @@ fn load_bars_from_dbn(path: &Path) -> Result<Vec<OHLCVBar>> {
 
 /// Decode OHLCV records from an already-opened DBN decoder.
 ///
-/// When a `.FUT` parent symbol is used, the DBN file may contain duplicate
-/// bars at the same timestamp from overlapping contracts (e.g. ESH5, ESM5 at
-/// roll dates). We deduplicate by timestamp, keeping the bar with the highest
-/// volume (front-month contract).
+/// **Calendar-spread filter:** when `stype_in=parent` is used to request
+/// `ES.FUT`, Databento returns BOTH outright contracts (`ESH4`, `ESM4`, ...)
+/// AND calendar-spread instruments (`ESH4-ESM4`, `ESM4-ESU4`, ...) in the
+/// same DBN stream. Spreads trade at the price *difference* between adjacent
+/// contracts (~$60-150 due to cost-of-carry) so their bars look like
+/// ridiculously low ES quotes (e.g. close=$61 vs prev=$5141). The
+/// `instrument_id → symbol` resolution comes from the DBN file's metadata
+/// `symbol_map()`; any symbol containing `-` is a spread and is rejected at
+/// parse time. Pre-fix this contamination was 23% of records on Q1 2024 ES
+/// and ~8800 spread bars survived dedup-by-volume across the full dataset
+/// because spreads occasionally carry higher minute-volume than the outright
+/// during low-liquidity overnight periods.
+///
+/// **Same-timestamp dedup:** within outright records, multiple contracts
+/// (front-month + back-month, e.g. `ESH4` + `ESM4` during rollover week)
+/// can both report bars at the same `ts_event`. Keep the highest-volume bar
+/// at each timestamp — that's the active front month.
 fn decode_ohlcv_records<R: std::io::Read>(
     decoder: &mut dbn::decode::dbn::Decoder<R>,
     path: &Path,
 ) -> Result<Vec<OHLCVBar>> {
+    use dbn::decode::DbnMetadata;
     use dbn::OhlcvMsg;
     use std::collections::BTreeMap;
 
     let price_scale = 1e-9_f64;
     let mut by_ts: BTreeMap<u64, OHLCVBar> = BTreeMap::new();
     let mut total_records = 0_usize;
+    let mut skipped_spread = 0_usize;
+
+    // Build instrument_id → symbol resolver from the DBN metadata. This is the
+    // authoritative source — the same map Databento's Python client uses for
+    // its `symbol` column. Empty for files that lack symbology mappings (older
+    // DBN versions or non-parent-symbol requests); in that case the spread
+    // filter below silently no-ops, and the bar-level sanity gate in
+    // `sanitize_bars` remains as a defense-in-depth backstop.
+    let symbol_map = decoder
+        .metadata()
+        .symbol_map()
+        .with_context(|| format!("Failed to build symbol map for {}", path.display()))?;
 
     while let Some(record) = decoder
         .decode_record::<OhlcvMsg>()
@@ -90,11 +116,32 @@ fn decode_ohlcv_records<R: std::io::Read>(
     {
         total_records += 1;
         let ts_nanos = record.hd.ts_event;
-        let volume = record.volume as f64;
+        let timestamp = nanos_to_datetime(ts_nanos);
 
+        // Resolve symbol via TsSymbolMap (date + instrument_id → symbol). Drop
+        // any record whose symbol contains a hyphen — those are calendar
+        // spread instruments and their prices are not on the underlying's
+        // scale. If the symbol can't be resolved (no mapping for this date),
+        // skip the record rather than guess.
+        let secs_since_epoch = (ts_nanos / 1_000_000_000) as i64;
+        let date = time::OffsetDateTime::from_unix_timestamp(secs_since_epoch)
+            .map(|dt| dt.date())
+            .ok();
+        let symbol = match date.and_then(|d| symbol_map.get(d, record.hd.instrument_id)) {
+            Some(s) => s.as_str(),
+            None => {
+                skipped_spread += 1;
+                continue;
+            }
+        };
+        if symbol.contains('-') {
+            skipped_spread += 1;
+            continue;
+        }
+
+        let volume = record.volume as f64;
         let existing_vol = by_ts.get(&ts_nanos).map(|b| b.volume).unwrap_or(0.0);
         if volume >= existing_vol {
-            let timestamp = nanos_to_datetime(ts_nanos);
             by_ts.insert(
                 ts_nanos,
                 OHLCVBar {
@@ -110,13 +157,15 @@ fn decode_ohlcv_records<R: std::io::Read>(
     }
 
     let bars: Vec<OHLCVBar> = by_ts.into_values().collect();
-    let deduped = total_records - bars.len();
-    if deduped > 0 {
+    let deduped = total_records - bars.len() - skipped_spread;
+    if skipped_spread > 0 || deduped > 0 {
         info!(
-            "  Deduplicated {}/{} bars by timestamp from {}",
-            deduped,
+            "  {}: {} records → {} bars (skipped {} spreads, deduped {} same-ts)",
+            path.display(),
             total_records,
-            path.display()
+            bars.len(),
+            skipped_spread,
+            deduped,
         );
     }
     Ok(bars)

docs/dqn-gpu-hot-path-audit.md

@@ -315,6 +315,28 @@ Final structural sweep that closes the residual `*_via_pinned` ctor / cold-path
 
 **Out of scope.** `mapped_pinned.rs` retains the `upload_f32_via_pinned` / `upload_i32_via_pinned` / `clone_to_device_f32_via_pinned` / `clone_to_device_i32_via_pinned` helper definitions — `trainers/ppo.rs` and a small set of remaining cold-path callers still depend on them; the helpers are deleted under task #290 once the call count globally hits zero. Per `feedback_no_hiding` no suppression markers were added.
 
+### Fix 26 — DBN spread-instrument filter (root cause of Bug 2 contamination) (2026-05-02)
+**Pinned the source of the 8,799 corrupt bars sanitize_bars caught.**
+
+Direct inspection of `ES.FUT_2024-Q1.dbn.zst` via `databento` Python client (offline, kubectl-cp from PVC):
+
+| Instrument type | Symbols | Records (Q1 file) | Price range |
+|---|---|---|---|
+| **Outright** (correct) | `ESH4`, `ESM4`, `ESU4`, `ESZ4`, `ESH5` | 43,353 (76.87%) | $5,063 – $5,478 |
+| **Calendar spreads** (poison) | `ESH4-ESM4`, `ESM4-ESU4`, `ESH4-ESU4`, `ESM4-ESZ4`, ... | 13,048 (23.13%) | $47.30 – $219.70 |
+
+Calendar spreads trade at the *price difference* between adjacent contracts (~$60-150 due to cost-of-carry roll basis). Databento's `stype_in=parent` symbology (used for `ES.FUT` requests) returns BOTH outright contracts AND every calendar-spread combination in the same DBN stream.
+
+**Why the contamination leaked through:** the legacy `decode_ohlcv_records` keyed records only by `ts_event`, then deduped by `volume` (highest-wins). Outrights typically dominate volume during regular hours, so most spread bars were dropped. But during low-liquidity overnight windows + active rollover periods, spreads hit higher minute-volume than the outright (samples show spread volumes up to 1554 contracts/min during ESH4→ESM4 transition vs near-zero outright volume in those minutes). 780 spread bars survived dedup in Q1 alone; ~8,800 across the full 2024-2026 dataset.
+
+**The fix** (`baseline_common/mod.rs::decode_ohlcv_records`): build `dbn::TsSymbolMap` from the DBN metadata once, resolve each record's `instrument_id` to its symbol, and skip any symbol containing `-` (the canonical spread separator). Same-timestamp dedup-by-volume continues to handle legitimate front/back-month overlap during rollover.
+
+**Why this is correct vs the bar-level sanitize:** the sanitize_bars layer (Fix 25) catches the *symptom* (close ratio out of bounds) and works for any instrument. The spread filter catches the *cause* — a spread between contracts where the underlying happens to have a small basis can sit within the [0.5, 2.0] sanitize ratio (e.g., short-tenor spread on a low-vol day might trade at $5/$5141 = 0.001 OR cleanly on the underlying scale during fast moves). The two layers are complementary: spread filter is precise but symbol-dependent; sanitize is universal but symptom-based.
+
+**Validation.** `cargo check -p ml --example train_baseline_rl` clean. Adds `time = "0.3"` to ml/Cargo.toml (required by `dbn::TsSymbolMap`'s `time::Date` API). Expected next-smoke effect: `decode_ohlcv_records` log shows `skipped N spreads` with N matching the per-file contamination count; `sanitize_bars` reports zero (or near-zero) drops since the spread filter caught everything upstream. Sanitize remains as defense-in-depth for unknown future contamination shapes.
+
+**Closes the chase.** Bug 2 root → ~8,800 spread bars from Databento's parent-symbol resolution. Caught structurally now at the DBN parser, not at z-score time.
+
 ### Fix 25 — Structural data-loading defense + DIAG_BUG2 cleanup (2026-05-02)
 **Goal: stop chasing one-off data-corruption bugs. Make corruption impossible to flow into training, regardless of source.**