gitadmin/foxhunt
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

refactor: precompute_features runs pure CPU — no DQNTrainer needed

Browse Source 
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) <noreply@anthropic.com>
This commit is contained in:
jgrusewski
2026-04-01 00:52:36 +02:00
parent b14e9cbd6b
commit 08d6b1a32d
4 changed files with 198 additions and 107 deletions
Download Patch File Download Diff File Expand all files Collapse all files

202
crates/ml/examples/precompute_features.rs
View File

@@ -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<OHLCVBar> = 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<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<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());

26
crates/ml/src/trainers/dqn/data_loading.rs
View File

@@ -31,7 +31,7 @@ fn is_zstd_file(path: &Path) -> Result<bool> {
}
/// Recursively collect all .dbn and .dbn.zst files from a directory and its subdirectories.
fn collect_dbn_files_recursive(dir: &Path) -> Vec<std::path::PathBuf> {
pub fn collect_dbn_files_recursive(dir: &Path) -> Vec<std::path::PathBuf> {
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<std::path::PathBuf> {
files
}
/// Extract OHLCV bars from a single DBN file — no DQNTrainer required.
pub fn extract_ohlcv_bars_from_dbn(file_path: &Path) -> Result<Vec<OHLCVBar>> {
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<Vec<OHLCVBar>> {
// 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<R: std::io::Read>(

75
crates/ml/src/trainers/dqn/features.rs
View File

@@ -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<Vec<FeatureVector>> {
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<FinancialFeatures> {
@@ -66,37 +102,14 @@ impl DQNTrainer {
bars: &[OHLCVBar],
_mbp10_snapshots: Option<&[data::providers::databento::mbp10::Mbp10Snapshot]>,
) -> Result<Vec<FeatureVector>> {
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

2
crates/ml/src/trainers/dqn/mod.rs
View File

@@ -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