From 7a0238296515e8da538e7ff07e2a305c7cef8959 Mon Sep 17 00:00:00 2001 From: jgrusewski Date: Fri, 20 Feb 2026 14:16:38 +0100 Subject: [PATCH] refactor(ml): split DQN trainer.rs into sub-modules Extract ~1,300 lines from the 4,755-line trainer.rs into four focused sub-modules to improve maintainability and code navigation: - monitoring.rs (290 lines): TrainingMonitor for per-epoch reward, action, and Q-value tracking with health validation - data_loading.rs (719 lines): Parquet/DBN data loading, MBP-10 OFI integration, feature caching, and preprocessing pipeline - risk.rs (145 lines): Volatility-adjusted epsilon, risk-adjusted rewards, Kelly criterion sizing, and risk tracker updates - features.rs (211 lines): Full feature extraction (51-dim), feature statistics calculation, z-score normalization, synthetic features All public API paths preserved via mod.rs re-exports. Fields accessed across module boundaries changed to pub(crate) visibility. Verified: cargo check --workspace (0 errors), cargo test -p ml --lib (1,817 passed, 0 failed). Co-Authored-By: Claude Opus 4.6 --- ml/src/trainers/dqn/data_loading.rs | 719 ++++++++++++++ ml/src/trainers/dqn/features.rs | 211 +++++ ml/src/trainers/dqn/mod.rs | 10 +- ml/src/trainers/dqn/monitoring.rs | 290 ++++++ ml/src/trainers/dqn/risk.rs | 145 +++ ml/src/trainers/dqn/trainer.rs | 1338 +-------------------------- 6 files changed, 1392 insertions(+), 1321 deletions(-) create mode 100644 ml/src/trainers/dqn/data_loading.rs create mode 100644 ml/src/trainers/dqn/features.rs create mode 100644 ml/src/trainers/dqn/monitoring.rs create mode 100644 ml/src/trainers/dqn/risk.rs diff --git a/ml/src/trainers/dqn/data_loading.rs b/ml/src/trainers/dqn/data_loading.rs new file mode 100644 index 000000000..4ba04bc49 --- /dev/null +++ b/ml/src/trainers/dqn/data_loading.rs @@ -0,0 +1,719 @@ +//! Data Loading for DQN Training +//! +//! Methods for loading market data from Parquet files and DBN files, +//! including feature extraction, preprocessing, and train/val splitting. + +use std::path::Path; + +use anyhow::{Context, Result}; +use candle_core::{Device, Tensor}; +use tracing::{debug, info, warn}; + +use crate::features::extraction::OHLCVBar; +use crate::preprocessing::{preprocess_prices, PreprocessConfig}; +use crate::training_pipeline::FinancialFeatures; +use crate::TrainingMetrics; + +use super::FeatureVector51; +use super::trainer::DQNTrainer; + +impl DQNTrainer { + /// Train DQN on market data from Parquet file (Wave 12 Group 3) + /// + /// # Arguments + /// + /// * `parquet_path` - Path to Parquet file containing OHLCV bars + /// * `checkpoint_callback` - Callback for saving checkpoints (epoch, model_data) -> `Result` + /// + /// # Returns + /// + /// Training metrics (loss, accuracy, gradient norms, Q-values) + pub async fn train_from_parquet( + &mut self, + parquet_path: &str, + checkpoint_callback: F, + ) -> Result + where + F: FnMut(usize, Vec, bool) -> Result + Send, + { + info!("Starting DQN training from Parquet file: {}", parquet_path); + + // Load market data from Parquet file (returns train/val split) + let (mut training_data, mut validation_data) = + self.load_training_data_from_parquet(parquet_path).await?; + + info!( + "Loaded {} training samples, {} validation samples", + training_data.len(), + validation_data.len() + ); + + + // Calculate feature statistics from all training samples + info!("📊 Calculating feature statistics from {} training samples...", training_data.len()); + let feature_stats = self.calculate_feature_statistics(&training_data)?; + self.feature_stats = Some(feature_stats.clone()); + info!("✅ Feature statistics calculated: {} features normalized", feature_stats.mean.len()); + + // Normalize all training and validation samples BEFORE training starts + info!("📊 Normalizing all samples with z-score normalization..."); + self.normalize_dataset(&mut training_data)?; + self.normalize_dataset(&mut validation_data)?; + info!("✅ Dataset normalization complete"); + + // Store normalized validation data for validation loss computation + self.val_data = validation_data; + + // Use the same training loop as DBN-based training + self.train_with_data_full_loop(training_data, checkpoint_callback) + .await + } + + /// Load training data from Parquet file (Wave 12 Group 3) + /// Returns (train_data, val_data) with 80/20 split + pub async fn load_training_data_from_parquet( + &mut self, + parquet_path: &str, + ) -> Result<( + Vec<(FeatureVector51, Vec)>, + Vec<(FeatureVector51, Vec)>, + )> { + use arrow::array::{Array, Float64Array, PrimitiveArray, UInt64Array}; + use arrow::datatypes::TimestampNanosecondType; + use arrow::record_batch::RecordBatch; + use parquet::arrow::arrow_reader::ParquetRecordBatchReaderBuilder; + use std::fs::File; + + info!("Loading Parquet file: {}", parquet_path); + + // TRY CACHE FIRST + if let Some(cache_dir) = &self.feature_cache_dir { + let parquet_path_obj = Path::new(parquet_path); + let mbp10 = Path::new("test_data/mbp10"); + + info!("🔍 Checking feature cache..."); + + // Calculate cache key + match crate::feature_cache::calculate_cache_key( + parquet_path_obj, + mbp10, + 50, // warmup period + ) { + Ok(cache_key) => { + // Try to load from cache + match crate::feature_cache::load_features_from_cache( + cache_dir, + &cache_key, + ).await { + Ok(Some(features)) => { + info!("🚀 Loaded {} feature vectors from cache", features.len()); + info!(" ⚡ Savings vs compute: ~2m 25s"); + + // Split into train/val (80/20) + let split_idx = (features.len() as f64 * 0.8) as usize; + let train_data: Vec<(FeatureVector51, Vec)> = features[..split_idx] + .iter() + .map(|f| (*f, vec![])) + .collect(); + let val_data: Vec<(FeatureVector51, Vec)> = features[split_idx..] + .iter() + .map(|f| (*f, vec![])) + .collect(); + + return Ok((train_data, val_data)); + } + Ok(None) => { + info!("⚠️ Cache miss, computing features from scratch..."); + } + Err(e) => { + warn!("⚠️ Cache load failed: {}, computing features...", e); + } + } + } + Err(e) => { + warn!("⚠️ Failed to calculate cache key: {}, skipping cache", e); + } + } + } + + // FALLBACK: Original feature extraction + info!("📊 Computing features from scratch..."); + + // Open Parquet file + let file = File::open(parquet_path) + .with_context(|| format!("Failed to open Parquet file: {}", parquet_path))?; + + // Create Parquet reader + let builder = ParquetRecordBatchReaderBuilder::try_new(file) + .with_context(|| "Failed to create Parquet reader")?; + + let reader = builder + .build() + .with_context(|| "Failed to build Parquet reader")?; + + // Read all batches + let mut all_ohlcv_bars = Vec::new(); + + for batch_result in reader { + let batch: RecordBatch = batch_result.with_context(|| "Failed to read record batch")?; + + // Extract columns by name (schema-agnostic approach) + // Required columns: timestamp_ns (or ts_event), open, high, low, close, volume + + // Try timestamp_ns first (our schema), fallback to ts_event (Databento schema) + let timestamp_col = batch + .column_by_name("timestamp_ns") + .or_else(|| batch.column_by_name("ts_event")) + .ok_or_else(|| { + anyhow::anyhow!( + "Missing timestamp column. Expected 'timestamp_ns' or 'ts_event'" + ) + })?; + + let timestamps = timestamp_col + .as_any() + .downcast_ref::>() + .ok_or_else(|| { + anyhow::anyhow!( + "Failed to downcast timestamp column. Expected Timestamp(Nanosecond), got: {:?}", + timestamp_col.data_type() + ) + })?; + + // Extract OHLCV columns by name + let opens = batch + .column_by_name("open") + .ok_or_else(|| anyhow::anyhow!("Missing 'open' column in Parquet schema"))? + .as_any() + .downcast_ref::() + .ok_or_else(|| anyhow::anyhow!("Invalid 'open' column type. Expected Float64"))?; + + let highs = batch + .column_by_name("high") + .ok_or_else(|| anyhow::anyhow!("Missing 'high' column in Parquet schema"))? + .as_any() + .downcast_ref::() + .ok_or_else(|| anyhow::anyhow!("Invalid 'high' column type. Expected Float64"))?; + + let lows = batch + .column_by_name("low") + .ok_or_else(|| anyhow::anyhow!("Missing 'low' column in Parquet schema"))? + .as_any() + .downcast_ref::() + .ok_or_else(|| anyhow::anyhow!("Invalid 'low' column type. Expected Float64"))?; + + let closes = batch + .column_by_name("close") + .ok_or_else(|| anyhow::anyhow!("Missing 'close' column in Parquet schema"))? + .as_any() + .downcast_ref::() + .ok_or_else(|| anyhow::anyhow!("Invalid 'close' column type. Expected Float64"))?; + + let volumes = batch + .column_by_name("volume") + .ok_or_else(|| anyhow::anyhow!("Missing 'volume' column in Parquet schema"))? + .as_any() + .downcast_ref::() + .ok_or_else(|| anyhow::anyhow!("Invalid 'volume' column type. Expected UInt64"))?; + + // Convert to OHLCVBar structs + for i in 0..batch.num_rows() { + let timestamp_ns = timestamps.value(i); + let timestamp = chrono::DateTime::from_timestamp_nanos(timestamp_ns); + + let bar = OHLCVBar { + timestamp, + open: opens.value(i), + high: highs.value(i), + low: lows.value(i), + close: closes.value(i), + volume: volumes.value(i) as f64, // Convert u64 to f64 + }; + all_ohlcv_bars.push(bar); + } + } + + info!( + "Successfully loaded {} OHLCV bars from Parquet file", + all_ohlcv_bars.len() + ); + + // Sort bars by timestamp (critical for rolling window feature extraction) + debug!("Sorting bars chronologically by timestamp..."); + all_ohlcv_bars.sort_by_key(|bar| bar.timestamp); + debug!("Bars sorted successfully"); + + // Wave 14 Agent 32: Preprocess close prices for stationarity + let preprocessed_closes = if self.hyperparams.enable_preprocessing { + info!("🔬 Preprocessing enabled: Applying log returns + windowed normalization + outlier clipping"); + + // Extract close prices + // WAVE 16E: Convert f64 to f32 for preprocessing (preprocessing expects f32 tensors) + let close_prices_f64: Vec = all_ohlcv_bars.iter().map(|b| b.close).collect(); + let close_prices_f32: Vec = close_prices_f64.iter().map(|&x| x as f32).collect(); + let device = Device::cuda_if_available(0).unwrap_or(Device::Cpu); + let close_tensor = + Tensor::from_slice(&close_prices_f32, (close_prices_f32.len(),), &device) + .context("Failed to create close price tensor")?; + + // Configure preprocessing + let preprocess_config = PreprocessConfig { + window_size: self.hyperparams.preprocessing_window, + clip_sigma: self.hyperparams.preprocessing_clip_sigma, + use_log_returns: true, + }; + + info!(" • Window size: {}", preprocess_config.window_size); + info!(" • Clip sigma: ±{:.1}σ", preprocess_config.clip_sigma); + + // Apply preprocessing + let preprocessed_tensor = preprocess_prices(&close_tensor, preprocess_config) + .context("Failed to preprocess prices")?; + + let preprocessed_vec: Vec = preprocessed_tensor + .to_vec1() + .context("Failed to convert preprocessed tensor to vec")?; + + // Convert f32 to f64 for consistency with existing pipeline + let preprocessed_f64: Vec = preprocessed_vec.iter().map(|&x| x as f64).collect(); + + // Compute statistics for validation + let warmup = preprocess_config.window_size as usize; + let post_warmup: Vec = preprocessed_f64[warmup..].to_vec(); + let mean = post_warmup.iter().sum::() / post_warmup.len() as f64; + let variance = post_warmup.iter().map(|&x| (x - mean).powi(2)).sum::() + / post_warmup.len() as f64; + let std = variance.sqrt(); + let max_abs = post_warmup.iter().map(|&x| x.abs()).fold(0.0f64, f64::max); + + debug!("✅ Preprocessing complete:"); + debug!(" • Mean: {:.6} (expected ~0 for normalized data)", mean); + debug!(" • Std: {:.4} (expected ~1 for normalized data)", std); + debug!( + " • Max absolute value: {:.4} (clipped at ±{:.1}σ)", + max_abs, preprocess_config.clip_sigma + ); + + Some(preprocessed_f64) + } else { + info!("⚠️ Preprocessing disabled: Using raw close prices (NON-STATIONARY)"); + None + }; + + // WAVE 2-A2: Load MBP-10 snapshots for OFI calculation + use data::providers::databento::dbn_parser::DbnParser; + let mbp10_dir = Path::new("test_data/mbp10"); + let mbp10_snapshots = if mbp10_dir.exists() { + info!("📊 Loading MBP-10 order book snapshots for OFI calculation..."); + // Load all .dbn files in the directory + let mut all_snapshots = Vec::new(); + if let Ok(entries) = std::fs::read_dir(mbp10_dir) { + let parser = DbnParser::new() + .context("Failed to create DBN parser for MBP-10 data")?; + + for entry in entries.flatten() { + let path = entry.path(); + // Only load .dbn files (not .zst compressed files) + if path.extension().and_then(|s| s.to_str()) == Some("dbn") { + match parser.parse_mbp10_file(&path).await { + Ok(mut snaps) => { + info!(" ✅ Loaded {} snapshots from {:?}", snaps.len(), path.file_name()); + all_snapshots.append(&mut snaps); + } + Err(e) => { + warn!(" ⚠️ Failed to load {:?}: {}", path.file_name(), e); + } + } + } + } + } + + if all_snapshots.is_empty() { + warn!("⚠️ No MBP-10 snapshots loaded. OFI features will be zeros."); + None + } else { + // Sort snapshots by timestamp for efficient lookup + all_snapshots.sort_by_key(|s| s.timestamp); + info!("✅ Total MBP-10 snapshots loaded: {} (sorted by timestamp)", all_snapshots.len()); + Some(all_snapshots) + } + } else { + warn!("⚠️ MBP-10 directory not found at {:?}. OFI features will be zeros.", mbp10_dir); + None + }; + + // Extract 54-feature vectors (technical indicators, OFI, time, statistical features) + info!("Extracting 51-feature vectors from OHLCV bars (51-feature architecture)..."); + let feature_vectors = self.extract_full_features(&all_ohlcv_bars, mbp10_snapshots.as_deref())?; + + info!( + "Extracted {} feature vectors (51 dimensions: technical indicators, time, statistical features (Proxy OFI removed))", + feature_vectors.len() + ); + + // Create training data pairs (features, target) + // Target: [preprocessed_current, preprocessed_next, raw_current, raw_next] + // WAVE 3 BUG FIX: Include raw prices for triple barrier tracker (needs actual market prices in cents) + // Wave 14 Agent 32: Use preprocessed closes if enabled + let mut training_data = Vec::new(); + for i in 0..feature_vectors.len().saturating_sub(1) { + let (preprocessed_current, preprocessed_next, raw_current, raw_next) = if let Some(ref preprocessed) = preprocessed_closes { + // Use preprocessed values for reward calculation + raw for barrier tracker + ( + preprocessed[i + 50], + preprocessed[i + 1 + 50], + all_ohlcv_bars[i + 50].close, + all_ohlcv_bars[i + 1 + 50].close, + ) + } else { + // Use raw prices for both (original behavior) + let raw_curr = all_ohlcv_bars[i + 50].close; + let raw_next = all_ohlcv_bars[i + 1 + 50].close; + (raw_curr, raw_next, raw_curr, raw_next) + }; + training_data.push((feature_vectors[i], vec![preprocessed_current, preprocessed_next, raw_current, raw_next])); + } + // Last sample targets itself + if !feature_vectors.is_empty() { + let idx = all_ohlcv_bars.len() - 1; + let (preprocessed_close, raw_close) = if let Some(ref preprocessed) = preprocessed_closes { + (preprocessed[idx], all_ohlcv_bars[idx].close) + } else { + let raw = all_ohlcv_bars[idx].close; + (raw, raw) + }; + training_data.push(( + feature_vectors[feature_vectors.len() - 1], + vec![preprocessed_close, preprocessed_close, raw_close, raw_close], + )); + } + + info!( + "Created {} total samples with 54-dim features", + training_data.len() + ); + + // Split training data 80/20 for train/validation + let split_idx = (training_data.len() * 80) / 100; + let train_data = training_data[..split_idx].to_vec(); + let val_data = training_data[split_idx..].to_vec(); + + info!( + "Split data - Training samples: {}, Validation samples: {}", + train_data.len(), + val_data.len() + ); + + Ok((train_data, val_data)) + } + + /// Load training data from DBN files using official dbn crate decoder + /// Returns (train_data, val_data) with 80/20 split + pub(crate) async fn load_training_data( + &mut self, + dbn_data_dir: &str, + ) -> Result<( + Vec<(FeatureVector51, Vec)>, + Vec<(FeatureVector51, Vec)>, + )> { + // Find all DBN files in directory + let dir_path = Path::new(dbn_data_dir); + if !dir_path.exists() { + return Err(anyhow::anyhow!( + "Data directory not found: {}", + dbn_data_dir + )); + } + + let dbn_files: Vec<_> = std::fs::read_dir(dir_path)? + .filter_map(|entry| entry.ok()) + .filter(|entry| entry.path().extension().and_then(|s| s.to_str()) == Some("dbn")) + .map(|entry| entry.path()) + .collect(); + + if dbn_files.is_empty() { + return Err(anyhow::anyhow!("No DBN files found in: {}", dbn_data_dir)); + } + + info!("Found {} DBN files to load", dbn_files.len()); + + let mut all_ohlcv_bars = Vec::new(); + + // Load and decode each DBN file to collect OHLCV bars + for (file_idx, file_path) in dbn_files.iter().enumerate() { + debug!( + "Loading DBN file {}/{}: {}", + file_idx + 1, + dbn_files.len(), + file_path.display() + ); + + // Extract raw OHLCV bars from file + let file_bars = self.extract_ohlcv_bars_from_dbn(file_path)?; + + debug!( + "Extracted {} OHLCV bars from {}", + file_bars.len(), + file_path.file_name().unwrap_or_default().to_string_lossy() + ); + + all_ohlcv_bars.extend(file_bars); + } + + if all_ohlcv_bars.is_empty() { + return Err(anyhow::anyhow!( + "No OHLCV bars extracted from DBN files. Check if files contain OHLCV messages." + )); + } + + info!( + "Successfully loaded {} OHLCV bars from {} DBN files", + all_ohlcv_bars.len(), + dbn_files.len() + ); + + // Sort bars by timestamp (critical for rolling window feature extraction) + debug!("Sorting bars chronologically by timestamp..."); + all_ohlcv_bars.sort_by_key(|bar| bar.timestamp); + debug!("Bars sorted successfully"); + + // Extract 54-feature vectors (technical indicators, Proxy OFI, time, statistical features) + // Note: DBN loader does not load MBP-10 data, so OFI features will be zeros + info!("Extracting 51-feature vectors from OHLCV bars (51-feature architecture)..."); + let feature_vectors = self.extract_full_features(&all_ohlcv_bars, None)?; + + info!( + "Extracted {} feature vectors (51 dimensions: technical indicators, time, statistical features (Proxy OFI removed))", + feature_vectors.len() + ); + + // Create training data pairs (features, target) + // Target: [current_close, next_close] for proper reward calculation + let mut training_data = Vec::new(); + for i in 0..feature_vectors.len().saturating_sub(1) { + let current_close = all_ohlcv_bars[i + 50].close; // +50 to account for warmup period + let next_close = all_ohlcv_bars[i + 1 + 50].close; + training_data.push((feature_vectors[i], vec![current_close, next_close])); + } + // Last sample targets itself + if !feature_vectors.is_empty() { + let idx = all_ohlcv_bars.len() - 1; + let current_close = all_ohlcv_bars[idx].close; + training_data.push(( + feature_vectors[feature_vectors.len() - 1], + vec![current_close, current_close], + )); + } + + info!( + "Created {} total samples with 54-dim features", + training_data.len() + ); + + // Split training data 80/20 for train/validation + let split_idx = (training_data.len() * 80) / 100; + let train_data = training_data[..split_idx].to_vec(); + let val_data = training_data[split_idx..].to_vec(); + + info!( + "Split data - Training samples: {}, Validation samples: {}", + train_data.len(), + val_data.len() + ); + + Ok((train_data, val_data)) + } + + /// Extract raw OHLCV bars from DBN file using official dbn crate decoder + /// + /// This replaces the custom parser that only extracted 2 messages (header metadata). + /// Now extracts all OHLCV bars (400-500+ records per file). + /// + /// Public for testing purposes. + pub fn extract_ohlcv_bars_from_dbn(&self, file_path: &Path) -> Result> { + use dbn::decode::dbn::Decoder; + use dbn::decode::{DbnMetadata, DecodeRecordRef}; + use std::fs::File; + use std::io::BufReader; + + let mut ohlcv_bars = Vec::new(); + + // Open file and create official DBN decoder + let file = File::open(file_path) + .with_context(|| format!("Failed to open DBN file: {:?}", file_path))?; + let reader = BufReader::new(file); + + let mut decoder = Decoder::new(reader) + .map_err(|e| anyhow::anyhow!("Failed to create DBN decoder: {}", e))?; + + // Read metadata (for logging) + let metadata = decoder.metadata(); + debug!( + "DBN file metadata: dataset={:?}, schema={:?}, symbols={:?}", + metadata.dataset, metadata.schema, metadata.symbols + ); + + // Decode all OHLCV records + let mut ohlcv_count = 0; + let mut other_count = 0; + let mut idx = 0; + + loop { + match decoder.decode_record_ref() { + Ok(Some(record)) => { + idx += 1; + + // Convert RecordRef to RecordRefEnum for pattern matching + let record_enum = record + .as_enum() + .map_err(|e| anyhow::anyhow!("Failed to convert record to enum: {}", e))?; + + match record_enum { + dbn::RecordRefEnum::Ohlcv(ohlcv) => { + ohlcv_count += 1; + + // Extract OHLCV values (prices are i64 scaled by 1e-9 per DBN spec, volume is u64) + let open_f64 = ohlcv.open as f64 * 1e-9; + let high_f64 = ohlcv.high as f64 * 1e-9; + let low_f64 = ohlcv.low as f64 * 1e-9; + let close_f64 = ohlcv.close as f64 * 1e-9; + let volume_u64 = ohlcv.volume; + + // WAVE 8 AGENT 36: Validate all price values are finite (not NaN/Inf) + // Skip bars with invalid data to prevent NaN propagation + if !open_f64.is_finite() + || !high_f64.is_finite() + || !low_f64.is_finite() + || !close_f64.is_finite() + { + debug!( + "Skipping OHLCV bar {} with non-finite values: open={}, high={}, low={}, close={}", + ohlcv_count, open_f64, high_f64, low_f64, close_f64 + ); + continue; + } + + // Log first few records for validation + if ohlcv_count <= 5 { + debug!( + "Raw OHLCV #{}: open={}, high={}, low={}, close={}", + ohlcv_count, ohlcv.open, ohlcv.high, ohlcv.low, ohlcv.close + ); + debug!( + "Scaled OHLCV #{}: open={:.6}, high={:.6}, low={:.6}, close={:.6}", + ohlcv_count, open_f64, high_f64, low_f64, close_f64 + ); + } + + // Convert timestamp from nanoseconds since epoch to DateTime + let timestamp_nanos = ohlcv.hd.ts_event as i64; + let timestamp_secs = timestamp_nanos / 1_000_000_000; + let timestamp_nanos_remainder = + (timestamp_nanos % 1_000_000_000) as u32; + let timestamp = chrono::DateTime::::from_timestamp( + timestamp_secs, + timestamp_nanos_remainder, + ) + .unwrap_or_else(|| chrono::Utc::now()); + + // Create OHLCVBar for feature extraction pipeline + let bar = OHLCVBar { + timestamp, + open: open_f64, + high: high_f64, + low: low_f64, + close: close_f64, + volume: volume_u64 as f64, + }; + + ohlcv_bars.push(bar); + }, + _ => { + other_count += 1; + if other_count <= 5 { + debug!("Skipping non-OHLCV record at index {}", idx); + } + }, + } + }, + Ok(None) => { + // End of stream + break; + }, + Err(e) => { + return Err(anyhow::anyhow!("Failed to decode record {}: {}", idx, e)); + }, + } + } + + info!( + "Extracted {} OHLCV bars from {:?} ({} other records skipped)", + ohlcv_count, + file_path.file_name().unwrap_or_default(), + other_count + ); + + Ok(ohlcv_bars) + } + + /// Create features from OHLCV data + pub(crate) fn create_ohlcv_features( + &self, + open: f64, + high: f64, + low: f64, + close: f64, + volume: u64, + ) -> Result { + use std::collections::HashMap; + + // Use absolute values for Price type (futures data can have negative values) + // For ML training, the absolute magnitude is what matters for feature extraction + let close_price = + common::Price::from_f64(close.abs()).unwrap_or_else(|_| common::Price::ZERO); + let open_price = + common::Price::from_f64(open.abs()).unwrap_or_else(|_| common::Price::ZERO); + let high_price = + common::Price::from_f64(high.abs()).unwrap_or_else(|_| common::Price::ZERO); + let low_price = common::Price::from_f64(low.abs()).unwrap_or_else(|_| common::Price::ZERO); + + // Calculate technical indicators + let mut indicators = HashMap::new(); + + // Price-based features + let price_range = high - low; + let body_size = (close - open).abs(); + let upper_shadow = high - close.max(open); + let lower_shadow = close.min(open) - low; + + indicators.insert("price_range".to_string(), price_range); + indicators.insert("body_size".to_string(), body_size); + indicators.insert("upper_shadow".to_string(), upper_shadow); + indicators.insert("lower_shadow".to_string(), lower_shadow); + indicators.insert("close_to_high".to_string(), (close - high).abs()); + indicators.insert("close_to_low".to_string(), (close - low).abs()); + + // Microstructure features + let spread_bps = ((high - low) / close * 10000.0) as i32; + let trade_intensity = volume as f64; + + Ok(FinancialFeatures { + prices: vec![open_price, high_price, low_price, close_price], + volumes: vec![volume as i64], + technical_indicators: indicators, + microstructure: crate::training_pipeline::MicrostructureFeatures { + spread_bps, + imbalance: 0.0, // Not available from OHLCV + trade_intensity, + vwap: close_price, // Approximate VWAP as close + }, + risk_metrics: crate::training_pipeline::RiskFeatures { + var_5pct: -0.02, // Placeholder + expected_shortfall: -0.03, + max_drawdown: -0.05, + sharpe_ratio: 1.0, + }, + timestamp: chrono::Utc::now(), + }) + } +} diff --git a/ml/src/trainers/dqn/features.rs b/ml/src/trainers/dqn/features.rs new file mode 100644 index 000000000..69f3ef462 --- /dev/null +++ b/ml/src/trainers/dqn/features.rs @@ -0,0 +1,211 @@ +//! Feature Extraction for DQN Training +//! +//! Methods for extracting 140-dimensional features (125 market + 3 portfolio +//! + 12 microstructure), calculating feature statistics, normalizing datasets, +//! and creating synthetic features for testing. + +use anyhow::Result; +use tracing::warn; + +use crate::features::extraction::OHLCVBar; +use crate::training_pipeline::FinancialFeatures; + +use super::statistics::FeatureStatistics; +use super::FeatureVector51; +use super::trainer::DQNTrainer; + +impl DQNTrainer { + /// Create synthetic features (placeholder for testing) + pub(crate) fn create_synthetic_features(&self, price: f64) -> Result { + use std::collections::HashMap; + + let price_obj = + common::Price::from_f64(price).unwrap_or_else(|_| common::Price::new(price).unwrap()); + + let mut indicators = HashMap::new(); + indicators.insert("rsi_14".to_string(), 50.0); + indicators.insert("sma_20".to_string(), price); + indicators.insert("ema_12".to_string(), price); + + Ok(FinancialFeatures { + prices: vec![price_obj; 4], + volumes: vec![1000], + technical_indicators: indicators, + microstructure: crate::training_pipeline::MicrostructureFeatures { + spread_bps: 10, + imbalance: 0.0, + trade_intensity: 100.0, + vwap: price_obj, + }, + risk_metrics: crate::training_pipeline::RiskFeatures { + var_5pct: -0.02, + expected_shortfall: -0.03, + max_drawdown: -0.05, + sharpe_ratio: 1.0, + }, + timestamp: chrono::Utc::now(), + }) + } + + /// WAVE 3.10: Extract 140 features (125 market + 3 portfolio + 12 microstructure) + /// + /// This method extracts 140 features for DQN state representation: + /// - 125 market features (price, technical indicators, volatility, etc.) + /// - 3 portfolio features (populated later via PortfolioTracker) + /// - 12 microstructure features (spread estimators, liquidity, order flow, market impact) + /// + /// The microstructure features are calculated on-the-fly from OHLCV data using + /// the calculators initialized in DQNTrainer::new(). + /// + /// # Arguments + /// + /// * `bars` - OHLCV bars for feature extraction + /// * `mbp10_snapshots` - Optional MBP-10 order book snapshots for OFI calculation + pub(crate) fn extract_full_features( + &mut self, + bars: &[OHLCVBar], + mbp10_snapshots: Option<&[data::providers::databento::mbp10::Mbp10Snapshot]>, + ) -> Result> { + use crate::features::extraction::FeatureExtractor; + + if bars.is_empty() { + anyhow::bail!("Cannot extract features from empty bar sequence"); + } + + const WARMUP_PERIOD: usize = 50; + if bars.len() < WARMUP_PERIOD { + anyhow::bail!( + "Insufficient data: {} bars provided, {} required for warmup", + bars.len(), + WARMUP_PERIOD + ); + } + + let mut extractor = FeatureExtractor::new(); + let mut feature_vectors = Vec::with_capacity(bars.len() - WARMUP_PERIOD); + + // Feed bars sequentially to build rolling windows + for (i, bar) in bars.iter().enumerate() { + extractor.update(bar)?; + + // WAVE 3.10: Update microstructure features with current bar + // Calculate timestamp in nanoseconds + let timestamp_ns = bar.timestamp.timestamp_nanos_opt().unwrap_or(0) as u64; + + // Update all 8 microstructure calculators (4 more already exist in microstructure.rs) + let hl_spread = self.micro_high_low_spread.update(bar.high, bar.low); + let _vw_spread = self.micro_vw_spread.update(hl_spread, bar.volume); + let _tick_count = self.micro_tick_count.update(bar.close); + let _inter_arrival = self.micro_inter_arrival.update(timestamp_ns); + let _buy_sell_imb = self.micro_buy_sell_imbalance.update(bar.close, bar.volume); + + // Kyle's Lambda: slow-updating (only updates every 5 minutes) + let return_pct = if self.last_close > 0.0 { + (bar.close - self.last_close) / self.last_close + } else { + 0.0 + }; + let signed_volume = (bar.close - bar.open).signum() * (bar.close * bar.volume).sqrt(); + let _kyle_lambda = self.micro_kyle_lambda.maybe_update(timestamp_ns, return_pct, signed_volume); + + let _price_impact = self.micro_price_impact.update(bar.high, bar.low, bar.close); + let _variance_ratio = self.micro_variance_ratio.update(return_pct); + + // Track last close for next iteration + self.last_close = bar.close; + + // Start extracting features after warmup + if i >= WARMUP_PERIOD { + // WAVE 10: Extract 43 base features + 8 OFI features (51 total, Proxy OFI removed) + // Features breakdown: + // - 0-4: OHLCV (5) + // - 5-9: Technical indicators (5) + // - 10-15: Price patterns (6) + // - 16-21: Volume features (6) + // - 22-26: Time-based (5) + // - 27-39: Statistical (13) + // - 40-42: Regime detection (3) + // - 43-50: OFI features (8) + // TOTAL: 51 features + + // Extract features with OFI if MBP-10 data available + let features_51 = if let Some(mbp10_data) = mbp10_snapshots { + // Calculate OFI features from MBP-10 order book data + use crate::features::mbp10_loader::get_snapshots_for_timestamp; + + let bar_timestamp_ns = bar.timestamp.timestamp_nanos_opt().unwrap_or(0) as u64; + let window = get_snapshots_for_timestamp(mbp10_data, bar_timestamp_ns, 100); + + if !window.is_empty() { + match extractor.extract_current_features_with_ofi(window) { + Ok(feats) => feats, + Err(e) => { + warn!("Failed to calculate OFI for bar {}: {}. Using zeros for OFI.", i, e); + let base_features_43 = extractor.extract_current_features_v2()?; + let mut feats = [0.0; 51]; + feats[..43].copy_from_slice(&base_features_43); + feats + } + } + } else { + // No MBP-10 snapshots for this timestamp + let base_features_43 = extractor.extract_current_features_v2()?; + let mut feats = [0.0; 51]; + feats[..43].copy_from_slice(&base_features_43); + feats + } + } else { + // Fallback: Use base 43 features + zero-padded OFI + let base_features_43 = extractor.extract_current_features_v2()?; + let mut feats = [0.0; 51]; + feats[..43].copy_from_slice(&base_features_43); + feats + }; + + feature_vectors.push(features_51); + } + } + + Ok(feature_vectors) + } + + /// Calculate feature statistics from training samples using Welford's algorithm + pub(crate) fn calculate_feature_statistics( + &self, + samples: &[(FeatureVector51, Vec)], + ) -> Result { + let mut stats = FeatureStatistics::new(54); + + for (feature_vec, _) in samples { + let features: Vec = feature_vec.iter().map(|&v| v as f32).collect(); + stats.update(&features); + } + + Ok(stats) + } + + /// Normalize all samples in a dataset using z-score normalization + pub(crate) fn normalize_dataset( + &mut self, + samples: &mut [(FeatureVector51, Vec)], + ) -> Result<()> { + if let Some(ref stats) = self.feature_stats { + for (feature_vec, _) in samples.iter_mut() { + // Convert to f32 for normalization + let features_f32: Vec = feature_vec.iter().map(|&v| v as f32).collect(); + + // Normalize with skip (indices 125-127 are portfolio placeholders) + let normalized = stats.normalize_with_skip(&features_f32, &[125, 126, 127]); + + // Convert back to f64 and update + for (i, &val) in normalized.iter().enumerate() { + feature_vec[i] = val as f64; + } + } + } else { + return Err(anyhow::anyhow!("Feature statistics not initialized")); + } + + Ok(()) + } +} diff --git a/ml/src/trainers/dqn/mod.rs b/ml/src/trainers/dqn/mod.rs index dea483da9..0b7828343 100644 --- a/ml/src/trainers/dqn/mod.rs +++ b/ml/src/trainers/dqn/mod.rs @@ -10,13 +10,21 @@ //! ## Module Structure //! //! - `config` - Hyperparameters and agent type configuration +//! - `data_loading` - Parquet/DBN data loading and preprocessing //! - `early_stopping` - Patience-based early stopping mechanism (WAVE 24) +//! - `features` - Feature extraction, statistics, and normalization +//! - `monitoring` - Per-epoch training monitor and validation +//! - `risk` - Adaptive risk management (volatility epsilon, Kelly criterion) //! - `statistics` - Feature normalization and Q-value monitoring -//! - `trainer` - Main DQNTrainer implementation +//! - `trainer` - Main DQNTrainer implementation (core training loop) mod config; +mod data_loading; mod early_stopping; +mod features; pub mod lr_scheduler; +mod monitoring; +mod risk; mod statistics; mod trainer; diff --git a/ml/src/trainers/dqn/monitoring.rs b/ml/src/trainers/dqn/monitoring.rs new file mode 100644 index 000000000..2f5084e17 --- /dev/null +++ b/ml/src/trainers/dqn/monitoring.rs @@ -0,0 +1,290 @@ +//! Training Monitor for per-epoch tracking and validation +//! +//! Tracks rewards, actions, Q-values, episode lengths, and validates +//! training health (constant rewards, action diversity, Q-value balance). + +use anyhow::Result; +use tracing::{debug, warn}; + +use crate::dqn::action_space::FactoredAction; + +/// Training monitor for per-epoch tracking and validation +pub(crate) struct TrainingMonitor { + pub(crate) epoch: usize, + pub(crate) reward_history: Vec, + pub(crate) action_counts: [usize; 45], // 5 exposure x 3 order x 3 urgency (FactoredAction) + pub(crate) q_value_sums: [f64; 45], // Sum of Q-values per action + pub(crate) q_value_counts: [usize; 45], // Count of Q-values per action + pub(crate) consecutive_constant_epochs: usize, + // Q-value range tracking (WAVE 9-11 production monitoring) + pub(crate) q_value_min: f64, + pub(crate) q_value_max: f64, + pub(crate) q_value_history: Vec, // Per-step Q-values for mean calculation + + // WAVE P2: Episode length tracking + pub(crate) episode_lengths: Vec, + pub(crate) episode_start_step: usize, + pub(crate) barrier_exit_counts: [usize; 4], // [profit, stop, time, boundary] +} + +impl TrainingMonitor { + pub(crate) fn new(epoch: usize) -> Self { + Self { + epoch, + reward_history: Vec::new(), + action_counts: [0; 45], + q_value_sums: [0.0; 45], + q_value_counts: [0; 45], + consecutive_constant_epochs: 0, + q_value_min: f64::INFINITY, + q_value_max: f64::NEG_INFINITY, + q_value_history: Vec::new(), + + // WAVE P2: Episode tracking + episode_lengths: Vec::new(), + episode_start_step: 0, + barrier_exit_counts: [0; 4], // [profit=0, stop=1, time=2, boundary=3] + } + } + + /// Add reward to tracking (with bounded history) + pub(crate) fn track_reward(&mut self, reward: f32) { + self.reward_history.push(reward); + // MEMORY LEAK FIX: Limit reward history to last 1000 entries per epoch + // Each trial has ~10-50 epochs, so this limits to ~10-50K entries total + // vs unbounded growth causing OOM at 10-30 trials + if self.reward_history.len() > 1000 { + self.reward_history.drain(0..500); // Remove oldest 500, keep newest 500 + } + } + + /// Add action to tracking + pub(crate) fn track_action(&mut self, action: &FactoredAction) { + let idx = action.to_index() as usize; // Returns 0-44 + self.action_counts[idx] += 1; + } + + /// Add Q-value to tracking + pub(crate) fn track_q_value(&mut self, action: &FactoredAction, q_value: f64) { + let idx = action.to_index() as usize; // Returns 0-44 + self.q_value_sums[idx] += q_value; + self.q_value_counts[idx] += 1; + } + + /// Track Q-value range for monitoring (WAVE 9-11 production) + pub(crate) fn track_q_value_range(&mut self, q_value: f64) { + if q_value < self.q_value_min { + self.q_value_min = q_value; + } + if q_value > self.q_value_max { + self.q_value_max = q_value; + } + self.q_value_history.push(q_value); + // MEMORY LEAK FIX: Limit Q-value history to last 1000 entries per epoch + if self.q_value_history.len() > 1000 { + self.q_value_history.drain(0..500); // Remove oldest 500, keep newest 500 + } + } + + /// Get Q-value statistics (min, max, mean) + pub(crate) fn get_q_value_stats(&self) -> (f64, f64, f64) { + if self.q_value_history.is_empty() { + return (0.0, 0.0, 0.0); + } + let mean = self.q_value_history.iter().sum::() / self.q_value_history.len() as f64; + (self.q_value_min, self.q_value_max, mean) + } + + /// Validate rewards are not constant + pub(crate) fn validate_rewards(&mut self) -> Result<()> { + if self.reward_history.is_empty() { + return Ok(()); + } + + let mean = self.reward_history.iter().sum::() / self.reward_history.len() as f32; + let variance = self + .reward_history + .iter() + .map(|r| (r - mean).powi(2)) + .sum::() + / self.reward_history.len() as f32; + let std = variance.sqrt(); + + // Check if all rewards are identical (std == 0) or nearly constant (std < 0.01) + if std < 0.01 { + self.consecutive_constant_epochs += 1; + + warn!( + "⚠️ CONSTANT REWARDS DETECTED at epoch {}! std={:.6}, mean={:.4}, consecutive_epochs={}", + self.epoch, std, mean, self.consecutive_constant_epochs + ); + + // Panic if constant for 5+ consecutive epochs (critical bug) + if self.consecutive_constant_epochs >= 5 { + return Err(anyhow::anyhow!( + "❌ CRITICAL: Constant rewards for {} consecutive epochs! std={:.6}, mean={:.4}\n\ + This indicates a reward calculation bug. Training aborted.", + self.consecutive_constant_epochs, std, mean + )); + } + } else { + // Reset counter if variance is healthy + self.consecutive_constant_epochs = 0; + } + + Ok(()) + } + + /// Validate action diversity + pub(crate) fn validate_action_diversity(&self) -> Result<()> { + let total_actions: usize = self.action_counts.iter().sum(); + + if total_actions == 0 { + return Ok(()); // No actions yet, skip validation + } + + // Check if any action is below diversity threshold + // Uniform distribution for 45 actions = 100/45 = 2.22% + // During exploration (epsilon=0.3): Expected ~0.7% per action + // Warn if action < 0.5% (truly neglected actions only) + for (i, &count) in self.action_counts.iter().enumerate() { + let percentage = (count as f64 / total_actions as f64) * 100.0; + + // Convert index to FactoredAction for proper display + if let Ok(action) = FactoredAction::from_index(i) { + let action_str = format!("{:?}", action); + + if percentage < 0.5 { + warn!( + "⚠️ LOW ACTION DIVERSITY at epoch {}: {} only {:.1}% ({}/{})", + self.epoch, action_str, percentage, count, total_actions + ); + } + } + } + + Ok(()) + } + + /// Validate Q-value balance across actions + pub(crate) fn validate_q_value_balance(&self) -> Result<()> { + // Calculate average Q-value per action + let mut avg_q_values = [0.0f64; 3]; + for i in 0..3 { + if self.q_value_counts[i] > 0 { + avg_q_values[i] = self.q_value_sums[i] / self.q_value_counts[i] as f64; + } + } + + // Check if BUY Q-values diverge > 1000 from SELL/HOLD + let buy_q = avg_q_values[0]; + let sell_q = avg_q_values[1]; + let hold_q = avg_q_values[2]; + + if (buy_q - sell_q).abs() > 1000.0 || (buy_q - hold_q).abs() > 1000.0 { + warn!( + "⚠️ Q-VALUE DIVERGENCE at epoch {}: BUY={:.2}, SELL={:.2}, HOLD={:.2}", + self.epoch, buy_q, sell_q, hold_q + ); + } + + Ok(()) + } + + /// Log action distribution every 10 epochs + pub(crate) fn log_action_distribution(&self) { + if self.epoch % 10 == 0 { + let total_actions: usize = self.action_counts.iter().sum(); + if total_actions > 0 { + // Sort action_counts by frequency (descending) + let mut sorted_actions: Vec<(usize, usize)> = self + .action_counts + .iter() + .enumerate() + .map(|(idx, &count)| (idx, count)) + .collect(); + sorted_actions.sort_by(|a, b| b.1.cmp(&a.1)); + + // Log top 5 most frequent actions (DEBUG level) + debug!( + "Action Distribution [Epoch {}] - Top 5 Actions:", + self.epoch + ); + for (idx, count) in sorted_actions.iter().take(5) { + if *count > 0 { + if let Ok(action) = FactoredAction::from_index(*idx) { + let pct = (*count as f64 / total_actions as f64) * 100.0; + debug!(" [{:2}] {:?}: {} ({:.1}%)", idx, action, count, pct); + } + } + } + + // Log average Q-values per action (top 5) (DEBUG level) + let mut avg_q = [0.0f64; 45]; + for i in 0..45 { + if self.q_value_counts[i] > 0 { + avg_q[i] = self.q_value_sums[i] / self.q_value_counts[i] as f64; + } + } + + debug!("Average Q-values [Epoch {}] - Top 5 Actions:", self.epoch); + for (idx, _count) in sorted_actions.iter().take(5) { + if self.q_value_counts[*idx] > 0 { + if let Ok(action) = FactoredAction::from_index(*idx) { + debug!(" [{:2}] {:?}: Q={:.4}", idx, action, avg_q[*idx]); + } + } + } + } + } + } + + /// Run all validations + pub(crate) fn validate_all(&mut self) -> Result<()> { + self.validate_rewards()?; + self.validate_action_diversity()?; + self.validate_q_value_balance()?; + self.log_action_distribution(); + Ok(()) + } + + /// Track episode end and record length/exit reason + pub(crate) fn track_episode_end(&mut self, current_step: usize, barrier_label: Option) { + let episode_length = current_step - self.episode_start_step; + self.episode_lengths.push(episode_length); + + // Count exit reason + match barrier_label { + Some(1) => self.barrier_exit_counts[0] += 1, // Profit target + Some(-1) => self.barrier_exit_counts[1] += 1, // Stop loss + Some(0) => self.barrier_exit_counts[2] += 1, // Time expiry + _ => self.barrier_exit_counts[3] += 1, // Data/time boundary + } + + // Reset for next episode + self.episode_start_step = current_step + 1; + } + + /// Get episode statistics + pub(crate) fn get_episode_stats(&self) -> (f64, f64, usize, usize, [usize; 4]) { + if self.episode_lengths.is_empty() { + return (0.0, 0.0, 0, 0, [0; 4]); + } + + let total = self.episode_lengths.len(); + let mean = self.episode_lengths.iter().sum::() as f64 / total as f64; + let min = *self.episode_lengths.iter().min().unwrap(); + let max = *self.episode_lengths.iter().max().unwrap(); + + // Calculate std dev + let variance = self.episode_lengths.iter() + .map(|&len| { + let diff = len as f64 - mean; + diff * diff + }) + .sum::() / total as f64; + let std_dev = variance.sqrt(); + + (mean, std_dev, min, max, self.barrier_exit_counts) + } +} diff --git a/ml/src/trainers/dqn/risk.rs b/ml/src/trainers/dqn/risk.rs new file mode 100644 index 000000000..9d54a37e8 --- /dev/null +++ b/ml/src/trainers/dqn/risk.rs @@ -0,0 +1,145 @@ +//! Adaptive Risk Management for DQN Training +//! +//! Helper methods for volatility-adjusted exploration, risk-adjusted rewards, +//! Kelly criterion position sizing, and risk tracker updates. + +use tracing::info; + +use super::trainer::DQNTrainer; + +impl DQNTrainer { + // WAVE 16S: Adaptive Risk Management Helper Methods + + /// Calculate volatility-adjusted epsilon for exploration + /// + /// Adjusts epsilon based on recent return volatility: + /// - Low volatility (<1%): reduce epsilon (exploit more) + /// - High volatility (>5%): increase epsilon (explore more) + /// - Moderate volatility: linear interpolation + pub(crate) fn calculate_volatility_adjusted_epsilon(&self, base_epsilon: f64) -> f64 { + if !self.hyperparams.enable_volatility_epsilon || self.volatility_returns.len() < 10 { + return base_epsilon; + } + + // Calculate volatility (standard deviation of returns) + let mean: f64 = self.volatility_returns.iter().sum::() / self.volatility_returns.len() as f64; + let variance: f64 = self.volatility_returns.iter() + .map(|x| (x - mean).powi(2)) + .sum::() / self.volatility_returns.len() as f64; + let volatility = variance.sqrt(); + + // Adjust epsilon based on volatility regime + let multiplier = if volatility < 0.01 { + 0.5 // Low volatility: exploit more (reduce epsilon) + } else if volatility > 0.05 { + 2.0 // High volatility: explore more (increase epsilon) + } else { + // Linear interpolation between 0.01 and 0.05 + 0.5 + (volatility - 0.01) / (0.05 - 0.01) * 1.5 + }; + + let adjusted_epsilon = (base_epsilon * multiplier).clamp(0.01, 1.0); + + info!( + "Volatility-adjusted epsilon: base={:.4} × {:.2} = {:.4} (vol={:.4})", + base_epsilon, multiplier, adjusted_epsilon, volatility + ); + + adjusted_epsilon + } + + /// Calculate risk-adjusted reward using Sharpe-like normalization + /// + /// Divides raw reward by rolling volatility of PnL to reward consistency + /// over large but volatile returns. + pub(crate) fn calculate_risk_adjusted_reward(&self, raw_reward: f64) -> f64 { + if !self.hyperparams.enable_risk_adjusted_rewards || self.pnl_history.len() < 10 { + return raw_reward; + } + + // Calculate PnL volatility + let mean: f64 = self.pnl_history.iter().sum::() / self.pnl_history.len() as f64; + let variance: f64 = self.pnl_history.iter() + .map(|x| (x - mean).powi(2)) + .sum::() / self.pnl_history.len() as f64; + let pnl_vol = variance.sqrt().max(1e-6); // Prevent division by zero + + // Sharpe-like: reward / volatility + raw_reward / pnl_vol + } + + /// Get Kelly fraction for position sizing + /// + /// Returns Kelly criterion position size (0.0-0.25) based on trade history. + /// Requires minimum trade history for statistical significance. + pub fn get_kelly_fraction(&self) -> f64 { + if !self.hyperparams.enable_kelly_sizing { + return 1.0; // Full position sizing (disabled) + } + + let kelly_opt = match &self.kelly_optimizer { + Some(opt) => opt, + None => return 1.0, + }; + + // Need minimum trades for Kelly calculation + if self.trade_history.len() < self.hyperparams.kelly_min_trades { + return 0.1; // Conservative default until enough history + } + + // Calculate win/loss statistics + let wins: Vec = self.trade_history.iter() + .filter(|&&r| r > 0.0) + .copied() + .collect(); + let losses: Vec = self.trade_history.iter() + .filter(|&&r| r < 0.0) + .map(|r| -r) + .collect(); + + let win_prob = wins.len() as f64 / self.trade_history.len() as f64; + let avg_win = if wins.is_empty() { + 0.01 + } else { + wins.iter().sum::() / wins.len() as f64 + }; + let avg_loss = if losses.is_empty() { + 0.01 + } else { + losses.iter().sum::() / losses.len() as f64 + }; + + // Calculate Kelly fraction + let kelly_result = kelly_opt.calculate_basic_kelly(win_prob, avg_win, avg_loss); + let kelly_fraction = kelly_result.unwrap_or(0.1); + + // Apply fractional Kelly (conservative) + let fractional_kelly = kelly_fraction * self.hyperparams.kelly_fractional; + + // Clamp to safety bounds + fractional_kelly.clamp(0.01, self.hyperparams.kelly_max_fraction) + } + + /// Update adaptive risk trackers with new market data + pub(crate) fn update_risk_trackers(&mut self, reward: f64, price_return: f64) { + // Update PnL history + self.pnl_history.push_back(reward); + if self.pnl_history.len() > 1000 { + self.pnl_history.pop_front(); + } + + // Update volatility tracker + self.volatility_returns.push_back(price_return); + if self.volatility_returns.len() > self.hyperparams.volatility_window { + self.volatility_returns.pop_front(); + } + + // Update trade history for Kelly + if reward.abs() > 1e-6 { // Only track non-zero rewards + self.trade_history.push_back(reward); + if self.trade_history.len() > 500 { + self.trade_history.pop_front(); + } + } + } +} diff --git a/ml/src/trainers/dqn/trainer.rs b/ml/src/trainers/dqn/trainer.rs index 959137132..8e1b30146 100644 --- a/ml/src/trainers/dqn/trainer.rs +++ b/ml/src/trainers/dqn/trainer.rs @@ -3,7 +3,6 @@ //! Main training loop and execution logic for Deep Q-Network. use std::collections::VecDeque; -use std::path::Path; use std::path::PathBuf; use std::sync::Arc; use std::time::Duration; @@ -30,10 +29,7 @@ use crate::dqn::reward::{RewardConfig, RewardFunction}; use crate::dqn::target_update::convergence_half_life; use crate::dqn::{Experience, TradingState}; use crate::evaluation::metrics::calculate_var_cvar; -use crate::features::extraction::OHLCVBar; -use crate::preprocessing::{preprocess_prices, PreprocessConfig}; use crate::trainers::TargetUpdateMode; -use crate::training_pipeline::FinancialFeatures; use crate::TrainingMetrics; use crate::features::microstructure_features::*; use crate::labeling::triple_barrier::{TripleBarrierEngine, PricePoint}; @@ -41,295 +37,16 @@ use crate::labeling::types::BarrierConfig; // Import from sibling modules use super::config::{DQNAgentType, DQNHyperparameters}; +use super::monitoring::TrainingMonitor; use super::statistics::{FeatureStatistics, QValueStats}; use super::{EPISODE_LENGTH, FeatureVector51}; -/// Training monitor for per-epoch tracking and validation -struct TrainingMonitor { - epoch: usize, - reward_history: Vec, - action_counts: [usize; 45], // 5 exposure × 3 order × 3 urgency (FactoredAction) - q_value_sums: [f64; 45], // Sum of Q-values per action - q_value_counts: [usize; 45], // Count of Q-values per action - consecutive_constant_epochs: usize, - // Q-value range tracking (WAVE 9-11 production monitoring) - q_value_min: f64, - q_value_max: f64, - q_value_history: Vec, // Per-step Q-values for mean calculation - - // WAVE P2: Episode length tracking - episode_lengths: Vec, - episode_start_step: usize, - barrier_exit_counts: [usize; 4], // [profit, stop, time, boundary] -} - -impl TrainingMonitor { - fn new(epoch: usize) -> Self { - Self { - epoch, - reward_history: Vec::new(), - action_counts: [0; 45], - q_value_sums: [0.0; 45], - q_value_counts: [0; 45], - consecutive_constant_epochs: 0, - q_value_min: f64::INFINITY, - q_value_max: f64::NEG_INFINITY, - q_value_history: Vec::new(), - - // WAVE P2: Episode tracking - episode_lengths: Vec::new(), - episode_start_step: 0, - barrier_exit_counts: [0; 4], // [profit=0, stop=1, time=2, boundary=3] - } - } - - /// Add reward to tracking (with bounded history) - fn track_reward(&mut self, reward: f32) { - self.reward_history.push(reward); - // MEMORY LEAK FIX: Limit reward history to last 1000 entries per epoch - // Each trial has ~10-50 epochs, so this limits to ~10-50K entries total - // vs unbounded growth causing OOM at 10-30 trials - if self.reward_history.len() > 1000 { - self.reward_history.drain(0..500); // Remove oldest 500, keep newest 500 - } - } - - /// Add action to tracking - fn track_action(&mut self, action: &FactoredAction) { - let idx = action.to_index() as usize; // Returns 0-44 - self.action_counts[idx] += 1; - } - - /// Add Q-value to tracking - fn track_q_value(&mut self, action: &FactoredAction, q_value: f64) { - let idx = action.to_index() as usize; // Returns 0-44 - self.q_value_sums[idx] += q_value; - self.q_value_counts[idx] += 1; - } - - /// Track Q-value range for monitoring (WAVE 9-11 production) - fn track_q_value_range(&mut self, q_value: f64) { - if q_value < self.q_value_min { - self.q_value_min = q_value; - } - if q_value > self.q_value_max { - self.q_value_max = q_value; - } - self.q_value_history.push(q_value); - // MEMORY LEAK FIX: Limit Q-value history to last 1000 entries per epoch - if self.q_value_history.len() > 1000 { - self.q_value_history.drain(0..500); // Remove oldest 500, keep newest 500 - } - } - - /// Get Q-value statistics (min, max, mean) - fn get_q_value_stats(&self) -> (f64, f64, f64) { - if self.q_value_history.is_empty() { - return (0.0, 0.0, 0.0); - } - let mean = self.q_value_history.iter().sum::() / self.q_value_history.len() as f64; - (self.q_value_min, self.q_value_max, mean) - } - - /// Validate rewards are not constant - fn validate_rewards(&mut self) -> Result<()> { - if self.reward_history.is_empty() { - return Ok(()); - } - - let mean = self.reward_history.iter().sum::() / self.reward_history.len() as f32; - let variance = self - .reward_history - .iter() - .map(|r| (r - mean).powi(2)) - .sum::() - / self.reward_history.len() as f32; - let std = variance.sqrt(); - - // Check if all rewards are identical (std == 0) or nearly constant (std < 0.01) - if std < 0.01 { - self.consecutive_constant_epochs += 1; - - warn!( - "⚠️ CONSTANT REWARDS DETECTED at epoch {}! std={:.6}, mean={:.4}, consecutive_epochs={}", - self.epoch, std, mean, self.consecutive_constant_epochs - ); - - // Panic if constant for 5+ consecutive epochs (critical bug) - if self.consecutive_constant_epochs >= 5 { - return Err(anyhow::anyhow!( - "❌ CRITICAL: Constant rewards for {} consecutive epochs! std={:.6}, mean={:.4}\n\ - This indicates a reward calculation bug. Training aborted.", - self.consecutive_constant_epochs, std, mean - )); - } - } else { - // Reset counter if variance is healthy - self.consecutive_constant_epochs = 0; - } - - Ok(()) - } - - /// Validate action diversity - fn validate_action_diversity(&self) -> Result<()> { - let total_actions: usize = self.action_counts.iter().sum(); - - if total_actions == 0 { - return Ok(()); // No actions yet, skip validation - } - - // Check if any action is below diversity threshold - // Uniform distribution for 45 actions = 100/45 = 2.22% - // During exploration (ε=0.3): Expected ~0.7% per action - // Warn if action < 0.5% (truly neglected actions only) - for (i, &count) in self.action_counts.iter().enumerate() { - let percentage = (count as f64 / total_actions as f64) * 100.0; - - // Convert index to FactoredAction for proper display - if let Ok(action) = FactoredAction::from_index(i) { - let action_str = format!("{:?}", action); - - if percentage < 0.5 { - warn!( - "⚠️ LOW ACTION DIVERSITY at epoch {}: {} only {:.1}% ({}/{})", - self.epoch, action_str, percentage, count, total_actions - ); - } - } - } - - Ok(()) - } - - /// Validate Q-value balance across actions - fn validate_q_value_balance(&self) -> Result<()> { - // Calculate average Q-value per action - let mut avg_q_values = [0.0f64; 3]; - for i in 0..3 { - if self.q_value_counts[i] > 0 { - avg_q_values[i] = self.q_value_sums[i] / self.q_value_counts[i] as f64; - } - } - - // Check if BUY Q-values diverge > 1000 from SELL/HOLD - let buy_q = avg_q_values[0]; - let sell_q = avg_q_values[1]; - let hold_q = avg_q_values[2]; - - if (buy_q - sell_q).abs() > 1000.0 || (buy_q - hold_q).abs() > 1000.0 { - warn!( - "⚠️ Q-VALUE DIVERGENCE at epoch {}: BUY={:.2}, SELL={:.2}, HOLD={:.2}", - self.epoch, buy_q, sell_q, hold_q - ); - } - - Ok(()) - } - - /// Log action distribution every 10 epochs - fn log_action_distribution(&self) { - if self.epoch % 10 == 0 { - let total_actions: usize = self.action_counts.iter().sum(); - if total_actions > 0 { - // Sort action_counts by frequency (descending) - let mut sorted_actions: Vec<(usize, usize)> = self - .action_counts - .iter() - .enumerate() - .map(|(idx, &count)| (idx, count)) - .collect(); - sorted_actions.sort_by(|a, b| b.1.cmp(&a.1)); - - // Log top 5 most frequent actions (DEBUG level) - debug!( - "Action Distribution [Epoch {}] - Top 5 Actions:", - self.epoch - ); - for (idx, count) in sorted_actions.iter().take(5) { - if *count > 0 { - if let Ok(action) = FactoredAction::from_index(*idx) { - let pct = (*count as f64 / total_actions as f64) * 100.0; - debug!(" [{:2}] {:?}: {} ({:.1}%)", idx, action, count, pct); - } - } - } - - // Log average Q-values per action (top 5) (DEBUG level) - let mut avg_q = [0.0f64; 45]; - for i in 0..45 { - if self.q_value_counts[i] > 0 { - avg_q[i] = self.q_value_sums[i] / self.q_value_counts[i] as f64; - } - } - - debug!("Average Q-values [Epoch {}] - Top 5 Actions:", self.epoch); - for (idx, _count) in sorted_actions.iter().take(5) { - if self.q_value_counts[*idx] > 0 { - if let Ok(action) = FactoredAction::from_index(*idx) { - debug!(" [{:2}] {:?}: Q={:.4}", idx, action, avg_q[*idx]); - } - } - } - } - } - } - - /// Run all validations - fn validate_all(&mut self) -> Result<()> { - self.validate_rewards()?; - self.validate_action_diversity()?; - self.validate_q_value_balance()?; - self.log_action_distribution(); - Ok(()) - } - - /// Track episode end and record length/exit reason - fn track_episode_end(&mut self, current_step: usize, barrier_label: Option) { - let episode_length = current_step - self.episode_start_step; - self.episode_lengths.push(episode_length); - - // Count exit reason - match barrier_label { - Some(1) => self.barrier_exit_counts[0] += 1, // Profit target - Some(-1) => self.barrier_exit_counts[1] += 1, // Stop loss - Some(0) => self.barrier_exit_counts[2] += 1, // Time expiry - _ => self.barrier_exit_counts[3] += 1, // Data/time boundary - } - - // Reset for next episode - self.episode_start_step = current_step + 1; - } - - /// Get episode statistics - fn get_episode_stats(&self) -> (f64, f64, usize, usize, [usize; 4]) { - if self.episode_lengths.is_empty() { - return (0.0, 0.0, 0, 0, [0; 4]); - } - - let total = self.episode_lengths.len(); - let mean = self.episode_lengths.iter().sum::() as f64 / total as f64; - let min = *self.episode_lengths.iter().min().unwrap(); - let max = *self.episode_lengths.iter().max().unwrap(); - - // Calculate std dev - let variance = self.episode_lengths.iter() - .map(|&len| { - let diff = len as f64 - mean; - diff * diff - }) - .sum::() / total as f64; - let std_dev = variance.sqrt(); - - (mean, std_dev, min, max, self.barrier_exit_counts) - } -} pub struct DQNTrainer { /// DQN agent agent: Arc>, /// Training hyperparameters - hyperparams: DQNHyperparameters, + pub(crate) hyperparams: DQNHyperparameters, /// Device (GPU or CPU) device: Device, /// Training metrics @@ -341,7 +58,7 @@ pub struct DQNTrainer { /// Best validation loss achieved so far best_val_loss: f64, /// Validation data for computing validation loss - val_data: Vec<(FeatureVector51, Vec)>, + pub(crate) val_data: Vec<(FeatureVector51, Vec)>, /// Validation loss history for early stopping val_loss_history: Vec, /// Epoch with best validation loss @@ -360,13 +77,13 @@ pub struct DQNTrainer { // WAVE 16S: Adaptive Risk Management Components /// Kelly criterion optimizer for position sizing (None if disabled) - kelly_optimizer: Option>, + pub(crate) kelly_optimizer: Option>, /// Trade history for Kelly calculation (wins/losses) - trade_history: VecDeque, + pub(crate) trade_history: VecDeque, /// Volatility tracker for epsilon adjustment (None if disabled) - volatility_returns: VecDeque, + pub(crate) volatility_returns: VecDeque, /// PnL history for Sharpe calculation (max 1000 entries) - pnl_history: VecDeque, + pub(crate) pnl_history: VecDeque, // Wave 16 Portfolio Features /// Enable action masking (filters invalid actions before Q-value computation) @@ -389,20 +106,20 @@ pub struct DQNTrainer { pub circuit_breaker: Option>, // WAVE 3.10: Microstructure Feature Calculators (12 features) - micro_high_low_spread: HighLowSpread, - micro_vw_spread: VolumeWeightedSpread, - micro_tick_count: TickCount, - micro_inter_arrival: InterArrivalTime, - micro_buy_sell_imbalance: BuySellImbalance, - micro_kyle_lambda: KyleLambda, - micro_price_impact: PriceImpact, - micro_variance_ratio: VarianceRatio, + pub(crate) micro_high_low_spread: HighLowSpread, + pub(crate) micro_vw_spread: VolumeWeightedSpread, + pub(crate) micro_tick_count: TickCount, + pub(crate) micro_inter_arrival: InterArrivalTime, + pub(crate) micro_buy_sell_imbalance: BuySellImbalance, + pub(crate) micro_kyle_lambda: KyleLambda, + pub(crate) micro_price_impact: PriceImpact, + pub(crate) micro_variance_ratio: VarianceRatio, // Note: Roll Measure, Corwin-Schultz, Amihud, VPIN already exist in ml/src/microstructure/ // We'll integrate those in the update logic /// Track last timestamp for inter-arrival time calculation last_timestamp_ns: u64, /// Track last close price for microstructure calculations - last_close: f64, + pub(crate) last_close: f64, // WAVE 1.1: Triple Barrier Integration /// Triple barrier engine for position exit labeling @@ -427,7 +144,7 @@ pub struct DQNTrainer { safety_step_counter: usize, /// Optional path to feature cache directory for faster hyperopt - feature_cache_dir: Option, + pub(crate) feature_cache_dir: Option, /// WAVE 24 (Agent 17): Patience-based early stopping for anti-overfitting early_stopping: super::early_stopping::EarlyStopping, @@ -1369,7 +1086,7 @@ impl DQNTrainer { Ok(metrics) } - async fn train_with_data_full_loop( + pub(crate) async fn train_with_data_full_loop( &mut self, training_data: Vec<(FeatureVector51, Vec)>, mut checkpoint_callback: F, @@ -2620,704 +2337,6 @@ impl DQNTrainer { Ok(metrics) } - /// Train DQN on market data from Parquet file (Wave 12 Group 3) - /// - /// # Arguments - /// - /// * `parquet_path` - Path to Parquet file containing OHLCV bars - /// * `checkpoint_callback` - Callback for saving checkpoints (epoch, model_data) -> `Result` - /// - /// # Returns - /// - /// Training metrics (loss, accuracy, gradient norms, Q-values) - pub async fn train_from_parquet( - &mut self, - parquet_path: &str, - checkpoint_callback: F, - ) -> Result - where - F: FnMut(usize, Vec, bool) -> Result + Send, - { - info!("Starting DQN training from Parquet file: {}", parquet_path); - - // Load market data from Parquet file (returns train/val split) - let (mut training_data, mut validation_data) = - self.load_training_data_from_parquet(parquet_path).await?; - - info!( - "Loaded {} training samples, {} validation samples", - training_data.len(), - validation_data.len() - ); - - - // Calculate feature statistics from all training samples - info!("📊 Calculating feature statistics from {} training samples...", training_data.len()); - let feature_stats = self.calculate_feature_statistics(&training_data)?; - self.feature_stats = Some(feature_stats.clone()); - info!("✅ Feature statistics calculated: {} features normalized", feature_stats.mean.len()); - - // Normalize all training and validation samples BEFORE training starts - info!("📊 Normalizing all samples with z-score normalization..."); - self.normalize_dataset(&mut training_data)?; - self.normalize_dataset(&mut validation_data)?; - info!("✅ Dataset normalization complete"); - - // Store normalized validation data for validation loss computation - self.val_data = validation_data; - - // Use the same training loop as DBN-based training - self.train_with_data_full_loop(training_data, checkpoint_callback) - .await - } - - /// Load training data from Parquet file (Wave 12 Group 3) - /// Returns (train_data, val_data) with 80/20 split - pub async fn load_training_data_from_parquet( - &mut self, - parquet_path: &str, - ) -> Result<( - Vec<(FeatureVector51, Vec)>, - Vec<(FeatureVector51, Vec)>, - )> { - use arrow::array::{Array, Float64Array, PrimitiveArray, UInt64Array}; - use arrow::datatypes::TimestampNanosecondType; - use arrow::record_batch::RecordBatch; - use parquet::arrow::arrow_reader::ParquetRecordBatchReaderBuilder; - use std::fs::File; - - info!("Loading Parquet file: {}", parquet_path); - - // TRY CACHE FIRST - if let Some(cache_dir) = &self.feature_cache_dir { - let parquet_path_obj = Path::new(parquet_path); - let mbp10 = Path::new("test_data/mbp10"); - - info!("🔍 Checking feature cache..."); - - // Calculate cache key - match crate::feature_cache::calculate_cache_key( - parquet_path_obj, - mbp10, - 50, // warmup period - ) { - Ok(cache_key) => { - // Try to load from cache - match crate::feature_cache::load_features_from_cache( - cache_dir, - &cache_key, - ).await { - Ok(Some(features)) => { - info!("🚀 Loaded {} feature vectors from cache", features.len()); - info!(" ⚡ Savings vs compute: ~2m 25s"); - - // Split into train/val (80/20) - let split_idx = (features.len() as f64 * 0.8) as usize; - let train_data: Vec<(FeatureVector51, Vec)> = features[..split_idx] - .iter() - .map(|f| (*f, vec![])) - .collect(); - let val_data: Vec<(FeatureVector51, Vec)> = features[split_idx..] - .iter() - .map(|f| (*f, vec![])) - .collect(); - - return Ok((train_data, val_data)); - } - Ok(None) => { - info!("⚠️ Cache miss, computing features from scratch..."); - } - Err(e) => { - warn!("⚠️ Cache load failed: {}, computing features...", e); - } - } - } - Err(e) => { - warn!("⚠️ Failed to calculate cache key: {}, skipping cache", e); - } - } - } - - // FALLBACK: Original feature extraction - info!("📊 Computing features from scratch..."); - - // Open Parquet file - let file = File::open(parquet_path) - .with_context(|| format!("Failed to open Parquet file: {}", parquet_path))?; - - // Create Parquet reader - let builder = ParquetRecordBatchReaderBuilder::try_new(file) - .with_context(|| "Failed to create Parquet reader")?; - - let reader = builder - .build() - .with_context(|| "Failed to build Parquet reader")?; - - // Read all batches - let mut all_ohlcv_bars = Vec::new(); - - for batch_result in reader { - let batch: RecordBatch = batch_result.with_context(|| "Failed to read record batch")?; - - // Extract columns by name (schema-agnostic approach) - // Required columns: timestamp_ns (or ts_event), open, high, low, close, volume - - // Try timestamp_ns first (our schema), fallback to ts_event (Databento schema) - let timestamp_col = batch - .column_by_name("timestamp_ns") - .or_else(|| batch.column_by_name("ts_event")) - .ok_or_else(|| { - anyhow::anyhow!( - "Missing timestamp column. Expected 'timestamp_ns' or 'ts_event'" - ) - })?; - - let timestamps = timestamp_col - .as_any() - .downcast_ref::>() - .ok_or_else(|| { - anyhow::anyhow!( - "Failed to downcast timestamp column. Expected Timestamp(Nanosecond), got: {:?}", - timestamp_col.data_type() - ) - })?; - - // Extract OHLCV columns by name - let opens = batch - .column_by_name("open") - .ok_or_else(|| anyhow::anyhow!("Missing 'open' column in Parquet schema"))? - .as_any() - .downcast_ref::() - .ok_or_else(|| anyhow::anyhow!("Invalid 'open' column type. Expected Float64"))?; - - let highs = batch - .column_by_name("high") - .ok_or_else(|| anyhow::anyhow!("Missing 'high' column in Parquet schema"))? - .as_any() - .downcast_ref::() - .ok_or_else(|| anyhow::anyhow!("Invalid 'high' column type. Expected Float64"))?; - - let lows = batch - .column_by_name("low") - .ok_or_else(|| anyhow::anyhow!("Missing 'low' column in Parquet schema"))? - .as_any() - .downcast_ref::() - .ok_or_else(|| anyhow::anyhow!("Invalid 'low' column type. Expected Float64"))?; - - let closes = batch - .column_by_name("close") - .ok_or_else(|| anyhow::anyhow!("Missing 'close' column in Parquet schema"))? - .as_any() - .downcast_ref::() - .ok_or_else(|| anyhow::anyhow!("Invalid 'close' column type. Expected Float64"))?; - - let volumes = batch - .column_by_name("volume") - .ok_or_else(|| anyhow::anyhow!("Missing 'volume' column in Parquet schema"))? - .as_any() - .downcast_ref::() - .ok_or_else(|| anyhow::anyhow!("Invalid 'volume' column type. Expected UInt64"))?; - - // Convert to OHLCVBar structs - for i in 0..batch.num_rows() { - let timestamp_ns = timestamps.value(i); - let timestamp = chrono::DateTime::from_timestamp_nanos(timestamp_ns); - - let bar = OHLCVBar { - timestamp, - open: opens.value(i), - high: highs.value(i), - low: lows.value(i), - close: closes.value(i), - volume: volumes.value(i) as f64, // Convert u64 to f64 - }; - all_ohlcv_bars.push(bar); - } - } - - info!( - "Successfully loaded {} OHLCV bars from Parquet file", - all_ohlcv_bars.len() - ); - - // Sort bars by timestamp (critical for rolling window feature extraction) - debug!("Sorting bars chronologically by timestamp..."); - all_ohlcv_bars.sort_by_key(|bar| bar.timestamp); - debug!("Bars sorted successfully"); - - // Wave 14 Agent 32: Preprocess close prices for stationarity - let preprocessed_closes = if self.hyperparams.enable_preprocessing { - info!("🔬 Preprocessing enabled: Applying log returns + windowed normalization + outlier clipping"); - - // Extract close prices - // WAVE 16E: Convert f64 to f32 for preprocessing (preprocessing expects f32 tensors) - let close_prices_f64: Vec = all_ohlcv_bars.iter().map(|b| b.close).collect(); - let close_prices_f32: Vec = close_prices_f64.iter().map(|&x| x as f32).collect(); - let device = Device::cuda_if_available(0).unwrap_or(Device::Cpu); - let close_tensor = - Tensor::from_slice(&close_prices_f32, (close_prices_f32.len(),), &device) - .context("Failed to create close price tensor")?; - - // Configure preprocessing - let preprocess_config = PreprocessConfig { - window_size: self.hyperparams.preprocessing_window, - clip_sigma: self.hyperparams.preprocessing_clip_sigma, - use_log_returns: true, - }; - - info!(" • Window size: {}", preprocess_config.window_size); - info!(" • Clip sigma: ±{:.1}σ", preprocess_config.clip_sigma); - - // Apply preprocessing - let preprocessed_tensor = preprocess_prices(&close_tensor, preprocess_config) - .context("Failed to preprocess prices")?; - - let preprocessed_vec: Vec = preprocessed_tensor - .to_vec1() - .context("Failed to convert preprocessed tensor to vec")?; - - // Convert f32 to f64 for consistency with existing pipeline - let preprocessed_f64: Vec = preprocessed_vec.iter().map(|&x| x as f64).collect(); - - // Compute statistics for validation - let warmup = preprocess_config.window_size as usize; - let post_warmup: Vec = preprocessed_f64[warmup..].to_vec(); - let mean = post_warmup.iter().sum::() / post_warmup.len() as f64; - let variance = post_warmup.iter().map(|&x| (x - mean).powi(2)).sum::() - / post_warmup.len() as f64; - let std = variance.sqrt(); - let max_abs = post_warmup.iter().map(|&x| x.abs()).fold(0.0f64, f64::max); - - debug!("✅ Preprocessing complete:"); - debug!(" • Mean: {:.6} (expected ~0 for normalized data)", mean); - debug!(" • Std: {:.4} (expected ~1 for normalized data)", std); - debug!( - " • Max absolute value: {:.4} (clipped at ±{:.1}σ)", - max_abs, preprocess_config.clip_sigma - ); - - Some(preprocessed_f64) - } else { - info!("⚠️ Preprocessing disabled: Using raw close prices (NON-STATIONARY)"); - None - }; - - // WAVE 2-A2: Load MBP-10 snapshots for OFI calculation - use data::providers::databento::dbn_parser::DbnParser; - let mbp10_dir = Path::new("test_data/mbp10"); - let mbp10_snapshots = if mbp10_dir.exists() { - info!("📊 Loading MBP-10 order book snapshots for OFI calculation..."); - // Load all .dbn files in the directory - let mut all_snapshots = Vec::new(); - if let Ok(entries) = std::fs::read_dir(mbp10_dir) { - let parser = DbnParser::new() - .context("Failed to create DBN parser for MBP-10 data")?; - - for entry in entries.flatten() { - let path = entry.path(); - // Only load .dbn files (not .zst compressed files) - if path.extension().and_then(|s| s.to_str()) == Some("dbn") { - match parser.parse_mbp10_file(&path).await { - Ok(mut snaps) => { - info!(" ✅ Loaded {} snapshots from {:?}", snaps.len(), path.file_name()); - all_snapshots.append(&mut snaps); - } - Err(e) => { - warn!(" ⚠️ Failed to load {:?}: {}", path.file_name(), e); - } - } - } - } - } - - if all_snapshots.is_empty() { - warn!("⚠️ No MBP-10 snapshots loaded. OFI features will be zeros."); - None - } else { - // Sort snapshots by timestamp for efficient lookup - all_snapshots.sort_by_key(|s| s.timestamp); - info!("✅ Total MBP-10 snapshots loaded: {} (sorted by timestamp)", all_snapshots.len()); - Some(all_snapshots) - } - } else { - warn!("⚠️ MBP-10 directory not found at {:?}. OFI features will be zeros.", mbp10_dir); - None - }; - - // Extract 54-feature vectors (technical indicators, OFI, time, statistical features) - info!("Extracting 51-feature vectors from OHLCV bars (51-feature architecture)..."); - let feature_vectors = self.extract_full_features(&all_ohlcv_bars, mbp10_snapshots.as_deref())?; - - info!( - "Extracted {} feature vectors (51 dimensions: technical indicators, time, statistical features (Proxy OFI removed))", - feature_vectors.len() - ); - - // Create training data pairs (features, target) - // Target: [preprocessed_current, preprocessed_next, raw_current, raw_next] - // WAVE 3 BUG FIX: Include raw prices for triple barrier tracker (needs actual market prices in cents) - // Wave 14 Agent 32: Use preprocessed closes if enabled - let mut training_data = Vec::new(); - for i in 0..feature_vectors.len().saturating_sub(1) { - let (preprocessed_current, preprocessed_next, raw_current, raw_next) = if let Some(ref preprocessed) = preprocessed_closes { - // Use preprocessed values for reward calculation + raw for barrier tracker - ( - preprocessed[i + 50], - preprocessed[i + 1 + 50], - all_ohlcv_bars[i + 50].close, - all_ohlcv_bars[i + 1 + 50].close, - ) - } else { - // Use raw prices for both (original behavior) - let raw_curr = all_ohlcv_bars[i + 50].close; - let raw_next = all_ohlcv_bars[i + 1 + 50].close; - (raw_curr, raw_next, raw_curr, raw_next) - }; - training_data.push((feature_vectors[i], vec![preprocessed_current, preprocessed_next, raw_current, raw_next])); - } - // Last sample targets itself - if !feature_vectors.is_empty() { - let idx = all_ohlcv_bars.len() - 1; - let (preprocessed_close, raw_close) = if let Some(ref preprocessed) = preprocessed_closes { - (preprocessed[idx], all_ohlcv_bars[idx].close) - } else { - let raw = all_ohlcv_bars[idx].close; - (raw, raw) - }; - training_data.push(( - feature_vectors[feature_vectors.len() - 1], - vec![preprocessed_close, preprocessed_close, raw_close, raw_close], - )); - } - - info!( - "Created {} total samples with 54-dim features", - training_data.len() - ); - - // Split training data 80/20 for train/validation - let split_idx = (training_data.len() * 80) / 100; - let train_data = training_data[..split_idx].to_vec(); - let val_data = training_data[split_idx..].to_vec(); - - info!( - "Split data - Training samples: {}, Validation samples: {}", - train_data.len(), - val_data.len() - ); - - Ok((train_data, val_data)) - } - - /// Load training data from DBN files using official dbn crate decoder - /// Returns (train_data, val_data) with 80/20 split - async fn load_training_data( - &mut self, - dbn_data_dir: &str, - ) -> Result<( - Vec<(FeatureVector51, Vec)>, - Vec<(FeatureVector51, Vec)>, - )> { - // Find all DBN files in directory - let dir_path = Path::new(dbn_data_dir); - if !dir_path.exists() { - return Err(anyhow::anyhow!( - "Data directory not found: {}", - dbn_data_dir - )); - } - - let dbn_files: Vec<_> = std::fs::read_dir(dir_path)? - .filter_map(|entry| entry.ok()) - .filter(|entry| entry.path().extension().and_then(|s| s.to_str()) == Some("dbn")) - .map(|entry| entry.path()) - .collect(); - - if dbn_files.is_empty() { - return Err(anyhow::anyhow!("No DBN files found in: {}", dbn_data_dir)); - } - - info!("Found {} DBN files to load", dbn_files.len()); - - let mut all_ohlcv_bars = Vec::new(); - - // Load and decode each DBN file to collect OHLCV bars - for (file_idx, file_path) in dbn_files.iter().enumerate() { - debug!( - "Loading DBN file {}/{}: {}", - file_idx + 1, - dbn_files.len(), - file_path.display() - ); - - // Extract raw OHLCV bars from file - let file_bars = self.extract_ohlcv_bars_from_dbn(file_path)?; - - debug!( - "Extracted {} OHLCV bars from {}", - file_bars.len(), - file_path.file_name().unwrap_or_default().to_string_lossy() - ); - - all_ohlcv_bars.extend(file_bars); - } - - if all_ohlcv_bars.is_empty() { - return Err(anyhow::anyhow!( - "No OHLCV bars extracted from DBN files. Check if files contain OHLCV messages." - )); - } - - info!( - "Successfully loaded {} OHLCV bars from {} DBN files", - all_ohlcv_bars.len(), - dbn_files.len() - ); - - // Sort bars by timestamp (critical for rolling window feature extraction) - debug!("Sorting bars chronologically by timestamp..."); - all_ohlcv_bars.sort_by_key(|bar| bar.timestamp); - debug!("Bars sorted successfully"); - - // Extract 54-feature vectors (technical indicators, Proxy OFI, time, statistical features) - // Note: DBN loader does not load MBP-10 data, so OFI features will be zeros - info!("Extracting 51-feature vectors from OHLCV bars (51-feature architecture)..."); - let feature_vectors = self.extract_full_features(&all_ohlcv_bars, None)?; - - info!( - "Extracted {} feature vectors (51 dimensions: technical indicators, time, statistical features (Proxy OFI removed))", - feature_vectors.len() - ); - - // Create training data pairs (features, target) - // Target: [current_close, next_close] for proper reward calculation - let mut training_data = Vec::new(); - for i in 0..feature_vectors.len().saturating_sub(1) { - let current_close = all_ohlcv_bars[i + 50].close; // +50 to account for warmup period - let next_close = all_ohlcv_bars[i + 1 + 50].close; - training_data.push((feature_vectors[i], vec![current_close, next_close])); - } - // Last sample targets itself - if !feature_vectors.is_empty() { - let idx = all_ohlcv_bars.len() - 1; - let current_close = all_ohlcv_bars[idx].close; - training_data.push(( - feature_vectors[feature_vectors.len() - 1], - vec![current_close, current_close], - )); - } - - info!( - "Created {} total samples with 54-dim features", - training_data.len() - ); - - // Split training data 80/20 for train/validation - let split_idx = (training_data.len() * 80) / 100; - let train_data = training_data[..split_idx].to_vec(); - let val_data = training_data[split_idx..].to_vec(); - - info!( - "Split data - Training samples: {}, Validation samples: {}", - train_data.len(), - val_data.len() - ); - - Ok((train_data, val_data)) - } - - /// Extract raw OHLCV bars from DBN file using official dbn crate decoder - /// - /// This replaces the custom parser that only extracted 2 messages (header metadata). - /// Now extracts all OHLCV bars (400-500+ records per file). - /// - /// Public for testing purposes. - pub fn extract_ohlcv_bars_from_dbn(&self, file_path: &Path) -> Result> { - use dbn::decode::dbn::Decoder; - use dbn::decode::{DbnMetadata, DecodeRecordRef}; - use std::fs::File; - use std::io::BufReader; - - let mut ohlcv_bars = Vec::new(); - - // Open file and create official DBN decoder - let file = File::open(file_path) - .with_context(|| format!("Failed to open DBN file: {:?}", file_path))?; - let reader = BufReader::new(file); - - let mut decoder = Decoder::new(reader) - .map_err(|e| anyhow::anyhow!("Failed to create DBN decoder: {}", e))?; - - // Read metadata (for logging) - let metadata = decoder.metadata(); - debug!( - "DBN file metadata: dataset={:?}, schema={:?}, symbols={:?}", - metadata.dataset, metadata.schema, metadata.symbols - ); - - // Decode all OHLCV records - let mut ohlcv_count = 0; - let mut other_count = 0; - let mut idx = 0; - - loop { - match decoder.decode_record_ref() { - Ok(Some(record)) => { - idx += 1; - - // Convert RecordRef to RecordRefEnum for pattern matching - let record_enum = record - .as_enum() - .map_err(|e| anyhow::anyhow!("Failed to convert record to enum: {}", e))?; - - match record_enum { - dbn::RecordRefEnum::Ohlcv(ohlcv) => { - ohlcv_count += 1; - - // Extract OHLCV values (prices are i64 scaled by 1e-9 per DBN spec, volume is u64) - let open_f64 = ohlcv.open as f64 * 1e-9; - let high_f64 = ohlcv.high as f64 * 1e-9; - let low_f64 = ohlcv.low as f64 * 1e-9; - let close_f64 = ohlcv.close as f64 * 1e-9; - let volume_u64 = ohlcv.volume; - - // WAVE 8 AGENT 36: Validate all price values are finite (not NaN/Inf) - // Skip bars with invalid data to prevent NaN propagation - if !open_f64.is_finite() - || !high_f64.is_finite() - || !low_f64.is_finite() - || !close_f64.is_finite() - { - debug!( - "Skipping OHLCV bar {} with non-finite values: open={}, high={}, low={}, close={}", - ohlcv_count, open_f64, high_f64, low_f64, close_f64 - ); - continue; - } - - // Log first few records for validation - if ohlcv_count <= 5 { - debug!( - "Raw OHLCV #{}: open={}, high={}, low={}, close={}", - ohlcv_count, ohlcv.open, ohlcv.high, ohlcv.low, ohlcv.close - ); - debug!( - "Scaled OHLCV #{}: open={:.6}, high={:.6}, low={:.6}, close={:.6}", - ohlcv_count, open_f64, high_f64, low_f64, close_f64 - ); - } - - // Convert timestamp from nanoseconds since epoch to DateTime - let timestamp_nanos = ohlcv.hd.ts_event as i64; - let timestamp_secs = timestamp_nanos / 1_000_000_000; - let timestamp_nanos_remainder = - (timestamp_nanos % 1_000_000_000) as u32; - let timestamp = chrono::DateTime::::from_timestamp( - timestamp_secs, - timestamp_nanos_remainder, - ) - .unwrap_or_else(|| chrono::Utc::now()); - - // Create OHLCVBar for feature extraction pipeline - let bar = OHLCVBar { - timestamp, - open: open_f64, - high: high_f64, - low: low_f64, - close: close_f64, - volume: volume_u64 as f64, - }; - - ohlcv_bars.push(bar); - }, - _ => { - other_count += 1; - if other_count <= 5 { - debug!("Skipping non-OHLCV record at index {}", idx); - } - }, - } - }, - Ok(None) => { - // End of stream - break; - }, - Err(e) => { - return Err(anyhow::anyhow!("Failed to decode record {}: {}", idx, e)); - }, - } - } - - info!( - "Extracted {} OHLCV bars from {:?} ({} other records skipped)", - ohlcv_count, - file_path.file_name().unwrap_or_default(), - other_count - ); - - Ok(ohlcv_bars) - } - - /// Create features from OHLCV data - fn create_ohlcv_features( - &self, - open: f64, - high: f64, - low: f64, - close: f64, - volume: u64, - ) -> Result { - use std::collections::HashMap; - - // Use absolute values for Price type (futures data can have negative values) - // For ML training, the absolute magnitude is what matters for feature extraction - let close_price = - common::Price::from_f64(close.abs()).unwrap_or_else(|_| common::Price::ZERO); - let open_price = - common::Price::from_f64(open.abs()).unwrap_or_else(|_| common::Price::ZERO); - let high_price = - common::Price::from_f64(high.abs()).unwrap_or_else(|_| common::Price::ZERO); - let low_price = common::Price::from_f64(low.abs()).unwrap_or_else(|_| common::Price::ZERO); - - // Calculate technical indicators - let mut indicators = HashMap::new(); - - // Price-based features - let price_range = high - low; - let body_size = (close - open).abs(); - let upper_shadow = high - close.max(open); - let lower_shadow = close.min(open) - low; - - indicators.insert("price_range".to_string(), price_range); - indicators.insert("body_size".to_string(), body_size); - indicators.insert("upper_shadow".to_string(), upper_shadow); - indicators.insert("lower_shadow".to_string(), lower_shadow); - indicators.insert("close_to_high".to_string(), (close - high).abs()); - indicators.insert("close_to_low".to_string(), (close - low).abs()); - - // Microstructure features - let spread_bps = ((high - low) / close * 10000.0) as i32; - let trade_intensity = volume as f64; - - Ok(FinancialFeatures { - prices: vec![open_price, high_price, low_price, close_price], - volumes: vec![volume as i64], - technical_indicators: indicators, - microstructure: crate::training_pipeline::MicrostructureFeatures { - spread_bps, - imbalance: 0.0, // Not available from OHLCV - trade_intensity, - vwap: close_price, // Approximate VWAP as close - }, - risk_metrics: crate::training_pipeline::RiskFeatures { - var_5pct: -0.02, // Placeholder - expected_shortfall: -0.03, - max_drawdown: -0.05, - sharpe_ratio: 1.0, - }, - timestamp: chrono::Utc::now(), - }) - } /// Convert 54-dim feature vector to TradingState /// @@ -3960,334 +2979,13 @@ impl DQNTrainer { Ok(()) } - // WAVE 16S: Adaptive Risk Management Helper Methods - /// Calculate volatility-adjusted epsilon for exploration - /// - /// Adjusts epsilon based on recent return volatility: - /// - Low volatility (<1%): reduce epsilon (exploit more) - /// - High volatility (>5%): increase epsilon (explore more) - /// - Moderate volatility: linear interpolation - fn calculate_volatility_adjusted_epsilon(&self, base_epsilon: f64) -> f64 { - if !self.hyperparams.enable_volatility_epsilon || self.volatility_returns.len() < 10 { - return base_epsilon; - } - - // Calculate volatility (standard deviation of returns) - let mean: f64 = self.volatility_returns.iter().sum::() / self.volatility_returns.len() as f64; - let variance: f64 = self.volatility_returns.iter() - .map(|x| (x - mean).powi(2)) - .sum::() / self.volatility_returns.len() as f64; - let volatility = variance.sqrt(); - - // Adjust epsilon based on volatility regime - let multiplier = if volatility < 0.01 { - 0.5 // Low volatility: exploit more (reduce epsilon) - } else if volatility > 0.05 { - 2.0 // High volatility: explore more (increase epsilon) - } else { - // Linear interpolation between 0.01 and 0.05 - 0.5 + (volatility - 0.01) / 0.04 * 1.5 - }; - - (base_epsilon * multiplier).clamp(0.05, 0.95) - } - - /// Calculate risk-adjusted reward using Sharpe ratio - /// - /// Amplifies rewards for consistent profitable strategies, - /// reduces rewards for volatile strategies. - fn calculate_risk_adjusted_reward(&self, raw_reward: f64) -> f64 { - if !self.hyperparams.enable_risk_adjusted_rewards || self.pnl_history.len() < 20 { - return raw_reward; - } - - // Calculate Sharpe ratio from PnL history - let mean: f64 = self.pnl_history.iter().sum::() / self.pnl_history.len() as f64; - let variance: f64 = self.pnl_history.iter() - .map(|x| (x - mean).powi(2)) - .sum::() / self.pnl_history.len() as f64; - let std_dev = variance.sqrt().max(1e-8); - let sharpe = mean / std_dev; - - // Apply Sharpe multiplier to reward - // Positive Sharpe amplifies reward, negative reduces it - raw_reward * sharpe - } - - /// Get Kelly fraction for position sizing - /// - /// Returns Kelly criterion position size (0.0-0.25) based on trade history. - /// Requires minimum trade history for statistical significance. - pub fn get_kelly_fraction(&self) -> f64 { - if !self.hyperparams.enable_kelly_sizing { - return 1.0; // Full position sizing (disabled) - } - - let kelly_opt = match &self.kelly_optimizer { - Some(opt) => opt, - None => return 1.0, - }; - - // Need minimum trades for Kelly calculation - if self.trade_history.len() < self.hyperparams.kelly_min_trades { - return 0.1; // Conservative default until enough history - } - - // Calculate win/loss statistics - let wins: Vec = self.trade_history.iter() - .filter(|&&r| r > 0.0) - .copied() - .collect(); - let losses: Vec = self.trade_history.iter() - .filter(|&&r| r < 0.0) - .map(|r| -r) - .collect(); - - let win_prob = wins.len() as f64 / self.trade_history.len() as f64; - let avg_win = if wins.is_empty() { - 0.01 - } else { - wins.iter().sum::() / wins.len() as f64 - }; - let avg_loss = if losses.is_empty() { - 0.01 - } else { - losses.iter().sum::() / losses.len() as f64 - }; - - // Calculate Kelly fraction - let kelly_result = kelly_opt.calculate_basic_kelly(win_prob, avg_win, avg_loss); - let kelly_fraction = kelly_result.unwrap_or(0.1); - - // Apply fractional Kelly (conservative) - let fractional_kelly = kelly_fraction * self.hyperparams.kelly_fractional; - - // Clamp to safety bounds - fractional_kelly.clamp(0.01, self.hyperparams.kelly_max_fraction) - } - - /// Update adaptive risk trackers with new market data - fn update_risk_trackers(&mut self, reward: f64, price_return: f64) { - // Update PnL history - self.pnl_history.push_back(reward); - if self.pnl_history.len() > 1000 { - self.pnl_history.pop_front(); - } - - // Update volatility tracker - self.volatility_returns.push_back(price_return); - if self.volatility_returns.len() > self.hyperparams.volatility_window { - self.volatility_returns.pop_front(); - } - - // Update trade history for Kelly - if reward.abs() > 1e-6 { // Only track non-zero rewards - self.trade_history.push_back(reward); - if self.trade_history.len() > 500 { - self.trade_history.pop_front(); - } - } - } - - /// Create synthetic features (placeholder for testing) - fn create_synthetic_features(&self, price: f64) -> Result { - use std::collections::HashMap; - - let price_obj = - common::Price::from_f64(price).unwrap_or_else(|_| common::Price::new(price).unwrap()); - - let mut indicators = HashMap::new(); - indicators.insert("rsi_14".to_string(), 50.0); - indicators.insert("sma_20".to_string(), price); - indicators.insert("ema_12".to_string(), price); - - Ok(FinancialFeatures { - prices: vec![price_obj; 4], - volumes: vec![1000], - technical_indicators: indicators, - microstructure: crate::training_pipeline::MicrostructureFeatures { - spread_bps: 10, - imbalance: 0.0, - trade_intensity: 100.0, - vwap: price_obj, - }, - risk_metrics: crate::training_pipeline::RiskFeatures { - var_5pct: -0.02, - expected_shortfall: -0.03, - max_drawdown: -0.05, - sharpe_ratio: 1.0, - }, - timestamp: chrono::Utc::now(), - }) - } /// Get current training metrics pub async fn get_metrics(&self) -> TrainingMetrics { self.metrics.read().await.clone() } - /// WAVE 3.10: Extract 140 features (125 market + 3 portfolio + 12 microstructure) - /// - /// This method extracts 140 features for DQN state representation: - /// - 125 market features (price, technical indicators, volatility, etc.) - /// - 3 portfolio features (populated later via PortfolioTracker) - /// - 12 microstructure features (spread estimators, liquidity, order flow, market impact) - /// - /// The microstructure features are calculated on-the-fly from OHLCV data using - /// the calculators initialized in DQNTrainer::new(). - /// - /// # Arguments - /// - /// * `bars` - OHLCV bars for feature extraction - /// * `mbp10_snapshots` - Optional MBP-10 order book snapshots for OFI calculation - fn extract_full_features( - &mut self, - bars: &[OHLCVBar], - mbp10_snapshots: Option<&[data::providers::databento::mbp10::Mbp10Snapshot]>, - ) -> Result> { - use crate::features::extraction::FeatureExtractor; - - if bars.is_empty() { - anyhow::bail!("Cannot extract features from empty bar sequence"); - } - - const WARMUP_PERIOD: usize = 50; - if bars.len() < WARMUP_PERIOD { - anyhow::bail!( - "Insufficient data: {} bars provided, {} required for warmup", - bars.len(), - WARMUP_PERIOD - ); - } - - let mut extractor = FeatureExtractor::new(); - let mut feature_vectors = Vec::with_capacity(bars.len() - WARMUP_PERIOD); - - // Feed bars sequentially to build rolling windows - for (i, bar) in bars.iter().enumerate() { - extractor.update(bar)?; - - // WAVE 3.10: Update microstructure features with current bar - // Calculate timestamp in nanoseconds - let timestamp_ns = bar.timestamp.timestamp_nanos_opt().unwrap_or(0) as u64; - - // Update all 8 microstructure calculators (4 more already exist in microstructure.rs) - let hl_spread = self.micro_high_low_spread.update(bar.high, bar.low); - let _vw_spread = self.micro_vw_spread.update(hl_spread, bar.volume); - let _tick_count = self.micro_tick_count.update(bar.close); - let _inter_arrival = self.micro_inter_arrival.update(timestamp_ns); - let _buy_sell_imb = self.micro_buy_sell_imbalance.update(bar.close, bar.volume); - - // Kyle's Lambda: slow-updating (only updates every 5 minutes) - let return_pct = if self.last_close > 0.0 { - (bar.close - self.last_close) / self.last_close - } else { - 0.0 - }; - let signed_volume = (bar.close - bar.open).signum() * (bar.close * bar.volume).sqrt(); - let _kyle_lambda = self.micro_kyle_lambda.maybe_update(timestamp_ns, return_pct, signed_volume); - - let _price_impact = self.micro_price_impact.update(bar.high, bar.low, bar.close); - let _variance_ratio = self.micro_variance_ratio.update(return_pct); - - // Track last close for next iteration - self.last_close = bar.close; - - // Start extracting features after warmup - if i >= WARMUP_PERIOD { - // WAVE 10: Extract 43 base features + 8 OFI features (51 total, Proxy OFI removed) - // Features breakdown: - // - 0-4: OHLCV (5) - // - 5-9: Technical indicators (5) - // - 10-15: Price patterns (6) - // - 16-21: Volume features (6) - // - 22-26: Time-based (5) - // - 27-39: Statistical (13) - // - 40-42: Regime detection (3) - // - 43-50: OFI features (8) - // TOTAL: 51 features - - // Extract features with OFI if MBP-10 data available - let features_51 = if let Some(mbp10_data) = mbp10_snapshots { - // Calculate OFI features from MBP-10 order book data - use crate::features::mbp10_loader::get_snapshots_for_timestamp; - - let bar_timestamp_ns = bar.timestamp.timestamp_nanos_opt().unwrap_or(0) as u64; - let window = get_snapshots_for_timestamp(mbp10_data, bar_timestamp_ns, 100); - - if !window.is_empty() { - match extractor.extract_current_features_with_ofi(window) { - Ok(feats) => feats, - Err(e) => { - warn!("Failed to calculate OFI for bar {}: {}. Using zeros for OFI.", i, e); - let base_features_43 = extractor.extract_current_features_v2()?; - let mut feats = [0.0; 51]; - feats[..43].copy_from_slice(&base_features_43); - feats - } - } - } else { - // No MBP-10 snapshots for this timestamp - let base_features_43 = extractor.extract_current_features_v2()?; - let mut feats = [0.0; 51]; - feats[..43].copy_from_slice(&base_features_43); - feats - } - } else { - // Fallback: Use base 43 features + zero-padded OFI - let base_features_43 = extractor.extract_current_features_v2()?; - let mut feats = [0.0; 51]; - feats[..43].copy_from_slice(&base_features_43); - feats - }; - - feature_vectors.push(features_51); - } - } - - Ok(feature_vectors) - } - - /// Calculate feature statistics from training samples using Welford's algorithm - fn calculate_feature_statistics( - &self, - samples: &[(FeatureVector51, Vec)], - ) -> Result { - let mut stats = FeatureStatistics::new(54); - - for (feature_vec, _) in samples { - let features: Vec = feature_vec.iter().map(|&v| v as f32).collect(); - stats.update(&features); - } - - Ok(stats) - } - - /// Normalize all samples in a dataset using z-score normalization - fn normalize_dataset( - &mut self, - samples: &mut [(FeatureVector51, Vec)], - ) -> Result<()> { - if let Some(ref stats) = self.feature_stats { - for (feature_vec, _) in samples.iter_mut() { - // Convert to f32 for normalization - let features_f32: Vec = feature_vec.iter().map(|&v| v as f32).collect(); - - // Normalize with skip (indices 125-127 are portfolio placeholders) - let normalized = stats.normalize_with_skip(&features_f32, &[125, 126, 127]); - - // Convert back to f64 and update - for (i, &val) in normalized.iter().enumerate() { - feature_vec[i] = val as f64; - } - } - } else { - return Err(anyhow::anyhow!("Feature statistics not initialized")); - } - - Ok(()) - } } #[cfg(test)]