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

fix(ofi): align per-bar OFI features in DBN training path and walk-forward evaluator

Browse Source 
The DBN loading path (load_training_data) never computed per-bar OFI —
it fell through to load_ofi_features_parallel() which returns one OFI
per MBP-10 snapshot (~14.5M for 895K bars). upload_ofi() then truncated
to num_bars, misaligning snapshot-level OFI with bar-level data.

- Add per-bar OFI computation to load_training_data() matching the
  Parquet path: iterate OHLCV bars, find nearest MBP-10 snapshot,
  calculate 8 OFI features via OFICalculator
- preload_data() now prefers loader's per-bar OFI over snapshot-level
  parallel loader
- evaluate_gpu() walk-forward uses real OFI with offset indexing
  instead of zero-padding 8 dimensions

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
jgrusewski
2026-03-12 19:44:24 +01:00
parent 1ed97579c1
commit 68eefbfc9e
2 changed files with 147 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

49
crates/ml/src/hyperopt/adapters/dqn.rs
View File

@@ -1280,12 +1280,16 @@ impl DQNTrainer {
.block_on(loader.load_training_data(data_path_str))
.map_err(|e| MLError::TrainingError(format!("Failed to preload data: {}", e)))?;
// Load OFI features from MBP-10 data (separate from OHLCV loading).
// The internal trainer's load_training_data() only loads OHLCV — OFI must be loaded
// explicitly via the adapter's load_ofi_features() which uses self.mbp10_data_dir.
let ofi_features = self.load_ofi_features();
// Extract per-bar OFI features computed by load_training_data().
// These are properly aligned 1:1 with OHLCV bars (NOT per-MBP-10-snapshot).
// Falls back to the parallel snapshot-level loader if the internal trainer
// didn't compute per-bar OFI (e.g. MBP-10 files unavailable at load time).
let ofi_features = loader.ofi_features.take().or_else(|| {
debug!("Per-bar OFI not computed by loader, trying parallel snapshot loader");
self.load_ofi_features()
});
if let Some(ref ofi) = ofi_features {
info!("OFI features preloaded: {} bars x 8 dims (VPIN, Kyle's Lambda, OFI, trade imbalance)", ofi.len());
info!("OFI features preloaded: {} bars x 8 dims (per-bar aligned)", ofi.len());
} else if self.mbp10_data_dir.is_some() {
warn!("MBP-10 data dir configured but no OFI features loaded — state_dim mismatch likely");
} else {
@@ -1680,6 +1684,27 @@ impl DQNTrainer {
let market_dim: usize = 42;
let feature_dim: usize = if ofi_enabled { 50 } else { market_dim };
// OFI feature overlay: preloaded OFI covers ALL bars (train+val), left-aligned.
// Validation bars start at index `train_len` in the global OFI array.
let ofi_offset = self
.preloaded_training_data
.as_ref()
.map_or(0, |d| d.len());
let ofi_ref = self.preloaded_ofi_features.as_deref();
if ofi_enabled {
let ofi_total = ofi_ref.map_or(0, |o| o.len());
let ofi_avail = ofi_total.saturating_sub(ofi_offset);
debug!(
ofi_offset,
ofi_total,
ofi_avail,
total_bars,
"walk-forward OFI overlay: {} of {} val bars have real OFI",
ofi_avail.min(total_bars),
total_bars,
);
}
let mut window_prices = Vec::with_capacity(window_count);
let mut window_features = Vec::with_capacity(window_count);
@@ -1711,11 +1736,17 @@ impl DQNTrainer {
))
})?;
let mut fv_f32: Vec<f32> = fv_slice.iter().map(|&v| v as f32).collect();
// For OFI models, zero-pad from 42 to 50. The model was trained with
// real OFI at 45-52 but zeros are the safe fallback (matches the
// `ofi_enabled && data_missing` path in feature_vector_to_state_with_ofi).
// Overlay real OFI features from preloaded MBP-10 data.
// Falls back to zero-padding only when OFI data is unavailable.
if ofi_enabled {
fv_f32.resize(feature_dim, 0.0);
let ofi_idx = ofi_offset + i;
if let Some(ofi) = ofi_ref.filter(|o| ofi_idx < o.len()) {
for &v in ofi.get(ofi_idx).iter().flat_map(|f| f.iter()) {
fv_f32.push(v as f32);
}
} else {
fv_f32.resize(feature_dim, 0.0);
}
}
features.push(fv_f32);
}

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

@@ -670,6 +670,113 @@ impl DQNTrainer {
feature_vectors.len()
);
// Compute per-bar OFI features from MBP-10 snapshots (8 features per bar)
if let Some(ref mbp10_dir_str) = self.hyperparams.mbp10_data_dir {
use crate::features::ofi_calculator::OFICalculator;
use crate::features::mbp10_loader::get_snapshots_for_timestamp;
use crate::features::trades_loader::get_trades_for_bar;
use data::providers::databento::dbn_parser::DbnParser;
let mbp10_dir = Path::new(mbp10_dir_str);
if mbp10_dir.exists() {
info!("Loading MBP-10 snapshots for per-bar OFI...");
let mut dbn_files = Vec::new();
collect_dbn_files_recursive(mbp10_dir)
.iter()
.for_each(|f| dbn_files.push(f.clone()));
dbn_files.sort();
if !dbn_files.is_empty() {
let mut all_snapshots = Vec::new();
if let Ok(ref parser) = DbnParser::new() {
for file in &dbn_files {
match parser.parse_mbp10_file(file).await {
Ok(snaps) => {
info!(" Loaded {} MBP-10 snapshots from {:?}", snaps.len(), file.file_name());
all_snapshots.extend(snaps);
}
Err(e) => warn!(" Failed to load MBP-10 from {:?}: {}", file.file_name(), e),
}
}
}
if !all_snapshots.is_empty() {
all_snapshots.sort_by_key(|s| s.timestamp);
info!("Computing per-bar OFI from {} MBP-10 snapshots...", all_snapshots.len());
// Load trade data for VPIN/Kyle's Lambda enrichment
let trades = if let Some(ref trades_dir_str) = self.hyperparams.trades_data_dir {
use crate::features::trades_loader::load_trades_sync;
let trades_dir = Path::new(trades_dir_str);
if trades_dir.exists() {
let mut all_trades = Vec::new();
let mut trade_files = Vec::new();
collect_dbn_files_recursive(trades_dir)
.iter()
.for_each(|f| trade_files.push(f.clone()));
trade_files.sort();
for path in &trade_files {
if let Ok(mut t) = load_trades_sync(path) {
all_trades.append(&mut t);
}
}
(!all_trades.is_empty()).then(|| {
all_trades.sort_by_key(|t| t.timestamp);
all_trades
})
} else {
None
}
} else {
None
};
let mut ofi_calculator = OFICalculator::new();
let mut ofi_per_bar = Vec::with_capacity(feature_vectors.len());
const WARMUP: usize = 50;
for i in 0..feature_vectors.len() {
let bar = &all_ohlcv_bars[i + WARMUP];
let bar_ts = bar.timestamp.timestamp_nanos_opt().unwrap_or(0) as u64;
if let Some(ref all_trades) = trades {
let bar_end_ts = all_ohlcv_bars
.get(i + WARMUP + 1)
.map(|b| b.timestamp.timestamp_nanos_opt().unwrap_or(0) as u64)
.unwrap_or(bar_ts + 60_000_000_000);
let bar_trades = get_trades_for_bar(all_trades, bar_ts, bar_end_ts);
for trade in bar_trades {
ofi_calculator.feed_trade(trade.price, trade.volume, trade.is_buy);
}
}
let window = get_snapshots_for_timestamp(&all_snapshots, bar_ts, 1);
if let Some(snap) = window.first() {
match ofi_calculator.calculate(snap) {
Ok(features) => {
if features.is_valid() {
ofi_per_bar.push(features.to_array());
} else {
ofi_per_bar.push([0.0; 8]);
}
}
Err(_) => ofi_per_bar.push([0.0; 8]),
}
} else {
ofi_per_bar.push([0.0; 8]);
}
}
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,
if ofi_per_bar.is_empty() { 0.0 } else { non_zero as f64 / ofi_per_bar.len() as f64 * 100.0 });
self.ofi_features = Some(ofi_per_bar);
}
}
}
}
// Create training data pairs (features, target)
// Target: [current_close, next_close] for proper reward calculation
let mut training_data = Vec::new();