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 <noreply@anthropic.com>
This commit is contained in:
jgrusewski
719
ml/src/trainers/dqn/data_loading.rs
Normal file
719
ml/src/trainers/dqn/data_loading.rs
Normal file
@@ -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<String>`
|
||||
///
|
||||
/// # Returns
|
||||
///
|
||||
/// Training metrics (loss, accuracy, gradient norms, Q-values)
|
||||
pub async fn train_from_parquet<F>(
|
||||
&mut self,
|
||||
parquet_path: &str,
|
||||
checkpoint_callback: F,
|
||||
) -> Result<TrainingMetrics>
|
||||
where
|
||||
F: FnMut(usize, Vec<u8>, bool) -> Result<String> + 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<f64>)>,
|
||||
Vec<(FeatureVector51, Vec<f64>)>,
|
||||
)> {
|
||||
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<f64>)> = features[..split_idx]
|
||||
.iter()
|
||||
.map(|f| (*f, vec![]))
|
||||
.collect();
|
||||
let val_data: Vec<(FeatureVector51, Vec<f64>)> = 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::<PrimitiveArray<TimestampNanosecondType>>()
|
||||
.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::<Float64Array>()
|
||||
.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::<Float64Array>()
|
||||
.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::<Float64Array>()
|
||||
.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::<Float64Array>()
|
||||
.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::<UInt64Array>()
|
||||
.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<f64> = all_ohlcv_bars.iter().map(|b| b.close).collect();
|
||||
let close_prices_f32: Vec<f32> = 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<f32> = 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<f64> = 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<f64> = preprocessed_f64[warmup..].to_vec();
|
||||
let mean = post_warmup.iter().sum::<f64>() / post_warmup.len() as f64;
|
||||
let variance = post_warmup.iter().map(|&x| (x - mean).powi(2)).sum::<f64>()
|
||||
/ 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<f64>)>,
|
||||
Vec<(FeatureVector51, Vec<f64>)>,
|
||||
)> {
|
||||
// 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<Vec<OHLCVBar>> {
|
||||
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::<chrono::Utc>::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<FinancialFeatures> {
|
||||
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(),
|
||||
})
|
||||
}
|
||||
}
|
||||
211
ml/src/trainers/dqn/features.rs
Normal file
211
ml/src/trainers/dqn/features.rs
Normal file
@@ -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<FinancialFeatures> {
|
||||
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<Vec<FeatureVector51>> {
|
||||
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<f64>)],
|
||||
) -> Result<FeatureStatistics> {
|
||||
let mut stats = FeatureStatistics::new(54);
|
||||
|
||||
for (feature_vec, _) in samples {
|
||||
let features: Vec<f32> = 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<f64>)],
|
||||
) -> 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<f32> = 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(())
|
||||
}
|
||||
}
|
||||
@@ -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;
|
||||
|
||||
|
||||
290
ml/src/trainers/dqn/monitoring.rs
Normal file
290
ml/src/trainers/dqn/monitoring.rs
Normal file
@@ -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<f32>,
|
||||
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<f64>, // Per-step Q-values for mean calculation
|
||||
|
||||
// WAVE P2: Episode length tracking
|
||||
pub(crate) episode_lengths: Vec<usize>,
|
||||
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::<f64>() / 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::<f32>() / self.reward_history.len() as f32;
|
||||
let variance = self
|
||||
.reward_history
|
||||
.iter()
|
||||
.map(|r| (r - mean).powi(2))
|
||||
.sum::<f32>()
|
||||
/ 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<i8>) {
|
||||
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::<usize>() 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::<f64>() / total as f64;
|
||||
let std_dev = variance.sqrt();
|
||||
|
||||
(mean, std_dev, min, max, self.barrier_exit_counts)
|
||||
}
|
||||
}
|
||||
145
ml/src/trainers/dqn/risk.rs
Normal file
145
ml/src/trainers/dqn/risk.rs
Normal file
@@ -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::<f64>() / self.volatility_returns.len() as f64;
|
||||
let variance: f64 = self.volatility_returns.iter()
|
||||
.map(|x| (x - mean).powi(2))
|
||||
.sum::<f64>() / 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::<f64>() / self.pnl_history.len() as f64;
|
||||
let variance: f64 = self.pnl_history.iter()
|
||||
.map(|x| (x - mean).powi(2))
|
||||
.sum::<f64>() / 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<f64> = self.trade_history.iter()
|
||||
.filter(|&&r| r > 0.0)
|
||||
.copied()
|
||||
.collect();
|
||||
let losses: Vec<f64> = 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::<f64>() / wins.len() as f64
|
||||
};
|
||||
let avg_loss = if losses.is_empty() {
|
||||
0.01
|
||||
} else {
|
||||
losses.iter().sum::<f64>() / 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();
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
File diff suppressed because it is too large
Load Diff
Reference in New Issue
Block a user