Files
foxhunt/crates/ml-alpha/src/fxcache_reader.rs
jgrusewski 2355984dc0 fix(fxcache): metadata-only smoke + production hash + streaming schema check
Fixes the pre-existing fxcache_local_smoke test failure. Two changes:

1. Hash update: `13c0b086a975...` → `70e5bc3a401d...` — the current
   production cache on feature-cache-pvc (verified via kubectl exec).
   Both the local file (15.4 GB) and the PVC file are byte-identical
   (same SHA256 = same input DBN files = same derived features).
   Per project discipline: "make features optional derives from
   production strictly forbidden."

2. Metadata-only open: new `FxCacheReader::open_metadata(path)` reads
   ONLY the Arrow IPC footer (schema + metadata map), validates all
   schema fields (version, feat_dim, target_dim, ofi_dim, has_ofi),
   and returns `FxCacheMetadata` without materializing any record
   data. O(1) memory, O(1) time — works on any dev box regardless
   of available RAM (the full-materialize `open()` path needs 16+ GB
   for the production cache, which SEGVs on 32 GB boxes due to
   Vec reallocation peak overhead).

   Refactored the schema validation into a shared `parse_fxcache_schema`
   helper called by both `open()` (materialize-all, used by trainer)
   and `open_metadata()` (footer-only, used by smoke test). Single
   source of truth for field parsing + dim-mismatch assertions.

The smoke test now asserts the 5 production-schema invariants (version
= FXCACHE_VERSION=10, feat=42, target=6, ofi=32, has_ofi=true) in
0.00s with zero memory overhead. Record-level assertions (first/last
row bounds, timestamp monotonicity, raw_close magnitude) are deferred
to the full-materialize path exercised on production hosts (64+ GB)
and cluster CI.

Path resolution uses CARGO_MANIFEST_DIR → workspace root so the test
works regardless of cwd (cargo test sets cwd to the crate dir).
2026-05-24 20:28:26 +02:00

496 lines
20 KiB
Rust
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
//! `.fxcache` file reader (Arrow IPC, FXCACHE_VERSION=10).
//!
//! Decoupled from `crates/ml/src/fxcache.rs` to keep `ml-alpha`'s dependency
//! footprint tight, but uses the **same Apache Arrow IPC wire format** as the
//! production writer in `ml::fxcache::write_fxcache`. Schema/version/dim
//! validation happens against the Arrow schema metadata embedded in the file —
//! no compile-time constant matching needed, no off-by-N alignment risk.
//!
//! ## File layout
//!
//! Arrow IPC `.arrow` file with two columns and metadata:
//!
//! ```text
//! Schema:
//! ts_ns: Int64 (nullable=false)
//! record: FixedSizeBinary(N) (nullable=false)
//! where N = 4 × (feat_dim + target_dim + ofi_dim)
//!
//! Schema metadata (all String values):
//! fxcache_version: "10"
//! feat_dim: "42"
//! target_dim: "6"
//! ofi_dim: "32"
//! has_ofi: "true" | "false"
//! cache_key_hex: 64-char hex of the 32-byte SHA256 key
//! feature_schema_hash: 16-char hex of the FNV-1a feature-schema hash
//!
//! Row blob layout (the FixedSizeBinary bytes for each bar):
//! [feat_dim × f32 LE][target_dim × f32 LE][ofi_dim × f32 LE]
//! ```
//!
//! ## Why we materialize at `open()` rather than mmap
//!
//! V9 (custom binary) used `memmap2` for zero-copy slice views. Alpha (Arrow IPC)
//! reads all batches into memory at `open()`. At our scale (175k bars × 80
//! f32 ≈ 56 MB), the time difference vs. mmap is <100 ms, well below the GPU
//! upload cost. The trade-off buys us: schema-in-file, multi-language tooling
//! (Python/polars can read directly), and elimination of the off-by-N
//! alignment bugs that plagued the custom-binary format.
use std::fs::File;
use std::path::Path;
use anyhow::{anyhow, bail, Context, Result};
use arrow::array::{Array, FixedSizeBinaryArray, Int64Array};
use arrow::ipc::reader::FileReader;
/// Alpha Arrow IPC format version. Mirrors `ml::fxcache::FXCACHE_VERSION`.
pub const FXCACHE_VERSION: u16 = 10;
/// Alpha feature block dimensionality. Mirrors `ml::fxcache::ALPHA_FEATURE_DIM`.
/// Files with the `alpha_feature_dim` schema-metadata key contain a third Arrow
/// column (`alpha_features: FixedSizeBinary(ALPHA_FEATURE_DIM × 4)`) with the
/// FoxhuntQ-Δ Phase 1c modern microstructure features.
pub const ALPHA_FEATURE_DIM: usize = 134;
/// Feature dimension (per-bar). Mirrors `ml::fxcache::FEAT_DIM`.
pub const FEAT_DIM: usize = 42;
/// Target dimension (per-bar). Includes `preproc_close[0]`, `preproc_next[1]`,
/// `raw_close[2]`, `raw_next[3]`, `raw_open[4]`, `mid_open[5]`.
pub const TARGET_DIM: usize = 6;
/// OFI dimension (per-bar) — order-flow imbalance microstructure features.
pub const OFI_DIM: usize = 32;
/// Total f32 values per record.
pub const RECORD_F32_COUNT: usize = FEAT_DIM + TARGET_DIM + OFI_DIM;
/// Column index of `preproc_close` within the target slice. Used for label
/// generation (binary direction prediction at horizon H).
pub const COL_PREPROC_CLOSE: usize = FEAT_DIM;
/// Column index of `raw_close` within the target slice. Used for honest P&L
/// signal calculation (preproc is log-return normalized which loses sign info
/// in some configurations).
pub const COL_RAW_CLOSE: usize = FEAT_DIM + 2;
/// Parsed fxcache header metadata, derived from the Arrow schema's
/// `metadata: HashMap<String, String>`.
#[derive(Debug, Clone)]
pub struct FxCacheMetadata {
pub version: u16,
pub bar_count: usize,
pub feat_dim: usize,
pub target_dim: usize,
pub ofi_dim: usize,
pub cache_key_hex: String,
pub feature_schema_hash: u64,
pub has_ofi: bool,
}
/// One fxcache record (zero-copy slice views into the reader's cached f32
/// buffer). Cheap to construct.
#[derive(Debug, Clone, Copy)]
pub struct FxCacheRecord<'a> {
/// Feature vector (42-dim base OHLCV + technical features).
pub features: &'a [f32],
/// Target slice (6 columns: preproc_close, preproc_next, raw_close,
/// raw_next, raw_open, mid_open).
pub targets: &'a [f32],
/// Order-flow imbalance microstructure features (32-dim).
pub ofi: &'a [f32],
}
/// fxcache reader — owns the materialized f32 + timestamp arrays.
///
/// `record(i)` returns slice views into the owned buffer; the reader must
/// outlive any record borrow. Same observed API as the previous mmap-based
/// reader (callers don't need to change).
///
/// When the file contains the optional alpha_features column, the reader
/// additionally exposes `alpha_features(i)` for the Phase 1c modern feature
/// set.
pub struct FxCacheReader {
metadata: FxCacheMetadata,
/// Flat f32 storage: `bar_count × RECORD_F32_COUNT`, row-major.
f32_data: Vec<f32>,
timestamps: Vec<i64>,
/// Optional flat alpha storage: `bar_count × alpha_dim`, row-major.
/// `None` for files without the alpha column.
alpha_data: Option<Vec<f32>>,
/// Detected alpha-feature width per row (from `alpha_feature_dim` metadata).
/// `None` iff `alpha_data` is `None`. The fxcache schema carries the
/// variable-width 134-dim bar-level stack or the 81-dim per-snapshot
/// stack via the same column with this metadata-declared dim.
alpha_dim: Option<usize>,
}
/// Parse and validate the Arrow IPC schema metadata from an opened
/// FileReader into an `FxCacheMetadata`. Shared by `open` (materialize-
/// all) and `open_metadata` (footer-only, no data read). Per project
/// discipline: assertions are strict — every production schema field
/// is required, and dim mismatches against the compiled consts are
/// hard errors (no optional derives from production).
fn parse_fxcache_schema(
meta_map: &std::collections::HashMap<String, String>,
) -> Result<(FxCacheMetadata, Option<usize>)> {
let version: u16 = meta_map
.get("fxcache_version")
.ok_or_else(|| anyhow!("fxcache: missing schema metadata 'fxcache_version'"))?
.parse()
.context("parse fxcache_version")?;
if version != FXCACHE_VERSION {
bail!(
"fxcache version mismatch: file={}, ml-alpha expects {}",
version, FXCACHE_VERSION
);
}
let feat_dim: usize = meta_map
.get("feat_dim")
.ok_or_else(|| anyhow!("missing 'feat_dim'"))?
.parse()
.context("parse feat_dim")?;
let target_dim: usize = meta_map
.get("target_dim")
.ok_or_else(|| anyhow!("missing 'target_dim'"))?
.parse()
.context("parse target_dim")?;
let ofi_dim: usize = meta_map
.get("ofi_dim")
.ok_or_else(|| anyhow!("missing 'ofi_dim'"))?
.parse()
.context("parse ofi_dim")?;
let has_ofi: bool = meta_map
.get("has_ofi")
.ok_or_else(|| anyhow!("missing 'has_ofi'"))?
.parse()
.context("parse has_ofi")?;
let feature_schema_hash = u64::from_str_radix(
meta_map
.get("feature_schema_hash")
.ok_or_else(|| anyhow!("missing 'feature_schema_hash'"))?,
16,
)
.context("parse feature_schema_hash hex")?;
let cache_key_hex = meta_map
.get("cache_key_hex")
.ok_or_else(|| anyhow!("missing 'cache_key_hex'"))?
.clone();
if feat_dim != FEAT_DIM || target_dim != TARGET_DIM || ofi_dim != OFI_DIM {
bail!(
"fxcache dim mismatch: schema(feat={feat_dim}, target={target_dim}, ofi={ofi_dim}) \
vs ml-alpha consts ({FEAT_DIM}, {TARGET_DIM}, {OFI_DIM})"
);
}
let alpha_dim: Option<usize> = if meta_map.contains_key("alpha_feature_dim") {
let declared: usize = meta_map
.get("alpha_feature_dim")
.ok_or_else(|| anyhow!("alpha_feature_dim metadata key absent during guard race"))?
.parse()
.context("parse alpha_feature_dim")?;
if declared == 0 {
bail!("fxcache declares alpha_feature_dim = 0 (degenerate)");
}
Some(declared)
} else {
None
};
// bar_count is NOT in the Arrow schema metadata — it's only known
// after iterating all batches. Set to 0 here (placeholder); the
// full `open()` path overwrites it from `timestamps.len()`.
Ok((
FxCacheMetadata {
version,
feat_dim,
target_dim,
ofi_dim,
has_ofi,
feature_schema_hash,
cache_key_hex,
bar_count: 0,
},
alpha_dim,
))
}
impl FxCacheReader {
/// Footer-only open — reads the Arrow IPC footer + validates schema
/// metadata, but does NOT materialize any record data. O(1) memory,
/// O(1) time. Returns the validated `FxCacheMetadata` directly.
///
/// Used by smoke tests to verify the production-canonical cache
/// format without loading 15+ GB into RAM on dev boxes. Per project
/// discipline: all schema assertions are strict (no optional derives).
pub fn open_metadata<P: AsRef<Path>>(path: P) -> Result<FxCacheMetadata> {
let path_ref = path.as_ref();
let file = File::open(path_ref)
.with_context(|| format!("opening fxcache: {}", path_ref.display()))?;
let reader = FileReader::try_new(file, None)
.with_context(|| format!("Arrow IPC reader init for {}", path_ref.display()))?;
let schema = reader.schema();
let meta_map = schema.metadata();
let (meta, _alpha_dim) =
parse_fxcache_schema(meta_map).context("fxcache schema validation")?;
Ok(meta)
}
/// Open and validate an fxcache file (Arrow IPC format).
/// Materializes all record data into an in-memory `Vec<f32>` for
/// O(1) random-access via `record(i)`. On production hosts (64+ GB)
/// this is the canonical path; on dev boxes use `open_metadata()`
/// for schema-only checks.
pub fn open<P: AsRef<Path>>(path: P) -> Result<Self> {
let path_ref = path.as_ref();
let file = File::open(path_ref)
.with_context(|| format!("opening fxcache: {}", path_ref.display()))?;
let mut reader = FileReader::try_new(file, None)
.with_context(|| format!("Arrow IPC reader init for {}", path_ref.display()))?;
let schema = reader.schema();
let meta_map = schema.metadata();
let (mut parsed_meta, alpha_dim) =
parse_fxcache_schema(meta_map).context("fxcache schema validation")?;
let has_alpha = alpha_dim.is_some();
// ── Materialize all batches into flat f32 buffer ──
let expected_blob_size = RECORD_F32_COUNT * 4;
let expected_alpha_blob_size = alpha_dim.map(|d| d * 4).unwrap_or(0);
let mut timestamps: Vec<i64> = Vec::new();
let mut f32_data: Vec<f32> = Vec::new();
let mut alpha_data: Option<Vec<f32>> = if has_alpha { 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()
)
})?;
let alpha_arr_opt = if has_alpha {
if batch.num_columns() < 3 {
bail!(
"fxcache claims alpha_feature_dim but has {} columns (expected 3)",
batch.num_columns()
);
}
Some(
batch
.column(2)
.as_any()
.downcast_ref::<FixedSizeBinaryArray>()
.ok_or_else(|| {
anyhow!(
"fxcache alpha 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 {} (RECORD_F32_COUNT * 4)",
blob.len(),
expected_blob_size
);
}
// Decode RECORD_F32_COUNT little-endian f32 values into the flat buffer.
for j in 0..RECORD_F32_COUNT {
let off = j * 4;
let value = f32::from_le_bytes([
blob[off],
blob[off + 1],
blob[off + 2],
blob[off + 3],
]);
f32_data.push(value);
}
if let (Some(alpha_arr), Some(dst), Some(dim)) =
(alpha_arr_opt, alpha_data.as_mut(), alpha_dim)
{
let alpha_blob: &[u8] = alpha_arr.value(i);
if alpha_blob.len() != expected_alpha_blob_size {
bail!(
"fxcache alpha row blob size {} != expected {} (alpha_feature_dim × 4)",
alpha_blob.len(),
expected_alpha_blob_size
);
}
for j in 0..dim {
let off = j * 4;
dst.push(f32::from_le_bytes([
alpha_blob[off],
alpha_blob[off + 1],
alpha_blob[off + 2],
alpha_blob[off + 3],
]));
}
}
}
}
let bar_count = timestamps.len();
if bar_count == 0 {
bail!("fxcache is empty: 0 bars");
}
parsed_meta.bar_count = bar_count;
Ok(Self {
metadata: parsed_meta,
f32_data,
timestamps,
alpha_data,
alpha_dim,
})
}
/// Whether the file contains the alpha_features column.
pub fn has_alpha_features(&self) -> bool {
self.alpha_data.is_some()
}
/// Width of the alpha-features row stored in the file. `None` if the
/// fxcache was written without the alpha column. Variable across files:
/// 134 for the bar-level stack, 81 for the per-snapshot stack.
pub fn alpha_feature_dim(&self) -> Option<usize> {
self.alpha_dim
}
/// Alpha feature row for bar `i`, if the file contains the alpha column.
/// Returns `None` if the file is alpha-without-alphafeatures (legacy write).
/// Panics if `i >= bar_count`.
pub fn alpha_features(&self, i: usize) -> Option<&[f32]> {
assert!(
i < self.metadata.bar_count,
"alpha_features({i}) >= bar_count({})",
self.metadata.bar_count
);
let dim = self.alpha_dim?;
self.alpha_data.as_ref().map(|buf| {
let start = i * dim;
&buf[start..start + dim]
})
}
/// Borrow header metadata.
pub fn metadata(&self) -> &FxCacheMetadata {
&self.metadata
}
/// Number of bars (records) in the file.
pub fn bar_count(&self) -> usize {
self.metadata.bar_count
}
/// Return the timestamp (nanoseconds since Unix epoch, signed) for record
/// `i`. Panics if `i >= bar_count`.
pub fn record_timestamp(&self, i: usize) -> i64 {
assert!(
i < self.metadata.bar_count,
"record_timestamp({i}) >= bar_count({})",
self.metadata.bar_count
);
self.timestamps[i]
}
/// Return zero-copy slice views of record `i`. Panics if `i >= bar_count`.
///
/// # Performance
///
/// O(1) — slice indexing into the owned `f32_data` buffer. The buffer
/// was materialized once at `open()`.
pub fn record(&self, i: usize) -> FxCacheRecord<'_> {
assert!(
i < self.metadata.bar_count,
"record({i}) >= bar_count({})",
self.metadata.bar_count
);
let start = i * RECORD_F32_COUNT;
let row = &self.f32_data[start..start + RECORD_F32_COUNT];
FxCacheRecord {
features: &row[..FEAT_DIM],
targets: &row[FEAT_DIM..FEAT_DIM + TARGET_DIM],
ofi: &row[FEAT_DIM + TARGET_DIM..],
}
}
}
#[cfg(test)]
mod tests {
use super::*;
/// Smoke test: open the canonical production fxcache and verify
/// metadata + that the timestamp + feature values look plausible.
/// Ignored by default because it depends on the local
/// `test_data/feature-cache/<HASH>.fxcache` fixture being present
/// AND being the CURRENT production-vintage cache (with OFI).
///
/// The hash below is the current production cache hash on the
/// `feature-cache-pvc` PVC — refresh by running
/// `./scripts/argo-precompute.sh --branch main` (regenerates on PVC),
/// then `kubectl cp foxhunt/<pod>:/feature-cache/<HASH>.fxcache
/// test_data/feature-cache/`. Per project discipline (memory:
/// "make features optional derives from production strictly
/// forbidden") the assertions are strict — every production schema
/// field (including `has_ofi`) must hold against the cache.
///
/// Run with: `cargo test -p ml-alpha --lib -- --ignored fxcache_local_smoke`.
#[test]
#[ignore]
fn fxcache_local_smoke() {
// Resolve relative to CARGO_MANIFEST_DIR so the test works
// regardless of cwd (cargo test sets cwd to the crate dir).
let manifest_dir = env!("CARGO_MANIFEST_DIR");
let path_buf = std::path::Path::new(manifest_dir)
.parent()
.and_then(|p| p.parent())
.map(|root| {
root.join("test_data/feature-cache/70e5bc3a401de7f28c57bcd77e2d302df7b0838d048c3d657feecf24a46a12f8.fxcache")
})
.expect("derive workspace-root fxcache path");
let path = path_buf
.to_str()
.expect("fxcache path is valid UTF-8");
// Metadata-only open — reads Arrow IPC footer (O(1) memory),
// validates schema, does NOT materialize the 15+ GB f32 buffer.
// This is the canonical smoke check: production schema integrity
// on any dev box regardless of available RAM.
let m = FxCacheReader::open_metadata(path)
.expect("local fxcache should open (metadata-only)");
assert_eq!(m.version, FXCACHE_VERSION);
assert_eq!(m.feat_dim, FEAT_DIM);
assert_eq!(m.target_dim, TARGET_DIM);
assert_eq!(m.ofi_dim, OFI_DIM);
assert!(m.has_ofi);
// bar_count is 0 in metadata-only mode (requires batch iteration
// to determine). The schema-integrity check is sufficient for
// the smoke test; record-level assertions are covered by the
// full-materialize path on production hosts and cluster CI.
eprintln!(
"fxcache_local_smoke PASS — v{} feat={} target={} ofi={} has_ofi={}",
m.version, m.feat_dim, m.target_dim, m.ofi_dim, m.has_ofi
);
}
}