foxhunt/crates/ml/examples/precompute_features.rs

#![allow(
    clippy::assertions_on_constants,
    clippy::assertions_on_result_states,
    clippy::clone_on_copy,
    clippy::decimal_literal_representation,
    clippy::doc_markdown,
    clippy::empty_line_after_doc_comments,
    clippy::field_reassign_with_default,
    clippy::get_unwrap,
    clippy::identity_op,
    clippy::inconsistent_digit_grouping,
    clippy::indexing_slicing,
    clippy::integer_division,
    clippy::len_zero,
    clippy::let_underscore_must_use,
    clippy::manual_div_ceil,
    clippy::manual_let_else,
    clippy::manual_range_contains,
    clippy::modulo_arithmetic,
    clippy::needless_range_loop,
    clippy::non_ascii_literal,
    clippy::redundant_clone,
    clippy::shadow_reuse,
    clippy::shadow_same,
    clippy::shadow_unrelated,
    clippy::single_match_else,
    clippy::str_to_string,
    clippy::string_slice,
    clippy::tests_outside_test_module,
    clippy::too_many_lines,
    clippy::unnecessary_wraps,
    clippy::unseparated_literal_suffix,
    clippy::use_debug,
    clippy::useless_vec,
    clippy::wildcard_enum_match_arm,
    clippy::else_if_without_else,
    clippy::expect_used,
    clippy::missing_const_for_fn,
    clippy::similar_names,
    clippy::type_complexity,
    clippy::collapsible_else_if,
    clippy::doc_lazy_continuation,
    clippy::items_after_test_module,
    clippy::map_clone,
    clippy::multiple_unsafe_ops_per_block,
    clippy::unwrap_or_default,
    clippy::assign_op_pattern,
    clippy::needless_borrow,
    clippy::println_empty_string,
    clippy::unnecessary_cast,
    clippy::used_underscore_binding,
    clippy::create_dir,
    clippy::implicit_saturating_sub,
    clippy::exit,
    clippy::expect_fun_call,
    clippy::too_many_arguments,
    clippy::unnecessary_map_or,
    clippy::unwrap_used,
    dead_code,
    unused_imports,
    unused_variables,
    clippy::cloned_ref_to_slice_refs,
    clippy::neg_multiply,
    clippy::while_let_loop,
    clippy::bool_assert_comparison,
    clippy::excessive_precision,
    clippy::trivially_copy_pass_by_ref,
    clippy::op_ref,
    clippy::redundant_closure,
    clippy::unnecessary_lazy_evaluations,
    clippy::if_then_some_else_none,
    clippy::unnecessary_to_owned,
    clippy::single_component_path_imports,
)]
//! Precompute Features — DBN to .fxcache pipeline
//!
//! Reads OHLCV + MBP-10 + trades DBN data, runs the full feature extraction
//! pipeline (standalone, no GPU required), and writes a `.fxcache` binary file for
//! zero-overhead GPU loading during training.
//!
//! Usage:
//!     cargo run -p ml --example precompute_features --release -- \
//!         --data-dir test_data/futures-baseline \
//!         --mbp10-data-dir test_data/futures-baseline-mbp10 \
//!         --trades-data-dir test_data/futures-baseline-trades \
//!         --output-dir test_data/feature-cache \
//!         --symbol ES.FUT \
//!         --yes

use std::path::{Path, PathBuf};
use std::time::Instant;

use anyhow::{Context, Result};
use clap::Parser;
use tracing::info;

use ml::trainers::dqn::{
    collect_dbn_files_filtered, collect_dbn_files_recursive, extract_features_from_bars,
};
use ml::features::extraction::OHLCVBar;

// ---------------------------------------------------------------------------
// CLI
// ---------------------------------------------------------------------------

#[derive(Debug, Parser)]
#[command(
    name = "precompute_features",
    about = "Precompute features from DBN data and write .fxcache binary"
)]
struct Opts {
    /// Directory containing OHLCV .dbn/.dbn.zst files (e.g. test_data/futures-baseline)
    #[arg(long)]
    data_dir: String,

    /// Directory containing MBP-10 .dbn/.dbn.zst files (optional)
    #[arg(long)]
    mbp10_data_dir: Option<String>,

    /// Directory containing trades .dbn/.dbn.zst files (optional)
    #[arg(long)]
    trades_data_dir: Option<String>,

    /// Output directory for .fxcache files (default: sibling of --data-dir)
    #[arg(long)]
    output_dir: Option<String>,

    /// Symbol name (used in output filename)
    #[arg(long, default_value = "ES.FUT")]
    symbol: String,

    /// Data-source identifier mixed into the SHA256 cache key.
    ///
    /// MUST match the `data_source` field of the training profile that will
    /// consume this fxcache (e.g. `dqn-production.toml: data_source = "mbp10"`,
    /// `dqn-smoketest.toml: data_source = "mbp10"`). Mismatch produces a
    /// silent cache MISS at training time → DBN-direct fallback uploads
    /// un-normalised features → aux-head label_scale picks up raw-price
    /// magnitudes → cascade of broken metrics + impossible Sharpe (see
    /// 2026-04-27 incident: train-h5gxb epoch-0 Sharpe=141 from this exact
    /// path mismatch).
    ///
    /// Defaults to `"mbp10"` — production, smoke, and localdev profiles all
    /// use MBP-10 microstructure data. Pass `--data-source ohlcv` only when
    /// intentionally generating a 1-minute candle cache for a profile that
    /// overrides `data_source = "ohlcv"`.
    #[arg(long, default_value = "mbp10")]
    data_source: String,

    /// Skip confirmation prompt
    #[arg(long)]
    yes: bool,
}

// ---------------------------------------------------------------------------
// Cache key computation
// ---------------------------------------------------------------------------

/// Compute a SHA256 cache key from all DBN files across the given directories.
///
// ---------------------------------------------------------------------------
// Main
// ---------------------------------------------------------------------------

#[tokio::main]
async fn main() -> Result<()> {
    // Initialize tracing
    tracing_subscriber::fmt()
        .with_env_filter(
            tracing_subscriber::EnvFilter::try_from_default_env()
                .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("info")),
        )
        .init();

    let opts = Opts::parse();

    let data_dir = PathBuf::from(&opts.data_dir);
    // Default output_dir: sibling of data_dir (e.g. test_data/futures-baseline → test_data/feature-cache)
    let output_dir = match opts.output_dir {
        Some(ref d) => PathBuf::from(d),
        None => data_dir.parent().unwrap_or(&data_dir).join("feature-cache"),
    };
    let mbp10_dir = opts.mbp10_data_dir.as_ref().map(PathBuf::from);
    let trades_dir = opts.trades_data_dir.as_ref().map(PathBuf::from);

    // ── Validate inputs ──────────────────────────────────────────────────────
    if !data_dir.exists() {
        anyhow::bail!("Data directory not found: {}", data_dir.display());
    }

    // ── Print config summary ─────────────────────────────────────────────────
    println!("================================================================================");
    println!("Precompute Features — DBN to .fxcache Pipeline");
    println!("================================================================================");
    println!();
    println!("Data dir:      {}", data_dir.display());
    if let Some(ref d) = mbp10_dir {
        println!("MBP-10 dir:    {}", d.display());
    }
    if let Some(ref d) = trades_dir {
        println!("Trades dir:    {}", d.display());
    }
    println!("Output dir:    {}", output_dir.display());
    println!("Symbol:        {}", opts.symbol);
    println!("Format:        f32 (v{}), OFI_DIM={}", ml::fxcache::FXCACHE_VERSION, ml::fxcache::OFI_DIM);
    println!();

    // ── Confirmation ─────────────────────────────────────────────────────────
    if !opts.yes {
        print!("Proceed with feature extraction? (yes/no): ");
        std::io::Write::flush(&mut std::io::stdout())?;

        let mut input = String::new();
        std::io::stdin().read_line(&mut input)?;
        let trimmed = input.trim();
        if !trimmed.eq_ignore_ascii_case("yes") && !trimmed.eq_ignore_ascii_case("y") {
            println!("Cancelled.");
            return Ok(());
        }
        println!();
    }

    let t0 = Instant::now();
    const WARMUP: usize = 50;

    // ── SP19 (2026-05-09) — producer-side multi-horizon reward blend ──
    // Path (B) of the multi-horizon reward augmentation spec. We blend
    // 1-bar / 5-bar / 30-bar log-returns into `tgt[1]` at fxcache write
    // time so kernel consumers see a single reward signal (no kernel
    // changes). `1/sqrt(N)` vol-scale correction inside the blend keeps
    // each horizon's log-return at unit-volatility-equivalent before
    // weighting under random-walk assumption.
    //
    // Why hardcoded weights here? `precompute_features` runs BEFORE
    // training so the ISV bus isn't initialised when the blend happens.
    // Slots [507..510) (`REWARD_HORIZON_WEIGHT_*BAR_INDEX`) are
    // reservations for a future Path (A) refactor that bumps TARGET_DIM
    // to carry per-horizon log-returns through the kernel pipeline; in
    // that world the consumer reads ISV at training-step time. For the
    // empirical hypothesis test ("does multi-horizon blend lift WR?")
    // equal-thirds is sufficient since varying weights requires fxcache
    // regen (~10-15 min on L40S) and per-batch adaptation is impossible
    // without the TARGET_DIM bump.
    //
    // Sentinel match: `SENTINEL_REWARD_HORIZON_WEIGHT_DEFAULT` in
    // `crates/ml/src/cuda_pipeline/sp14_isv_slots.rs` equals
    // `1.0_f32 / 3.0_f32 = 0.333_333_343`. The producer here uses f64
    // `1.0 / 3.0` for numerical accuracy in the blend itself; the f32
    // ISV slot value is what a future consumer reads, and matches
    // bit-identically when cast.
    const LOOKAHEAD_HORIZON_MAX: usize = 30;
    const SP19_HORIZON_BLEND_WEIGHTS: [f64; 3] = [
        1.0_f64 / 3.0_f64,
        1.0_f64 / 3.0_f64,
        1.0_f64 / 3.0_f64,
    ];

    // ── Early exit if cache already exists AND matches current version ────────
    let hex_key_early = ml::feature_cache::calculate_dbn_cache_key_full(
        &data_dir,
        mbp10_dir.as_deref(),
        trades_dir.as_deref(),
        &opts.symbol,
        &opts.data_source,
    ).context("Failed to compute cache key")?;
    let early_check_path = output_dir.join(format!("{hex_key_early}.fxcache"));
    if early_check_path.exists() {
        match ml::fxcache::load_fxcache(&early_check_path) {
            Ok(cached) if cached.has_ofi || mbp10_dir.is_none() => {
                // has_ofi flag is necessary but not sufficient — verify actual OFI data is non-zero
                let ofi_nonzero = cached.ofi.iter().filter(|r| r.iter().any(|&v| v != 0.0)).count();
                if mbp10_dir.is_some() && ofi_nonzero == 0 {
                    println!("Cache has_ofi=true but OFI data is all zeros ({} bars) — deleting stale cache",
                        cached.ofi.len());
                    std::fs::remove_file(&early_check_path).ok();
                } else {
                    let size_mb = std::fs::metadata(&early_check_path)
                        .map(|m| m.len() as f64 / 1_048_576.0)
                        .unwrap_or(0.0);
                    println!("Cache already exists: {} ({:.1} MB, has_ofi={}, ofi_nonzero={}) — skipping extraction",
                        early_check_path.display(), size_mb, cached.has_ofi, ofi_nonzero);
                    return Ok(());
                }
            }
            Ok(_) => {
                println!("Cache exists but has_ofi=false while MBP-10 data available — deleting stale cache");
                std::fs::remove_file(&early_check_path).ok();
            }
            Err(e) => {
                println!("Cache exists but failed validation: {e} — deleting stale cache");
                std::fs::remove_file(&early_check_path).ok();
            }
        }
    }

    // ── Step 1: Build volume bars from trades data ────────────────────────────
    let t1 = Instant::now();

    // Use trades_data_dir if provided, otherwise try data_dir for trades
    let trades_source = trades_dir.as_ref().unwrap_or(&data_dir);

    info!("Loading trades from {} (symbol={})...", trades_source.display(), opts.symbol);
    let mut trade_files = collect_dbn_files_filtered(trades_source, Some(&opts.symbol));
    if trade_files.is_empty() {
        // Fallback: try data_dir if trades_dir was specified but had no files
        if trades_dir.is_some() {
            trade_files = collect_dbn_files_filtered(&data_dir, Some(&opts.symbol));
        }
        if trade_files.is_empty() {
            anyhow::bail!(
                "No trades DBN files found for symbol '{}' in {} or {}",
                opts.symbol,
                trades_source.display(),
                data_dir.display()
            );
        }
    }
    trade_files.sort();
    info!("Found {} trades/OHLCV DBN files for symbol '{}'", trade_files.len(), opts.symbol);

    // Load trades PER FILE and filter front-month within each file.
    // Each quarterly DBN file has a different front-month contract (e.g. ESH24, ESM24).
    // Filtering globally would select only ONE quarter's contract.
    let mut all_front_month_trades: Vec<ml::features::DbnTrade> = Vec::new();
    let mut total_raw = 0_usize;
    for file in &trade_files {
        match ml::features::load_trades_sync(file) {
            Ok(trades) => {
                let raw_count = trades.len();
                total_raw += raw_count;
                let filtered = ml::features::filter_front_month(&trades);
                info!("  {} -> {} trades, {} front-month",
                    file.file_name().unwrap_or_default().to_string_lossy(),
                    raw_count, filtered.len());
                all_front_month_trades.extend(filtered);
            }
            Err(e) => {
                tracing::warn!("  Failed {:?}: {e}", file.file_name());
            }
        }
    }
    all_front_month_trades.sort_by_key(|t| t.timestamp);
    info!("Total trades: {} raw, {} front-month", total_raw, all_front_month_trades.len());

    // Build volume bars
    let all_bars = ml::features::build_volume_bars(&all_front_month_trades, ml::features::DEFAULT_VOLUME_BAR_SIZE);
    info!("Volume bars built: {} bars ({} contracts/bar) in {:.1}s",
        all_bars.len(), ml::features::DEFAULT_VOLUME_BAR_SIZE, t1.elapsed().as_secs_f64());

    // SP19 Path (B) (2026-05-09): need at least `WARMUP + LOOKAHEAD_HORIZON_MAX + 1`
    // bars so the 30-bar log-return at `i = 0` (`all_bars[WARMUP + 30]`) is
    // valid AND we still produce at least one output row.
    if all_bars.len() < WARMUP + LOOKAHEAD_HORIZON_MAX + 1 {
        anyhow::bail!(
            "Insufficient data: {} volume bars, need at least {} (WARMUP={WARMUP} + LOOKAHEAD_HORIZON_MAX={LOOKAHEAD_HORIZON_MAX} + 1)",
            all_bars.len(), WARMUP + LOOKAHEAD_HORIZON_MAX + 1
        );
    }

    // ── Price continuity check (regression gate) ────────────────────────
    let mut max_jump_pct = 0.0_f64;
    let mut jump_count = 0_usize;
    for w in all_bars.windows(2) {
        let pct = ((w[1].close - w[0].close) / w[0].close).abs() * 100.0;
        if pct > 0.5 {
            jump_count += 1;
        }
        max_jump_pct = max_jump_pct.max(pct);
    }
    info!("Price continuity: max_jump={:.3}%, jumps>0.5%={}/{}", max_jump_pct, jump_count, all_bars.len() - 1);
    if jump_count as f64 / all_bars.len() as f64 > 0.01 {
        anyhow::bail!(
            "PRICE CONTINUITY FAILED: {} of {} bars have >0.5% price jump — likely interleaved contracts",
            jump_count, all_bars.len()
        );
    }

    // ── Step 2: Extract 42-dim feature vectors ──────────────────────────────
    let t1 = Instant::now();
    info!("Extracting features...");
    let feature_vectors = extract_features_from_bars(&all_bars)
        .context("Feature extraction failed")?;
    info!("Extracted {} feature vectors in {:.1}s", feature_vectors.len(), t1.elapsed().as_secs_f64());

    // ── Step 3: Build targets [preproc_close, preproc_next, raw_close, raw_next, raw_open, mid_price_open]
    //   [0:1] = log-return-normalized close/next (network input; matches contract
    //           in experience_kernels.cu:1556 + cuda_pipeline/mod.rs:508)
    //   [2:3] = raw dollar prices for portfolio simulation P&L + tx costs
    //   [4]   = raw open price
    //   [5]   = mid-price at bar open (MBP-10 midpoint; fallback to raw_open below)
    //
    // Prior implementation wrote raw_curr/raw_next to slots [0:1] in violation of
    // the documented contract — the network-input columns ended up containing raw
    // prices (~$5000+ for ES futures). Fixed here so the fxcache matches its own
    // documented schema.
    //
    // SP19 Path (B) (2026-05-09): `tgt[1]` (preproc_next) now holds the
    // multi-horizon-blended log-return rather than the 1-bar log-return.
    // The blend mixes 1-bar / 5-bar / 30-bar log-returns at equal-thirds
    // weights with `1/sqrt(N)` vol-scale correction. The valid range
    // shrinks by `LOOKAHEAD_HORIZON_MAX` bars (we need
    // `all_bars[i + WARMUP + 30]` for the 30-bar log-return).
    let n = feature_vectors.len().saturating_sub(LOOKAHEAD_HORIZON_MAX);
    let features: Vec<[f64; 42]> = feature_vectors[..n].to_vec();
    let mut targets: Vec<[f64; 6]> = Vec::with_capacity(n);
    for i in 0..n {
        let raw_curr = all_bars[i + WARMUP].close;
        let raw_next = all_bars[i + WARMUP + 1].close;
        let raw_open = all_bars[i + WARMUP].open;
        // prev_close: 1-bar lag for log-return computation. WARMUP ≥ 50 so
        // all_bars[i + WARMUP - 1] is always valid.
        let prev_close = all_bars[i + WARMUP - 1].close;
        let preproc_close = if prev_close > 0.0 {
            (raw_curr / prev_close).ln()
        } else { 0.0 };
        // ── SP19 Path (B) — multi-horizon blended log-return ──
        // 1-bar / 5-bar / 30-bar log-returns with `1/sqrt(N)` vol-scale
        // correction applied INSIDE the blend (before weighting). Tail
        // bars are excluded by the `n = len - LOOKAHEAD_HORIZON_MAX`
        // trim above so `all_bars[i + WARMUP + 30]` is always valid.
        let raw_5bar  = all_bars[i + WARMUP +  5].close;
        let raw_30bar = all_bars[i + WARMUP + 30].close;
        let log_return_1bar  = if raw_curr > 0.0 { (raw_next  / raw_curr).ln() } else { 0.0 };
        let log_return_5bar  = if raw_curr > 0.0 { (raw_5bar  / raw_curr).ln() / (5.0_f64).sqrt()  } else { 0.0 };
        let log_return_30bar = if raw_curr > 0.0 { (raw_30bar / raw_curr).ln() / (30.0_f64).sqrt() } else { 0.0 };
        let preproc_next = SP19_HORIZON_BLEND_WEIGHTS[0] * log_return_1bar
                         + SP19_HORIZON_BLEND_WEIGHTS[1] * log_return_5bar
                         + SP19_HORIZON_BLEND_WEIGHTS[2] * log_return_30bar;
        // mid_price_open will be filled from MBP-10 data below if available;
        // default to raw_open (OHLCV-only fallback).
        targets.push([preproc_close, preproc_next, raw_curr, raw_next, raw_open, raw_open]);
    }

    // ── Step 3b: Build per-bar timestamps ─────────────────────────────────
    let timestamps: Vec<i64> = (0..n)
        .map(|i| all_bars[i + WARMUP].timestamp.timestamp_nanos_opt().unwrap_or(0))
        .collect();

    // ── Step 4: Compute OFI from MBP-10 + trades ────────────────────────────
    const OFI_DIM: usize = ml_core::state_layout::OFI_DIM;
    let t2 = Instant::now();
    let ofi: Vec<[f64; OFI_DIM]> = if let Some(ref mbp10_path) = mbp10_dir {
        use ml::features::mbp10_loader::load_ofi_features_parallel;
        use ml::features::trades_loader::load_trades_sync;
        use ml::features::ofi_calculator::OFICalculator;

        info!("Loading MBP-10 snapshots from {}...", mbp10_path.display());
        let mut mbp10_files = collect_dbn_files_recursive(mbp10_path);
        mbp10_files.sort();

        if mbp10_files.is_empty() {
            info!("No MBP-10 files found, OFI will be zeros");
            vec![[0.0; OFI_DIM]; n]
        } else {
            // Load MBP-10 snapshots (parallel)
            use data::providers::databento::dbn_parser::DbnParser;
            let per_file: Vec<Vec<_>> = {
                use rayon::prelude::*;
                mbp10_files.par_iter().filter_map(|file| {
                    let parser = match DbnParser::new() {
                        Ok(p) => p,
                        Err(e) => { tracing::warn!("Parser init failed: {e}"); return None; }
                    };
                    let mut snapshots = Vec::new();
                    match parser.parse_mbp10_streaming(file, 100, |snap| {
                        snapshots.push(snap.clone());
                    }) {
                        Ok(_) => {
                            info!("  {} -> {} snapshots", file.file_name().unwrap_or_default().to_string_lossy(), snapshots.len());
                            Some(snapshots)
                        }
                        Err(e) => {
                            tracing::warn!("  Failed {:?}: {e}", file.file_name());
                            None
                        }
                    }
                }).collect()
            };

            let mut all_snapshots = Vec::new();
            for snaps in per_file { all_snapshots.extend(snaps); }
            all_snapshots.sort_by_key(|s| s.timestamp);
            info!("Loaded {} MBP-10 snapshots in {:.1}s", all_snapshots.len(), t2.elapsed().as_secs_f64());

            // Load trades (parallel)
            let all_trades = if let Some(ref tdir) = trades_dir {
                let mut trade_files = collect_dbn_files_recursive(tdir);
                trade_files.sort();
                let per_file_trades: Vec<Vec<_>> = {
                    use rayon::prelude::*;
                    trade_files.par_iter().filter_map(|path| {
                        match load_trades_sync(path) {
                            Ok(t) => { info!("  {} trades from {:?}", t.len(), path.file_name()); Some(t) }
                            Err(e) => { tracing::warn!("  Failed {:?}: {e}", path.file_name()); None }
                        }
                    }).collect()
                };
                let mut trades = Vec::new();
                for t in per_file_trades { trades.extend(t); }
                trades.sort_by_key(|t| t.timestamp);
                if trades.is_empty() { None } else { Some(trades) }
            } else {
                None
            };

            // Compute per-bar OFI
            use ml::features::trades_loader::get_trades_for_bar;
            use ml::features::ofi_calculator::MicrostructureState;
            let mut calculator = OFICalculator::new();
            let mut ofi_per_bar = Vec::with_capacity(n);

            for i in 0..n {
                let bar = &all_bars[i + WARMUP];
                // Bar timestamp = close-time of bar (volume bars: last trade in bucket;
                // see `crates/ml-features/src/trades_loader.rs:124-176`). The OFI window
                // for bar t is therefore the *formation interval*
                // `(close(bar_{t-1}), close(bar_t)]` — strictly historical relative to
                // the policy's decision time at bar t. The previous implementation used
                // `[close(bar_t), close(bar_{t+1}))` which leaks bar t+1's microstructure
                // (audit `docs/lookahead-bias-audit-2026-04-28.md` §3, 31/32 OFI dims).
                // Mirrors the backward-looking convention already used by
                // `log_bar_duration` below at lines 567-575.
                let bar_ts = bar.timestamp.timestamp_nanos_opt().unwrap_or(0) as u64;
                let bar_start_ts = if i + WARMUP >= 1 {
                    all_bars[i + WARMUP - 1].timestamp.timestamp_nanos_opt().unwrap_or(0) as u64
                } else {
                    // No prior bar — emit empty window. log_bar_duration uses the same
                    // sentinel pathway (60s default) for bar 0; here we simply skip
                    // population so OFI[0] stays zero (consistent with existing
                    // first-bar delta handling at lines 558-562).
                    bar_ts
                };
                let bar_duration_ns = bar_ts.saturating_sub(bar_start_ts);

                let mut micro_state = MicrostructureState::new(bar_start_ts, bar_duration_ns);

                // Feed trades for this bar window (OFICalculator + MicrostructureState).
                // Window is half-open at the start, closed at the end:
                // `(close(bar_{t-1}), close(bar_t)]` so trade-at-close (which formed bar t)
                // is included.
                if let Some(ref trades) = all_trades {
                    for trade in get_trades_for_bar(trades, bar_start_ts, bar_ts.saturating_add(1)) {
                        calculator.feed_trade(trade.price, trade.volume, trade.is_buy);
                        micro_state.update_trade(trade.price, trade.volume, trade.is_buy, trade.timestamp);
                    }
                }

                // Get ALL MBP-10 snapshots within this bar window for tick-level features.
                // Binary search for bar range: (bar_start_ts, bar_ts] — formation interval.
                let snap_start = all_snapshots.partition_point(|s| s.timestamp <= bar_start_ts);
                let snap_end = all_snapshots.partition_point(|s| s.timestamp <= bar_ts);
                let bar_snapshots = &all_snapshots[snap_start..snap_end];

                // Feed every snapshot to MicrostructureState for tick-level resolution
                for snap in bar_snapshots {
                    micro_state.update_snapshot(snap);
                }

                // Use the LAST snapshot in the bar for OFI calculation (matches prior behavior)
                let last_snap = bar_snapshots.last()
                    .or_else(|| {
                        // Fallback: nearest snapshot at/after bar_ts (prior behavior)
                        all_snapshots.get(snap_start)
                    });

                if let Some(snap) = last_snap {
                    match calculator.calculate(snap) {
                        Ok(f) if f.is_valid() => {
                            // Feed OFI-derived values to MicrostructureState
                            micro_state.update_ofi_derived(f.ofi_level1, f.vpin);

                            let arr8 = f.to_array();
                            let micro_12 = micro_state.snapshot();
                            // v5 OFI layout (see state_layout.rs OFI slot map):
                            //   [0..8)    raw OFI (8 features)
                            //   [8..16)   lag-1 deltas (filled in post-loop)
                            //   [16]      book_aggression (filled below)
                            //   [17]      log_bar_duration (filled in post-loop)
                            //   [18]      ofi_acceleration (micro_12[10])
                            //   [19]      toxicity_gradient (micro_12[11])
                            //   [20..30)  MicrostructureState::snapshot()[0..10]
                            //   [30]      order_count_imbalance ((Σbid_ct − Σask_ct) / Σ(bid_ct+ask_ct))
                            //   [31]      microprice_residual ((weighted_mid − mid) / mid)
                            let mut ofi_row = [0.0_f64; OFI_DIM];
                            ofi_row[..8].copy_from_slice(&arr8);
                            ofi_row[18] = micro_12[10]; // ofi_acceleration
                            ofi_row[19] = micro_12[11]; // toxicity_gradient
                            for k in 0..10 {
                                ofi_row[20 + k] = micro_12[k];
                            }
                            // TLOB-novel slots: derive directly from Mbp10Snapshot counts/prices.
                            // order_count_imbalance: count-based analog to depth_imbalance.
                            let (bid_ct_sum, ask_ct_sum) = snap.levels.iter().fold((0u64, 0u64), |(b, a), l| {
                                (b + l.bid_ct as u64, a + l.ask_ct as u64)
                            });
                            let total_ct = bid_ct_sum + ask_ct_sum;
                            let order_count_imbalance = if total_ct > 0 {
                                (bid_ct_sum as f64 - ask_ct_sum as f64) / total_ct as f64
                            } else {
                                0.0
                            };
                            ofi_row[30] = order_count_imbalance.clamp(-1.0, 1.0);
                            // microprice_residual: (weighted_mid − mid) / mid; dimensionless.
                            // Mbp10Snapshot::weighted_mid_price() uses top-of-book volume-weighted mid.
                            let mid = snap.mid_price();
                            let wmid = snap.weighted_mid_price();
                            let microprice_residual = if mid > 0.0 && mid.is_finite() && wmid.is_finite() {
                                ((wmid - mid) / mid).clamp(-0.01, 0.01)
                            } else {
                                0.0
                            };
                            ofi_row[31] = microprice_residual;
                            ofi_per_bar.push(ofi_row);
                        }
                        _ => ofi_per_bar.push([0.0; OFI_DIM]),
                    }
                } else {
                    ofi_per_bar.push([0.0; OFI_DIM]);
                }

                // ── Book aggression: center-of-mass asymmetry from MBP-10 depth ──
                // Uses the last snapshot in the bar window (same as OFI calculation above)
                if let Some(snap) = bar_snapshots.last() {
                    let mut buy_weighted_sum = 0.0_f64;
                    let mut buy_total = 0.0_f64;
                    let mut sell_weighted_sum = 0.0_f64;
                    let mut sell_total = 0.0_f64;
                    for (level, pair) in snap.levels.iter().enumerate().take(10) {
                        let bid_size = pair.bid_sz as f64;
                        let ask_size = pair.ask_sz as f64;
                        buy_weighted_sum += (level + 1) as f64 * bid_size;
                        buy_total += bid_size;
                        sell_weighted_sum += (level + 1) as f64 * ask_size;
                        sell_total += ask_size;
                    }
                    let buy_com = if buy_total > 0.0 { buy_weighted_sum / buy_total } else { 5.5 };
                    let sell_com = if sell_total > 0.0 { sell_weighted_sum / sell_total } else { 5.5 };
                    let book_aggression = (sell_com - buy_com) / 10.0; // normalize to [-1, 1]
                    if let Some(last) = ofi_per_bar.last_mut() {
                        last[16] = book_aggression;
                    }
                }
            }

            // ── OFI temporal deltas: delta[bar] = ofi[bar] - ofi[bar-1] ──
            // Computed after all bars so we have the full sequence
            for i in (1..ofi_per_bar.len()).rev() {
                for k in 0..8 {
                    ofi_per_bar[i][8 + k] = ofi_per_bar[i][k] - ofi_per_bar[i - 1][k];
                }
            }
            // First bar has no previous — deltas are zero (already initialized)

            // ── Log bar duration: ln(duration_secs) / 10.0 ──
            for i in 0..n {
                let duration_secs = if i > 0 {
                    let ts_cur = all_bars[i + WARMUP].timestamp;
                    let ts_prev = all_bars[i + WARMUP - 1].timestamp;
                    (ts_cur - ts_prev).num_seconds() as f64
                } else {
                    60.0 // default 1 minute for first bar
                };
                let log_duration = duration_secs.max(0.1).ln() / 10.0;
                ofi_per_bar[i][17] = log_duration;
            }

            let delta_nonzero = ofi_per_bar.iter().filter(|f| f[8..16].iter().any(|&v| v != 0.0)).count();
            let book_nonzero = ofi_per_bar.iter().filter(|f| f[16] != 0.0).count();
            let micro_nonzero = ofi_per_bar.iter().filter(|f| f[20..30].iter().any(|&v| v != 0.0)).count();
            let tlob_nonzero = ofi_per_bar.iter().filter(|f| f[30] != 0.0 || f[31] != 0.0).count();
            info!("OFI v5: deltas_nonzero={}/{}, book_aggression_nonzero={}/{}, microstructure_nonzero={}/{}, tlob_novel_nonzero={}/{}",
                delta_nonzero, n, book_nonzero, n, micro_nonzero, n, tlob_nonzero, n);

            // Fill targets[i][5] with MBP-10 midpoint at bar open (mid_price_open)
            let mut mid_fill_count = 0_usize;
            for i in 0..n {
                let bar = &all_bars[i + WARMUP];
                let bar_ts = bar.timestamp.timestamp_nanos_opt().unwrap_or(0) as u64;

                // Find first MBP-10 snapshot at or after bar open
                let snap_idx = all_snapshots.partition_point(|s| s.timestamp < bar_ts);
                if let Some(snap) = all_snapshots.get(snap_idx) {
                    let mid = snap.mid_price();
                    if mid > 0.0 {
                        targets[i][5] = mid;
                        mid_fill_count += 1;
                    }
                }
            }
            info!("mid_price_open filled from MBP-10: {}/{} bars", mid_fill_count, n);

            let non_zero = ofi_per_bar.iter().filter(|f| f.iter().any(|&v| v != 0.0)).count();
            info!("OFI computed: {} bars, {} non-zero ({:.1}%) in {:.1}s",
                ofi_per_bar.len(), non_zero,
                if ofi_per_bar.is_empty() { 0.0 } else { non_zero as f64 / ofi_per_bar.len() as f64 * 100.0 },
                t2.elapsed().as_secs_f64());
            ofi_per_bar
        }
    } else {
        info!("No MBP-10 directory, OFI will be zeros (log_bar_duration still computed)");
        let mut ofi_per_bar = vec![[0.0_f64; OFI_DIM]; n];
        // Even without MBP-10 data, compute log_bar_duration from bar timestamps
        for i in 0..n {
            let duration_secs = if i > 0 {
                let ts_cur = all_bars[i + WARMUP].timestamp;
                let ts_prev = all_bars[i + WARMUP - 1].timestamp;
                (ts_cur - ts_prev).num_seconds() as f64
            } else {
                60.0 // default 1 minute for first bar
            };
            let log_duration = duration_secs.max(0.1).ln() / 10.0;
            ofi_per_bar[i][17] = log_duration;
        }
        ofi_per_bar
    };

    let total_len = n;

    // ── Compute cache key ────────────────────────────────────────────────────
    let hex_key = ml::feature_cache::calculate_dbn_cache_key_full(
        &data_dir,
        mbp10_dir.as_deref(),
        trades_dir.as_deref(),
        &opts.symbol,
        &opts.data_source,
    ).context("Failed to compute cache key")?;
    let cache_key: [u8; 32] = hex::decode(&hex_key)
        .context("Invalid hex key")?
        .try_into()
        .map_err(|_| anyhow::anyhow!("Cache key wrong length"))?;

    // ── Normalize features (z-score) ──────────────────────────────────────────
    // Raw features contain OHLCV prices (up to 25,000). Normalize once at
    // precompute time so every consumer
    // (training, hyperopt, inference) gets consistent normalized features.
    let norm_stats = ml::walk_forward::NormStats::from_features(&features);
    let features = norm_stats.normalize_batch(&features);
    info!("Features z-score normalized ({} bars × 42 dims)", features.len());

    // Defence-in-depth gate: never write a poisoned fxcache to disk. If anything
    // upstream produced an out-of-bounds value, fail loudly here rather than
    // letting it silently propagate to every future training run that reads
    // this cache. Same gate the fxcache loader and DBN fallback enforce.
    ml::walk_forward::validate_normalized_features(&features, "precompute_features writer")?;

    // ── Write .fxcache ───────────────────────────────────────────────────────
    std::fs::create_dir_all(&output_dir)
        .with_context(|| format!("Failed to create output directory: {}", output_dir.display()))?;

    let output_path = output_dir.join(format!("{hex_key}.fxcache"));

    // Save NormStats alongside cache for inference denormalization
    let norm_path = output_dir.join(format!("{hex_key}.norm_stats.json"));
    let norm_json = serde_json::to_string_pretty(&norm_stats)
        .context("Failed to serialize NormStats")?;
    std::fs::write(&norm_path, &norm_json)
        .with_context(|| format!("Failed to write NormStats: {}", norm_path.display()))?;
    info!("NormStats saved to {}", norm_path.display());

    let t2 = Instant::now();
    info!("Writing .fxcache to {}...", output_path.display());

    let has_ofi = mbp10_dir.is_some();
    let bytes_written = ml::fxcache::write_fxcache(
        &output_path,
        &features,
        &targets,
        &ofi,
        &timestamps,
        cache_key,
        has_ofi,
    )
    .context("Failed to write .fxcache file")?;

    let write_secs = t2.elapsed().as_secs_f64();
    let total_secs = t0.elapsed().as_secs_f64();

    // ── Summary ──────────────────────────────────────────────────────────────
    println!();
    println!("================================================================================");
    println!("PRECOMPUTE SUMMARY");
    println!("================================================================================");
    println!();
    println!("Bars:          {}", total_len);
    println!("Features:      42-dim");
    println!("Targets:       6-dim");
    println!("OFI:           {}-dim ({})", OFI_DIM, if has_ofi { "from MBP-10" } else { "zero-padded" });
    println!("Format:        f32 (v{}), OFI_DIM={}", ml::fxcache::FXCACHE_VERSION, ml::fxcache::OFI_DIM);
    println!("Cache key:     {}", hex_key);
    println!("Output:        {}", output_path.display());
    println!("Size:          {:.2} MB", bytes_written as f64 / 1_048_576.0);
    println!("Write time:    {:.1}s", write_secs);
    println!("Total time:    {:.1}s", total_secs);
    println!();

    Ok(())
}