From 689231d6cb5af03cdbaa0a7fa3e8e5c09edfa6b2 Mon Sep 17 00:00:00 2001 From: jgrusewski Date: Sat, 21 Feb 2026 12:40:41 +0100 Subject: [PATCH] feat(data): integrate MBP10/OFI features into training data pipeline Add load_ofi_features() helper to both DQN and PPO trainer adapters that loads MBP10 snapshots from a sibling mbp10/ directory, computes 8-slot OFI feature vectors via OFICalculator, and overlays them onto positions 43-50 of the training feature arrays. Gracefully falls back to zero-padded features when MBP10 data is not available. Co-Authored-By: Claude Opus 4.6 --- ml/src/hyperopt/adapters/dqn.rs | 93 +++++++++++++++++++++++++++-- ml/src/hyperopt/adapters/ppo.rs | 101 ++++++++++++++++++++++++++++++-- 2 files changed, 186 insertions(+), 8 deletions(-) diff --git a/ml/src/hyperopt/adapters/dqn.rs b/ml/src/hyperopt/adapters/dqn.rs index 386d08b85..e8a090f50 100644 --- a/ml/src/hyperopt/adapters/dqn.rs +++ b/ml/src/hyperopt/adapters/dqn.rs @@ -1318,11 +1318,77 @@ impl DQNTrainer { self.extract_features_and_targets(&all_bars) } + /// Attempt to load OFI features from MBP10 data. + /// Returns None if MBP10 data is not available or on error. + fn load_ofi_features(&self) -> Option> { + use crate::features::mbp10_loader::load_mbp10_snapshots_sync; + use crate::features::ofi_calculator::OFICalculator; + + let mbp10_dir = std::path::Path::new(&self.dbn_data_dir) + .parent() + .map(|p| p.join("mbp10"))?; + + if !mbp10_dir.exists() { + info!( + "No MBP10 directory at {}, OFI features will be zero-padded", + mbp10_dir.display() + ); + return None; + } + + let mbp10_files = collect_dbn_files_recursive(&mbp10_dir); + if mbp10_files.is_empty() { + info!("No MBP10 .dbn files found in {}", mbp10_dir.display()); + return None; + } + + info!( + "Loading OFI features from {} MBP10 files", + mbp10_files.len() + ); + + let mut calculator = OFICalculator::new(); + let mut all_ofi: Vec<[f64; 8]> = Vec::new(); + + for file in &mbp10_files { + match load_mbp10_snapshots_sync(file) { + Ok(snapshots) => { + for snapshot in &snapshots { + match calculator.calculate(snapshot) { + Ok(features) => all_ofi.push(features.to_array()), + Err(e) => { + tracing::warn!("OFI calculation failed: {}", e); + } + } + } + } + Err(e) => { + tracing::warn!( + "Failed to load MBP10 file {}: {}", + file.display(), + e + ); + } + } + } + + if all_ofi.is_empty() { + return None; + } + + info!( + "Computed {} OFI feature vectors from MBP10 data", + all_ofi.len() + ); + Some(all_ofi) + } + /// Extract 51-feature vectors from OHLCV bars and create training data /// /// Uses the production feature extraction API (extract_ml_features) to /// generate 51-feature vectors from OHLCV bars. Creates dummy rewards /// for DQN training (actual rewards are computed during training). + /// When MBP10 data is available, OFI features are overlaid at positions 43-50. /// /// # Arguments /// @@ -1364,14 +1430,33 @@ impl DQNTrainer { info!("Extracted {} feature vectors", feature_vectors.len()); + // Load OFI features if MBP10 data is available + let ofi_features = self.load_ofi_features(); + let ofi_count = ofi_features.as_ref().map_or(0, |v| v.len()); + if ofi_count > 0 { + info!( + "Overlaying {} OFI features onto positions 43-50", + ofi_count + ); + } + // Convert to [f32; 54] and create dummy rewards (actual rewards computed during training) let training_data: Vec<([f32; 54], f64)> = feature_vectors .into_iter() - .map(|vec_f64| { - // Convert [f64; 54] to [f32; 54] + .enumerate() + .map(|(i, vec_f64)| { + // Convert [f64; 51] to [f32; 54] (51 market features + 3 portfolio state zeros) let mut vec_f32 = [0.0_f32; 54]; - for (i, &val) in vec_f64.iter().enumerate() { - vec_f32[i] = val as f32; + for (j, &val) in vec_f64.iter().enumerate() { + vec_f32[j] = val as f32; + } + // Overlay OFI features at positions 43-50 if available + if let Some(ref ofi) = ofi_features { + if let Some(ofi_arr) = ofi.get(i) { + for (j, &val) in ofi_arr.iter().enumerate() { + vec_f32[43 + j] = val as f32; + } + } } (vec_f32, 0.0_f64) // Dummy reward (actual rewards computed during training) }) diff --git a/ml/src/hyperopt/adapters/ppo.rs b/ml/src/hyperopt/adapters/ppo.rs index ed3a332ad..09e45486e 100644 --- a/ml/src/hyperopt/adapters/ppo.rs +++ b/ml/src/hyperopt/adapters/ppo.rs @@ -766,7 +766,73 @@ impl PPOTrainer { self.extract_features_and_targets(&all_bars) } - /// Extract 51-feature vectors and targets from OHLCV bars + /// Attempt to load OFI features from MBP10 data. + /// Returns None if MBP10 data is not available or on error. + fn load_ofi_features(&self) -> Option> { + use crate::features::mbp10_loader::load_mbp10_snapshots_sync; + use crate::features::ofi_calculator::OFICalculator; + + let mbp10_dir = std::path::Path::new(&self.dbn_data_dir) + .parent() + .map(|p| p.join("mbp10"))?; + + if !mbp10_dir.exists() { + info!( + "No MBP10 directory at {}, OFI features will be zero-padded", + mbp10_dir.display() + ); + return None; + } + + let mbp10_files = collect_dbn_files_recursive(&mbp10_dir); + if mbp10_files.is_empty() { + info!("No MBP10 .dbn files found in {}", mbp10_dir.display()); + return None; + } + + info!( + "Loading OFI features from {} MBP10 files", + mbp10_files.len() + ); + + let mut calculator = OFICalculator::new(); + let mut all_ofi: Vec<[f64; 8]> = Vec::new(); + + for file in &mbp10_files { + match load_mbp10_snapshots_sync(file) { + Ok(snapshots) => { + for snapshot in &snapshots { + match calculator.calculate(snapshot) { + Ok(features) => all_ofi.push(features.to_array()), + Err(e) => { + tracing::warn!("OFI calculation failed: {}", e); + } + } + } + } + Err(e) => { + tracing::warn!( + "Failed to load MBP10 file {}: {}", + file.display(), + e + ); + } + } + } + + if all_ofi.is_empty() { + return None; + } + + info!( + "Computed {} OFI feature vectors from MBP10 data", + all_ofi.len() + ); + Some(all_ofi) + } + + /// Extract 51-feature vectors and targets from OHLCV bars. + /// When MBP10 data is available, OFI features are overlaid at positions 43-50. fn extract_features_and_targets( &self, ohlcv_bars: &[crate::features::extraction::OHLCVBar], @@ -786,15 +852,25 @@ impl PPOTrainer { )); } - // Extract features using production API (returns Vec<[f64; 54]>) + // Extract features using production API (returns Vec<[f64; 51]>) let feature_vectors = extract_ml_features(ohlcv_bars) .map_err(|e| anyhow::anyhow!("Feature extraction failed: {}", e))?; info!( - "Extracted {} feature vectors (54 dimensions each)", + "Extracted {} feature vectors (51 dimensions each)", feature_vectors.len() ); + // Load OFI features if MBP10 data is available + let ofi_features = self.load_ofi_features(); + let ofi_count = ofi_features.as_ref().map_or(0, |v| v.len()); + if ofi_count > 0 { + info!( + "Overlaying {} OFI features onto positions 43-50", + ofi_count + ); + } + // Create training data pairs (features, target) // Target: Next bar's close price (autoregressive prediction) // Features are 51-dim market + 3 zero-padded portfolio state = 54 total @@ -811,17 +887,34 @@ impl PPOTrainer { for (j, &val) in feature_vectors[i].iter().enumerate() { feature_array[j] = val as f32; } + // Overlay OFI features at positions 43-50 if available + if let Some(ref ofi) = ofi_features { + if let Some(ofi_arr) = ofi.get(i) { + for (j, &val) in ofi_arr.iter().enumerate() { + feature_array[43 + j] = val as f32; + } + } + } training_data.push((feature_array, next_close)); } } // Last sample targets itself if !feature_vectors.is_empty() { + let last_idx = feature_vectors.len() - 1; let last_close = ohlcv_bars[ohlcv_bars.len() - 1].close; let mut feature_array = [0.0f32; 54]; - for (j, &val) in feature_vectors[feature_vectors.len() - 1].iter().enumerate() { + for (j, &val) in feature_vectors[last_idx].iter().enumerate() { feature_array[j] = val as f32; } + // Overlay OFI features at positions 43-50 if available + if let Some(ref ofi) = ofi_features { + if let Some(ofi_arr) = ofi.get(last_idx) { + for (j, &val) in ofi_arr.iter().enumerate() { + feature_array[43 + j] = val as f32; + } + } + } training_data.push((feature_array, last_close)); }