Files
foxhunt/crates/ml/src/fxcache.rs
2026-04-10 18:45:49 +02:00

628 lines
21 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 — 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 (64 bytes) │
//! │ magic [u8; 8] = b"FXCACHE\0" │
//! │ version u16 = 1 (f32) │
//! │ feat_dim u16 = 42 │
//! │ target_dim u16 = 4 │
//! │ ofi_dim u16 = 8 │
//! │ bar_count u64 │
//! │ cache_key [u8; 32] (SHA256 raw bytes) │
//! │ reserved [u8; 8] │
//! └───────────────────────────────────────────────────┘
//! │ Body (bar_count records) │
//! │ Each record starts with an i64 timestamp (ns). │
//! │ Version 1: [i64 ts][54 × f32] = 224 bytes/bar │
//! └───────────────────────────────────────────────────┘
//! ```
use anyhow::{bail, Context, Result};
use std::io::{BufReader, BufWriter, Read, Write};
use std::path::{Path, PathBuf};
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 = 64;
/// Feature vector dimensionality.
const FEAT_DIM: usize = 42;
/// Target vector dimensionality (close, next_close, raw_close, raw_next).
const TARGET_DIM: usize = 4;
/// OFI vector dimensionality (8 MBP-10 order-flow imbalance levels).
const OFI_DIM: usize = 8;
/// Total f64 values per record: features + targets + OFI = 42 + 4 + 8 = 54.
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 ───────────────────────────────────────────────────────────────────
/// 64-byte fixed header for `.fxcache` files.
#[derive(Debug, Clone)]
pub struct FxCacheHeader {
/// Magic bytes: `b"FXCACHE\0"`.
pub magic: [u8; 8],
/// Format version: 1 = f32.
pub version: u16,
/// Feature dimension (42).
pub feat_dim: u16,
/// Target dimension (4).
pub target_dim: u16,
/// OFI dimension (8).
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],
/// 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,
reserved,
}
}
/// Validate header integrity.
pub fn validate(&self) -> Result<()> {
if self.magic != FXCACHE_MAGIC {
bail!(
"Invalid FxCache magic: expected {:?}, got {:?}",
FXCACHE_MAGIC,
self.magic
);
}
if self.version != 1 {
bail!(
"Unsupported FxCache version: {} (expected 1)",
self.version
);
}
if self.feat_dim as usize != FEAT_DIM {
bail!(
"Feature dimension mismatch: expected {}, got {}",
FEAT_DIM,
self.feat_dim
);
}
if self.target_dim as usize != TARGET_DIM {
bail!(
"Target dimension mismatch: expected {}, got {}",
TARGET_DIM,
self.target_dim
);
}
if self.ofi_dim as usize != OFI_DIM {
bail!(
"OFI dimension mismatch: expected {}, got {}",
OFI_DIM,
self.ofi_dim
);
}
if self.bar_count == 0 {
bail!("FxCache bar_count is zero — empty cache files are not valid");
}
Ok(())
}
/// Serialize header to 64 bytes (little-endian).
fn to_bytes(&self) -> [u8; HEADER_SIZE] {
let mut buf = [0u8; HEADER_SIZE];
buf[0..8].copy_from_slice(&self.magic);
buf[8..10].copy_from_slice(&self.version.to_le_bytes());
buf[10..12].copy_from_slice(&self.feat_dim.to_le_bytes());
buf[12..14].copy_from_slice(&self.target_dim.to_le_bytes());
buf[14..16].copy_from_slice(&self.ofi_dim.to_le_bytes());
buf[16..24].copy_from_slice(&self.bar_count.to_le_bytes());
buf[24..56].copy_from_slice(&self.cache_key);
buf[56..64].copy_from_slice(&self.reserved);
buf
}
/// Deserialize header from 64 bytes (little-endian).
fn from_bytes(buf: &[u8; HEADER_SIZE]) -> Self {
let mut magic = [0u8; 8];
magic.copy_from_slice(&buf[0..8]);
let version = u16::from_le_bytes([buf[8], buf[9]]);
let feat_dim = u16::from_le_bytes([buf[10], buf[11]]);
let target_dim = u16::from_le_bytes([buf[12], buf[13]]);
let ofi_dim = u16::from_le_bytes([buf[14], buf[15]]);
let bar_count = u64::from_le_bytes([
buf[16], buf[17], buf[18], buf[19], buf[20], buf[21], buf[22], buf[23],
]);
let mut cache_key = [0u8; 32];
cache_key.copy_from_slice(&buf[24..56]);
let mut reserved = [0u8; 8];
reserved.copy_from_slice(&buf[56..64]);
Self {
magic,
version,
feat_dim,
target_dim,
ofi_dim,
bar_count,
cache_key,
reserved,
}
}
}
// ── 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 (4 elements each).
pub targets: Vec<[f64; TARGET_DIM]>,
/// OFI vectors, one per bar (8 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,
}
// ── 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 4-element target vectors
/// * `ofi` — Slice of 8-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).
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,
) -> 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)");
}
// Create parent directories
if let Some(parent) = path.parent() {
std::fs::create_dir_all(parent)
.with_context(|| format!("Failed to create parent dirs for {:?}", path))?;
}
let header = FxCacheHeader::new(1, bar_count as u64, cache_key, has_ofi);
header.validate()?;
let file = std::fs::File::create(path)
.with_context(|| format!("Failed to create FxCache file {:?}", path))?;
let mut writer = BufWriter::new(file);
// Write header
writer
.write_all(&header.to_bytes())
.context("Failed to write FxCache header")?;
// Write body (f32 format)
let body_bytes = write_body_f32(&mut writer, features, targets, ofi, timestamps)?;
writer.flush().context("Failed to flush FxCache writer")?;
let total_bytes = HEADER_SIZE as u64 + body_bytes;
info!(
"FxCache written: {} bars, v1 (f32), {:.2} MB -> {:?}",
bar_count,
total_bytes as f64 / 1_048_576.0,
path
);
Ok(total_bytes)
}
/// Write body in f32 format (version 1): [i64 ts] + 54 × f32 = 224 bytes per bar.
fn write_body_f32(
writer: &mut BufWriter<std::fs::File>,
features: &[[f64; FEAT_DIM]],
targets: &[[f64; TARGET_DIM]],
ofi: &[[f64; OFI_DIM]],
timestamps: &[i64],
) -> Result<u64> {
let bytes_per_bar = 8 + RECORD_F32_COUNT * 4; // i64 timestamp + f32 data
let total = features.len() as u64 * bytes_per_bar as u64;
for i in 0..features.len() {
writer.write_all(&timestamps[i].to_le_bytes())?;
for &v in &features[i] {
writer.write_all(&(v as f32).to_le_bytes())?;
}
for &v in &targets[i] {
writer.write_all(&(v as f32).to_le_bytes())?;
}
for &v in &ofi[i] {
writer.write_all(&(v as f32).to_le_bytes())?;
}
}
Ok(total)
}
// ── Reader ───────────────────────────────────────────────────────────────────
/// Load an `.fxcache` file into memory.
///
/// Reads the 64-byte header, validates it, then reads the f32 body,
/// converting each f32 back to f64 on load.
///
/// # Arguments
///
/// * `path` — Path to the `.fxcache` file
///
/// # Returns
///
/// Fully parsed `FxCacheData` with features, targets, OFI, cache key, and bar count.
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 file_len = file
.metadata()
.with_context(|| format!("Failed to stat FxCache file {:?}", path))?
.len();
let mut reader = BufReader::new(file);
// Read header
let mut header_buf = [0u8; HEADER_SIZE];
reader
.read_exact(&mut header_buf)
.context("Failed to read FxCache header")?;
let header = FxCacheHeader::from_bytes(&header_buf);
header.validate()?;
let bar_count = header.bar_count as usize;
// Sanity-check file size: [i64 ts] + 54 × f32 = 224 bytes/bar
let expected_body = bar_count as u64 * (8 + RECORD_F32_COUNT as u64 * 4);
let expected_total = HEADER_SIZE as u64 + expected_body;
if file_len < expected_total {
bail!(
"FxCache file truncated: expected {} bytes, got {}",
expected_total,
file_len
);
}
// Read body (f32 format)
let (timestamps, features, targets, ofi) = read_body_f32(&mut reader, bar_count)?;
info!(
"FxCache loaded: {} bars, v1 (f32) from {:?}",
bar_count,
path
);
let has_ofi = header.reserved[0] == 1;
Ok(FxCacheData {
timestamps,
features,
targets,
ofi,
cache_key: header.cache_key,
bar_count,
has_ofi,
})
}
/// Read body in f32 format (version 1), converting f32 back to f64 on load.
fn read_body_f32(
reader: &mut BufReader<std::fs::File>,
bar_count: usize,
) -> Result<(Vec<i64>, Vec<[f64; FEAT_DIM]>, Vec<[f64; TARGET_DIM]>, Vec<[f64; OFI_DIM]>)> {
let mut timestamps = Vec::with_capacity(bar_count);
let mut features = Vec::with_capacity(bar_count);
let mut targets = Vec::with_capacity(bar_count);
let mut ofi = Vec::with_capacity(bar_count);
let mut i64_buf = [0u8; 8];
let mut f32_buf = [0u8; 4];
for _ in 0..bar_count {
reader.read_exact(&mut i64_buf)?;
timestamps.push(i64::from_le_bytes(i64_buf));
let mut feat = [0.0_f64; FEAT_DIM];
for slot in &mut feat {
reader.read_exact(&mut f32_buf)?;
*slot = f32::from_le_bytes(f32_buf) as f64;
}
features.push(feat);
let mut tgt = [0.0_f64; TARGET_DIM];
for slot in &mut tgt {
reader.read_exact(&mut f32_buf)?;
*slot = f32::from_le_bytes(f32_buf) as f64;
}
targets.push(tgt);
let mut ofi_row = [0.0_f64; OFI_DIM];
for slot in &mut ofi_row {
reader.read_exact(&mut f32_buf)?;
*slot = f32::from_le_bytes(f32_buf) as f64;
}
ofi.push(ofi_row);
}
Ok((timestamps, features, targets, ofi))
}
// ── Finder ───────────────────────────────────────────────────────────────────
#[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; 4]; 10];
let ofi = vec![[0.5_f64; 8]; 10];
let timestamps = vec![1_i64; 10];
let key = [0u8; 32];
write_fxcache(&path, &features, &targets, &ofi, &timestamps, key, true).unwrap();
let loaded = load_fxcache(&path).unwrap();
assert!(loaded.has_ofi, "has_ofi should be true");
assert_eq!(loaded.bar_count, 10);
}
#[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; 4]; 10];
let ofi = vec![[0.0_f64; 8]; 10];
let timestamps = vec![1_i64; 10];
let key = [0u8; 32];
write_fxcache(&path, &features, &targets, &ofi, &timestamps, key, false).unwrap();
let loaded = load_fxcache(&path).unwrap();
assert!(!loaded.has_ofi, "has_ofi should be false");
}
}
/// 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
}
/// 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 (`"ohlcv"` or `"mbp10"`)
/// * `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,
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)
let key_hex = crate::feature_cache::calculate_dbn_cache_key_full(
data_dir, mbp10_dir, trades_dir, symbol, data_source,
)
.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,
"ohlcv",
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; 4]; 5];
let ofi = vec![[0.0_f64; 8]; 5];
let timestamps = vec![1_i64; 5];
write_fxcache(&path, &features, &targets, &ofi, &timestamps, [0u8; 32], false)
.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,
"ohlcv",
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());
}
}