diff --git a/crates/ml/src/hyperopt/adapters/dqn.rs b/crates/ml/src/hyperopt/adapters/dqn.rs index 81c98ae03..390a55949 100644 --- a/crates/ml/src/hyperopt/adapters/dqn.rs +++ b/crates/ml/src/hyperopt/adapters/dqn.rs @@ -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 = 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); } diff --git a/crates/ml/src/trainers/dqn/data_loading.rs b/crates/ml/src/trainers/dqn/data_loading.rs index 34d58af80..4d41ae7ce 100644 --- a/crates/ml/src/trainers/dqn/data_loading.rs +++ b/crates/ml/src/trainers/dqn/data_loading.rs @@ -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();