8434737a69
Stops chasing one-off corruption bugs. Three+ historical fixes patched specific writers (#191 fxcache column-0, Bug 1 target schema, label_scale=5443 leaks, today's state[0] heavy-tail). Each new corruption shape found the next hole. This installs a structural defense so corruption is REJECTED at the data-loading boundary regardless of source. Five independent layers, each mandatory: 1. Bar-level sanity (baseline_common::sanitize_bars): drop bars with non-positive OHLC, high<low, non-finite values, or close/prev_close outside [0.5, 2.0]. Catches DBN parse glitches, broker tick errors, near-zero-open bars at source. 2. safe_log_return result clamp (extraction.rs:1246): ratio.ln().clamp(±0.1). Real-market 1-bar log returns rarely exceed ±0.05; ±0.1 traps every legitimate move while rejecting the corruption shape (corrupt bar with bar.open ≈ 0 → ln = -30 → previously normalized to -30000-magnitude state[0] outliers). 3. validate_features pre-norm bound (extraction.rs:538): |val| ≤ 5.0 post-extraction. Pre-norm features come from safe_normalize ([0,1]/[-1,1]), safe_clip (max ±3), or clamped log-returns (±0.1); ±5 catches extractor invariant breaks. 4. NormStats::normalize post-norm clamp (walk_forward.rs:688): ((val - mean) / std).clamp(±20.0). Even if upstream produces outliers, every value uploaded to GPU is bounded. 5. Shared validate_normalized_features gate (walk_forward.rs): single source-of-truth invariant enforced at THREE sites: - fxcache fast path (after discover_and_load) - DBN fallback (after normalize_batch) - precompute writer (before fxcache write — never persist a poisoned cache) Removed: DIAG_BUG2 + DIAG_BUG2_v2 one-shot diagnostics (~125 lines of host-side download + outlier scan in training_loop.rs). Replaced by structural defense — instrumentation isn't needed when corruption can't reach state[0]. FEATURE_SCHEMA_HASH auto-bumps via build.rs FNV-1a hash over SCHEMA_FILES (extraction.rs included). All pre-fix .fxcache files on PVC are invalidated at load time; ensure-fxcache regen produces clean cache with new clamps applied. Why this finally closes the chapter: per-writer fixes are reactive (land after corruption hits prod). Boundary validation is proactive — every future regression to extraction or normalization trips the gate at load, not at epoch 5 of a 50-epoch run. The 5 layers are independent: a bug in any one leaves the others as backstop. Validation: cargo check -p ml --all-targets --offline clean. NormStats unit tests (walk_forward.rs:706+) still pass — clamp + validate are additive; existing test inputs are well within bounds. Refs: SP5 Bug 2 (state[0] std=570 outliers on smoke-test-xb78r), historical #191 #210 #214 #193 #195 chains. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
746 lines
35 KiB
Rust
746 lines
35 KiB
Rust
#![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<String>,
|
||
|
||
/// Directory containing trades .dbn/.dbn.zst files (optional)
|
||
#[arg(long)]
|
||
trades_data_dir: Option<String>,
|
||
|
||
/// Output directory for .fxcache files (default: sibling of --data-dir)
|
||
#[arg(long)]
|
||
output_dir: Option<String>,
|
||
|
||
/// Symbol name (used in output filename)
|
||
#[arg(long, default_value = "ES.FUT")]
|
||
symbol: String,
|
||
|
||
/// Data-source identifier mixed into the SHA256 cache key.
|
||
///
|
||
/// MUST match the `data_source` field of the training profile that will
|
||
/// consume this fxcache (e.g. `dqn-production.toml: data_source = "mbp10"`,
|
||
/// `dqn-smoketest.toml: data_source = "mbp10"`). Mismatch produces a
|
||
/// silent cache MISS at training time → DBN-direct fallback uploads
|
||
/// un-normalised features → aux-head label_scale picks up raw-price
|
||
/// magnitudes → cascade of broken metrics + impossible Sharpe (see
|
||
/// 2026-04-27 incident: train-h5gxb epoch-0 Sharpe=141 from this exact
|
||
/// path mismatch).
|
||
///
|
||
/// Defaults to `"mbp10"` — production, smoke, and localdev profiles all
|
||
/// use MBP-10 microstructure data. Pass `--data-source ohlcv` only when
|
||
/// intentionally generating a 1-minute candle cache for a profile that
|
||
/// overrides `data_source = "ohlcv"`.
|
||
#[arg(long, default_value = "mbp10")]
|
||
data_source: 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,
|
||
&opts.data_source,
|
||
).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<ml::features::DbnTrade> = 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]
|
||
// [0:1] = log-return-normalized close/next (network input; matches contract
|
||
// in experience_kernels.cu:1556 + cuda_pipeline/mod.rs:508)
|
||
// [2:3] = raw dollar prices for portfolio simulation P&L + tx costs
|
||
// [4] = raw open price
|
||
// [5] = mid-price at bar open (MBP-10 midpoint; fallback to raw_open below)
|
||
//
|
||
// Prior implementation wrote raw_curr/raw_next to slots [0:1] in violation of
|
||
// the documented contract — the network-input columns ended up containing raw
|
||
// prices (~$5000+ for ES futures). Fixed here so the fxcache matches its own
|
||
// documented schema.
|
||
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;
|
||
// prev_close: 1-bar lag for log-return computation. WARMUP ≥ 50 so
|
||
// all_bars[i + WARMUP - 1] is always valid.
|
||
let prev_close = all_bars[i + WARMUP - 1].close;
|
||
let preproc_close = if prev_close > 0.0 {
|
||
(raw_curr / prev_close).ln()
|
||
} else { 0.0 };
|
||
let preproc_next = if raw_curr > 0.0 {
|
||
(raw_next / raw_curr).ln()
|
||
} else { 0.0 };
|
||
// mid_price_open will be filled from MBP-10 data below if available;
|
||
// default to raw_open (OHLCV-only fallback).
|
||
targets.push([preproc_close, preproc_next, raw_curr, raw_next, raw_open, raw_open]);
|
||
}
|
||
|
||
// ── Step 3b: Build per-bar timestamps ─────────────────────────────────
|
||
let timestamps: Vec<i64> = (0..n)
|
||
.map(|i| all_bars[i + WARMUP].timestamp.timestamp_nanos_opt().unwrap_or(0))
|
||
.collect();
|
||
|
||
// ── Step 4: Compute OFI from MBP-10 + trades ────────────────────────────
|
||
const OFI_DIM: usize = ml_core::state_layout::OFI_DIM;
|
||
let t2 = Instant::now();
|
||
let ofi: Vec<[f64; OFI_DIM]> = 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; OFI_DIM]; 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 {
|
||
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];
|
||
// Bar timestamp = close-time of bar (volume bars: last trade in bucket;
|
||
// see `crates/ml-features/src/trades_loader.rs:124-176`). The OFI window
|
||
// for bar t is therefore the *formation interval*
|
||
// `(close(bar_{t-1}), close(bar_t)]` — strictly historical relative to
|
||
// the policy's decision time at bar t. The previous implementation used
|
||
// `[close(bar_t), close(bar_{t+1}))` which leaks bar t+1's microstructure
|
||
// (audit `docs/lookahead-bias-audit-2026-04-28.md` §3, 31/32 OFI dims).
|
||
// Mirrors the backward-looking convention already used by
|
||
// `log_bar_duration` below at lines 567-575.
|
||
let bar_ts = bar.timestamp.timestamp_nanos_opt().unwrap_or(0) as u64;
|
||
let bar_start_ts = if i + WARMUP >= 1 {
|
||
all_bars[i + WARMUP - 1].timestamp.timestamp_nanos_opt().unwrap_or(0) as u64
|
||
} else {
|
||
// No prior bar — emit empty window. log_bar_duration uses the same
|
||
// sentinel pathway (60s default) for bar 0; here we simply skip
|
||
// population so OFI[0] stays zero (consistent with existing
|
||
// first-bar delta handling at lines 558-562).
|
||
bar_ts
|
||
};
|
||
let bar_duration_ns = bar_ts.saturating_sub(bar_start_ts);
|
||
|
||
let mut micro_state = MicrostructureState::new(bar_start_ts, bar_duration_ns);
|
||
|
||
// Feed trades for this bar window (OFICalculator + MicrostructureState).
|
||
// Window is half-open at the start, closed at the end:
|
||
// `(close(bar_{t-1}), close(bar_t)]` so trade-at-close (which formed bar t)
|
||
// is included.
|
||
if let Some(ref trades) = all_trades {
|
||
for trade in get_trades_for_bar(trades, bar_start_ts, bar_ts.saturating_add(1)) {
|
||
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_start_ts, bar_ts] — formation interval.
|
||
let snap_start = all_snapshots.partition_point(|s| s.timestamp <= bar_start_ts);
|
||
let snap_end = all_snapshots.partition_point(|s| s.timestamp <= bar_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();
|
||
// v5 OFI layout (see state_layout.rs OFI slot map):
|
||
// [0..8) raw OFI (8 features)
|
||
// [8..16) lag-1 deltas (filled in post-loop)
|
||
// [16] book_aggression (filled below)
|
||
// [17] log_bar_duration (filled in post-loop)
|
||
// [18] ofi_acceleration (micro_12[10])
|
||
// [19] toxicity_gradient (micro_12[11])
|
||
// [20..30) MicrostructureState::snapshot()[0..10]
|
||
// [30] order_count_imbalance ((Σbid_ct − Σask_ct) / Σ(bid_ct+ask_ct))
|
||
// [31] microprice_residual ((weighted_mid − mid) / mid)
|
||
let mut ofi_row = [0.0_f64; OFI_DIM];
|
||
ofi_row[..8].copy_from_slice(&arr8);
|
||
ofi_row[18] = micro_12[10]; // ofi_acceleration
|
||
ofi_row[19] = micro_12[11]; // toxicity_gradient
|
||
for k in 0..10 {
|
||
ofi_row[20 + k] = micro_12[k];
|
||
}
|
||
// TLOB-novel slots: derive directly from Mbp10Snapshot counts/prices.
|
||
// order_count_imbalance: count-based analog to depth_imbalance.
|
||
let (bid_ct_sum, ask_ct_sum) = snap.levels.iter().fold((0u64, 0u64), |(b, a), l| {
|
||
(b + l.bid_ct as u64, a + l.ask_ct as u64)
|
||
});
|
||
let total_ct = bid_ct_sum + ask_ct_sum;
|
||
let order_count_imbalance = if total_ct > 0 {
|
||
(bid_ct_sum as f64 - ask_ct_sum as f64) / total_ct as f64
|
||
} else {
|
||
0.0
|
||
};
|
||
ofi_row[30] = order_count_imbalance.clamp(-1.0, 1.0);
|
||
// microprice_residual: (weighted_mid − mid) / mid; dimensionless.
|
||
// Mbp10Snapshot::weighted_mid_price() uses top-of-book volume-weighted mid.
|
||
let mid = snap.mid_price();
|
||
let wmid = snap.weighted_mid_price();
|
||
let microprice_residual = if mid > 0.0 && mid.is_finite() && wmid.is_finite() {
|
||
((wmid - mid) / mid).clamp(-0.01, 0.01)
|
||
} else {
|
||
0.0
|
||
};
|
||
ofi_row[31] = microprice_residual;
|
||
ofi_per_bar.push(ofi_row);
|
||
}
|
||
_ => ofi_per_bar.push([0.0; OFI_DIM]),
|
||
}
|
||
} else {
|
||
ofi_per_bar.push([0.0; OFI_DIM]);
|
||
}
|
||
|
||
// ── 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]
|
||
if let Some(last) = ofi_per_bar.last_mut() {
|
||
last[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();
|
||
let micro_nonzero = ofi_per_bar.iter().filter(|f| f[20..30].iter().any(|&v| v != 0.0)).count();
|
||
let tlob_nonzero = ofi_per_bar.iter().filter(|f| f[30] != 0.0 || f[31] != 0.0).count();
|
||
info!("OFI v5: deltas_nonzero={}/{}, book_aggression_nonzero={}/{}, microstructure_nonzero={}/{}, tlob_novel_nonzero={}/{}",
|
||
delta_nonzero, n, book_nonzero, n, micro_nonzero, n, tlob_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; OFI_DIM]; 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,
|
||
&opts.data_source,
|
||
).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());
|
||
|
||
// Defence-in-depth gate: never write a poisoned fxcache to disk. If anything
|
||
// upstream produced an out-of-bounds value, fail loudly here rather than
|
||
// letting it silently propagate to every future training run that reads
|
||
// this cache. Same gate the fxcache loader and DBN fallback enforce.
|
||
ml::walk_forward::validate_normalized_features(&features, "precompute_features writer")?;
|
||
|
||
// ── 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: {}-dim ({})", OFI_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(())
|
||
}
|