diag(fxt-data-audit): predecoded MBP-10 sidecar audit binary

One-shot diagnostic that loads a .dbn.zst via
load_or_predecode_mbp10 and reports symbol distribution, per-symbol
price stats (min/p1/p50/p99/max for bid_px[0]/ask_px[0]), outlier
counts (zero-price, huge-price >$100k, zero-price-with-size),
and first-record samples per symbol.

Built to investigate the 18% sentinel-laden trade residue in the
ISV stop-controller cluster smokes (97 zero, 85 i32::MAX, 14 weird-
other). The controller path is structurally correct; remaining bad
records hypothesized to originate from source MBP-10 data (mixed
symbols, corrupt records, or unreported instrument families).

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

Cargo.toml

@@ -147,6 +147,7 @@ members = [
     # CLI binary
     "bin/fxt",
     "bin/fxt-backtest",
+    "bin/fxt-data-audit",
     # Services
     "services/backtesting_service",
     "services/broker_gateway_service",

bin/fxt-data-audit/Cargo.toml

@@ -0,0 +1,19 @@
+[package]
+name = "fxt-data-audit"
+version.workspace = true
+edition.workspace = true
+rust-version.workspace = true
+authors.workspace = true
+license.workspace = true
+repository.workspace = true
+homepage.workspace = true
+documentation.workspace = true
+publish.workspace = true
+keywords.workspace = true
+categories.workspace = true
+description = "Diagnostic audit of predecoded MBP-10 sidecar files — symbol distribution, price outliers, schema oddities."
+
+[dependencies]
+ml-features = { path = "../../crates/ml-features" }
+data = { path = "../../crates/data" }
+anyhow.workspace = true

bin/fxt-data-audit/src/main.rs

@@ -0,0 +1,155 @@
+//! Diagnostic audit of predecoded MBP-10 sidecar files.
+//!
+//! Usage: `fxt-data-audit <source.dbn.zst> <predecoded_dir>`
+//!
+//! Loads via `ml_features::predecoded::load_or_predecode_mbp10` and reports:
+//!   * Total snapshots in the file.
+//!   * Symbol distribution (count by `Mbp10Snapshot.symbol`).
+//!   * Per-symbol price stats: min/max/p1/p50/p99 of `bid_px[0]` and `ask_px[0]`
+//!     (raw i64 from the parser; conversion to f64 = /1e9 nanoprice).
+//!   * Outlier counts: bid_px[0]=0, bid_px[0]>1e14 (=100k$ price after /1e9),
+//!     ask_px[0]=0, ask_px[0]>1e14.
+//!   * "Sized but un-priced" anomalies: bid_sz[0]>0 AND bid_px[0]=0 (and ask).
+//!
+//! Run on the cluster with the training-data + feature-cache PVCs mounted.
+
+use anyhow::Result;
+use data::providers::databento::mbp10::BidAskPair;
+use ml_features::predecoded::load_or_predecode_mbp10;
+use std::collections::BTreeMap;
+use std::path::PathBuf;
+
+fn percentile(sorted: &[i64], p: f64) -> i64 {
+    if sorted.is_empty() {
+        return 0;
+    }
+    let idx = ((sorted.len() as f64) * p).clamp(0.0, sorted.len() as f64 - 1.0) as usize;
+    sorted[idx]
+}
+
+fn main() -> Result<()> {
+    let args: Vec<String> = std::env::args().collect();
+    if args.len() < 3 {
+        eprintln!("usage: {} <source.dbn.zst> <predecoded_dir>", args[0]);
+        std::process::exit(2);
+    }
+    let source = PathBuf::from(&args[1]);
+    let predecoded_dir = PathBuf::from(&args[2]);
+
+    eprintln!("loading: {} (predecoded_dir={})", source.display(), predecoded_dir.display());
+    let snapshots = load_or_predecode_mbp10(&source, &predecoded_dir)
+        .map_err(|e| anyhow::anyhow!("load_or_predecode_mbp10: {e}"))?;
+    eprintln!("loaded: {} snapshots", snapshots.len());
+
+    // Symbol distribution.
+    let mut symbol_counts: BTreeMap<String, usize> = BTreeMap::new();
+    for s in &snapshots {
+        *symbol_counts.entry(s.symbol.clone()).or_default() += 1;
+    }
+    println!("=== symbol distribution ===");
+    for (sym, n) in &symbol_counts {
+        println!("  {:>20} : {}", sym, n);
+    }
+
+    // Per-symbol price stats + outliers.
+    println!();
+    println!("=== per-symbol price stats (raw i64, /1e9 nanoprice = display $) ===");
+    println!(
+        "{:>12} | {:>10} | {:>14} {:>14} {:>14} {:>14} {:>14} | {:>14} {:>14} {:>14} {:>14} {:>14}",
+        "symbol", "n", "bid_min", "bid_p1", "bid_p50", "bid_p99", "bid_max",
+        "ask_min", "ask_p1", "ask_p50", "ask_p99", "ask_max"
+    );
+    let mut total_zero_bid = 0usize;
+    let mut total_zero_ask = 0usize;
+    let mut total_huge_bid = 0usize;
+    let mut total_huge_ask = 0usize;
+    let mut total_zero_bid_sized = 0usize;
+    let mut total_zero_ask_sized = 0usize;
+    let huge_threshold: i64 = 100_000_000_000_000; // 1e14 raw → /1e9 = $100k. ES never at $100k.
+
+    for (sym, _) in &symbol_counts {
+        let mut bid_px: Vec<i64> = Vec::new();
+        let mut ask_px: Vec<i64> = Vec::new();
+        let mut zero_bid = 0usize;
+        let mut zero_ask = 0usize;
+        let mut huge_bid = 0usize;
+        let mut huge_ask = 0usize;
+        let mut zero_bid_sized = 0usize;
+        let mut zero_ask_sized = 0usize;
+        for s in &snapshots {
+            if &s.symbol != sym {
+                continue;
+            }
+            if let Some(l0) = s.levels.first() {
+                if l0.bid_px == 0 {
+                    zero_bid += 1;
+                    if l0.bid_sz > 0 {
+                        zero_bid_sized += 1;
+                    }
+                } else if l0.bid_px > huge_threshold {
+                    huge_bid += 1;
+                }
+                if l0.ask_px == 0 {
+                    zero_ask += 1;
+                    if l0.ask_sz > 0 {
+                        zero_ask_sized += 1;
+                    }
+                } else if l0.ask_px > huge_threshold {
+                    huge_ask += 1;
+                }
+                bid_px.push(l0.bid_px);
+                ask_px.push(l0.ask_px);
+            }
+        }
+        bid_px.sort_unstable();
+        ask_px.sort_unstable();
+        let bid_min = *bid_px.first().unwrap_or(&0);
+        let bid_max = *bid_px.last().unwrap_or(&0);
+        let ask_min = *ask_px.first().unwrap_or(&0);
+        let ask_max = *ask_px.last().unwrap_or(&0);
+        println!(
+            "{:>12} | {:>10} | {:>14} {:>14} {:>14} {:>14} {:>14} | {:>14} {:>14} {:>14} {:>14} {:>14}",
+            sym, bid_px.len(),
+            bid_min, percentile(&bid_px, 0.01), percentile(&bid_px, 0.50),
+            percentile(&bid_px, 0.99), bid_max,
+            ask_min, percentile(&ask_px, 0.01), percentile(&ask_px, 0.50),
+            percentile(&ask_px, 0.99), ask_max
+        );
+        total_zero_bid += zero_bid;
+        total_zero_ask += zero_ask;
+        total_huge_bid += huge_bid;
+        total_huge_ask += huge_ask;
+        total_zero_bid_sized += zero_bid_sized;
+        total_zero_ask_sized += zero_ask_sized;
+    }
+
+    println!();
+    println!("=== outlier counts (across all symbols) ===");
+    println!("  bid_px[0]=0           : {}", total_zero_bid);
+    println!("  bid_px[0]=0 AND bid_sz[0]>0  : {} (Bug C-b source)", total_zero_bid_sized);
+    println!("  bid_px[0]>1e14 (>$100k) : {}", total_huge_bid);
+    println!("  ask_px[0]=0           : {}", total_zero_ask);
+    println!("  ask_px[0]=0 AND ask_sz[0]>0  : {} (Bug C-b source)", total_zero_ask_sized);
+    println!("  ask_px[0]>1e14 (>$100k) : {}", total_huge_ask);
+
+    // Spot-check the price scale for the first record of the largest-symbol group.
+    println!();
+    println!("=== sanity check: first record per symbol (raw → /1e9 display) ===");
+    let mut seen: BTreeMap<String, bool> = BTreeMap::new();
+    for s in &snapshots {
+        if seen.contains_key(&s.symbol) {
+            continue;
+        }
+        if let Some(l0) = s.levels.first() {
+            let bid_f = BidAskPair::price_to_f64(l0.bid_px);
+            let ask_f = BidAskPair::price_to_f64(l0.ask_px);
+            println!(
+                "  {:>20} ts={} bid_px={} (={:.4}) bid_sz={} ask_px={} (={:.4}) ask_sz={}",
+                s.symbol, s.timestamp, l0.bid_px, bid_f, l0.bid_sz, l0.ask_px, ask_f, l0.ask_sz
+            );
+        }
+        seen.insert(s.symbol.clone(), true);
+    }
+
+    Ok(())
+}