From a4ba8bddaa525daa5fddba0c908f305d82c8b7bd Mon Sep 17 00:00:00 2001 From: jgrusewski Date: Wed, 1 Apr 2026 19:29:41 +0200 Subject: [PATCH] feat: fxcache stores per-bar timestamps for walk-forward windowing Each record now starts with an i64 timestamp (nanoseconds since epoch) before the feature/target/OFI data. v1 records grow from 432 to 440 bytes, v2 from 112 to 120 bytes. The timestamp is always i64 even in bf16 mode. train_baseline_rl reconstructs bars with real timestamps instead of placeholders so the walk-forward windower can split by month. Co-Authored-By: Claude Opus 4.6 (1M context) --- crates/ml/examples/precompute_features.rs | 6 ++ crates/ml/examples/train_baseline_rl.rs | 5 +- crates/ml/src/fxcache.rs | 73 +++++++++++++++-------- crates/ml/tests/fxcache_roundtrip_test.rs | 32 ++++++++-- 4 files changed, 84 insertions(+), 32 deletions(-) diff --git a/crates/ml/examples/precompute_features.rs b/crates/ml/examples/precompute_features.rs index 99b719b09..1aaee512d 100644 --- a/crates/ml/examples/precompute_features.rs +++ b/crates/ml/examples/precompute_features.rs @@ -239,6 +239,11 @@ async fn main() -> Result<()> { targets.push([raw_curr, raw_next, raw_curr, raw_next]); } + // ── Step 3b: Build per-bar timestamps ───────────────────────────────── + let timestamps: Vec = (0..n) + .map(|i| all_bars[i + WARMUP].timestamp.timestamp_nanos_opt().unwrap_or(0)) + .collect(); + // ── Step 4: Compute OFI from MBP-10 + trades ──────────────────────────── let t2 = Instant::now(); let ofi: Vec<[f64; 8]> = if let Some(ref mbp10_path) = mbp10_dir { @@ -375,6 +380,7 @@ async fn main() -> Result<()> { &features, &targets, &ofi, + ×tamps, cache_key, opts.bf16, ) diff --git a/crates/ml/examples/train_baseline_rl.rs b/crates/ml/examples/train_baseline_rl.rs index bca8098df..6a0732f8d 100644 --- a/crates/ml/examples/train_baseline_rl.rs +++ b/crates/ml/examples/train_baseline_rl.rs @@ -856,14 +856,13 @@ fn run_training(args: &Args) -> Result> { }); let (bars, all_features) = if let Some(cached) = fxcache_data { - // Reconstruct bars from cached targets (raw_close at index 2) + // Reconstruct bars from cached targets (raw_close at index 2) + real timestamps let n = cached.bar_count; let mut bars = Vec::with_capacity(n); for i in 0..n { let close = cached.targets[i][2]; // raw_close - let next_close = cached.targets[i][3]; // raw_next_close bars.push(ml::features::extraction::OHLCVBar { - timestamp: chrono::Utc::now(), // placeholder — walk-forward uses index not timestamp + timestamp: chrono::DateTime::from_timestamp_nanos(cached.timestamps[i]), open: close, high: close, low: close, diff --git a/crates/ml/src/fxcache.rs b/crates/ml/src/fxcache.rs index 38b9ea572..f1c1473d2 100644 --- a/crates/ml/src/fxcache.rs +++ b/crates/ml/src/fxcache.rs @@ -19,8 +19,9 @@ //! │ reserved [u8; 8] │ //! └───────────────────────────────────────────────────┘ //! │ Body (bar_count records) │ -//! │ Version 1: 54 × f64 = 432 bytes/bar │ -//! │ Version 2: 56 × bf16 = 112 bytes/bar │ +//! │ Each record starts with an i64 timestamp (ns). │ +//! │ Version 1: [i64 ts][54 × f64] = 440 bytes/bar │ +//! │ Version 2: [i64 ts][56 × bf16] = 120 bytes/bar │ //! │ (54 data + 2 zero-padding) │ //! └───────────────────────────────────────────────────┘ //! ``` @@ -185,6 +186,8 @@ impl FxCacheHeader { /// In-memory representation of an FxCache file's contents. #[derive(Debug)] pub struct FxCacheData { + /// Per-bar timestamps (nanoseconds since Unix epoch). + pub timestamps: Vec, /// Feature vectors, one per bar (42 elements each). pub features: Vec<[f64; FEAT_DIM]>, /// Target vectors, one per bar (4 elements each). @@ -203,12 +206,13 @@ pub struct FxCacheData { /// /// # 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 -/// * `cache_key` — SHA256 key (raw 32 bytes) -/// * `bf16` — If true, write version 2 (bf16); otherwise version 1 (f64) +/// * `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) +/// * `bf16` — If true, write version 2 (bf16); otherwise version 1 (f64) /// /// # Returns /// @@ -218,16 +222,18 @@ pub fn write_fxcache( features: &[[f64; FEAT_DIM]], targets: &[[f64; TARGET_DIM]], ofi: &[[f64; OFI_DIM]], + timestamps: &[i64], cache_key: [u8; 32], bf16: bool, ) -> Result { let bar_count = features.len(); - if targets.len() != bar_count || ofi.len() != bar_count { + if targets.len() != bar_count || ofi.len() != bar_count || timestamps.len() != bar_count { bail!( - "Length mismatch: features={}, targets={}, ofi={}", + "Length mismatch: features={}, targets={}, ofi={}, timestamps={}", bar_count, targets.len(), - ofi.len() + ofi.len(), + timestamps.len() ); } if bar_count == 0 { @@ -255,9 +261,9 @@ pub fn write_fxcache( // Write body let body_bytes: u64 = if bf16 { - write_body_bf16(&mut writer, features, targets, ofi)? + write_body_bf16(&mut writer, features, targets, ofi, timestamps)? } else { - write_body_f64(&mut writer, features, targets, ofi)? + write_body_f64(&mut writer, features, targets, ofi, timestamps)? }; writer.flush().context("Failed to flush FxCache writer")?; @@ -276,17 +282,19 @@ pub fn write_fxcache( Ok(total_bytes) } -/// Write body in f64 format (version 1): 54 f64 values = 432 bytes per bar. +/// Write body in f64 format (version 1): [i64 ts] + 54 f64 values = 440 bytes per bar. fn write_body_f64( writer: &mut BufWriter, features: &[[f64; FEAT_DIM]], targets: &[[f64; TARGET_DIM]], ofi: &[[f64; OFI_DIM]], + timestamps: &[i64], ) -> Result { - let bytes_per_bar = RECORD_F64_COUNT * 8; + let bytes_per_bar = 8 + RECORD_F64_COUNT * 8; // i64 timestamp + f64 data let total = features.len() as u64 * bytes_per_bar as u64; for i in 0..features.len() { + writer.write_all(×tamps[i].to_le_bytes())?; for &v in &features[i] { writer.write_all(&v.to_le_bytes())?; } @@ -301,19 +309,22 @@ fn write_body_f64( Ok(total) } -/// Write body in bf16 format (version 2): 56 bf16 values = 112 bytes per bar. +/// Write body in bf16 format (version 2): [i64 ts] + 56 bf16 values = 120 bytes per bar. /// (54 data values + 2 zero-padding for 4-byte alignment.) +/// Timestamp is always i64 (8 bytes) — nanosecond precision requires 64 bits. fn write_body_bf16( writer: &mut BufWriter, features: &[[f64; FEAT_DIM]], targets: &[[f64; TARGET_DIM]], ofi: &[[f64; OFI_DIM]], + timestamps: &[i64], ) -> Result { - let bytes_per_bar = RECORD_BF16_COUNT * 2; + let bytes_per_bar = 8 + RECORD_BF16_COUNT * 2; // i64 timestamp + bf16 data let total = features.len() as u64 * bytes_per_bar as u64; let zero = f16::ZERO; for i in 0..features.len() { + writer.write_all(×tamps[i].to_le_bytes())?; for &v in &features[i] { writer.write_all(&f16::from_f64(v).to_le_bytes())?; } @@ -364,11 +375,11 @@ pub fn load_fxcache(path: &Path) -> Result { let bar_count = header.bar_count as usize; - // Sanity-check file size + // Sanity-check file size (each record has an i64 timestamp prefix) let expected_body = if header.version == 1 { - bar_count as u64 * RECORD_F64_COUNT as u64 * 8 + bar_count as u64 * (8 + RECORD_F64_COUNT as u64 * 8) } else { - bar_count as u64 * RECORD_BF16_COUNT as u64 * 2 + bar_count as u64 * (8 + RECORD_BF16_COUNT as u64 * 2) }; let expected_total = HEADER_SIZE as u64 + expected_body; if file_len < expected_total { @@ -380,7 +391,7 @@ pub fn load_fxcache(path: &Path) -> Result { } // Read body - let (features, targets, ofi) = if header.version == 1 { + let (timestamps, features, targets, ofi) = if header.version == 1 { read_body_f64(&mut reader, bar_count)? } else { read_body_bf16(&mut reader, bar_count)? @@ -395,6 +406,7 @@ pub fn load_fxcache(path: &Path) -> Result { ); Ok(FxCacheData { + timestamps, features, targets, ofi, @@ -407,13 +419,18 @@ pub fn load_fxcache(path: &Path) -> Result { fn read_body_f64( reader: &mut BufReader, bar_count: usize, -) -> Result<(Vec<[f64; FEAT_DIM]>, Vec<[f64; TARGET_DIM]>, Vec<[f64; OFI_DIM]>)> { +) -> Result<(Vec, 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 f64_buf = [0u8; 8]; 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 f64_buf)?; @@ -436,20 +453,26 @@ fn read_body_f64( ofi.push(ofi_row); } - Ok((features, targets, ofi)) + Ok((timestamps, features, targets, ofi)) } /// Read body in bf16 format (version 2), converting to f64 on load. +/// Timestamp is always read as i64 (8 bytes) regardless of bf16 mode. fn read_body_bf16( reader: &mut BufReader, bar_count: usize, -) -> Result<(Vec<[f64; FEAT_DIM]>, Vec<[f64; TARGET_DIM]>, Vec<[f64; OFI_DIM]>)> { +) -> Result<(Vec, 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 bf16_buf = [0u8; 2]; 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 bf16_buf)?; @@ -476,7 +499,7 @@ fn read_body_bf16( reader.read_exact(&mut bf16_buf)?; } - Ok((features, targets, ofi)) + Ok((timestamps, features, targets, ofi)) } // ── Finder ─────────────────────────────────────────────────────────────────── diff --git a/crates/ml/tests/fxcache_roundtrip_test.rs b/crates/ml/tests/fxcache_roundtrip_test.rs index 0309ed4a7..9688b45d2 100644 --- a/crates/ml/tests/fxcache_roundtrip_test.rs +++ b/crates/ml/tests/fxcache_roundtrip_test.rs @@ -47,6 +47,13 @@ fn make_ofi(n: usize) -> Vec<[f64; 8]> { .collect() } +/// Deterministic per-bar timestamps (1-minute intervals starting 2022-01-01). +fn make_timestamps(n: usize) -> Vec { + (0..n) + .map(|i| 1640995200_000_000_000i64 + i as i64 * 60_000_000_000) + .collect() +} + /// A fixed 32-byte cache key for tests. fn test_cache_key() -> [u8; 32] { let mut key = [0u8; 32]; @@ -68,10 +75,11 @@ fn test_fxcache_f64_roundtrip() { let features = make_features(n); let targets = make_targets(n); let ofi = make_ofi(n); + let timestamps = make_timestamps(n); let key = test_cache_key(); let bytes_written = - write_fxcache(&path, &features, &targets, &ofi, key, false).unwrap(); + write_fxcache(&path, &features, &targets, &ofi, ×tamps, key, false).unwrap(); assert!(bytes_written > 0, "expected nonzero bytes written"); let data: FxCacheData = load_fxcache(&path).unwrap(); @@ -80,9 +88,14 @@ fn test_fxcache_f64_roundtrip() { assert_eq!(data.features.len(), n); assert_eq!(data.targets.len(), n); assert_eq!(data.ofi.len(), n); + assert_eq!(data.timestamps.len(), n); // Bit-exact match for f64 roundtrip. for i in 0..n { + assert_eq!( + data.timestamps[i], timestamps[i], + "timestamp mismatch at bar {i}" + ); assert_eq!( data.features[i], features[i], "feature mismatch at bar {i}" @@ -106,16 +119,25 @@ fn test_fxcache_bf16_roundtrip() { let features = make_features(n); let targets = make_targets(n); let ofi = make_ofi(n); + let timestamps = make_timestamps(n); let key = test_cache_key(); let bytes_written = - write_fxcache(&path, &features, &targets, &ofi, key, true).unwrap(); + write_fxcache(&path, &features, &targets, &ofi, ×tamps, key, true).unwrap(); assert!(bytes_written > 0, "expected nonzero bytes written"); let data: FxCacheData = load_fxcache(&path).unwrap(); assert_eq!(data.bar_count, n); assert_eq!(data.cache_key, key); + // Timestamps are always i64 — bit-exact even in bf16 mode. + for i in 0..n { + assert_eq!( + data.timestamps[i], timestamps[i], + "timestamp mismatch at bar {i}" + ); + } + // bf16 tolerance checks. for i in 0..n { for j in 0..42 { @@ -161,8 +183,9 @@ fn test_fxcache_find() { let features = make_features(5); let targets = make_targets(5); let ofi = make_ofi(5); + let timestamps = make_timestamps(5); - write_fxcache(&path, &features, &targets, &ofi, key, false).unwrap(); + write_fxcache(&path, &features, &targets, &ofi, ×tamps, key, false).unwrap(); // Should find the file. let found = find_fxcache(dir.path(), &key); @@ -205,9 +228,10 @@ fn test_fxcache_empty() { let features: Vec<[f64; 42]> = vec![]; let targets: Vec<[f64; 4]> = vec![]; let ofi: Vec<[f64; 8]> = vec![]; + let timestamps: Vec = vec![]; let key = test_cache_key(); - let err = write_fxcache(&path, &features, &targets, &ofi, key, false).unwrap_err(); + let err = write_fxcache(&path, &features, &targets, &ofi, ×tamps, key, false).unwrap_err(); let msg = format!("{err:#}"); assert!( msg.contains("0 bars") || msg.contains("empty"),