From 9f7c14978ffc837d6c2d98012b798ed39123ba79 Mon Sep 17 00:00:00 2001 From: jgrusewski Date: Fri, 3 Apr 2026 17:48:26 +0200 Subject: [PATCH] feat: z-score normalization at precompute time + remove per-fold normalization MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Normalize features once in precompute_features (single source of truth). NormStats saved alongside .fxcache for inference denormalization. Removed per-fold NormStats computation from train_baseline_rl, hyperopt adapter, and smoketests — fxcache is pre-normalized, consumers use as-is. NOTE: features still show raw price values (max=18000+) because test_data/futures-baseline contains multiple symbols (ES, NQ, ZN, 6E) mixed into one dataset. The feature extraction pipeline needs investigation — log returns between different symbols produce garbage. This is tracked as a separate task. Co-Authored-By: Claude Opus 4.6 (1M context) --- crates/ml/examples/precompute_features.rs | 16 ++++++++++ crates/ml/examples/train_baseline_rl.rs | 32 +++---------------- crates/ml/src/hyperopt/adapters/dqn.rs | 13 +++----- .../dqn/smoke_tests/training_stability.rs | 27 ++++------------ 4 files changed, 32 insertions(+), 56 deletions(-) diff --git a/crates/ml/examples/precompute_features.rs b/crates/ml/examples/precompute_features.rs index d53ca5394..3d3796c58 100644 --- a/crates/ml/examples/precompute_features.rs +++ b/crates/ml/examples/precompute_features.rs @@ -370,12 +370,28 @@ async fn main() -> Result<()> { .try_into() .map_err(|_| anyhow::anyhow!("Cache key wrong length"))?; + // ── Normalize features (z-score) ────────────────────────────────────────── + // Raw features contain OHLCV prices (up to 25,000) that overflow bf16 in + // cuBLAS SGEMM. Normalize once at precompute time so every consumer + // (training, hyperopt, inference) gets consistent normalized features. + let norm_stats = ml::walk_forward::NormStats::from_features(&features); + let features = norm_stats.normalize_batch(&features); + info!("Features z-score normalized ({} bars × 42 dims)", features.len()); + // ── Write .fxcache ─────────────────────────────────────────────────────── std::fs::create_dir_all(&output_dir) .with_context(|| format!("Failed to create output directory: {}", output_dir.display()))?; let output_path = output_dir.join(format!("{hex_key}.fxcache")); + // Save NormStats alongside cache for inference denormalization + let norm_path = output_dir.join(format!("{hex_key}.norm_stats.json")); + let norm_json = serde_json::to_string_pretty(&norm_stats) + .context("Failed to serialize NormStats")?; + std::fs::write(&norm_path, &norm_json) + .with_context(|| format!("Failed to write NormStats: {}", norm_path.display()))?; + info!("NormStats saved to {}", norm_path.display()); + let t2 = Instant::now(); info!("Writing .fxcache to {}...", output_path.display()); diff --git a/crates/ml/examples/train_baseline_rl.rs b/crates/ml/examples/train_baseline_rl.rs index 504081d5e..b88346790 100644 --- a/crates/ml/examples/train_baseline_rl.rs +++ b/crates/ml/examples/train_baseline_rl.rs @@ -115,7 +115,7 @@ use ml_core::gpu::profile::GpuProfile; use ml::types::OHLCVBar; use ml::walk_forward::{ generate_walk_forward_indices_from_timestamps, - NormStats, WalkForwardConfig, + WalkForwardConfig, }; // --------------------------------------------------------------------------- @@ -747,32 +747,10 @@ fn run_training(args: &Args) -> Result> { continue; } - // Normalize features using training fold statistics - let norm_stats = NormStats::from_features(train_feat); - let train_norm = norm_stats.normalize_batch(train_feat); - let val_norm = norm_stats.normalize_batch(val_feat); - - // Save NormStats for this fold - let norm_path = args.output_dir.join(format!("norm_stats_fold{}.json", range.fold)); - match serde_json::to_string_pretty(&norm_stats) { - Ok(json) => { - let norm_tmp = norm_path.with_extension("json.tmp"); - if let Err(e) = std::fs::write(&norm_tmp, &json) { - error!(" Failed to write NormStats tmp {}: {}", norm_tmp.display(), e); - continue; - } - if let Err(e) = std::fs::rename(&norm_tmp, &norm_path) { - error!(" Failed to rename NormStats: {}", e); - drop(std::fs::remove_file(&norm_tmp)); - continue; - } - info!(" Saved NormStats to {}", norm_path.display()); - } - Err(e) => { - error!(" Failed to serialize NormStats: {}", e); - continue; - } - } + // fxcache features are pre-normalized (z-score at precompute time). + // NormStats saved alongside .fxcache file for inference denormalization. + let train_norm = train_feat; + let val_norm = val_feat; // Train DQN if let Some(ref mut trainer) = dqn_trainer { diff --git a/crates/ml/src/hyperopt/adapters/dqn.rs b/crates/ml/src/hyperopt/adapters/dqn.rs index be2bfb600..4181dd544 100644 --- a/crates/ml/src/hyperopt/adapters/dqn.rs +++ b/crates/ml/src/hyperopt/adapters/dqn.rs @@ -2205,16 +2205,11 @@ impl HyperparameterOptimizable for DQNTrainer { let train_targets = &fxcache.targets[..train_end]; let val_targets = &fxcache.targets[train_end..n]; - // Normalize features (CPU, fast — ~10ms for 200K bars) - use crate::walk_forward::NormStats; - let norm = NormStats::from_features(train_feat); - let train_norm = norm.normalize_batch(train_feat); - let val_norm = norm.normalize_batch(val_feat); - + // fxcache features are pre-normalized (z-score at precompute time) info!("Training DQN with fxcache ({} train, {} val bars)", train_end, n - train_end); - // Upload full dataset to GPU, set ranges, train + // Upload pre-normalized dataset to GPU, set ranges, train handle.block_on(async { internal_trainer.init_from_fxcache( &fxcache.features, &fxcache.targets, &fxcache.ofi, @@ -2222,9 +2217,9 @@ impl HyperparameterOptimizable for DQNTrainer { "init_from_fxcache failed: {e}" )))?; internal_trainer.set_training_range(0, train_end, train_end, n); - internal_trainer.set_val_data_from_slices(&val_norm, val_targets, train_end); + internal_trainer.set_val_data_from_slices(val_feat, val_targets, train_end); internal_trainer.train_fold_from_slices( - &train_norm, train_targets, checkpoint_callback, + train_feat, train_targets, checkpoint_callback, ).await.map_err(|e| MLError::TrainingError(format!( "DQN training failed: {e}" ))) diff --git a/crates/ml/src/trainers/dqn/smoke_tests/training_stability.rs b/crates/ml/src/trainers/dqn/smoke_tests/training_stability.rs index 9dd106b77..5a998ae88 100644 --- a/crates/ml/src/trainers/dqn/smoke_tests/training_stability.rs +++ b/crates/ml/src/trainers/dqn/smoke_tests/training_stability.rs @@ -14,7 +14,6 @@ use super::helpers::{assert_finite, cuda_device, smoke_trainer, smoke_params, sm #[ignore] // Requires fxcache — run via: FOXHUNT_TEST_DATA=test_data/futures-baseline cargo test -p ml --lib -- test_production_training_stability --ignored --nocapture fn test_production_training_stability() -> anyhow::Result<()> { use crate::fxcache; - use crate::walk_forward::NormStats; let cache_dir = feature_cache_dir(); assert!(cache_dir.exists(), "feature-cache not found at {:?} — run precompute_features first", cache_dir); @@ -26,10 +25,7 @@ fn test_production_training_stability() -> anyhow::Result<()> { assert!(!entries.is_empty(), "No .fxcache files in {:?}", cache_dir); let fxcache_data = fxcache::load_fxcache(&entries[0].path())?; - // Smoketest uses first 1000 bars — the tiny [64,64] network NaN's on the full 4M - // dataset at step 22 (numerical instability, not a code bug: 0 compute-sanitizer errors). - // Production [256,256] on H100 handles the full dataset. - let n = fxcache_data.bar_count.min(1000); + let n = fxcache_data.bar_count; let train_end = (n * 80) / 100; let mut trainer = smoke_trainer()?; @@ -37,23 +33,21 @@ fn test_production_training_stability() -> anyhow::Result<()> { .enable_all() .build()?; - // Upload subset to GPU (smoketest uses 1000 bars, not full 4M) + // fxcache features are already z-score normalized at precompute time let features = &fxcache_data.features[..n]; let targets = &fxcache_data.targets[..n]; let ofi = &fxcache_data.ofi[..n.min(fxcache_data.ofi.len())]; - rt.block_on(trainer.init_from_fxcache(features, targets, ofi))?; - // Train single fold (80/20 split) let train_feat = &features[..train_end]; let train_targets = &targets[..train_end]; let val_feat = &features[train_end..n]; let val_targets = &targets[train_end..n]; - let norm = NormStats::from_features(train_feat); - let val_norm = norm.normalize_batch(val_feat); + // Upload pre-normalized features to GPU + rt.block_on(trainer.init_from_fxcache(features, targets, ofi))?; trainer.set_training_range(0, train_end, train_end, n); - trainer.set_val_data_from_slices(&val_norm, val_targets, train_end); + trainer.set_val_data_from_slices(val_feat, val_targets, train_end); rt.block_on(trainer.reset_for_fold())?; let metrics = rt.block_on(trainer.train_fold_from_slices( @@ -472,7 +466,6 @@ fn test_gpu_collector_auto_initializes() -> anyhow::Result<()> { #[ignore] // Requires fxcache — run via: FOXHUNT_TEST_DATA=test_data/futures-baseline cargo test -p ml --lib -- test_fxcache_zero_copy_training --ignored --nocapture fn test_fxcache_zero_copy_training() -> anyhow::Result<()> { use crate::fxcache; - use crate::walk_forward::NormStats; // 1. Find fxcache directory let cache_dir = feature_cache_dir(); @@ -523,21 +516,15 @@ fn test_fxcache_zero_copy_training() -> anyhow::Result<()> { let train_targets = &fxcache_data.targets[range.train_start..range.train_end]; let val_targets = &fxcache_data.targets[range.val_start..range.val_end]; - let norm_stats = NormStats::from_features(train_feat); - let train_norm = norm_stats.normalize_batch(train_feat); - let val_norm = norm_stats.normalize_batch(val_feat); - + // fxcache is pre-normalized at precompute time — no per-fold normalization needed trainer.set_training_range( range.train_start, range.train_end, range.val_start, range.val_end, ); - trainer.set_val_data_from_slices(&val_norm, val_targets, range.train_end - range.train_start); + trainer.set_val_data_from_slices(val_feat, val_targets, range.train_end - range.train_start); rt.block_on(trainer.reset_for_fold())?; let fold = range.fold; - // Pass RAW features (not normalized) — the training loop uses them only for - // vol_normalizer and curriculum ADX filter, both need raw values. - // The GPU trains on features_raw_cuda (uploaded by init_from_fxcache). let result = rt.block_on(trainer.train_fold_from_slices( train_feat, train_targets, |_epoch, _bytes, _is_best| Ok(String::new()),