foxhunt/crates/ml/src/fxcache.rs

//! FxCache — Flat Binary Feature Cache for Zero-Overhead GPU Loading
//!
//! Provides a compact binary format for pre-computed DQN training data
//! (features + targets + OFI vectors) designed for direct GPU upload
//! without parsing overhead.
//!
//! ## Format
//!
//! ```text
//! ┌─────────────────────────────────────────────────────────────┐
//! │ FxCacheHeader (72 bytes)                                    │
//! │  magic               [u8; 8]   = b"FXCACHE\0"               │
//! │  version             u16       = 6 (f32 + schema-hash)      │
//! │  feat_dim            u16       = 42                         │
//! │  target_dim          u16       = 6                          │
//! │  ofi_dim             u16       = 32                         │
//! │  bar_count           u64                                    │
//! │  cache_key           [u8; 32]  (SHA256 raw bytes)           │
//! │  feature_schema_hash u64       (FNV-1a from build.rs)       │
//! │  reserved            [u8; 8]                                │
//! └─────────────────────────────────────────────────────────────┘
//! │ Body (bar_count records)                                    │
//! │  Each record starts with an i64 timestamp (ns).             │
//! │  Version 6: [i64 ts][80 × f32] = 328 bytes/bar              │
//! └─────────────────────────────────────────────────────────────┘
//! ```

use std::collections::HashMap;
use std::path::{Path, PathBuf};
use std::sync::Arc;

use anyhow::{anyhow, bail, Context, Result};
use arrow::array::{Array, FixedSizeBinaryArray, FixedSizeBinaryBuilder, Int64Array};
use arrow::datatypes::{DataType, Field, Schema};
use arrow::ipc::reader::FileReader;
use arrow::ipc::writer::FileWriter;
use arrow::record_batch::RecordBatch;
use tracing::{debug, info};

// ── Constants ────────────────────────────────────────────────────────────────

/// Magic bytes identifying an FxCache file.
const FXCACHE_MAGIC: [u8; 8] = *b"FXCACHE\0";

/// Header size in bytes (fixed).
const HEADER_SIZE: usize = 72;

/// Cache format version. Bump on ANY *wire-format* change (header layout,
/// body record shape). Schema-level changes (feature column semantics,
/// dimensionality) are tracked automatically by `FEATURE_SCHEMA_HASH` —
/// callers do NOT need to bump this constant when they edit
/// `features/extraction.rs` or `ml-core::state_layout`.
/// Stale cache files with wrong version OR wrong schema hash are
/// auto-detected and rejected by `validate()`. The ensure-fxcache Argo step
/// catches the error and regenerates.
/// v5: OFI_DIM 20→32 — adds 10 MicrostructureState slots (ofi_trajectory,
///     realized_variance, hawkes_intensity, book_pressure, spread_dynamics,
///     aggression_ratio, queue_depletion_asymmetry, order_count_flux,
///     intra_bar_momentum, regime_score) + 2 TLOB-novel slots
///     (order_count_imbalance, microprice_residual).
/// v6: header grows 64→72 bytes, adds `feature_schema_hash: u64` between
///     `cache_key` and `reserved`. Stamped at write time from the
///     compile-time `FEATURE_SCHEMA_HASH` const (built from the bytes of
///     extraction.rs / fxcache.rs / state_layout.rs by `build.rs`).
///     Recover-from-stale path is identical: validate() bails, Argo regens.
/// v7: OFI window alignment fix — OFI features at bar t now use the formation
///     interval `(close(bar_{t-1}), close(bar_t)]` instead of the leaked
///     `[close(bar_t), close(bar_{t+1}))`. Wire-format unchanged; semantic
///     contract changed. Per audit `docs/lookahead-bias-audit-2026-04-28.md`
///     §3 the previous layout leaked bar t+1's microstructure into 31/32 OFI
///     dims at bar t. Caches written before this fix carry contaminated OFI
///     and MUST be regenerated. Strict-checked via `validate()`.
/// v8: Target column semantic fix — slots [0:1] (`preproc_close`/`preproc_next`)
///     now hold log-return-normalized values per the documented contract in
///     `experience_kernels.cu:1556` and `cuda_pipeline/mod.rs:508`. Prior writers
///     (`precompute_features.rs:360`, `data_loading.rs:510`) stored raw prices
///     in all 6 target slots, leaving the network-input columns at ~$5000+ ES
///     futures price level instead of unit-scale log returns. Wire-format
///     unchanged; semantic contract enforced. Caches written before this fix
///     carry raw prices in `preproc_*` columns and MUST be regenerated.
/// v9: SP19 Path (B) producer-side multi-horizon reward blend.
///     `tgt[1]` (`preproc_next`) now holds a 1-bar / 5-bar / 30-bar
///     log-return blend at equal-thirds weights with `1/sqrt(N)`
///     vol-scale correction, instead of the prior 1-bar log-return
///     alone. Producers (`precompute_features.rs`, `data_loading.rs`)
///     trim the final `LOOKAHEAD_HORIZON_MAX = 30` bars from the dataset
///     since the 30-bar log-return needs `close[i + 30]`. Wire-format
///     unchanged (`TARGET_DIM = 6`); semantic contract for `tgt[1]` is
///     a different scalar value composition. Caches written before this
///     fix carry the 1-bar-only `preproc_next` and MUST be regenerated.
///     Per `feedback_no_partial_refactor` both producer call sites
///     change atomically; kernel consumers of `tgt[1]` are unchanged.
// alpha (FoxhuntQ-Δ Phase 1c, 2026-05-14): switched on-disk format from the
// custom 72-byte-header + flat-binary body to **Apache Arrow IPC** with
// schema-in-file. Added optional `alpha_features: FixedSizeBinary(134×4)`
// column for the modern feature stack consumed by ml-alpha. v9-only readers
// (DQN) continue to work — they read the unchanged `ts_ns + record` columns
// and ignore `alpha_features`. Bump from 9 → 10 forces stale-cache regeneration.
pub const FXCACHE_VERSION: u16 = 10;

/// Compile-time fingerprint over the feature-schema source files. Bumps
/// automatically whenever `features/extraction.rs`, `fxcache.rs`, or
/// `ml-core::state_layout` change (any byte — whitespace counts).
///
/// Encoded into the fxcache header at write time and strict-checked at load
/// time, so caches built against a different schema fail validation and
/// trigger automatic regeneration via `precompute_features`. This removes
/// the manual "remember to bump `FXCACHE_VERSION` on schema change" ritual.
///
/// The actual value is set by `build.rs` (FNV-1a 64-bit). Stable across
/// rust versions and machines (unlike `std::hash::DefaultHasher`).
pub const FEATURE_SCHEMA_HASH: u64 = {
    // build.rs emits the hash as a decimal u64 string; `from_str_radix` is
    // const since rust 1.83 and the workspace MSRV is 1.85.
    match u64::from_str_radix(env!("FEATURE_SCHEMA_HASH"), 10) {
        Ok(v) => v,
        Err(_) => panic!("build.rs emitted invalid u64 for FEATURE_SCHEMA_HASH"),
    }
};

/// Feature vector dimensionality.
const FEAT_DIM: usize = 42;

/// Target vector dimensionality. Mirrors the named column constants below.
///
/// Public so consumer kernels and host-side readers can index by name rather
/// than literal stride. Two latent bugs converged in the cancelled 50-epoch
/// run `train-multi-seed-p5qzw` because consumers hardcoded stride 4 (left
/// over from pre-`063fd2716`) and column 0 (pre-`5a5dd0fed` raw_close, post-
/// fix preproc_close). Centralizing the layout here so future renames force
/// every consumer to update at the same call site.
pub const TARGET_DIM: usize = 6;

/// Column 0: `preproc_close` — log-return-normalized close (network input).
pub const TARGET_PREPROC_CLOSE: usize = 0;

/// Column 1: `preproc_next` — log-return-normalized next-bar close.
pub const TARGET_PREPROC_NEXT: usize = 1;

/// Column 2: `raw_close` — raw close price for portfolio simulation P&L + tx.
pub const TARGET_RAW_CLOSE: usize = 2;

/// Column 3: `raw_next` — raw next-bar close. **FUTURE information** — must
/// not be used for any reward, P&L, equity, or portfolio computation. Kept
/// in the layout for legacy back-compat; consumers should `(void)` this slot
/// to suppress unused-element warnings.
pub const TARGET_RAW_NEXT: usize = 3;

/// Column 4: `raw_open` — raw open price (OHLCV).
pub const TARGET_RAW_OPEN: usize = 4;

/// Column 5: `mid_price_open` — MBP-10 midpoint at bar open; falls back to
/// `raw_open` when no MBP-10 snapshot is available at the bar boundary.
pub const TARGET_MID_OPEN: usize = 5;

/// OFI vector dimensionality — mirrors `ml_core::state_layout::OFI_DIM`
/// (20 legacy slots + 12 new microstructure slots).
pub const OFI_DIM: usize = ml_core::state_layout::OFI_DIM;

/// Alpha feature block dimensionality — the 2026 SOTA feature stack authored in
/// `ml-features` (Blocks A-W combined).
///
/// Decomposition:
/// - Blocks A-E (factory wired, replacing old TA features): 50 dims
/// - Blocks F-W (new authoring in ml-features): 84 dims
/// - Total: 134 dims per bar
///
/// Stored as an **additional Arrow column** alongside the existing
/// `record` (features+targets+ofi) column. DQN consumers ignore this column
/// (they read `record` only); ml-alpha consumers read this column for the
/// FoxhuntQ-Δ Phase 1c smoke. **Additive design**: no breaking change to the
/// v9 schema; v9-only readers and writers continue to work.
pub const ALPHA_FEATURE_DIM: usize = 134;

/// Total f64 values per record: features + targets + OFI = 42 + 6 + 20 = 68.
const RECORD_F64_COUNT: usize = FEAT_DIM + TARGET_DIM + OFI_DIM;

/// Total f32 values per record (same count as f64, no padding needed).
const RECORD_F32_COUNT: usize = RECORD_F64_COUNT;

// ── Header ───────────────────────────────────────────────────────────────────

/// 72-byte fixed header for `.fxcache` files (v6+).
#[derive(Debug, Clone)]
pub struct FxCacheHeader {
    /// Magic bytes: `b"FXCACHE\0"`.
    pub magic: [u8; 8],
    /// Format version — see `FXCACHE_VERSION`.
    pub version: u16,
    /// Feature dimension (42).
    pub feat_dim: u16,
    /// Target dimension (6).
    pub target_dim: u16,
    /// OFI dimension (see `OFI_DIM`).
    pub ofi_dim: u16,
    /// Number of bars (records) in the file.
    pub bar_count: u64,
    /// SHA256 cache key (raw 32 bytes).
    pub cache_key: [u8; 32],
    /// Compile-time feature-schema fingerprint — see `FEATURE_SCHEMA_HASH`.
    /// Mismatch on load means the cache was built against different feature
    /// extraction / state-layout / fxcache-format source than the current
    /// binary. Strict-checked in `validate()`; regen via `precompute_features`.
    pub feature_schema_hash: u64,
    /// Reserved for future use.
    pub reserved: [u8; 8],
}

impl FxCacheHeader {
    /// Create a new header with the given parameters.
    pub fn new(version: u16, bar_count: u64, cache_key: [u8; 32], has_ofi: bool) -> Self {
        let mut reserved = [0u8; 8];
        reserved[0] = if has_ofi { 1 } else { 0 };
        Self {
            magic: FXCACHE_MAGIC,
            version,
            feat_dim: FEAT_DIM as u16,
            target_dim: TARGET_DIM as u16,
            ofi_dim: OFI_DIM as u16,
            bar_count,
            cache_key,
            feature_schema_hash: FEATURE_SCHEMA_HASH,
            reserved,
        }
    }

    /// Validate header integrity.
    ///
    /// Strictly enforces the current `FXCACHE_VERSION` AND the current
    /// `FEATURE_SCHEMA_HASH`. Any older version (including v4 with
    /// OFI_DIM=20, v5 without schema-hash) or any cache built against a
    /// different feature-extractor / state-layout source is rejected —
    /// regenerate via `precompute_features`.
    pub fn validate(&self) -> Result<()> {
        if self.magic != FXCACHE_MAGIC {
            bail!(
                "Invalid FxCache magic: expected {:?}, got {:?}",
                FXCACHE_MAGIC,
                self.magic
            );
        }
        if self.version != FXCACHE_VERSION {
            bail!(
                "Stale FxCache version: {} (expected {}). Delete and regenerate.",
                self.version, FXCACHE_VERSION
            );
        }
        if self.feat_dim as usize != FEAT_DIM {
            bail!(
                "Feature dimension mismatch: expected {}, got {}",
                FEAT_DIM,
                self.feat_dim
            );
        }
        let td = self.target_dim as usize;
        if td != TARGET_DIM {
            bail!(
                "Target dimension mismatch: expected {}, got {}",
                TARGET_DIM,
                td
            );
        }
        if self.ofi_dim as usize != OFI_DIM {
            bail!(
                "OFI dimension mismatch: expected {}, got {}",
                OFI_DIM,
                self.ofi_dim
            );
        }
        if self.feature_schema_hash != FEATURE_SCHEMA_HASH {
            bail!(
                "Stale FxCache feature schema: hash {:#018x} (expected {:#018x}). \
                 Source files defining feature extraction / state layout / \
                 fxcache format have changed since this cache was built. \
                 Delete and regenerate via precompute_features.",
                self.feature_schema_hash, FEATURE_SCHEMA_HASH
            );
        }
        if self.bar_count == 0 {
            bail!("FxCache bar_count is zero — empty cache files are not valid");
        }
        Ok(())
    }

}

// ── Data ─────────────────────────────────────────────────────────────────────

/// In-memory representation of an FxCache file's contents.
#[derive(Debug)]
pub struct FxCacheData {
    /// Per-bar timestamps (nanoseconds since Unix epoch).
    pub timestamps: Vec<i64>,
    /// Feature vectors, one per bar (42 elements each).
    pub features: Vec<[f64; FEAT_DIM]>,
    /// Target vectors, one per bar (6 elements each).
    pub targets: Vec<[f64; TARGET_DIM]>,
    /// OFI vectors, one per bar (`OFI_DIM` elements each).
    pub ofi: Vec<[f64; OFI_DIM]>,
    /// SHA256 cache key (raw 32 bytes).
    pub cache_key: [u8; 32],
    /// Number of bars.
    pub bar_count: usize,
    /// Explicit flag: true if OFI was computed from real MBP-10 data.
    /// False means OFI is zero-filled (no MBP-10 data was available during precompute).
    pub has_ofi: bool,
    /// Alpha feature block — `ALPHA_FEATURE_DIM` (134) elements per bar.
    ///
    /// `Some(rows)` when the fxcache file contains the alpha column (added in
    /// FoxhuntQ-Δ Phase 1c). `None` for legacy v9-only writes or files
    /// generated before alpha features were authored. ml-alpha consumers
    /// branch on this; DQN consumers ignore it.
    pub alpha_features: Option<Vec<Vec<f32>>>,
}

// ── Writer ───────────────────────────────────────────────────────────────────

/// Write feature/target/OFI data to an `.fxcache` binary file.
///
/// # Arguments
///
/// * `path`       — Output file path (parent directories are created automatically)
/// * `features`   — Slice of 42-element feature vectors
/// * `targets`    — Slice of 6-element target vectors
/// * `ofi`        — Slice of `OFI_DIM`-element OFI vectors
/// * `timestamps` — Per-bar timestamps (nanoseconds since Unix epoch)
/// * `cache_key`  — SHA256 key (raw 32 bytes)
/// * `has_ofi`    — If true, OFI was computed from real MBP-10 data; false means zero-filled
///
/// # Returns
///
/// Total bytes written (header + body).
/// Write fxcache data via **Arrow IPC** to `.arrow` file format.
///
/// FXCACHE_VERSION=10: replaces the custom 72-byte-header + flat-binary body format with
/// Apache Arrow IPC. Schema/dim metadata is embedded in the Arrow schema
/// (no off-by-N alignment risk), and the wire format is inspectable from
/// Python/pandas/polars/R.
///
/// File layout:
/// - Standard Arrow IPC `.arrow` file framing (magic + length-prefixed schema
///   + record batches + footer)
/// - Schema with two fields:
///   - `ts_ns: Int64` — per-bar timestamps
///   - `record: FixedSizeBinary(N)` — row blob of f32 LE bytes
///     `[feat_dim×f32][target_dim×f32][ofi_dim×f32]`
/// - Schema metadata HashMap (all values as Strings, version+dims+cache_key
///   hex + has_ofi + feature_schema_hash hex)
///
/// # Arguments / Returns — unchanged from the prior custom-binary writer.
pub fn write_fxcache(
    path: &Path,
    features: &[[f64; FEAT_DIM]],
    targets: &[[f64; TARGET_DIM]],
    ofi: &[[f64; OFI_DIM]],
    timestamps: &[i64],
    cache_key: [u8; 32],
    has_ofi: bool,
    alpha_features: Option<&[Vec<f32>]>,
) -> Result<u64> {
    let bar_count = features.len();
    if targets.len() != bar_count || ofi.len() != bar_count || timestamps.len() != bar_count {
        bail!(
            "Length mismatch: features={}, targets={}, ofi={}, timestamps={}",
            bar_count,
            targets.len(),
            ofi.len(),
            timestamps.len()
        );
    }
    if bar_count == 0 {
        bail!("Cannot write empty FxCache (0 bars)");
    }
    // Alpha-features column accepts a variable feature width — the dim is
    // inferred from row 0 and written to metadata so the reader can recover
    // it. This lets the same fxcache schema carry the 134-dim bar-level alpha
    // stack OR the 81-dim per-snapshot stack without a parallel format.
    let alpha_dim_opt: Option<usize> = if let Some(alpha) = alpha_features {
        if alpha.is_empty() {
            bail!("alpha_features provided but empty (must match bar_count)");
        }
        if alpha.len() != bar_count {
            bail!(
                "Alpha features row count {} != bar_count {}",
                alpha.len(),
                bar_count
            );
        }
        let dim = alpha[0].len();
        if dim == 0 {
            bail!("Alpha row 0 has zero features — refusing to write empty-dim column");
        }
        for (i, row) in alpha.iter().enumerate() {
            if row.len() != dim {
                bail!(
                    "Alpha row {i} has {} features, expected {} (inferred from row 0)",
                    row.len(),
                    dim
                );
            }
        }
        Some(dim)
    } else {
        None
    };

    if let Some(parent) = path.parent() {
        std::fs::create_dir_all(parent)
            .with_context(|| format!("Failed to create parent dirs for {:?}", path))?;
    }

    // Build row blobs: each bar is `4 × (feat_dim + target_dim + ofi_dim)` bytes
    // of little-endian f32. The fixed-size binary type guarantees alignment
    // and saves us from any per-row size-prefix overhead.
    let record_bytes_per_bar = 4 * (FEAT_DIM + TARGET_DIM + OFI_DIM);
    let mut blob_builder =
        FixedSizeBinaryBuilder::with_capacity(bar_count, record_bytes_per_bar as i32);
    let mut row_buf = vec![0_u8; record_bytes_per_bar];
    for i in 0..bar_count {
        let mut off = 0;
        for &v in &features[i] {
            row_buf[off..off + 4].copy_from_slice(&(v as f32).to_le_bytes());
            off += 4;
        }
        for &v in &targets[i] {
            row_buf[off..off + 4].copy_from_slice(&(v as f32).to_le_bytes());
            off += 4;
        }
        for &v in &ofi[i] {
            row_buf[off..off + 4].copy_from_slice(&(v as f32).to_le_bytes());
            off += 4;
        }
        blob_builder
            .append_value(&row_buf)
            .with_context(|| format!("append fxcache row {i}"))?;
    }
    let blob_array = blob_builder.finish();
    let ts_array = Int64Array::from(timestamps.to_vec());

    // Build the optional alpha_features column. Each row is ALPHA_FEATURE_DIM × f32 LE
    // packed into a FixedSizeBinary blob. We use a separate Arrow column rather
    // than extending the existing `record` column so v9-only consumers (DQN
    // data loader) keep working unchanged — they read `record`, ignore
    // `alpha_features` entirely.
    let alpha_blob_array_opt = if let (Some(alpha), Some(dim)) = (alpha_features, alpha_dim_opt) {
        let alpha_blob_bytes = dim * 4;
        let mut alpha_builder =
            FixedSizeBinaryBuilder::with_capacity(bar_count, alpha_blob_bytes as i32);
        let mut alpha_row_buf = vec![0_u8; alpha_blob_bytes];
        for (i, row) in alpha.iter().enumerate() {
            let mut off = 0;
            for &val in row {
                alpha_row_buf[off..off + 4].copy_from_slice(&val.to_le_bytes());
                off += 4;
            }
            alpha_builder
                .append_value(&alpha_row_buf)
                .with_context(|| format!("append alpha row {i}"))?;
        }
        Some(alpha_builder.finish())
    } else {
        None
    };

    // Schema metadata: everything that used to live in the 72-byte header now
    // becomes named String key-values in the Arrow schema. Multi-language
    // readers (pandas, polars, R) can introspect this directly.
    let mut meta: HashMap<String, String> = HashMap::new();
    meta.insert("fxcache_version".to_owned(), FXCACHE_VERSION.to_string());
    meta.insert("feat_dim".to_owned(), FEAT_DIM.to_string());
    meta.insert("target_dim".to_owned(), TARGET_DIM.to_string());
    meta.insert("ofi_dim".to_owned(), OFI_DIM.to_string());
    meta.insert("has_ofi".to_owned(), has_ofi.to_string());
    meta.insert("cache_key_hex".to_owned(), hex_encode_32(&cache_key));
    meta.insert(
        "feature_schema_hash".to_owned(),
        format!("{FEATURE_SCHEMA_HASH:016x}"),
    );
    if let Some(dim) = alpha_dim_opt {
        meta.insert("alpha_feature_dim".to_owned(), dim.to_string());
    }

    // Schema: ts_ns + record (always), optionally + alpha_features
    let mut fields = vec![
        Field::new("ts_ns", DataType::Int64, false),
        Field::new(
            "record",
            DataType::FixedSizeBinary(record_bytes_per_bar as i32),
            false,
        ),
    ];
    let mut batch_columns: Vec<Arc<dyn Array>> = vec![Arc::new(ts_array), Arc::new(blob_array)];
    if let (Some(alpha_arr), Some(dim)) = (alpha_blob_array_opt, alpha_dim_opt) {
        fields.push(Field::new(
            "alpha_features",
            DataType::FixedSizeBinary((dim * 4) as i32),
            false,
        ));
        batch_columns.push(Arc::new(alpha_arr));
    }

    let schema = Schema::new_with_metadata(fields, meta);
    let schema_arc = Arc::new(schema);

    let batch = RecordBatch::try_new(schema_arc.clone(), batch_columns)
        .context("build fxcache RecordBatch")?;

    let file = std::fs::File::create(path)
        .with_context(|| format!("Failed to create FxCache file {:?}", path))?;
    let mut writer = FileWriter::try_new(file, schema_arc.as_ref())
        .context("create Arrow IPC FileWriter")?;
    writer.write(&batch).context("write Arrow batch")?;
    writer.finish().context("finish Arrow IPC writer")?;

    let total_bytes = std::fs::metadata(path)
        .with_context(|| format!("Failed to stat written fxcache {:?}", path))?
        .len();

    info!(
        "FxCache (Arrow IPC) written: {} bars, v{} (feat={} target={} ofi={}), {:.2} MB -> {:?}",
        bar_count,
        FXCACHE_VERSION,
        FEAT_DIM,
        TARGET_DIM,
        OFI_DIM,
        total_bytes as f64 / 1_048_576.0,
        path
    );

    Ok(total_bytes)
}

// ── Reader ───────────────────────────────────────────────────────────────────

/// Load an `.fxcache` (Arrow IPC) file into memory.
///
/// Reads the Arrow IPC schema (schema metadata replaces the v9 72-byte
/// header), validates dims + version + feature_schema_hash against the
/// current compile-time constants, then materializes all batches into the
/// `FxCacheData` row-oriented in-memory layout (preserves the existing
/// DQN data-loader contract).
pub fn load_fxcache(path: &Path) -> Result<FxCacheData> {
    let file = std::fs::File::open(path)
        .with_context(|| format!("Failed to open FxCache file {:?}", path))?;
    let mut reader = FileReader::try_new(file, None)
        .context("open Arrow IPC reader for FxCache")?;
    let schema = reader.schema();
    let meta = schema.metadata();

    let version: u16 = meta
        .get("fxcache_version")
        .ok_or_else(|| anyhow!("FxCache: missing schema metadata 'fxcache_version'"))?
        .parse()
        .context("parse fxcache_version")?;
    if version != FXCACHE_VERSION {
        bail!(
            "Stale FxCache version: {} (expected {}). Delete and regenerate.",
            version, FXCACHE_VERSION
        );
    }
    let feat_dim: usize = meta
        .get("feat_dim")
        .ok_or_else(|| anyhow!("FxCache: missing 'feat_dim'"))?
        .parse()
        .context("parse feat_dim")?;
    let target_dim: usize = meta
        .get("target_dim")
        .ok_or_else(|| anyhow!("FxCache: missing 'target_dim'"))?
        .parse()
        .context("parse target_dim")?;
    let ofi_dim: usize = meta
        .get("ofi_dim")
        .ok_or_else(|| anyhow!("FxCache: missing 'ofi_dim'"))?
        .parse()
        .context("parse ofi_dim")?;
    let has_ofi: bool = meta
        .get("has_ofi")
        .ok_or_else(|| anyhow!("FxCache: missing 'has_ofi'"))?
        .parse()
        .context("parse has_ofi")?;
    let feature_schema_hash: u64 = u64::from_str_radix(
        meta.get("feature_schema_hash")
            .ok_or_else(|| anyhow!("FxCache: missing 'feature_schema_hash'"))?,
        16,
    )
    .context("parse feature_schema_hash hex")?;

    if feat_dim != FEAT_DIM {
        bail!("FxCache feat_dim mismatch: expected {}, got {}", FEAT_DIM, feat_dim);
    }
    if target_dim != TARGET_DIM {
        bail!(
            "FxCache target_dim mismatch: expected {}, got {}",
            TARGET_DIM, target_dim
        );
    }
    if ofi_dim != OFI_DIM {
        bail!("FxCache ofi_dim mismatch: expected {}, got {}", OFI_DIM, ofi_dim);
    }
    if feature_schema_hash != FEATURE_SCHEMA_HASH {
        bail!(
            "Stale FxCache feature schema: hash {:#018x} (expected {:#018x}). \
             Source files defining feature extraction / state layout / fxcache format \
             have changed since this cache was built. Delete and regenerate via precompute_features.",
            feature_schema_hash, FEATURE_SCHEMA_HASH
        );
    }

    let cache_key_hex = meta
        .get("cache_key_hex")
        .ok_or_else(|| anyhow!("FxCache: missing 'cache_key_hex'"))?;
    let cache_key = hex_decode_32(cache_key_hex)?;

    // Detect whether the file contains the alpha features column (added in
    // FoxhuntQ-Δ Phase 1c). Determined by the presence of the
    // `alpha_feature_dim` schema metadata key. v9-only files (or alpha files
    // written without alpha features) don't have this key — load_fxcache
    // returns `alpha_features: None`.
    let alpha_dim_opt: Option<usize> = if meta.contains_key("alpha_feature_dim") {
        let declared: usize = meta
            .get("alpha_feature_dim")
            .ok_or_else(|| anyhow!("FxCache: alpha_feature_dim missing (guard race)"))?
            .parse()
            .context("parse alpha_feature_dim")?;
        if declared == 0 {
            bail!("FxCache alpha_feature_dim = 0 (degenerate)");
        }
        Some(declared)
    } else {
        None
    };
    let has_alpha_features = alpha_dim_opt.is_some();

    // Decode all batches into row-oriented FxCacheData
    let feat_byte_count = FEAT_DIM * 4;
    let target_byte_count = TARGET_DIM * 4;
    let ofi_byte_count = OFI_DIM * 4;
    let expected_blob_size = feat_byte_count + target_byte_count + ofi_byte_count;
    let alpha_blob_size = alpha_dim_opt.map(|d| d * 4).unwrap_or(0);
    let alpha_dim_decoded = alpha_dim_opt.unwrap_or(0);

    let mut timestamps: Vec<i64> = Vec::new();
    let mut features: Vec<[f64; FEAT_DIM]> = Vec::new();
    let mut targets: Vec<[f64; TARGET_DIM]> = Vec::new();
    let mut ofi: Vec<[f64; OFI_DIM]> = Vec::new();
    let mut alpha_features: Option<Vec<Vec<f32>>> = if has_alpha_features { Some(Vec::new()) } else { None };

    for batch_result in reader.by_ref() {
        let batch = batch_result.context("read Arrow batch")?;
        let ts_arr = batch
            .column(0)
            .as_any()
            .downcast_ref::<Int64Array>()
            .ok_or_else(|| anyhow!("FxCache column 0 should be Int64, got {:?}", batch.column(0).data_type()))?;
        let blob_arr = batch
            .column(1)
            .as_any()
            .downcast_ref::<FixedSizeBinaryArray>()
            .ok_or_else(|| {
                anyhow!(
                    "FxCache column 1 should be FixedSizeBinary, got {:?}",
                    batch.column(1).data_type()
                )
            })?;
        // Optional alpha_features column (index 2). If has_alpha_features but the
        // column is missing, we treat this as a corrupt file (metadata claims
        // alpha but column absent).
        let alpha_arr_opt = if has_alpha_features {
            if batch.num_columns() < 3 {
                bail!(
                    "FxCache claims alpha_feature_dim but RecordBatch has {} columns (expected 3)",
                    batch.num_columns()
                );
            }
            Some(
                batch
                    .column(2)
                    .as_any()
                    .downcast_ref::<FixedSizeBinaryArray>()
                    .ok_or_else(|| {
                        anyhow!(
                            "FxCache alpha_features column should be FixedSizeBinary, got {:?}",
                            batch.column(2).data_type()
                        )
                    })?,
            )
        } else {
            None
        };

        for i in 0..batch.num_rows() {
            timestamps.push(ts_arr.value(i));
            let blob: &[u8] = blob_arr.value(i);
            if blob.len() != expected_blob_size {
                bail!(
                    "FxCache row blob size {} != expected {} (feat+target+ofi × 4 bytes)",
                    blob.len(),
                    expected_blob_size
                );
            }

            let mut feat = [0.0_f64; FEAT_DIM];
            for (j, slot) in feat.iter_mut().enumerate() {
                let off = j * 4;
                *slot = f32::from_le_bytes([blob[off], blob[off + 1], blob[off + 2], blob[off + 3]])
                    as f64;
            }
            features.push(feat);

            let mut tgt = [0.0_f64; TARGET_DIM];
            for (j, slot) in tgt.iter_mut().enumerate() {
                let off = feat_byte_count + j * 4;
                *slot = f32::from_le_bytes([blob[off], blob[off + 1], blob[off + 2], blob[off + 3]])
                    as f64;
            }
            targets.push(tgt);

            let mut ofi_row = [0.0_f64; OFI_DIM];
            for (j, slot) in ofi_row.iter_mut().enumerate() {
                let off = feat_byte_count + target_byte_count + j * 4;
                *slot = f32::from_le_bytes([blob[off], blob[off + 1], blob[off + 2], blob[off + 3]])
                    as f64;
            }
            ofi.push(ofi_row);

            // Decode the alpha_features row if the column is present.
            if let (Some(alpha_arr), Some(alpha_dst)) = (alpha_arr_opt, alpha_features.as_mut()) {
                let alpha_blob: &[u8] = alpha_arr.value(i);
                if alpha_blob.len() != alpha_blob_size {
                    bail!(
                        "FxCache alpha_features row blob size {} != expected {} (alpha_feature_dim × 4)",
                        alpha_blob.len(),
                        alpha_blob_size
                    );
                }
                let mut alpha_row = Vec::with_capacity(alpha_dim_decoded);
                for j in 0..alpha_dim_decoded {
                    let off = j * 4;
                    alpha_row.push(f32::from_le_bytes([
                        alpha_blob[off],
                        alpha_blob[off + 1],
                        alpha_blob[off + 2],
                        alpha_blob[off + 3],
                    ]));
                }
                alpha_dst.push(alpha_row);
            }
        }
    }

    let bar_count = features.len();
    if bar_count == 0 {
        bail!("FxCache contains no bars");
    }

    info!(
        "FxCache (Arrow IPC) loaded: {} bars, v{} from {:?} (alpha_features={})",
        bar_count,
        version,
        path,
        alpha_features.is_some()
    );

    debug!(
        "FxCache cache_key={} feat_dim={} target_dim={} ofi_dim={} has_ofi={} alpha_features={}",
        cache_key_hex,
        FEAT_DIM,
        TARGET_DIM,
        OFI_DIM,
        has_ofi,
        alpha_features.is_some()
    );

    Ok(FxCacheData {
        timestamps,
        features,
        targets,
        ofi,
        cache_key,
        bar_count,
        has_ofi,
        alpha_features,
    })
}

// ── Cache-key hex helpers ────────────────────────────────────────────────────

fn hex_encode_32(bytes: &[u8; 32]) -> String {
    bytes.iter().map(|b| format!("{b:02x}")).collect()
}

fn hex_decode_32(s: &str) -> Result<[u8; 32]> {
    if s.len() != 64 {
        bail!(
            "FxCache cache_key_hex must be 64 chars (32 bytes hex-encoded), got {}",
            s.len()
        );
    }
    let mut out = [0_u8; 32];
    for (i, byte_chars) in s.as_bytes().chunks_exact(2).take(32).enumerate() {
        let hex_str = std::str::from_utf8(byte_chars)
            .map_err(|_| anyhow!("Non-UTF-8 in cache_key_hex at byte {i}"))?;
        out[i] = u8::from_str_radix(hex_str, 16)
            .map_err(|_| anyhow!("Invalid hex in cache_key_hex: '{hex_str}' at byte {i}"))?;
    }
    Ok(out)
}

// ── Finder ───────────────────────────────────────────────────────────────────

#[cfg(test)]
mod target_layout_tests {
    use super::*;

    /// Compile-time guard: the named target columns are dense and exhaust
    /// the layout. If a future patch reorders or removes a slot, this fires.
    #[test]
    fn target_columns_dense_and_exhaustive() {
        let cols = [
            TARGET_PREPROC_CLOSE,
            TARGET_PREPROC_NEXT,
            TARGET_RAW_CLOSE,
            TARGET_RAW_NEXT,
            TARGET_RAW_OPEN,
            TARGET_MID_OPEN,
        ];
        for (i, &c) in cols.iter().enumerate() {
            assert_eq!(c, i, "Column constant out of order at index {i}");
        }
        assert_eq!(cols.len(), TARGET_DIM);
    }
}

#[cfg(test)]
mod has_ofi_tests {
    use super::*;

    #[test]
    fn test_has_ofi_roundtrip_true() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("test_ofi_true.fxcache");
        let features = vec![[1.0_f64; 42]; 10];
        let targets = vec![[0.0_f64; 6]; 10];
        let ofi = vec![[0.5_f64; OFI_DIM]; 10];
        let timestamps = vec![1_i64; 10];
        let key = [0u8; 32];

        write_fxcache(&path, &features, &targets, &ofi, &timestamps, key, true, None).unwrap();
        let loaded = load_fxcache(&path).unwrap();
        assert!(loaded.has_ofi, "has_ofi should be true");
        assert_eq!(loaded.bar_count, 10);
        assert!(loaded.alpha_features.is_none(), "no alpha column when None passed");
    }

    #[test]
    fn test_has_ofi_roundtrip_false() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("test_ofi_false.fxcache");
        let features = vec![[1.0_f64; 42]; 10];
        let targets = vec![[0.0_f64; 6]; 10];
        let ofi = vec![[0.0_f64; OFI_DIM]; 10];
        let timestamps = vec![1_i64; 10];
        let key = [0u8; 32];

        write_fxcache(&path, &features, &targets, &ofi, &timestamps, key, false, None).unwrap();
        let loaded = load_fxcache(&path).unwrap();
        assert!(!loaded.has_ofi, "has_ofi should be false");
        assert!(loaded.alpha_features.is_none());
    }

    #[test]
    fn test_alpha_features_roundtrip() {
        let dir = tempfile::tempdir().unwrap();
        let path = dir.path().join("test_alpha.fxcache");
        let n = 5_usize;
        let features = vec![[1.0_f64; 42]; n];
        let targets = vec![[0.0_f64; 6]; n];
        let ofi = vec![[0.0_f64; OFI_DIM]; n];
        let timestamps = vec![1_i64; n];
        let key = [0u8; 32];
        let alpha: Vec<Vec<f32>> = (0..n)
            .map(|i| (0..ALPHA_FEATURE_DIM).map(|j| (i * 1000 + j) as f32).collect())
            .collect();

        write_fxcache(&path, &features, &targets, &ofi, &timestamps, key, true, Some(&alpha)).unwrap();
        let loaded = load_fxcache(&path).unwrap();
        assert_eq!(loaded.bar_count, n);
        let alpha_loaded = loaded.alpha_features.expect("alpha_features should be Some after writing with Some(&alpha)");
        assert_eq!(alpha_loaded.len(), n);
        for (i, row) in alpha_loaded.iter().enumerate() {
            assert_eq!(row.len(), ALPHA_FEATURE_DIM);
            for (j, &val) in row.iter().enumerate() {
                let expected = (i * 1000 + j) as f32;
                assert!(
                    (val - expected).abs() < 1e-6,
                    "alpha row {i} col {j}: expected {expected}, got {val}"
                );
            }
        }
    }
}

/// Find an `.fxcache` file by hex-encoded cache key in a cache directory.
///
/// Looks for a file named `<hex_key>.fxcache` in `cache_dir`.
///
/// # Arguments
///
/// * `cache_dir` — Directory to search
/// * `cache_key` — Raw 32-byte SHA256 cache key
///
/// # Returns
///
/// `Some(path)` if the file exists, `None` otherwise.
pub fn find_fxcache(cache_dir: &Path, cache_key: &[u8; 32]) -> Option<PathBuf> {
    let hex_key = hex::encode(cache_key);
    let candidate = cache_dir.join(format!("{hex_key}.fxcache"));
    if candidate.exists() {
        debug!("FxCache found: {:?}", candidate);
        Some(candidate)
    } else {
        debug!("FxCache miss: {:?}", candidate);
        None
    }
}

/// Resolve fxcache directory: explicit override > env var > walk-up sibling.
pub fn resolve_cache_dir(data_dir: &Path, override_dir: Option<&Path>) -> Option<PathBuf> {
    // 1. Explicit override (CLI --feature-cache-dir)
    if let Some(dir) = override_dir {
        if dir.exists() {
            return Some(dir.to_path_buf());
        }
    }

    // 2. Environment variable
    if let Ok(dir) = std::env::var("FOXHUNT_FEATURE_CACHE_DIR") {
        let p = PathBuf::from(dir);
        if p.exists() {
            return Some(p);
        }
    }

    // 3. Walk up from data_dir to find sibling feature-cache/
    let mut dir = data_dir;
    loop {
        if let Some(parent) = dir.parent() {
            let candidate = parent.join("feature-cache");
            if candidate.exists() {
                return Some(candidate);
            }
            if parent == dir {
                break;
            }
            dir = parent;
        } else {
            break;
        }
    }

    None
}

/// Returns the path to the `{hex_key}.norm_stats.json` file that sits
/// alongside the `.fxcache` for the given cache key.
///
/// `precompute_features` writes this file at the same time as the fxcache
/// (see `precompute_features.rs:607`). Supervised training writes a per-fold
/// copy into the output dir for evaluation; RL training should use this
/// helper to produce the same per-fold copies (`norm_stats_fold{N}.json`)
/// so `evaluate_baseline` can find them. Returns `None` if the cache
/// directory can't be resolved.
pub fn norm_stats_path_for_key(
    data_dir: &Path,
    cache_dir_override: Option<&Path>,
    cache_key: &[u8; 32],
) -> Option<PathBuf> {
    let cache_dir = resolve_cache_dir(data_dir, cache_dir_override)?;
    let hex_key = hex::encode(cache_key);
    Some(cache_dir.join(format!("{hex_key}.norm_stats.json")))
}

/// Discover and load an fxcache file. Single source of truth for all callers.
///
/// Cache dir priority: `cache_dir_override` > `FOXHUNT_FEATURE_CACHE_DIR` env > walk-up sibling.
/// Strict key match only — returns `None` on miss. No "most recent" fallback.
///
/// # Arguments
/// * `data_dir` — Base data directory (e.g. `test_data/futures-baseline`)
/// * `symbol` — Trading symbol (e.g. `"ES.FUT"`)
/// * `mbp10_dir` — Optional MBP-10 order book data directory
/// * `trades_dir` — Optional trades data directory
/// * `data_source` — Data source mode (`"mbp10"` or `"ohlcv"`)
/// * `imbalance_bar_threshold` — Imbalance bar formation threshold (must match producer's value)
/// * `imbalance_bar_ewma_alpha` — Imbalance bar EWMA alpha (must match producer's value)
/// * `volume_bar_size` — Volume bar contracts/bar (must match producer's value)
/// * `cache_dir_override` — Explicit cache directory (from CLI `--feature-cache-dir`)
pub fn discover_and_load(
    data_dir: &Path,
    symbol: &str,
    mbp10_dir: Option<&Path>,
    trades_dir: Option<&Path>,
    data_source: &str,
    imbalance_bar_threshold: f64,
    imbalance_bar_ewma_alpha: f64,
    volume_bar_size: u64,
    cache_dir_override: Option<&Path>,
) -> Option<FxCacheData> {
    // 1. Resolve cache directory
    let cache_dir = resolve_cache_dir(data_dir, cache_dir_override)?;

    // 2. Compute cache key (includes symbol + data_source + bar params)
    let key_hex = crate::feature_cache::calculate_dbn_cache_key_full(
        data_dir, mbp10_dir, trades_dir, symbol, data_source,
        imbalance_bar_threshold, imbalance_bar_ewma_alpha, volume_bar_size,
    )
    .ok()?;
    let key: [u8; 32] = hex::decode(&key_hex).ok()?.try_into().ok()?;

    // 3. Strict key match — no fallback to "most recent"
    let path = find_fxcache(&cache_dir, &key)?;

    // 4. Load and return
    match load_fxcache(&path) {
        Ok(data) => {
            info!(
                "fxcache hit: {} bars, OFI={} from {:?}",
                data.bar_count, data.has_ofi, path
            );
            Some(data)
        }
        Err(e) => {
            tracing::warn!("fxcache load failed: {e}");
            None
        }
    }
}

// ── Tests ────────────────────────────────────────────────────────────────────

#[cfg(test)]
mod discover_tests {
    use super::*;

    #[test]
    fn test_discover_returns_none_for_nonexistent_dir() {
        let result = discover_and_load(
            Path::new("/nonexistent/path"),
            "ES.FUT",
            None,
            None,
            "mbp10",
            0.5,
            0.1,
            100,
            None,
        );
        assert!(result.is_none());
    }

    #[test]
    fn test_discover_returns_none_on_key_mismatch() {
        let dir = tempfile::tempdir().unwrap();
        let cache_dir = dir.path().join("feature-cache");
        std::fs::create_dir_all(&cache_dir).unwrap();

        // Write a cache file with a known dummy key
        let path = cache_dir.join(
            "0000000000000000000000000000000000000000000000000000000000000000.fxcache",
        );
        let features = vec![[1.0_f64; 42]; 5];
        let targets = vec![[0.0_f64; 6]; 5];
        let ofi = vec![[0.0_f64; OFI_DIM]; 5];
        let timestamps = vec![1_i64; 5];
        write_fxcache(&path, &features, &targets, &ofi, &timestamps, [0u8; 32], false, None)
            .unwrap();

        // Try to discover with a real data_dir (different key) — should NOT match
        let result = discover_and_load(
            Path::new("test_data/futures-baseline"),
            "ES.FUT",
            None,
            None,
            "mbp10",
            0.5,
            0.1,
            100,
            Some(&cache_dir),
        );
        // Strict match only — no "most recent" fallback
        assert!(result.is_none(), "Wrong key should not match (strict mode)");
    }

    #[test]
    fn test_resolve_cache_dir_explicit_override() {
        let dir = tempfile::tempdir().unwrap();
        let result = resolve_cache_dir(Path::new("/some/data"), Some(dir.path()));
        assert_eq!(result, Some(dir.path().to_path_buf()));
    }

    #[test]
    fn test_resolve_cache_dir_nonexistent_override_falls_through() {
        let result = resolve_cache_dir(
            Path::new("/some/data"),
            Some(Path::new("/nonexistent/override")),
        );
        // Falls through to env var / walk-up (both will fail here)
        assert!(result.is_none());
    }
}