#![allow( clippy::assertions_on_constants, clippy::assertions_on_result_states, clippy::clone_on_copy, clippy::decimal_literal_representation, clippy::doc_markdown, clippy::empty_line_after_doc_comments, clippy::field_reassign_with_default, clippy::get_unwrap, clippy::identity_op, clippy::inconsistent_digit_grouping, clippy::indexing_slicing, clippy::integer_division, clippy::len_zero, clippy::let_underscore_must_use, clippy::manual_div_ceil, clippy::manual_let_else, clippy::manual_range_contains, clippy::modulo_arithmetic, clippy::needless_range_loop, clippy::non_ascii_literal, clippy::redundant_clone, clippy::shadow_reuse, clippy::shadow_same, clippy::shadow_unrelated, clippy::single_match_else, clippy::str_to_string, clippy::string_slice, clippy::tests_outside_test_module, clippy::too_many_lines, clippy::unnecessary_wraps, clippy::unseparated_literal_suffix, clippy::use_debug, clippy::useless_vec, clippy::wildcard_enum_match_arm, clippy::else_if_without_else, clippy::expect_used, clippy::missing_const_for_fn, clippy::similar_names, clippy::type_complexity, clippy::collapsible_else_if, clippy::doc_lazy_continuation, clippy::items_after_test_module, clippy::map_clone, clippy::multiple_unsafe_ops_per_block, clippy::unwrap_or_default, clippy::assign_op_pattern, clippy::needless_borrow, clippy::println_empty_string, clippy::unnecessary_cast, clippy::used_underscore_binding, clippy::create_dir, clippy::implicit_saturating_sub, clippy::exit, clippy::expect_fun_call, clippy::too_many_arguments, clippy::unnecessary_map_or, clippy::unwrap_used, dead_code, unused_imports, unused_variables, clippy::cloned_ref_to_slice_refs, clippy::neg_multiply, clippy::while_let_loop, clippy::bool_assert_comparison, clippy::excessive_precision, clippy::trivially_copy_pass_by_ref, clippy::op_ref, clippy::redundant_closure, clippy::unnecessary_lazy_evaluations, clippy::if_then_some_else_none, clippy::unnecessary_to_owned, clippy::single_component_path_imports, )] //! Precompute Features — DBN to .fxcache pipeline //! //! Reads OHLCV + MBP-10 + trades DBN data, runs the full feature extraction //! pipeline (standalone, no GPU required), and writes a `.fxcache` binary file for //! zero-overhead GPU loading during training. //! //! Usage: //! cargo run -p ml --example precompute_features --release -- \ //! --data-dir test_data/futures-baseline \ //! --mbp10-data-dir test_data/futures-baseline-mbp10 \ //! --trades-data-dir test_data/futures-baseline-trades \ //! --output-dir test_data/feature-cache \ //! --symbol ES.FUT \ //! --yes use std::path::{Path, PathBuf}; use std::time::Instant; use anyhow::{Context, Result}; use clap::Parser; use tracing::info; use ml::trainers::dqn::{ collect_dbn_files_filtered, collect_dbn_files_recursive, extract_features_from_bars, }; use ml::features::extraction::OHLCVBar; // --------------------------------------------------------------------------- // CLI // --------------------------------------------------------------------------- #[derive(Debug, Parser)] #[command( name = "precompute_features", about = "Precompute features from DBN data and write .fxcache binary" )] struct Opts { /// Directory containing OHLCV .dbn/.dbn.zst files (e.g. test_data/futures-baseline) #[arg(long)] data_dir: String, /// Directory containing MBP-10 .dbn/.dbn.zst files (optional) #[arg(long)] mbp10_data_dir: Option, /// Directory containing trades .dbn/.dbn.zst files (optional) #[arg(long)] trades_data_dir: Option, /// Output directory for .fxcache files (default: sibling of --data-dir) #[arg(long)] output_dir: Option, /// Symbol name (used in output filename) #[arg(long, default_value = "ES.FUT")] symbol: String, /// Skip confirmation prompt #[arg(long)] yes: bool, } // --------------------------------------------------------------------------- // Cache key computation // --------------------------------------------------------------------------- /// Compute a SHA256 cache key from all DBN files across the given directories. /// // --------------------------------------------------------------------------- // Main // --------------------------------------------------------------------------- #[tokio::main] async fn main() -> Result<()> { // Initialize tracing tracing_subscriber::fmt() .with_env_filter( tracing_subscriber::EnvFilter::try_from_default_env() .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("info")), ) .init(); let opts = Opts::parse(); let data_dir = PathBuf::from(&opts.data_dir); // Default output_dir: sibling of data_dir (e.g. test_data/futures-baseline → test_data/feature-cache) let output_dir = match opts.output_dir { Some(ref d) => PathBuf::from(d), None => data_dir.parent().unwrap_or(&data_dir).join("feature-cache"), }; let mbp10_dir = opts.mbp10_data_dir.as_ref().map(PathBuf::from); let trades_dir = opts.trades_data_dir.as_ref().map(PathBuf::from); // ── Validate inputs ────────────────────────────────────────────────────── if !data_dir.exists() { anyhow::bail!("Data directory not found: {}", data_dir.display()); } // ── Print config summary ───────────────────────────────────────────────── println!("================================================================================"); println!("Precompute Features — DBN to .fxcache Pipeline"); println!("================================================================================"); println!(); println!("Data dir: {}", data_dir.display()); if let Some(ref d) = mbp10_dir { println!("MBP-10 dir: {}", d.display()); } if let Some(ref d) = trades_dir { println!("Trades dir: {}", d.display()); } println!("Output dir: {}", output_dir.display()); println!("Symbol: {}", opts.symbol); println!("Format: f32 (v{}), OFI_DIM={}", ml::fxcache::FXCACHE_VERSION, ml::fxcache::OFI_DIM); println!(); // ── Confirmation ───────────────────────────────────────────────────────── if !opts.yes { print!("Proceed with feature extraction? (yes/no): "); std::io::Write::flush(&mut std::io::stdout())?; let mut input = String::new(); std::io::stdin().read_line(&mut input)?; let trimmed = input.trim(); if !trimmed.eq_ignore_ascii_case("yes") && !trimmed.eq_ignore_ascii_case("y") { println!("Cancelled."); return Ok(()); } println!(); } let t0 = Instant::now(); const WARMUP: usize = 50; // ── Early exit if cache already exists AND matches current version ──────── let hex_key_early = ml::feature_cache::calculate_dbn_cache_key_full( &data_dir, mbp10_dir.as_deref(), trades_dir.as_deref(), &opts.symbol, "ohlcv", ).context("Failed to compute cache key")?; let early_check_path = output_dir.join(format!("{hex_key_early}.fxcache")); if early_check_path.exists() { match ml::fxcache::load_fxcache(&early_check_path) { Ok(cached) if cached.has_ofi || mbp10_dir.is_none() => { // has_ofi flag is necessary but not sufficient — verify actual OFI data is non-zero let ofi_nonzero = cached.ofi.iter().filter(|r| r.iter().any(|&v| v != 0.0)).count(); if mbp10_dir.is_some() && ofi_nonzero == 0 { println!("Cache has_ofi=true but OFI data is all zeros ({} bars) — deleting stale cache", cached.ofi.len()); std::fs::remove_file(&early_check_path).ok(); } else { let size_mb = std::fs::metadata(&early_check_path) .map(|m| m.len() as f64 / 1_048_576.0) .unwrap_or(0.0); println!("Cache already exists: {} ({:.1} MB, has_ofi={}, ofi_nonzero={}) — skipping extraction", early_check_path.display(), size_mb, cached.has_ofi, ofi_nonzero); return Ok(()); } } Ok(_) => { println!("Cache exists but has_ofi=false while MBP-10 data available — deleting stale cache"); std::fs::remove_file(&early_check_path).ok(); } Err(e) => { println!("Cache exists but failed validation: {e} — deleting stale cache"); std::fs::remove_file(&early_check_path).ok(); } } } // ── Step 1: Build volume bars from trades data ──────────────────────────── let t1 = Instant::now(); // Use trades_data_dir if provided, otherwise try data_dir for trades let trades_source = trades_dir.as_ref().unwrap_or(&data_dir); info!("Loading trades from {} (symbol={})...", trades_source.display(), opts.symbol); let mut trade_files = collect_dbn_files_filtered(trades_source, Some(&opts.symbol)); if trade_files.is_empty() { // Fallback: try data_dir if trades_dir was specified but had no files if trades_dir.is_some() { trade_files = collect_dbn_files_filtered(&data_dir, Some(&opts.symbol)); } if trade_files.is_empty() { anyhow::bail!( "No trades DBN files found for symbol '{}' in {} or {}", opts.symbol, trades_source.display(), data_dir.display() ); } } trade_files.sort(); info!("Found {} trades/OHLCV DBN files for symbol '{}'", trade_files.len(), opts.symbol); // Load trades PER FILE and filter front-month within each file. // Each quarterly DBN file has a different front-month contract (e.g. ESH24, ESM24). // Filtering globally would select only ONE quarter's contract. let mut all_front_month_trades: Vec = Vec::new(); let mut total_raw = 0_usize; for file in &trade_files { match ml::features::load_trades_sync(file) { Ok(trades) => { let raw_count = trades.len(); total_raw += raw_count; let filtered = ml::features::filter_front_month(&trades); info!(" {} -> {} trades, {} front-month", file.file_name().unwrap_or_default().to_string_lossy(), raw_count, filtered.len()); all_front_month_trades.extend(filtered); } Err(e) => { tracing::warn!(" Failed {:?}: {e}", file.file_name()); } } } all_front_month_trades.sort_by_key(|t| t.timestamp); info!("Total trades: {} raw, {} front-month", total_raw, all_front_month_trades.len()); // Build volume bars let all_bars = ml::features::build_volume_bars(&all_front_month_trades, ml::features::DEFAULT_VOLUME_BAR_SIZE); info!("Volume bars built: {} bars ({} contracts/bar) in {:.1}s", all_bars.len(), ml::features::DEFAULT_VOLUME_BAR_SIZE, t1.elapsed().as_secs_f64()); if all_bars.len() < WARMUP + 2 { anyhow::bail!( "Insufficient data: {} volume bars, need at least {} (WARMUP={WARMUP} + 2)", all_bars.len(), WARMUP + 2 ); } // ── Price continuity check (regression gate) ──────────────────────── let mut max_jump_pct = 0.0_f64; let mut jump_count = 0_usize; for w in all_bars.windows(2) { let pct = ((w[1].close - w[0].close) / w[0].close).abs() * 100.0; if pct > 0.5 { jump_count += 1; } max_jump_pct = max_jump_pct.max(pct); } info!("Price continuity: max_jump={:.3}%, jumps>0.5%={}/{}", max_jump_pct, jump_count, all_bars.len() - 1); if jump_count as f64 / all_bars.len() as f64 > 0.01 { anyhow::bail!( "PRICE CONTINUITY FAILED: {} of {} bars have >0.5% price jump — likely interleaved contracts", jump_count, all_bars.len() ); } // ── Step 2: Extract 42-dim feature vectors ────────────────────────────── let t1 = Instant::now(); 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()); // ── Step 3: Build targets [preproc_close, preproc_next, raw_close, raw_next, raw_open, mid_price_open] 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; 6]> = 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; let raw_open = all_bars[i + WARMUP].open; // mid_price_open will be filled from MBP-10 data below if available; // default to raw_open (OHLCV-only fallback). targets.push([raw_curr, raw_next, raw_curr, raw_next, raw_open, raw_open]); } // ── Step 3b: Build per-bar timestamps ───────────────────────────────── let timestamps: Vec = (0..n) .map(|i| all_bars[i + WARMUP].timestamp.timestamp_nanos_opt().unwrap_or(0)) .collect(); // ── Step 4: Compute OFI from MBP-10 + trades ──────────────────────────── let t2 = Instant::now(); let ofi: Vec<[f64; 20]> = 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; 20]; 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 { None }; // Compute per-bar OFI use ml::features::trades_loader::get_trades_for_bar; use ml::features::ofi_calculator::MicrostructureState; 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; 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); let bar_duration_ns = bar_end_ts - bar_ts; let mut micro_state = MicrostructureState::new(bar_ts, bar_duration_ns); // Feed trades for this bar window (OFICalculator + MicrostructureState) if let Some(ref trades) = all_trades { for trade in get_trades_for_bar(trades, bar_ts, bar_end_ts) { calculator.feed_trade(trade.price, trade.volume, trade.is_buy); micro_state.update_trade(trade.price, trade.volume, trade.is_buy, trade.timestamp); } } // Get ALL MBP-10 snapshots within this bar window for tick-level features. // Binary search for bar range: [bar_ts, bar_end_ts) let snap_start = all_snapshots.partition_point(|s| s.timestamp < bar_ts); let snap_end = all_snapshots.partition_point(|s| s.timestamp < bar_end_ts); let bar_snapshots = &all_snapshots[snap_start..snap_end]; // Feed every snapshot to MicrostructureState for tick-level resolution for snap in bar_snapshots { micro_state.update_snapshot(snap); } // Use the LAST snapshot in the bar for OFI calculation (matches prior behavior) let last_snap = bar_snapshots.last() .or_else(|| { // Fallback: nearest snapshot at/after bar_ts (prior behavior) all_snapshots.get(snap_start) }); if let Some(snap) = last_snap { match calculator.calculate(snap) { Ok(f) if f.is_valid() => { // Feed OFI-derived values to MicrostructureState micro_state.update_ofi_derived(f.ofi_level1, f.vpin); let arr8 = f.to_array(); let micro_12 = micro_state.snapshot(); // v4 OFI layout: [0..8) raw OFI, [8..16) deltas (filled below), // [16] book_aggression (filled below), [17] log_bar_duration (filled below), // [18..20) ofi_acceleration + toxicity_gradient from microstructure let mut arr20 = [0.0_f64; 20]; arr20[..8].copy_from_slice(&arr8); // Slots [8..16) reserved for OFI deltas (computed after all bars) // Slot [16] reserved for book_aggression (computed below) // Slot [17] reserved for log_bar_duration (computed below) // Keep micro_12[10..12) (ofi_acceleration, toxicity_gradient) at [18..20) arr20[18] = micro_12[10]; // ofi_acceleration arr20[19] = micro_12[11]; // toxicity_gradient ofi_per_bar.push(arr20); } _ => ofi_per_bar.push([0.0; 20]), } } else { ofi_per_bar.push([0.0; 20]); } // ── Book aggression: center-of-mass asymmetry from MBP-10 depth ── // Uses the last snapshot in the bar window (same as OFI calculation above) if let Some(snap) = bar_snapshots.last() { let mut buy_weighted_sum = 0.0_f64; let mut buy_total = 0.0_f64; let mut sell_weighted_sum = 0.0_f64; let mut sell_total = 0.0_f64; for (level, pair) in snap.levels.iter().enumerate().take(10) { let bid_size = pair.bid_sz as f64; let ask_size = pair.ask_sz as f64; buy_weighted_sum += (level + 1) as f64 * bid_size; buy_total += bid_size; sell_weighted_sum += (level + 1) as f64 * ask_size; sell_total += ask_size; } let buy_com = if buy_total > 0.0 { buy_weighted_sum / buy_total } else { 5.5 }; let sell_com = if sell_total > 0.0 { sell_weighted_sum / sell_total } else { 5.5 }; let book_aggression = (sell_com - buy_com) / 10.0; // normalize to [-1, 1] ofi_per_bar.last_mut().unwrap()[16] = book_aggression; } } // ── OFI temporal deltas: delta[bar] = ofi[bar] - ofi[bar-1] ── // Computed after all bars so we have the full sequence for i in (1..ofi_per_bar.len()).rev() { for k in 0..8 { ofi_per_bar[i][8 + k] = ofi_per_bar[i][k] - ofi_per_bar[i - 1][k]; } } // First bar has no previous — deltas are zero (already initialized) // ── Log bar duration: ln(duration_secs) / 10.0 ── for i in 0..n { let duration_secs = if i > 0 { let ts_cur = all_bars[i + WARMUP].timestamp; let ts_prev = all_bars[i + WARMUP - 1].timestamp; (ts_cur - ts_prev).num_seconds() as f64 } else { 60.0 // default 1 minute for first bar }; let log_duration = duration_secs.max(0.1).ln() / 10.0; ofi_per_bar[i][17] = log_duration; } let delta_nonzero = ofi_per_bar.iter().filter(|f| f[8..16].iter().any(|&v| v != 0.0)).count(); let book_nonzero = ofi_per_bar.iter().filter(|f| f[16] != 0.0).count(); info!("OFI v4: deltas_nonzero={}/{}, book_aggression_nonzero={}/{}", delta_nonzero, n, book_nonzero, n); // Fill targets[i][5] with MBP-10 midpoint at bar open (mid_price_open) let mut mid_fill_count = 0_usize; for i in 0..n { let bar = &all_bars[i + WARMUP]; let bar_ts = bar.timestamp.timestamp_nanos_opt().unwrap_or(0) as u64; // Find first MBP-10 snapshot at or after bar open let snap_idx = all_snapshots.partition_point(|s| s.timestamp < bar_ts); if let Some(snap) = all_snapshots.get(snap_idx) { let mid = snap.mid_price(); if mid > 0.0 { targets[i][5] = mid; mid_fill_count += 1; } } } info!("mid_price_open filled from MBP-10: {}/{} bars", mid_fill_count, n); 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 } } else { info!("No MBP-10 directory, OFI will be zeros (log_bar_duration still computed)"); let mut ofi_per_bar = vec![[0.0_f64; 20]; n]; // Even without MBP-10 data, compute log_bar_duration from bar timestamps for i in 0..n { let duration_secs = if i > 0 { let ts_cur = all_bars[i + WARMUP].timestamp; let ts_prev = all_bars[i + WARMUP - 1].timestamp; (ts_cur - ts_prev).num_seconds() as f64 } else { 60.0 // default 1 minute for first bar }; let log_duration = duration_secs.max(0.1).ln() / 10.0; ofi_per_bar[i][17] = log_duration; } ofi_per_bar }; let total_len = n; // ── Compute cache key ──────────────────────────────────────────────────── let hex_key = ml::feature_cache::calculate_dbn_cache_key_full( &data_dir, mbp10_dir.as_deref(), trades_dir.as_deref(), &opts.symbol, "ohlcv", ).context("Failed to compute cache key")?; let cache_key: [u8; 32] = hex::decode(&hex_key) .context("Invalid hex key")? .try_into() .map_err(|_| anyhow::anyhow!("Cache key wrong length"))?; // ── Normalize features (z-score) ────────────────────────────────────────── // Raw features contain OHLCV prices (up to 25,000). 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()); let has_ofi = mbp10_dir.is_some(); let bytes_written = ml::fxcache::write_fxcache( &output_path, &features, &targets, &ofi, ×tamps, cache_key, has_ofi, ) .context("Failed to write .fxcache file")?; let write_secs = t2.elapsed().as_secs_f64(); let total_secs = t0.elapsed().as_secs_f64(); // ── Summary ────────────────────────────────────────────────────────────── println!(); println!("================================================================================"); println!("PRECOMPUTE SUMMARY"); println!("================================================================================"); println!(); println!("Bars: {}", total_len); println!("Features: 42-dim"); println!("Targets: 6-dim"); println!("OFI: 20-dim ({})", if has_ofi { "from MBP-10" } else { "zero-padded" }); println!("Format: f32 (v{}), OFI_DIM={}", ml::fxcache::FXCACHE_VERSION, ml::fxcache::OFI_DIM); println!("Cache key: {}", hex_key); println!("Output: {}", output_path.display()); println!("Size: {:.2} MB", bytes_written as f64 / 1_048_576.0); println!("Write time: {:.1}s", write_secs); println!("Total time: {:.1}s", total_secs); println!(); Ok(()) }