From 429967dcd1ad7aab523f944b8a26231d63e393ef Mon Sep 17 00:00:00 2001 From: jgrusewski Date: Mon, 20 Apr 2026 08:11:03 +0200 Subject: [PATCH] feat: precompute OFI deltas + book aggression + bar duration in fxcache MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit FXCACHE_VERSION 3→4. Three precomputed features added to OFI region: - ofi[8..16): temporal deltas (ofi[bar] - ofi[bar-1]) for 8 features - ofi[16]: book aggression (MBP-10 10-level center-of-mass asymmetry) - ofi[17]: log bar duration (imbalance bar formation time, normalized) Previously 10 of 18 OFI embed MLP inputs were zero. Now all 18 have real data: raw_ofi(8) + delta_ofi(8) + book_aggression(1) + log_duration(1). Added read_state_sample() diagnostic for GPU state verification. Legacy v3 fxcache handled by zeroing new slots (v2→v4 graceful upgrade). Co-Authored-By: Claude Opus 4.6 (1M context) --- crates/ml/examples/precompute_features.rs | 75 ++++++++++++++++++- crates/ml/src/fxcache.rs | 33 +++++--- crates/ml/src/trainers/dqn/data_loading.rs | 35 +++++++++ crates/ml/src/trainers/dqn/fused_training.rs | 15 ++++ .../src/trainers/dqn/trainer/constructor.rs | 10 ++- 5 files changed, 152 insertions(+), 16 deletions(-) diff --git a/crates/ml/examples/precompute_features.rs b/crates/ml/examples/precompute_features.rs index 27000b151..4ef893469 100644 --- a/crates/ml/examples/precompute_features.rs +++ b/crates/ml/examples/precompute_features.rs @@ -443,9 +443,17 @@ async fn main() -> Result<()> { let arr8 = f.to_array(); let micro_12 = micro_state.snapshot(); + // v4 OFI layout: [0..8) raw OFI, [8..16) deltas (filled below), + // [16] book_aggression (filled below), [17] log_bar_duration (filled below), + // [18..20) ofi_acceleration + toxicity_gradient from microstructure let mut arr20 = [0.0_f64; 20]; arr20[..8].copy_from_slice(&arr8); - arr20[8..20].copy_from_slice(µ_12); + // Slots [8..16) reserved for OFI deltas (computed after all bars) + // Slot [16] reserved for book_aggression (computed below) + // Slot [17] reserved for log_bar_duration (computed below) + // Keep micro_12[10..12) (ofi_acceleration, toxicity_gradient) at [18..20) + arr20[18] = micro_12[10]; // ofi_acceleration + arr20[19] = micro_12[11]; // toxicity_gradient ofi_per_bar.push(arr20); } _ => ofi_per_bar.push([0.0; 20]), @@ -453,8 +461,56 @@ async fn main() -> Result<()> { } else { ofi_per_bar.push([0.0; 20]); } + + // ── Book aggression: center-of-mass asymmetry from MBP-10 depth ── + // Uses the last snapshot in the bar window (same as OFI calculation above) + if let Some(snap) = bar_snapshots.last() { + let mut buy_weighted_sum = 0.0_f64; + let mut buy_total = 0.0_f64; + let mut sell_weighted_sum = 0.0_f64; + let mut sell_total = 0.0_f64; + for (level, pair) in snap.levels.iter().enumerate().take(10) { + let bid_size = pair.bid_sz as f64; + let ask_size = pair.ask_sz as f64; + buy_weighted_sum += (level + 1) as f64 * bid_size; + buy_total += bid_size; + sell_weighted_sum += (level + 1) as f64 * ask_size; + sell_total += ask_size; + } + let buy_com = if buy_total > 0.0 { buy_weighted_sum / buy_total } else { 5.5 }; + let sell_com = if sell_total > 0.0 { sell_weighted_sum / sell_total } else { 5.5 }; + let book_aggression = (sell_com - buy_com) / 10.0; // normalize to [-1, 1] + ofi_per_bar.last_mut().unwrap()[16] = book_aggression; + } } + // ── OFI temporal deltas: delta[bar] = ofi[bar] - ofi[bar-1] ── + // Computed after all bars so we have the full sequence + for i in (1..ofi_per_bar.len()).rev() { + for k in 0..8 { + ofi_per_bar[i][8 + k] = ofi_per_bar[i][k] - ofi_per_bar[i - 1][k]; + } + } + // First bar has no previous — deltas are zero (already initialized) + + // ── Log bar duration: ln(duration_secs) / 10.0 ── + for i in 0..n { + let duration_secs = if i > 0 { + let ts_cur = all_bars[i + WARMUP].timestamp; + let ts_prev = all_bars[i + WARMUP - 1].timestamp; + (ts_cur - ts_prev).num_seconds() as f64 + } else { + 60.0 // default 1 minute for first bar + }; + let log_duration = duration_secs.max(0.1).ln() / 10.0; + ofi_per_bar[i][17] = log_duration; + } + + let delta_nonzero = ofi_per_bar.iter().filter(|f| f[8..16].iter().any(|&v| v != 0.0)).count(); + let book_nonzero = ofi_per_bar.iter().filter(|f| f[16] != 0.0).count(); + info!("OFI v4: deltas_nonzero={}/{}, book_aggression_nonzero={}/{}", + delta_nonzero, n, book_nonzero, n); + // Fill targets[i][5] with MBP-10 midpoint at bar open (mid_price_open) let mut mid_fill_count = 0_usize; for i in 0..n { @@ -481,8 +537,21 @@ async fn main() -> Result<()> { ofi_per_bar } } else { - info!("No MBP-10 directory, OFI will be zeros"); - vec![[0.0; 20]; n] + info!("No MBP-10 directory, OFI will be zeros (log_bar_duration still computed)"); + let mut ofi_per_bar = vec![[0.0_f64; 20]; n]; + // Even without MBP-10 data, compute log_bar_duration from bar timestamps + for i in 0..n { + let duration_secs = if i > 0 { + let ts_cur = all_bars[i + WARMUP].timestamp; + let ts_prev = all_bars[i + WARMUP - 1].timestamp; + (ts_cur - ts_prev).num_seconds() as f64 + } else { + 60.0 // default 1 minute for first bar + }; + let log_duration = duration_secs.max(0.1).ln() / 10.0; + ofi_per_bar[i][17] = log_duration; + } + ofi_per_bar }; let total_len = n; diff --git a/crates/ml/src/fxcache.rs b/crates/ml/src/fxcache.rs index a4553e13d..3c063c712 100644 --- a/crates/ml/src/fxcache.rs +++ b/crates/ml/src/fxcache.rs @@ -10,7 +10,7 @@ //! ┌───────────────────────────────────────────────────┐ //! │ FxCacheHeader (64 bytes) │ //! │ magic [u8; 8] = b"FXCACHE\0" │ -//! │ version u16 = 3 (f32) │ +//! │ version u16 = 4 (f32) │ //! │ feat_dim u16 = 42 │ //! │ target_dim u16 = 6 │ //! │ ofi_dim u16 = 20 │ @@ -20,7 +20,7 @@ //! └───────────────────────────────────────────────────┘ //! │ Body (bar_count records) │ //! │ Each record starts with an i64 timestamp (ns). │ -//! │ Version 3: [i64 ts][68 × f32] = 280 bytes/bar │ +//! │ Version 4: [i64 ts][68 × f32] = 280 bytes/bar │ //! └───────────────────────────────────────────────────┘ //! ``` @@ -40,7 +40,7 @@ const HEADER_SIZE: usize = 64; /// Cache format version. Bump on ANY format change (OFI_DIM, FEAT_DIM, etc.). /// Stale cache files with wrong version are auto-detected and rejected by validate(). /// The ensure-fxcache Argo step catches the error and regenerates. -pub const FXCACHE_VERSION: u16 = 3; // v3: TARGET_DIM 4→6 (added raw_open, mid_price_open) +pub const FXCACHE_VERSION: u16 = 4; // v4: precomputed OFI deltas + book_aggression + log_duration in ofi[8..18) /// Feature vector dimensionality. const FEAT_DIM: usize = 42; @@ -112,10 +112,10 @@ impl FxCacheHeader { self.magic ); } - // Accept v2 (legacy 4-target) and v3 (current 6-target) - if self.version != FXCACHE_VERSION && self.version != 2 { + // Accept v2 (legacy 4-target), v3 (6-target, no precomputed deltas), and v4 (current) + if self.version != FXCACHE_VERSION && self.version != 2 && self.version != 3 { bail!( - "Stale FxCache version: {} (expected {} or 2). Delete and regenerate.", + "Stale FxCache version: {} (expected {}, 3, or 2). Delete and regenerate.", self.version, FXCACHE_VERSION ); } @@ -366,18 +366,31 @@ pub fn load_fxcache(path: &Path) -> Result { ); } - // Read body (f32 format) — handles v2 (4-target) and v3 (6-target) - let (timestamps, features, targets, ofi) = if is_legacy_v2 { + // Read body (f32 format) — handles v2 (4-target), v3 (6-target, microstructure OFI), and v4 (current) + let (timestamps, features, targets, mut ofi) = if is_legacy_v2 { read_body_f32_legacy_v2(&mut reader, bar_count)? } else { read_body_f32(&mut reader, bar_count)? }; + // v3 files have microstructure features at ofi[8..18) instead of the + // precomputed deltas/book_aggression/log_duration that v4 expects. + // Zero out those positions so the OFI embed MLP gets safe zeros + // rather than semantically wrong microstructure data. + let is_legacy_v3 = header.version == 3; + if is_legacy_v3 { + for row in &mut ofi { + for k in 8..18 { + row[k] = 0.0; + } + } + } + info!( "FxCache loaded: {} bars, v{} (f32{}) from {:?}", bar_count, header.version, - if is_legacy_v2 { ", legacy 4→6 padded" } else { "" }, + if is_legacy_v2 { ", legacy 4→6 padded" } else if is_legacy_v3 { ", legacy v3 ofi[8..18) zeroed" } else { "" }, path ); @@ -394,7 +407,7 @@ pub fn load_fxcache(path: &Path) -> Result { }) } -/// Read body in f32 format (v3 / current), converting f32 back to f64 on load. +/// Read body in f32 format (v3/v4 / current), converting f32 back to f64 on load. fn read_body_f32( reader: &mut BufReader, bar_count: usize, diff --git a/crates/ml/src/trainers/dqn/data_loading.rs b/crates/ml/src/trainers/dqn/data_loading.rs index c498b994f..1a2b993e8 100644 --- a/crates/ml/src/trainers/dqn/data_loading.rs +++ b/crates/ml/src/trainers/dqn/data_loading.rs @@ -400,6 +400,22 @@ impl DQNTrainer { let arr8 = features.to_array(); let mut arr20 = [0.0_f64; 20]; arr20[..8].copy_from_slice(&arr8); + // Book aggression from MBP-10 depth + let mut buy_ws = 0.0_f64; + let mut buy_t = 0.0_f64; + let mut sell_ws = 0.0_f64; + let mut sell_t = 0.0_f64; + for (lv, pair) in snap.levels.iter().enumerate().take(10) { + let bs = pair.bid_sz as f64; + let as_ = pair.ask_sz as f64; + buy_ws += (lv + 1) as f64 * bs; + buy_t += bs; + sell_ws += (lv + 1) as f64 * as_; + sell_t += as_; + } + let buy_com = if buy_t > 0.0 { buy_ws / buy_t } else { 5.5 }; + let sell_com = if sell_t > 0.0 { sell_ws / sell_t } else { 5.5 }; + arr20[16] = (sell_com - buy_com) / 10.0; ofi_per_bar.push(arr20); } else { ofi_per_bar.push([0.0; 20]); @@ -412,6 +428,25 @@ impl DQNTrainer { } } + // Compute OFI temporal deltas: delta[bar] = ofi[bar] - ofi[bar-1] + for i in (1..ofi_per_bar.len()).rev() { + for k in 0..8 { + ofi_per_bar[i][8 + k] = ofi_per_bar[i][k] - ofi_per_bar[i - 1][k]; + } + } + + // Compute log bar duration + for i in 0..ofi_per_bar.len() { + let duration_secs = if i > 0 { + let ts_cur = all_ohlcv_bars[i + WARMUP].timestamp; + let ts_prev = all_ohlcv_bars[i + WARMUP - 1].timestamp; + (ts_cur - ts_prev).num_seconds() as f64 + } else { + 60.0 + }; + ofi_per_bar[i][17] = duration_secs.max(0.1).ln() / 10.0; + } + let non_zero = ofi_per_bar.iter().filter(|f| f.iter().any(|&v| v != 0.0)).count(); info!("Per-bar OFI computed: {} total, {} non-zero ({:.1}%)", ofi_per_bar.len(), non_zero, diff --git a/crates/ml/src/trainers/dqn/fused_training.rs b/crates/ml/src/trainers/dqn/fused_training.rs index 0066fbf6a..140ed18f9 100644 --- a/crates/ml/src/trainers/dqn/fused_training.rs +++ b/crates/ml/src/trainers/dqn/fused_training.rs @@ -2449,6 +2449,21 @@ impl FusedTrainingCtx { self.trainer.params().raw_ptr() } + /// Read one sample's state vector from the GPU states_buf (DtoH, 96 floats). + /// Diagnostic use only -- negligible cost. Returns the padded state vector. + pub(crate) fn read_state_sample(&self, sample_idx: usize) -> anyhow::Result> { + let sd = self.trainer.config().state_dim; + let sd_padded = (sd + 127) & !127; + let states = self.trainer.states_buf(); + let host: Vec = self.stream.clone_dtoh(states) + .map_err(|e| anyhow::anyhow!("states DtoH for diagnostic: {e}"))?; + let offset = sample_idx * sd_padded; + if offset + sd > host.len() { + anyhow::bail!("sample_idx {} out of range (host len={}, sd={})", sample_idx, host.len(), sd); + } + Ok(host[offset..offset + sd].to_vec()) + } + /// Compute a simple checksum of the flat params buffer (GPU→CPU readback). /// For determinism diagnostics only — synchronizes the stream. pub(crate) fn params_checksum(&self) -> anyhow::Result { diff --git a/crates/ml/src/trainers/dqn/trainer/constructor.rs b/crates/ml/src/trainers/dqn/trainer/constructor.rs index 0b314ce40..392558e8a 100644 --- a/crates/ml/src/trainers/dqn/trainer/constructor.rs +++ b/crates/ml/src/trainers/dqn/trainer/constructor.rs @@ -207,9 +207,13 @@ impl DQNTrainer { // Create DQN configuration // State layout (GPU pipeline): // [0..42) 42 market features (OHLCV, technical, patterns, volume, time, statistical) - // [42..50) 8 portfolio features (position, P&L, drawdown, etc.) - // [50..66) 16 multi-timeframe features (4 windows x 4 features) - // [66..86) 20 OFI features (from MBP-10 order book data, always loaded) + // [42..56) 14 portfolio features (8 base + 6 plan) + // [56..72) 16 multi-timeframe features (4 windows x 4 features) + // [66..74) 8 raw OFI features (bid_ask_spread, depth_imbalance, ..., vpin, kyle_lambda) + // [74..82) 8 OFI temporal deltas (delta[bar] = ofi[bar] - ofi[bar-1]) + // [82] book_aggression (center-of-mass asymmetry from MBP-10 depth) + // [83] log_bar_duration (ln(duration_secs) / 10.0) + // [84..86) 2 remaining microstructure (ofi_acceleration, toxicity_gradient) // Raw state_dim: 92 (42+14+16+20). OFI always enabled. // market_dim: always 42 (bottleneck compresses only base market features). // OFI features bypass bottleneck via portfolio_dim (fed directly to shared trunk).