gitadmin/foxhunt
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

feat: precompute OFI deltas + book aggression + bar duration in fxcache

Browse Source 
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) <noreply@anthropic.com>
This commit is contained in:
jgrusewski
2026-04-20 08:11:03 +02:00
parent f961b6ad64
commit 429967dcd1
5 changed files with 152 additions and 16 deletions
Download Patch File Download Diff File Expand all files Collapse all files

75
crates/ml/examples/precompute_features.rs
View File

@@ -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(&micro_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;

33
crates/ml/src/fxcache.rs
View File

@@ -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<FxCacheData> {
);
}
// 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<FxCacheData> {
})
}
/// 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<std::fs::File>,
bar_count: usize,

35
crates/ml/src/trainers/dqn/data_loading.rs
View File

@@ -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,

15
crates/ml/src/trainers/dqn/fused_training.rs
View File

@@ -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<Vec<f32>> {
let sd = self.trainer.config().state_dim;
let sd_padded = (sd + 127) & !127;
let states = self.trainer.states_buf();
let host: Vec<f32> = 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<u64> {

10
crates/ml/src/trainers/dqn/trainer/constructor.rs
View File

@@ -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).