From 08d6b1a32dabfc782917fae4c2b6591cf08b2877 Mon Sep 17 00:00:00 2001 From: jgrusewski Date: Wed, 1 Apr 2026 00:52:36 +0200 Subject: [PATCH] =?UTF-8?q?refactor:=20precompute=5Ffeatures=20runs=20pure?= =?UTF-8?q?=20CPU=20=E2=80=94=20no=20DQNTrainer=20needed?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Extract standalone functions from DQNTrainer: - extract_features_from_bars(): 42-dim feature extraction - extract_ohlcv_bars_from_dbn(): DBN to OHLCVBar conversion - collect_dbn_files_recursive(): file discovery DQNTrainer delegates to these, ensuring consistency. Precompute binary no longer initializes CUDA — runs on any CPU node (ci-compile-cpu at EUR 0.85/hr vs GPU at EUR 12.60/hr). Co-Authored-By: Claude Opus 4.6 (1M context) --- crates/ml/examples/precompute_features.rs | 202 ++++++++++++++------- crates/ml/src/trainers/dqn/data_loading.rs | 26 +-- crates/ml/src/trainers/dqn/features.rs | 75 ++++---- crates/ml/src/trainers/dqn/mod.rs | 2 + 4 files changed, 198 insertions(+), 107 deletions(-) diff --git a/crates/ml/examples/precompute_features.rs b/crates/ml/examples/precompute_features.rs index f8959b449..0e742f80f 100644 --- a/crates/ml/examples/precompute_features.rs +++ b/crates/ml/examples/precompute_features.rs @@ -75,7 +75,7 @@ //! Precompute Features — DBN to .fxcache pipeline //! //! Reads OHLCV + MBP-10 + trades DBN data, runs the full feature extraction -//! pipeline via DQNTrainer, and writes a `.fxcache` binary file for +//! pipeline (standalone, no GPU required), and writes a `.fxcache` binary file for //! zero-overhead GPU loading during training. //! //! Usage: @@ -96,7 +96,10 @@ use clap::Parser; use sha2::{Digest, Sha256}; use tracing::info; -use ml::trainers::dqn::{DQNHyperparameters, DQNTrainer}; +use ml::trainers::dqn::{ + collect_dbn_files_recursive, extract_features_from_bars, extract_ohlcv_bars_from_dbn, +}; +use ml::features::extraction::OHLCVBar; // --------------------------------------------------------------------------- // CLI @@ -256,75 +259,153 @@ async fn main() -> Result<()> { println!(); } - // ── Construct DQNTrainer ───────────────────────────────────────────────── let t0 = Instant::now(); - info!("Constructing DQNTrainer with default hyperparams..."); - let mut hyperparams = DQNHyperparameters::default(); - hyperparams.buffer_size = 100_000; - if let Some(ref d) = opts.mbp10_data_dir { - hyperparams.mbp10_data_dir = Some(d.clone()); - } - if let Some(ref d) = opts.trades_data_dir { - hyperparams.trades_data_dir = Some(d.clone()); + // ── Step 1: Load OHLCV bars from DBN files ────────────────────────────── + info!("Loading OHLCV bars from {}...", data_dir.display()); + let mut dbn_files = collect_dbn_files_recursive(&data_dir); + dbn_files.sort(); + info!("Found {} DBN files", dbn_files.len()); + + let mut all_bars: Vec = Vec::new(); + for file in &dbn_files { + match extract_ohlcv_bars_from_dbn(file) { + Ok(bars) => all_bars.extend(bars), + Err(e) => tracing::warn!("Failed to load {:?}: {e}", file.file_name()), + } } + all_bars.sort_by_key(|b| b.timestamp); + info!("Loaded {} OHLCV bars in {:.1}s", all_bars.len(), t0.elapsed().as_secs_f64()); - let mut trainer = DQNTrainer::new(hyperparams) - .context("Failed to construct DQNTrainer")?; - - info!("DQNTrainer constructed in {:.1}s", t0.elapsed().as_secs_f64()); - - // ── Load training data ─────────────────────────────────────────────────── + // ── Step 2: Extract 42-dim feature vectors ────────────────────────────── let t1 = Instant::now(); - info!("Loading training data from {}...", data_dir.display()); + info!("Extracting features..."); + let feature_vectors = extract_features_from_bars(&all_bars) + .context("Feature extraction failed")?; + info!("Extracted {} feature vectors in {:.1}s", feature_vectors.len(), t1.elapsed().as_secs_f64()); - let data_dir_str = opts.data_dir.clone(); - let (train_data, val_data) = trainer - .load_training_data(&data_dir_str) - .await - .context("Failed to load training data")?; - - let train_len = train_data.len(); - let val_len = val_data.len(); - let total_len = train_len + val_len; - - info!( - "Loaded {} bars ({} train + {} val) in {:.1}s", - total_len, - train_len, - val_len, - t1.elapsed().as_secs_f64() - ); - - // ── Merge train + val into ordered sequence ────────────────────────────── - let mut features: Vec<[f64; 42]> = Vec::with_capacity(total_len); - let mut targets: Vec<[f64; 4]> = Vec::with_capacity(total_len); - - for (feat, tgt) in train_data.iter().chain(val_data.iter()) { - features.push(*feat); - let mut t = [0.0_f64; 4]; - for (i, v) in tgt.iter().enumerate().take(4) { - t[i] = *v; - } - targets.push(t); + // ── Step 3: Build targets [preproc_close, preproc_next, raw_close, raw_next] + const WARMUP: usize = 50; + let n = feature_vectors.len().saturating_sub(1); // need next bar for target + let features: Vec<[f64; 42]> = feature_vectors[..n].to_vec(); + let mut targets: Vec<[f64; 4]> = Vec::with_capacity(n); + for i in 0..n { + let raw_curr = all_bars[i + WARMUP].close; + let raw_next = all_bars[i + WARMUP + 1].close; + targets.push([raw_curr, raw_next, raw_curr, raw_next]); } - // ── Extract OFI features ───────────────────────────────────────────────── - let ofi: Vec<[f64; 8]> = match trainer.ofi_features { - Some(ref ofi_arc) => { - let ofi_slice: &[[f64; 8]] = &**ofi_arc; - if ofi_slice.len() >= total_len { - ofi_slice[..total_len].to_vec() + // ── Step 4: Compute OFI from MBP-10 + trades ──────────────────────────── + let t2 = Instant::now(); + let ofi: Vec<[f64; 8]> = if let Some(ref mbp10_path) = mbp10_dir { + use ml::features::mbp10_loader::load_ofi_features_parallel; + use ml::features::trades_loader::load_trades_sync; + use ml::features::ofi_calculator::OFICalculator; + + info!("Loading MBP-10 snapshots from {}...", mbp10_path.display()); + let mut mbp10_files = collect_dbn_files_recursive(mbp10_path); + mbp10_files.sort(); + + if mbp10_files.is_empty() { + info!("No MBP-10 files found, OFI will be zeros"); + vec![[0.0; 8]; n] + } else { + // Load MBP-10 snapshots (parallel) + use data::providers::databento::dbn_parser::DbnParser; + let per_file: Vec> = { + use rayon::prelude::*; + mbp10_files.par_iter().filter_map(|file| { + let parser = match DbnParser::new() { + Ok(p) => p, + Err(e) => { tracing::warn!("Parser init failed: {e}"); return None; } + }; + let mut snapshots = Vec::new(); + match parser.parse_mbp10_streaming(file, 100, |snap| { + snapshots.push(snap.clone()); + }) { + Ok(_) => { + info!(" {} -> {} snapshots", file.file_name().unwrap_or_default().to_string_lossy(), snapshots.len()); + Some(snapshots) + } + Err(e) => { + tracing::warn!(" Failed {:?}: {e}", file.file_name()); + None + } + } + }).collect() + }; + + let mut all_snapshots = Vec::new(); + for snaps in per_file { all_snapshots.extend(snaps); } + all_snapshots.sort_by_key(|s| s.timestamp); + info!("Loaded {} MBP-10 snapshots in {:.1}s", all_snapshots.len(), t2.elapsed().as_secs_f64()); + + // Load trades (parallel) + let all_trades = if let Some(ref tdir) = trades_dir { + let mut trade_files = collect_dbn_files_recursive(tdir); + trade_files.sort(); + let per_file_trades: Vec> = { + use rayon::prelude::*; + trade_files.par_iter().filter_map(|path| { + match load_trades_sync(path) { + Ok(t) => { info!(" {} trades from {:?}", t.len(), path.file_name()); Some(t) } + Err(e) => { tracing::warn!(" Failed {:?}: {e}", path.file_name()); None } + } + }).collect() + }; + let mut trades = Vec::new(); + for t in per_file_trades { trades.extend(t); } + trades.sort_by_key(|t| t.timestamp); + if trades.is_empty() { None } else { Some(trades) } } else { - // Pad with zeros if OFI is shorter than feature count - let mut padded = ofi_slice.to_vec(); - padded.resize(total_len, [0.0; 8]); - padded + None + }; + + // Compute per-bar OFI + use ml::features::mbp10_loader::get_snapshots_for_timestamp; + use ml::features::trades_loader::get_trades_for_bar; + let mut calculator = OFICalculator::new(); + let mut ofi_per_bar = Vec::with_capacity(n); + + for i in 0..n { + let bar = &all_bars[i + WARMUP]; + let bar_ts = bar.timestamp.timestamp_nanos_opt().unwrap_or(0) as u64; + + // Feed trades for this bar window + if let Some(ref trades) = all_trades { + let bar_end_ts = all_bars.get(i + WARMUP + 1) + .map(|b| b.timestamp.timestamp_nanos_opt().unwrap_or(0) as u64) + .unwrap_or(bar_ts + 60_000_000_000); + for trade in get_trades_for_bar(trades, bar_ts, bar_end_ts) { + calculator.feed_trade(trade.price, trade.volume, trade.is_buy); + } + } + + let window = get_snapshots_for_timestamp(&all_snapshots, bar_ts, 1); + if let Some(snap) = window.first() { + match calculator.calculate(snap) { + Ok(f) if f.is_valid() => ofi_per_bar.push(f.to_array()), + _ => ofi_per_bar.push([0.0; 8]), + } + } else { + ofi_per_bar.push([0.0; 8]); + } } + + let non_zero = ofi_per_bar.iter().filter(|f| f.iter().any(|&v| v != 0.0)).count(); + info!("OFI computed: {} bars, {} non-zero ({:.1}%) in {:.1}s", + ofi_per_bar.len(), non_zero, + if ofi_per_bar.is_empty() { 0.0 } else { non_zero as f64 / ofi_per_bar.len() as f64 * 100.0 }, + t2.elapsed().as_secs_f64()); + ofi_per_bar } - None => vec![[0.0; 8]; total_len], + } else { + info!("No MBP-10 directory, OFI will be zeros"); + vec![[0.0; 8]; n] }; + let total_len = n; + // ── Compute cache key ──────────────────────────────────────────────────── let mut key_dirs: Vec<&Path> = vec![data_dir.as_path()]; if let Some(ref d) = mbp10_dir { @@ -364,12 +445,11 @@ async fn main() -> Result<()> { println!("PRECOMPUTE SUMMARY"); println!("================================================================================"); println!(); + let has_ofi = ofi.iter().any(|o| o.iter().any(|&v| v != 0.0)); println!("Bars: {}", total_len); - println!(" Train: {}", train_len); - println!(" Validation: {}", val_len); println!("Features: 42-dim"); println!("Targets: 4-dim"); - println!("OFI: 8-dim ({})", if trainer.ofi_features.is_some() { "from MBP-10" } else { "zero-padded" }); + println!("OFI: 8-dim ({})", if has_ofi { "from MBP-10" } else { "zero-padded" }); println!("Format: {} (v{})", if opts.bf16 { "bf16" } else { "f64" }, if opts.bf16 { 2 } else { 1 }); println!("Cache key: {}", hex_key); println!("Output: {}", output_path.display()); diff --git a/crates/ml/src/trainers/dqn/data_loading.rs b/crates/ml/src/trainers/dqn/data_loading.rs index cb71b4672..14571b7f8 100644 --- a/crates/ml/src/trainers/dqn/data_loading.rs +++ b/crates/ml/src/trainers/dqn/data_loading.rs @@ -31,7 +31,7 @@ fn is_zstd_file(path: &Path) -> Result { } /// Recursively collect all .dbn and .dbn.zst files from a directory and its subdirectories. -fn collect_dbn_files_recursive(dir: &Path) -> Vec { +pub fn collect_dbn_files_recursive(dir: &Path) -> Vec { let mut files = Vec::new(); if let Ok(entries) = std::fs::read_dir(dir) { for entry in entries.flatten() { @@ -50,6 +50,19 @@ fn collect_dbn_files_recursive(dir: &Path) -> Vec { files } +/// Extract OHLCV bars from a single DBN file — no DQNTrainer required. +pub fn extract_ohlcv_bars_from_dbn(file_path: &Path) -> Result> { + if is_zstd_file(file_path)? { + let mut decoder = dbn::decode::DbnDecoder::from_zstd_file(file_path) + .map_err(|e| anyhow::anyhow!("Failed to create zstd DBN decoder for {}: {}", file_path.display(), e))?; + DQNTrainer::decode_ohlcv_records(&mut decoder, file_path) + } else { + let mut decoder = dbn::decode::DbnDecoder::from_file(file_path) + .map_err(|e| anyhow::anyhow!("Failed to create DBN decoder for {}: {}", file_path.display(), e))?; + DQNTrainer::decode_ohlcv_records(&mut decoder, file_path) + } +} + impl DQNTrainer { /// Train DQN on market data from Parquet file (Wave 12 Group 3) /// @@ -979,16 +992,7 @@ impl DQNTrainer { /// /// Public for testing purposes. pub fn extract_ohlcv_bars_from_dbn(&self, file_path: &Path) -> Result> { - // Detect zstd by magic bytes (0x28B52FFD) — file extension is unreliable - if is_zstd_file(file_path)? { - let mut decoder = dbn::decode::DbnDecoder::from_zstd_file(file_path) - .map_err(|e| anyhow::anyhow!("Failed to create zstd DBN decoder for {}: {}", file_path.display(), e))?; - Self::decode_ohlcv_records(&mut decoder, file_path) - } else { - let mut decoder = dbn::decode::DbnDecoder::from_file(file_path) - .map_err(|e| anyhow::anyhow!("Failed to create DBN decoder for {}: {}", file_path.display(), e))?; - Self::decode_ohlcv_records(&mut decoder, file_path) - } + extract_ohlcv_bars_from_dbn(file_path) } fn decode_ohlcv_records( diff --git a/crates/ml/src/trainers/dqn/features.rs b/crates/ml/src/trainers/dqn/features.rs index 04f32b9fc..5f7186f28 100644 --- a/crates/ml/src/trainers/dqn/features.rs +++ b/crates/ml/src/trainers/dqn/features.rs @@ -14,6 +14,42 @@ use super::statistics::FeatureStatistics; use crate::features::extraction::FeatureVector; use super::trainer::DQNTrainer; +/// Standalone feature extraction from OHLCV bars — no DQNTrainer required. +/// +/// Extracts 42-dim feature vectors using the same `FeatureExtractor` pipeline +/// as `DQNTrainer::extract_full_features()`, ensuring consistency between +/// precompute and training paths. +pub fn extract_features_from_bars(bars: &[OHLCVBar]) -> 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); + + for (i, bar) in bars.iter().enumerate() { + extractor.update(bar)?; + + if i >= WARMUP_PERIOD { + let features = extractor.extract_current_features()?; + feature_vectors.push(features); + } + } + + Ok(feature_vectors) +} + impl DQNTrainer { /// Create synthetic features (placeholder for testing) pub(crate) fn create_synthetic_features(&self, price: f64) -> Result { @@ -66,37 +102,14 @@ impl DQNTrainer { 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)?; - - // Update microstructure calculators + // Update microstructure calculators (to be wired into feature vector) + for bar in bars { let timestamp_ns = bar.timestamp.timestamp_nanos_opt().unwrap_or(0) as u64; - 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); - let return_pct = if self.last_close > 0.0 { (bar.close - self.last_close) / self.last_close } else { @@ -104,21 +117,13 @@ impl DQNTrainer { }; 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); - self.last_close = bar.close; - - // Start extracting features after warmup - if i >= WARMUP_PERIOD { - // Extract 42 market features (40 base + 2 regime: ADX, CUSUM) - let features_40 = extractor.extract_current_features()?; - feature_vectors.push(features_40); - } } - Ok(feature_vectors) + // Core 42-dim feature extraction (shared with precompute binary) + extract_features_from_bars(bars) } /// Calculate feature statistics from training samples using Welford's algorithm diff --git a/crates/ml/src/trainers/dqn/mod.rs b/crates/ml/src/trainers/dqn/mod.rs index a22eeba5f..ce1b50265 100644 --- a/crates/ml/src/trainers/dqn/mod.rs +++ b/crates/ml/src/trainers/dqn/mod.rs @@ -38,6 +38,8 @@ pub use config::{DQNAgentType, DQNHyperparameters}; pub use early_stopping::EarlyStopping; pub use lr_scheduler::{LRDecayType, LRScheduler}; pub use statistics::{FeatureStatistics, QValueStats}; +pub use data_loading::{collect_dbn_files_recursive, extract_ohlcv_bars_from_dbn}; +pub use features::extract_features_from_bars; pub use trainer::DQNTrainer; // Re-export constants