3b1ed72eaa
Sequence builder now walks fine→coarse scales from the anchor's forward edge, aggregating each window into one pseudo-snapshot via aggregate_window. seq_len field deleted (total_positions() replaces all reads). Greenfield — no legacy single-scale path. External callers updated in subsequent tasks.
829 lines
39 KiB
Rust
829 lines
39 KiB
Rust
//! Phase A data loader.
|
||
//!
|
||
//! Reads predecoded MBP-10 sidecars (already on disk from earlier session
|
||
//! work — see `crates/ml-features/src/predecoded.rs`). Per spec Section 4:
|
||
//! yields multi-resolution windows of `Mbp10RawInput` (length =
|
||
//! `multi_resolution.total_positions()`) plus N_HORIZONS binary labels
|
||
//! computed via `multi_horizon_labels::generate_labels`.
|
||
//!
|
||
//! Memory discipline: snapshots stay on the host side here (CPU control
|
||
//! flow). The trainer (Task 13) uploads them into GPU buffers via the
|
||
//! `CfcTrunk::update_input_buffers` mapped-pinned path before each
|
||
//! captured-graph replay.
|
||
|
||
use std::path::PathBuf;
|
||
|
||
use anyhow::{Context, Result};
|
||
use data::providers::databento::mbp10::{BidAskPair, Mbp10Snapshot};
|
||
use ml_features::predecoded::load_or_predecode_mbp10;
|
||
|
||
/// Re-export so downstream crates (ml-backtesting, tests) can construct a
|
||
/// `MultiHorizonLoaderConfig` without taking a direct dependency on `data`.
|
||
pub use data::providers::databento::dbn_parser::InstrumentFilter;
|
||
use rand::{Rng, SeedableRng};
|
||
use rand_chacha::ChaCha8Rng;
|
||
|
||
use crate::cfc::snap_features::{Mbp10RawInput, ES_TICK_SIZE, REGIME_DIM};
|
||
use crate::heads::N_HORIZONS;
|
||
use crate::multi_horizon_labels::{generate_labels, generate_outcome_labels_ab};
|
||
|
||
/// EMA smoothing factors. Half-life ≈ ln(2)/α snapshots.
|
||
///
|
||
/// Empirical ES MBP-10 median inter-event Δt = 74µs (NOT 250ms as the
|
||
/// previous comments claimed — mean Δt is 55ms but the distribution is
|
||
/// extremely heavy-tailed so the median dominates EMA dynamics). At that
|
||
/// rate the half-lives in wall-clock are roughly:
|
||
///
|
||
/// * `ALPHA_FAST_2P6MS` (α=0.02) → ~35 snapshots × 74µs ≈ 2.6ms
|
||
/// — microstructure-burst scale
|
||
/// * `ALPHA_MED_100MS` (α=0.0005) → ~1400 snapshots × 74µs ≈ 100ms
|
||
/// — 1-2-second clustering scale
|
||
///
|
||
/// Values preserved (validated by all prior training); the names are
|
||
/// updated to reflect what they actually do. True macro-regime EMAs
|
||
/// (multi-second / multi-minute half-lives) would need α ~5e-6 or
|
||
/// smaller and are deliberately NOT added here — that's a separate
|
||
/// feature-engineering decision.
|
||
const ALPHA_FAST_2P6MS: f32 = 0.02;
|
||
const ALPHA_MED_100MS: f32 = 0.0005;
|
||
const REGIME_EPS: f32 = 1e-6;
|
||
|
||
/// Compute per-snapshot regime features via a single forward EMA pass
|
||
/// over the file. Returns `[regime; REGIME_DIM]` aligned to `snapshots`.
|
||
///
|
||
/// Sentinel-bootstrap (per [[pearl_first_observation_bootstrap]]): the
|
||
/// first observation replaces the EMA directly (no zero-bias warmup).
|
||
/// Variance EMAs use Welford-style: track ema_x and ema_x² then
|
||
/// var = max(ema_x² - ema_x², 0). All transforms are monotone /
|
||
/// log-compressed so no tuned constants leak in.
|
||
fn compute_regime_features(snapshots: &[Mbp10Snapshot]) -> Vec<[f32; REGIME_DIM]> {
|
||
let n = snapshots.len();
|
||
let mut out: Vec<[f32; REGIME_DIM]> = vec![[0.0; REGIME_DIM]; n];
|
||
if n == 0 { return out; }
|
||
|
||
let mut ema_mid_fast = f32::NAN;
|
||
let mut ema_mid_med = f32::NAN;
|
||
let mut ema_mid2_fast = f32::NAN; // E[mid²] for var estimate
|
||
let mut ema_mid2_med = f32::NAN;
|
||
let mut ema_spread = f32::NAN; // ticks
|
||
let mut ema_trades = f32::NAN; // per-snapshot delta count
|
||
let mut prev_trade_count: u32 = 0;
|
||
|
||
for (k, snap) in snapshots.iter().enumerate() {
|
||
let mid = mid_price_f32(snap);
|
||
let l0 = snap.levels.first().copied().unwrap_or_else(BidAskPair::empty);
|
||
let bid_p = BidAskPair::price_to_f64(l0.bid_px) as f32;
|
||
let ask_p = BidAskPair::price_to_f64(l0.ask_px) as f32;
|
||
let spread_t = if ask_p > bid_p { (ask_p - bid_p) / ES_TICK_SIZE } else { 0.0 };
|
||
let trades: u32 = if k == 0 { 0 } else { snap.trade_count.saturating_sub(prev_trade_count) };
|
||
prev_trade_count = snap.trade_count;
|
||
|
||
// Update EMAs with sentinel-bootstrap.
|
||
if ema_mid_fast.is_nan() {
|
||
ema_mid_fast = mid; ema_mid_med = mid;
|
||
ema_mid2_fast = mid * mid; ema_mid2_med = mid * mid;
|
||
ema_spread = spread_t; ema_trades = trades as f32;
|
||
} else {
|
||
ema_mid_fast = (1.0 - ALPHA_FAST_2P6MS) * ema_mid_fast + ALPHA_FAST_2P6MS * mid;
|
||
ema_mid_med = (1.0 - ALPHA_MED_100MS) * ema_mid_med + ALPHA_MED_100MS * mid;
|
||
ema_mid2_fast = (1.0 - ALPHA_FAST_2P6MS) * ema_mid2_fast + ALPHA_FAST_2P6MS * mid * mid;
|
||
ema_mid2_med = (1.0 - ALPHA_MED_100MS) * ema_mid2_med + ALPHA_MED_100MS * mid * mid;
|
||
ema_spread = (1.0 - ALPHA_FAST_2P6MS) * ema_spread + ALPHA_FAST_2P6MS * spread_t;
|
||
ema_trades = (1.0 - ALPHA_FAST_2P6MS) * ema_trades + ALPHA_FAST_2P6MS * trades as f32;
|
||
}
|
||
|
||
let var_fast = (ema_mid2_fast - ema_mid_fast * ema_mid_fast).max(0.0);
|
||
let var_med = (ema_mid2_med - ema_mid_med * ema_mid_med ).max(0.0);
|
||
let std_fast = (var_fast + REGIME_EPS).sqrt();
|
||
let std_med = (var_med + REGIME_EPS).sqrt();
|
||
|
||
out[k][0] = (mid - ema_mid_fast) / std_fast; // z-score fast (~2.6ms)
|
||
out[k][1] = (mid - ema_mid_med) / std_med; // z-score med (~100ms)
|
||
out[k][2] = (ema_mid_fast - ema_mid_med) / std_med; // trend signal
|
||
out[k][3] = (1.0 + var_med.sqrt() * 1e4).ln(); // log-compressed med vol
|
||
out[k][4] = (1.0 + ema_spread).ln(); // liquidity regime
|
||
out[k][5] = (1.0 + ema_trades * 100.0).ln(); // activity regime
|
||
}
|
||
out
|
||
}
|
||
|
||
#[derive(Clone, Debug)]
|
||
pub struct MultiHorizonLoaderConfig {
|
||
/// Explicit list of .dbn / .dbn.zst MBP-10 files to load. Order
|
||
/// is preserved (no internal shuffle) so callers can use file
|
||
/// ordering to control temporal train / val splits. Use
|
||
/// [`discover_mbp10_files_sorted`] to enumerate a directory in
|
||
/// chronological-filename order.
|
||
pub files: Vec<PathBuf>,
|
||
/// Directory where predecoded sidecars live (`load_or_predecode_mbp10`
|
||
/// arg — typically the same dir as `files` or a sibling).
|
||
pub predecoded_dir: PathBuf,
|
||
/// Multi-resolution input window. `total_positions()` defines the
|
||
/// sequence length the trainer sees; `required_lookback_ticks()` is
|
||
/// the per-anchor pre-edge snapshot count the loader needs from each
|
||
/// source file.
|
||
pub multi_resolution: crate::data::aggregation::MultiResolutionConfig,
|
||
/// Forward-horizon offsets in snapshots. Length tracks
|
||
/// `ml_alpha::heads::N_HORIZONS` so the loader's label layout matches
|
||
/// the per-horizon heads.
|
||
pub horizons: [usize; N_HORIZONS],
|
||
/// Soft cap on total sequences yielded across all source files.
|
||
pub n_max_sequences: usize,
|
||
/// Deterministic seed for sequence-anchor selection within files.
|
||
/// (Files themselves are taken in the order given in `files`; this
|
||
/// seed only randomises which snapshot inside each file becomes
|
||
/// the start of each yielded sequence.)
|
||
pub seed: u64,
|
||
/// When `true`, skip per-file forward-label precomputation (saves
|
||
/// ~half the file-load cost) and enable chronological inference
|
||
/// streaming via `next_inference_input()`. Used by the ml-backtesting
|
||
/// LOB backtest harness; trainer always passes `false`. See
|
||
/// `docs/superpowers/specs/2026-05-18-real-lob-integration-design.md` §1.
|
||
pub inference_only: bool,
|
||
/// Round-trip cost in price units used for D-style outcome label
|
||
/// generation (Phase B aux supervision). ES default: 0.5 (= 2 ticks
|
||
/// × $0.25 per tick). The cost is subtracted from realized PnL before
|
||
/// the 1.5×|max-adverse-excursion| penalty; setting it to 0.0 makes
|
||
/// the aux head learn a cost-free outcome and rarely matches the
|
||
/// inference-time trading objective.
|
||
pub outcome_label_cost: f32,
|
||
/// Selects which `instrument_id` records the MBP-10 predecoder keeps.
|
||
/// See [`InstrumentFilter`] for variants.
|
||
///
|
||
/// ES.FUT DBN archives are parent-symbol expansions: the same file
|
||
/// carries the front-month contract (e.g. `ESH4` for Q1 2024,
|
||
/// instrument_id 17077), back months (`ESM4`, etc.), plus calendar
|
||
/// spreads. Without a filter the loader interleaves all of them and
|
||
/// computes ΔP across instruments — silently inflating K-step labels
|
||
/// by thousands of dollars at the contract boundaries. For ES,
|
||
/// `FrontMonth` auto-detects the dominant id per file (handles the
|
||
/// quarterly roll); single-instrument datasets can use `All`.
|
||
pub instrument_filter: InstrumentFilter,
|
||
}
|
||
|
||
/// Default round-trip cost (price units) baked into D-style outcome labels
|
||
/// for ES contracts. 2 ticks × $0.25/tick = $0.50 in price units.
|
||
pub const DEFAULT_OUTCOME_LABEL_COST_ES: f32 = 0.5;
|
||
|
||
/// Discover MBP-10 files under `root` and return them sorted by filename.
|
||
/// Filenames follow the `ES.FUT_<YEAR>-Q<n>.dbn.zst` convention, so
|
||
/// lexicographic sort = chronological. Caller can then slice for
|
||
/// walk-forward CV.
|
||
pub fn discover_mbp10_files_sorted(root: &std::path::Path) -> Result<Vec<PathBuf>> {
|
||
let mut files: Vec<PathBuf> = std::fs::read_dir(root)
|
||
.with_context(|| format!("read mbp10 root {}", root.display()))?
|
||
.filter_map(|e| e.ok())
|
||
.map(|e| e.path())
|
||
.filter(|p| {
|
||
let s = p.to_string_lossy();
|
||
s.ends_with(".dbn.zst") || s.ends_with(".dbn")
|
||
})
|
||
.collect();
|
||
anyhow::ensure!(!files.is_empty(), "no DBN files under {}", root.display());
|
||
files.sort();
|
||
Ok(files)
|
||
}
|
||
|
||
pub struct LabeledSequence {
|
||
/// `multi_resolution.total_positions()` raw or aggregated snapshot inputs
|
||
/// in chronological (oldest-first) order.
|
||
pub snapshots: Vec<Mbp10RawInput>,
|
||
/// `[N_HORIZONS][total_positions]` BCE labels. NaN at positions where the
|
||
/// forward window is outside the source file (right edge) or where
|
||
/// the move is a tied price (drop, mirror `generate_labels` semantics).
|
||
pub labels: [Vec<f32>; N_HORIZONS],
|
||
/// `[N_HORIZONS][total_positions]` Candidate-B prof-binary labels for the
|
||
/// long direction (1.0 if `ΔP_K - 2×cost > 0`, 0.0 otherwise, NaN at right
|
||
/// edge or non-finite price). See [`crate::multi_horizon_labels::generate_outcome_labels_ab`].
|
||
pub outcome_prof_long: [Vec<f32>; N_HORIZONS],
|
||
/// `[N_HORIZONS][total_positions]` Candidate-B prof-binary labels for the
|
||
/// short direction.
|
||
pub outcome_prof_short: [Vec<f32>; N_HORIZONS],
|
||
/// `[N_HORIZONS][total_positions]` Candidate-B σ-normalized signed-return
|
||
/// targets for the long direction. NaN where prof_long==0 or σ_K not yet warmed.
|
||
pub outcome_size_long: [Vec<f32>; N_HORIZONS],
|
||
/// `[N_HORIZONS][total_positions]` Candidate-B σ-normalized signed-return
|
||
/// targets for the short direction. Same NaN semantics as `outcome_size_long`.
|
||
pub outcome_size_short: [Vec<f32>; N_HORIZONS],
|
||
/// `[N_HORIZONS][total_positions]` rolling 1000-step Welford std of K-step
|
||
/// log-returns at each snapshot. NaN during the warmup window. Used
|
||
/// downstream for σ-normalized size targets and any in-trainer scaling.
|
||
pub sigma_k: [Vec<f32>; N_HORIZONS],
|
||
/// Per-file class imbalance for `pos_weight` BCE balancing.
|
||
/// Length = `2 * N_HORIZONS`, laid out as
|
||
/// `[long_h0, long_h1, ..., long_hN, short_h0, ..., short_hN]`.
|
||
/// File-level statistic — every sequence yielded from the same source
|
||
/// file carries the same vector.
|
||
pub pos_fraction: Vec<f32>,
|
||
}
|
||
|
||
/// Loaded-file cache. Holds the deserialized snapshots + pre-computed
|
||
/// per-horizon labels so subsequent `next_sequence()` calls on the
|
||
/// same file avoid the multi-second bincode deserialization.
|
||
struct LoadedFile {
|
||
snapshots: Vec<Mbp10Snapshot>,
|
||
/// `labels_full[h]` holds the full-stream BCE label vector for horizon
|
||
/// `cfg.horizons[h]`, indexed the same as `snapshots`. Sliced per
|
||
/// anchor on each `next_sequence()` call.
|
||
labels_full: [Vec<f32>; N_HORIZONS],
|
||
/// `outcome_prof_long_full[h]` holds the full-stream Candidate-B
|
||
/// prof-binary long-direction label vector for horizon `cfg.horizons[h]`.
|
||
/// NaN at right-edge positions. Sliced per anchor alongside `labels_full`.
|
||
outcome_prof_long_full: [Vec<f32>; N_HORIZONS],
|
||
/// `outcome_prof_short_full[h]` — short-direction counterpart to
|
||
/// `outcome_prof_long_full`.
|
||
outcome_prof_short_full: [Vec<f32>; N_HORIZONS],
|
||
/// `outcome_size_long_full[h]` holds the full-stream Candidate-B
|
||
/// σ-normalized size target for the long direction. NaN where
|
||
/// prof_long==0 OR σ_K not yet warmed.
|
||
outcome_size_long_full: [Vec<f32>; N_HORIZONS],
|
||
/// `outcome_size_short_full[h]` — short-direction counterpart.
|
||
outcome_size_short_full: [Vec<f32>; N_HORIZONS],
|
||
/// `sigma_k_full[h]` rolling 1000-step Welford std of K-step log-returns.
|
||
/// NaN during the per-horizon warmup window.
|
||
sigma_k_full: [Vec<f32>; N_HORIZONS],
|
||
/// File-level positive-class fraction per (direction, horizon) for BCE
|
||
/// pos_weight balancing. Length = `2 * N_HORIZONS`, laid out as
|
||
/// `[long_h0..hN, short_h0..hN]`. Copied verbatim into every yielded
|
||
/// `LabeledSequence`.
|
||
pos_fraction: Vec<f32>,
|
||
/// Per-snapshot regime context computed once per file via a forward
|
||
/// EMA cascade. See [`compute_regime_features`]. Sliced per anchor.
|
||
regime_full: Vec<[f32; REGIME_DIM]>,
|
||
}
|
||
|
||
pub struct MultiHorizonLoader {
|
||
cfg: MultiHorizonLoaderConfig,
|
||
rng: ChaCha8Rng,
|
||
yielded: usize,
|
||
/// All MBP-10 files preloaded into memory at construction. Each
|
||
/// `LoadedFile` carries its deserialized snapshots + precomputed
|
||
/// per-horizon labels + regime features. Per-epoch iteration is
|
||
/// pure memory access — no disk IO after the initial preload.
|
||
///
|
||
/// Memory cost at full ES.FUT dataset (9 quarters, ~45M snapshots
|
||
/// total): ~13-15 GB. Worth it: per-epoch wall time drops from
|
||
/// ~8 min (file-IO-bound) to ~1 min (compute-bound) on L40S, a
|
||
/// ~5-7× speedup that compounds for longer runs.
|
||
files_loaded: Vec<LoadedFile>,
|
||
/// Chronological cursor for `next_inference_input()`. Only meaningful
|
||
/// when `cfg.inference_only == true`.
|
||
inference_file_idx: usize,
|
||
inference_snap_idx: usize,
|
||
}
|
||
|
||
impl MultiHorizonLoader {
|
||
/// Construct + preload the MBP-10 files listed in `cfg.files`,
|
||
/// in the order given (no shuffle — file order is the caller's
|
||
/// responsibility, see [`discover_mbp10_files_sorted`]). Slow
|
||
/// (~50s × N_files), done once at construction. Subsequent
|
||
/// `next_sequence` calls are pure memory.
|
||
pub fn new(cfg: &MultiHorizonLoaderConfig) -> Result<Self> {
|
||
anyhow::ensure!(
|
||
!cfg.files.is_empty(),
|
||
"MultiHorizonLoaderConfig.files is empty — pass at least one file"
|
||
);
|
||
let files: Vec<PathBuf> = cfg.files.clone();
|
||
let rng = ChaCha8Rng::seed_from_u64(cfg.seed);
|
||
|
||
let max_horizon = *cfg.horizons.iter().max().expect("non-empty horizons");
|
||
// Multi-resolution lookback: each anchor needs `required_lookback_ticks`
|
||
// raw snapshots BEFORE the forward edge to fill all aggregation scales.
|
||
let lookback = cfg.multi_resolution.required_lookback_ticks();
|
||
let min_size = if cfg.inference_only {
|
||
lookback.max(1)
|
||
} else {
|
||
lookback + max_horizon + 1
|
||
};
|
||
let mut files_loaded: Vec<LoadedFile> = Vec::with_capacity(files.len());
|
||
for path in &files {
|
||
let snapshots = load_or_predecode_mbp10(path, &cfg.predecoded_dir, cfg.instrument_filter)
|
||
.with_context(|| format!(
|
||
"load mbp10 {} (filter={:?})", path.display(), cfg.instrument_filter
|
||
))?;
|
||
if snapshots.len() < min_size {
|
||
tracing::warn!(
|
||
path = %path.display(),
|
||
snapshots = snapshots.len(),
|
||
min_size,
|
||
"skipping file (too few snapshots for required_lookback_ticks + max_horizon)",
|
||
);
|
||
continue;
|
||
}
|
||
let mut labels_full: [Vec<f32>; N_HORIZONS] = Default::default();
|
||
let mut outcome_prof_long_full: [Vec<f32>; N_HORIZONS] = Default::default();
|
||
let mut outcome_prof_short_full: [Vec<f32>; N_HORIZONS] = Default::default();
|
||
let mut outcome_size_long_full: [Vec<f32>; N_HORIZONS] = Default::default();
|
||
let mut outcome_size_short_full: [Vec<f32>; N_HORIZONS] = Default::default();
|
||
let mut sigma_k_full: [Vec<f32>; N_HORIZONS] = Default::default();
|
||
// Default pos_fraction is zero per direction/horizon; populated below in
|
||
// training mode. Length is fixed so downstream broadcast is uniform.
|
||
let mut pos_fraction: Vec<f32> = vec![0.0_f32; 2 * N_HORIZONS];
|
||
if !cfg.inference_only {
|
||
let prices: Vec<f32> = snapshots.iter().map(mid_price_f32).collect();
|
||
for (h_idx, &h) in cfg.horizons.iter().enumerate() {
|
||
let mut full = vec![f32::NAN; snapshots.len()];
|
||
let raw = generate_labels(&prices, h);
|
||
for (i, &t) in raw.valid_indices.iter().enumerate() {
|
||
full[t] = raw.labels[i];
|
||
}
|
||
labels_full[h_idx] = full;
|
||
}
|
||
// Candidate-B aux supervision: paired prof-binary + σ-normalized
|
||
// size-regression targets per (direction, horizon). Replaces the
|
||
// old D-style asymmetric loss-aversion labels.
|
||
let outcome = generate_outcome_labels_ab(
|
||
&prices,
|
||
&cfg.horizons,
|
||
cfg.outcome_label_cost,
|
||
)
|
||
.with_context(|| format!(
|
||
"generate_outcome_labels_ab for {} (cost={})",
|
||
path.display(),
|
||
cfg.outcome_label_cost,
|
||
))?;
|
||
outcome_prof_long_full = outcome.y_prof_long;
|
||
outcome_prof_short_full = outcome.y_prof_short;
|
||
outcome_size_long_full = outcome.y_size_long;
|
||
outcome_size_short_full = outcome.y_size_short;
|
||
sigma_k_full = outcome.sigma_k;
|
||
pos_fraction = outcome.pos_fraction;
|
||
}
|
||
let regime_full = compute_regime_features(&snapshots);
|
||
files_loaded.push(LoadedFile {
|
||
snapshots,
|
||
labels_full,
|
||
outcome_prof_long_full,
|
||
outcome_prof_short_full,
|
||
outcome_size_long_full,
|
||
outcome_size_short_full,
|
||
sigma_k_full,
|
||
pos_fraction,
|
||
regime_full,
|
||
});
|
||
}
|
||
anyhow::ensure!(
|
||
!files_loaded.is_empty(),
|
||
"no files had enough snapshots ({}+ required) across {} input files",
|
||
min_size, cfg.files.len()
|
||
);
|
||
tracing::info!(
|
||
n_files = files_loaded.len(),
|
||
total_snapshots = files_loaded.iter().map(|lf| lf.snapshots.len()).sum::<usize>(),
|
||
"MultiHorizonLoader preloaded all files",
|
||
);
|
||
Ok(Self {
|
||
cfg: cfg.clone(),
|
||
rng,
|
||
yielded: 0,
|
||
files_loaded,
|
||
inference_file_idx: 0,
|
||
inference_snap_idx: 0,
|
||
})
|
||
}
|
||
|
||
/// Reset per-epoch state: re-seed RNG (so anchor sampling differs
|
||
/// each epoch) and zero the yielded counter. The preloaded file
|
||
/// inventory is preserved — no disk IO.
|
||
pub fn reset(&mut self, seed: u64) {
|
||
self.rng = ChaCha8Rng::seed_from_u64(seed);
|
||
self.yielded = 0;
|
||
self.inference_file_idx = 0;
|
||
self.inference_snap_idx = 0;
|
||
}
|
||
|
||
/// Return a clone of the first chronological snapshot from the
|
||
/// first loaded file as an `Mbp10RawInput`. Used by the LOB
|
||
/// backtest harness to seed the first-call eager warmup of
|
||
/// `PerceptionTrainer::forward_only` (P5 ships CUDA Graph
|
||
/// capture of the inference dispatch chain). Pure read — no
|
||
/// cursor mutation. Available in both training and inference
|
||
/// modes.
|
||
pub fn peek_first(&self) -> Result<Mbp10RawInput> {
|
||
let first_file = self.files_loaded.first()
|
||
.context("loader has no loaded files")?;
|
||
let first_snap = first_file.snapshots.first()
|
||
.context("first loaded file has no snapshots")?;
|
||
// No previous snapshot exists; pass the first as both cur/prev.
|
||
// This yields prev_mid == cur_mid + trade_signed_vol == 0 + prev_ts_ns == ts_ns,
|
||
// which is the natural "stream start" semantics for inference seeding.
|
||
Ok(convert(first_snap, first_snap, first_file.regime_full[0]))
|
||
}
|
||
|
||
/// Iterator-style accessor for inference mode: walks every loaded
|
||
/// snapshot across every loaded file in chronological order.
|
||
/// Returns `Ok(None)` when the stream is exhausted. Unlike
|
||
/// [`next_sequence`], does NOT slice fixed-length windows.
|
||
/// A1: decision_stride removed; every snapshot is yielded consecutively.
|
||
/// Mutates the inference cursor. Errors if `cfg.inference_only` is false.
|
||
pub fn next_inference_input(&mut self) -> Result<Option<Mbp10RawInput>> {
|
||
anyhow::ensure!(
|
||
self.cfg.inference_only,
|
||
"next_inference_input requires MultiHorizonLoaderConfig.inference_only = true"
|
||
);
|
||
loop {
|
||
let Some(file) = self.files_loaded.get(self.inference_file_idx) else {
|
||
return Ok(None);
|
||
};
|
||
if self.inference_snap_idx >= file.snapshots.len() {
|
||
self.inference_file_idx += 1;
|
||
self.inference_snap_idx = 0;
|
||
continue;
|
||
}
|
||
let cur = &file.snapshots[self.inference_snap_idx];
|
||
// For the very first snapshot, use itself as prev (matches peek_first).
|
||
let prev = if self.inference_snap_idx == 0 {
|
||
cur
|
||
} else {
|
||
&file.snapshots[self.inference_snap_idx - 1]
|
||
};
|
||
let regime = file.regime_full[self.inference_snap_idx];
|
||
let out = convert(cur, prev, regime);
|
||
self.inference_snap_idx += 1;
|
||
self.yielded += 1;
|
||
return Ok(Some(out));
|
||
}
|
||
}
|
||
|
||
pub fn n_files(&self) -> usize {
|
||
self.files_loaded.len()
|
||
}
|
||
|
||
pub fn yielded(&self) -> usize {
|
||
self.yielded
|
||
}
|
||
|
||
pub fn next_sequence(&mut self) -> Result<Option<LabeledSequence>> {
|
||
self.next_sequence_random()
|
||
}
|
||
|
||
fn next_sequence_random(&mut self) -> Result<Option<LabeledSequence>> {
|
||
if self.yielded >= self.cfg.n_max_sequences {
|
||
return Ok(None);
|
||
}
|
||
let max_horizon = *self.cfg.horizons.iter().max().expect("non-empty horizons");
|
||
// A1: decision_stride removed; training sequences are always consecutive
|
||
// snapshots (stride = 1). dt_s = 1.0 in all CfC forward passes.
|
||
|
||
// Pick a random file from the preloaded inventory, then a random
|
||
// anchor within that file. Uniform-across-files sampling — each
|
||
// file contributes ~n_max_sequences / n_files samples per epoch.
|
||
let n_files = self.files_loaded.len();
|
||
let file_idx = self.rng.gen_range(0..n_files);
|
||
let lf = &self.files_loaded[file_idx];
|
||
let needed = self.cfg.multi_resolution.total_positions() + max_horizon;
|
||
anyhow::ensure!(
|
||
lf.snapshots.len() > needed,
|
||
"file too short for total_positions={} max_horizon={}: {} <= {}",
|
||
self.cfg.multi_resolution.total_positions(), max_horizon, lf.snapshots.len(), needed
|
||
);
|
||
let max_anchor = lf.snapshots.len() - needed;
|
||
let anchor: usize = self.rng.gen_range(0..max_anchor);
|
||
|
||
let mut labels: [Vec<f32>; N_HORIZONS] = Default::default();
|
||
let mut outcome_prof_long: [Vec<f32>; N_HORIZONS] = Default::default();
|
||
let mut outcome_prof_short: [Vec<f32>; N_HORIZONS] = Default::default();
|
||
let mut outcome_size_long: [Vec<f32>; N_HORIZONS] = Default::default();
|
||
let mut outcome_size_short: [Vec<f32>; N_HORIZONS] = Default::default();
|
||
let mut sigma_k: [Vec<f32>; N_HORIZONS] = Default::default();
|
||
for h in 0..N_HORIZONS {
|
||
let mut row = Vec::with_capacity(self.cfg.multi_resolution.total_positions());
|
||
let mut row_prof_long = Vec::with_capacity(self.cfg.multi_resolution.total_positions());
|
||
let mut row_prof_short = Vec::with_capacity(self.cfg.multi_resolution.total_positions());
|
||
let mut row_size_long = Vec::with_capacity(self.cfg.multi_resolution.total_positions());
|
||
let mut row_size_short = Vec::with_capacity(self.cfg.multi_resolution.total_positions());
|
||
let mut row_sigma = Vec::with_capacity(self.cfg.multi_resolution.total_positions());
|
||
for k in 0..self.cfg.multi_resolution.total_positions() {
|
||
row.push(lf.labels_full[h][anchor + k]);
|
||
row_prof_long.push(lf.outcome_prof_long_full[h][anchor + k]);
|
||
row_prof_short.push(lf.outcome_prof_short_full[h][anchor + k]);
|
||
row_size_long.push(lf.outcome_size_long_full[h][anchor + k]);
|
||
row_size_short.push(lf.outcome_size_short_full[h][anchor + k]);
|
||
row_sigma.push(lf.sigma_k_full[h][anchor + k]);
|
||
}
|
||
labels[h] = row;
|
||
outcome_prof_long[h] = row_prof_long;
|
||
outcome_prof_short[h] = row_prof_short;
|
||
outcome_size_long[h] = row_size_long;
|
||
outcome_size_short[h] = row_size_short;
|
||
sigma_k[h] = row_sigma;
|
||
}
|
||
// Multi-resolution window construction. The forward edge is
|
||
// `anchor + total_positions` exclusive (the snapshot AT
|
||
// `anchor + total_positions - 1` is the newest position in the
|
||
// emitted sequence). We walk fine→coarse from the edge backward
|
||
// so each scale's positions cover earlier-in-time windows than
|
||
// the previous scale, then reverse so the emitted sequence is
|
||
// chronological (oldest-first).
|
||
let scales = self.cfg.multi_resolution.scales();
|
||
let total = self.cfg.multi_resolution.total_positions();
|
||
let edge = anchor + total; // exclusive upper bound
|
||
|
||
let mut sequence_rev: Vec<Mbp10RawInput> = Vec::with_capacity(total);
|
||
let mut cursor = edge;
|
||
for &(scale, count) in scales {
|
||
for _ in 0..count {
|
||
let window_end = cursor; // exclusive
|
||
let window_start = window_end - scale; // inclusive
|
||
let mut window_raws: Vec<Mbp10RawInput> = Vec::with_capacity(scale);
|
||
for idx in window_start..window_end {
|
||
let cur = &lf.snapshots[idx];
|
||
let prev_idx = if idx == 0 { 0 } else { idx - 1 };
|
||
let prev = &lf.snapshots[prev_idx];
|
||
window_raws.push(convert(cur, prev, lf.regime_full[idx]));
|
||
}
|
||
if window_raws.len() == 1 {
|
||
sequence_rev.push(window_raws.into_iter().next().unwrap());
|
||
} else {
|
||
sequence_rev.push(
|
||
crate::data::aggregation::aggregate_window(window_raws.as_slice()),
|
||
);
|
||
}
|
||
cursor = window_start;
|
||
}
|
||
}
|
||
let sequence: Vec<Mbp10RawInput> = sequence_rev.into_iter().rev().collect();
|
||
debug_assert_eq!(sequence.len(), total);
|
||
|
||
self.yielded += 1;
|
||
Ok(Some(LabeledSequence {
|
||
snapshots: sequence,
|
||
labels,
|
||
outcome_prof_long,
|
||
outcome_prof_short,
|
||
outcome_size_long,
|
||
outcome_size_short,
|
||
sigma_k,
|
||
pos_fraction: lf.pos_fraction.clone(),
|
||
}))
|
||
}
|
||
}
|
||
|
||
fn mid_price_f32(s: &Mbp10Snapshot) -> f32 {
|
||
let l = s.levels.first().copied().unwrap_or_else(BidAskPair::empty);
|
||
let bid = BidAskPair::price_to_f64(l.bid_px);
|
||
let ask = BidAskPair::price_to_f64(l.ask_px);
|
||
(0.5 * (bid + ask)) as f32
|
||
}
|
||
|
||
fn convert(
|
||
cur: &Mbp10Snapshot,
|
||
prev: &Mbp10Snapshot,
|
||
regime: [f32; REGIME_DIM],
|
||
) -> Mbp10RawInput {
|
||
let mut bid_px = [0.0f32; 10];
|
||
let mut bid_sz = [0.0f32; 10];
|
||
let mut ask_px = [0.0f32; 10];
|
||
let mut ask_sz = [0.0f32; 10];
|
||
for (i, lvl) in cur.levels.iter().take(10).enumerate() {
|
||
bid_px[i] = BidAskPair::price_to_f64(lvl.bid_px) as f32;
|
||
bid_sz[i] = lvl.bid_sz as f32;
|
||
ask_px[i] = BidAskPair::price_to_f64(lvl.ask_px) as f32;
|
||
ask_sz[i] = lvl.ask_sz as f32;
|
||
}
|
||
let prev_mid = mid_price_f32(prev);
|
||
let trade_count = cur.trade_count.saturating_sub(prev.trade_count);
|
||
let trade_signed_vol = infer_signed_trade_flow(cur, prev);
|
||
Mbp10RawInput {
|
||
bid_px,
|
||
bid_sz,
|
||
ask_px,
|
||
ask_sz,
|
||
prev_mid,
|
||
trade_signed_vol,
|
||
trade_count,
|
||
ts_ns: cur.timestamp,
|
||
prev_ts_ns: prev.timestamp,
|
||
regime,
|
||
}
|
||
}
|
||
|
||
/// Infer signed trade flow at L1 from MBP-10 snapshot deltas without
|
||
/// the separate trades stream. Convention matches `Mbp10RawInput::
|
||
/// trade_signed_vol`: positive = buyer-initiated (aggressive cross of
|
||
/// the ask), negative = seller-initiated (aggressive hit of the bid).
|
||
///
|
||
/// Heuristic (Databento-standard tick rule applied to L1):
|
||
/// * ask_px[0] unchanged AND ask_sz[0] decreased
|
||
/// → aggressive buys consumed ask depth; add (prev.ask_sz - cur.ask_sz).
|
||
/// * bid_px[0] unchanged AND bid_sz[0] decreased
|
||
/// → aggressive sells consumed bid depth; subtract (prev.bid_sz - cur.bid_sz).
|
||
/// * ask_px[0] moved up (best ask cleared completely)
|
||
/// → infer full prev.ask_sz lifted; add prev.ask_sz.
|
||
/// * bid_px[0] moved down (best bid cleared completely)
|
||
/// → infer full prev.bid_sz hit; subtract prev.bid_sz.
|
||
/// * Pure cancellations (size shrank but price moved away from us)
|
||
/// ambiguous; ignore.
|
||
///
|
||
/// This is a lower-bound estimator — won't catch trades that crossed
|
||
/// multiple levels (those show up only via the deeper-level deltas)
|
||
/// nor trades against hidden / off-book liquidity. Acceptable for v1
|
||
/// queue-decay signal; production deployments should layer in the
|
||
/// trades-stream loader for ground-truth signed flow.
|
||
fn infer_signed_trade_flow(cur: &Mbp10Snapshot, prev: &Mbp10Snapshot) -> f32 {
|
||
let cur_l1 = cur.levels.first().copied().unwrap_or_else(BidAskPair::empty);
|
||
let prev_l1 = prev.levels.first().copied().unwrap_or_else(BidAskPair::empty);
|
||
|
||
let cur_bid_px = BidAskPair::price_to_f64(cur_l1.bid_px) as f32;
|
||
let cur_bid_sz = cur_l1.bid_sz as f32;
|
||
let cur_ask_px = BidAskPair::price_to_f64(cur_l1.ask_px) as f32;
|
||
let cur_ask_sz = cur_l1.ask_sz as f32;
|
||
let prev_bid_px = BidAskPair::price_to_f64(prev_l1.bid_px) as f32;
|
||
let prev_bid_sz = prev_l1.bid_sz as f32;
|
||
let prev_ask_px = BidAskPair::price_to_f64(prev_l1.ask_px) as f32;
|
||
let prev_ask_sz = prev_l1.ask_sz as f32;
|
||
|
||
let mut buyer_init = 0.0_f32;
|
||
let mut seller_init = 0.0_f32;
|
||
|
||
// Same-price size decrement = aggressive cross at that price.
|
||
if (cur_ask_px - prev_ask_px).abs() < f32::EPSILON && cur_ask_sz < prev_ask_sz {
|
||
buyer_init += prev_ask_sz - cur_ask_sz;
|
||
}
|
||
if (cur_bid_px - prev_bid_px).abs() < f32::EPSILON && cur_bid_sz < prev_bid_sz {
|
||
seller_init += prev_bid_sz - cur_bid_sz;
|
||
}
|
||
// Best price moved away — infer the previous level fully cleared.
|
||
if cur_ask_px > prev_ask_px && prev_ask_sz > 0.0 {
|
||
buyer_init += prev_ask_sz;
|
||
}
|
||
if cur_bid_px < prev_bid_px && prev_bid_sz > 0.0 {
|
||
seller_init += prev_bid_sz;
|
||
}
|
||
|
||
buyer_init - seller_init
|
||
}
|
||
|
||
#[cfg(test)]
|
||
mod trade_flow_tests {
|
||
use super::*;
|
||
|
||
fn snap_with_l1(ts: u64, bid_px: f64, bid_sz: u32, ask_px: f64, ask_sz: u32, trade_count: u32) -> Mbp10Snapshot {
|
||
let mut levels = vec![BidAskPair::empty(); 10];
|
||
levels[0] = BidAskPair {
|
||
bid_px: BidAskPair::price_from_f64(bid_px),
|
||
bid_sz,
|
||
bid_ct: 1,
|
||
ask_px: BidAskPair::price_from_f64(ask_px),
|
||
ask_sz,
|
||
ask_ct: 1,
|
||
};
|
||
Mbp10Snapshot::new("ES.FUT".into(), ts, levels, 0, trade_count)
|
||
}
|
||
|
||
#[test]
|
||
fn pure_cancel_no_size_change_reports_zero_flow() {
|
||
let prev = snap_with_l1(0, 5500.00, 100, 5500.25, 100, 0);
|
||
let cur = snap_with_l1(1, 5500.00, 100, 5500.25, 100, 0);
|
||
assert_eq!(infer_signed_trade_flow(&cur, &prev), 0.0);
|
||
}
|
||
|
||
#[test]
|
||
fn ask_size_shrank_unchanged_price_reports_positive_buyer_flow() {
|
||
let prev = snap_with_l1(0, 5500.00, 100, 5500.25, 100, 0);
|
||
let cur = snap_with_l1(1, 5500.00, 100, 5500.25, 60, 5);
|
||
// 100 → 60 = aggressive buys consumed 40 lots at the ask.
|
||
assert_eq!(infer_signed_trade_flow(&cur, &prev), 40.0);
|
||
}
|
||
|
||
#[test]
|
||
fn bid_size_shrank_unchanged_price_reports_negative_seller_flow() {
|
||
let prev = snap_with_l1(0, 5500.00, 100, 5500.25, 100, 0);
|
||
let cur = snap_with_l1(1, 5500.00, 70, 5500.25, 100, 3);
|
||
assert_eq!(infer_signed_trade_flow(&cur, &prev), -30.0);
|
||
}
|
||
|
||
#[test]
|
||
fn ask_price_moved_up_reports_full_prev_ask_as_buyer_flow() {
|
||
// best ask cleared — infer prev.ask_sz fully lifted.
|
||
let prev = snap_with_l1(0, 5500.00, 100, 5500.25, 50, 0);
|
||
let cur = snap_with_l1(1, 5500.00, 100, 5500.50, 100, 1);
|
||
assert_eq!(infer_signed_trade_flow(&cur, &prev), 50.0);
|
||
}
|
||
|
||
#[test]
|
||
fn bid_price_moved_down_reports_full_prev_bid_as_seller_flow() {
|
||
let prev = snap_with_l1(0, 5500.00, 80, 5500.25, 100, 0);
|
||
let cur = snap_with_l1(1, 5499.75, 100, 5500.25, 100, 1);
|
||
assert_eq!(infer_signed_trade_flow(&cur, &prev), -80.0);
|
||
}
|
||
|
||
#[test]
|
||
fn mixed_buyer_and_seller_within_same_snapshot_nets_correctly() {
|
||
let prev = snap_with_l1(0, 5500.00, 100, 5500.25, 100, 0);
|
||
// ask shrank by 25 (aggressive buys), bid shrank by 10 (aggressive sells).
|
||
let cur = snap_with_l1(1, 5500.00, 90, 5500.25, 75, 3);
|
||
assert_eq!(infer_signed_trade_flow(&cur, &prev), 15.0);
|
||
}
|
||
}
|
||
|
||
#[cfg(test)]
|
||
mod inference_mode_tests {
|
||
use super::*;
|
||
|
||
fn test_fixture_cfg(inference_only: bool) -> Option<MultiHorizonLoaderConfig> {
|
||
let root = std::env::var("FOXHUNT_TEST_DATA").ok()?;
|
||
let mbp10 = std::path::PathBuf::from(&root).join("ES.FUT");
|
||
if !mbp10.exists() { return None; }
|
||
let files = discover_mbp10_files_sorted(&mbp10).ok()?;
|
||
Some(MultiHorizonLoaderConfig {
|
||
files,
|
||
predecoded_dir: mbp10.clone(),
|
||
multi_resolution: crate::data::aggregation::MultiResolutionConfig::from_scales(vec![(1, 1)]).unwrap(),
|
||
horizons: crate::heads::HORIZONS,
|
||
n_max_sequences: 1,
|
||
seed: 0xCAFEF00D,
|
||
inference_only,
|
||
outcome_label_cost: DEFAULT_OUTCOME_LABEL_COST_ES,
|
||
instrument_filter: InstrumentFilter::All,
|
||
})
|
||
}
|
||
|
||
/// Try to construct a loader; return None if no usable data on disk
|
||
/// (placeholder fixtures = empty sidecars = 0 snapshots per file).
|
||
/// Treat that as "skip — real-data path".
|
||
fn try_loader(inference_only: bool) -> Option<MultiHorizonLoader> {
|
||
let cfg = test_fixture_cfg(inference_only)?;
|
||
match MultiHorizonLoader::new(&cfg) {
|
||
Ok(l) => Some(l),
|
||
Err(e) => {
|
||
eprintln!("skipping: fixture data not usable ({e})");
|
||
None
|
||
}
|
||
}
|
||
}
|
||
|
||
/// Task 1.1 verification: inference_only=true must NOT precompute labels.
|
||
#[test]
|
||
#[ignore = "requires populated FOXHUNT_TEST_DATA"]
|
||
fn inference_only_skips_label_precompute() {
|
||
let Some(loader) = try_loader(true) else { return };
|
||
assert!(loader.files_loaded.iter().all(|f| f.labels_full.iter().all(|v| v.is_empty())),
|
||
"inference_only=true must leave labels_full empty");
|
||
}
|
||
|
||
/// Task 1.1 paired check: inference_only=false populates labels.
|
||
#[test]
|
||
#[ignore = "requires populated FOXHUNT_TEST_DATA"]
|
||
fn training_mode_populates_labels() {
|
||
let Some(loader) = try_loader(false) else { return };
|
||
assert!(loader.files_loaded.iter().any(|f| f.labels_full.iter().any(|v| !v.is_empty())),
|
||
"inference_only=false must precompute labels for at least one file");
|
||
}
|
||
|
||
/// Task 1.2 verification: peek_first returns the first snapshot
|
||
/// converted to Mbp10RawInput with non-zero ts_ns + cur==prev semantics.
|
||
#[test]
|
||
#[ignore = "requires populated FOXHUNT_TEST_DATA"]
|
||
fn peek_first_returns_first_chronological_snapshot() {
|
||
let Some(loader) = try_loader(true) else { return };
|
||
let first = loader.peek_first().expect("peek_first ok");
|
||
assert!(first.ts_ns > 0);
|
||
assert_eq!(first.ts_ns, first.prev_ts_ns);
|
||
assert_eq!(first.trade_signed_vol, 0.0);
|
||
}
|
||
|
||
/// Task 1.3 verification: chronological iteration in inference mode.
|
||
#[test]
|
||
#[ignore = "requires populated FOXHUNT_TEST_DATA"]
|
||
fn next_inference_input_yields_chronological() {
|
||
let Some(mut loader) = try_loader(true) else { return };
|
||
let snap0 = loader.next_inference_input().expect("snap 0").expect("not none");
|
||
let snap1 = loader.next_inference_input().expect("snap 1").expect("not none");
|
||
let snap2 = loader.next_inference_input().expect("snap 2").expect("not none");
|
||
assert!(snap0.ts_ns <= snap1.ts_ns);
|
||
assert!(snap1.ts_ns <= snap2.ts_ns);
|
||
assert_eq!(snap0.ts_ns, snap0.prev_ts_ns);
|
||
assert_eq!(snap1.prev_ts_ns, snap0.ts_ns);
|
||
assert_eq!(snap2.prev_ts_ns, snap1.ts_ns);
|
||
}
|
||
|
||
/// Negative case: next_inference_input on a non-inference loader must error.
|
||
/// Tests can always construct an empty-files config to exercise this
|
||
/// branch without depending on real data.
|
||
#[test]
|
||
fn next_inference_input_errors_when_not_inference_mode() {
|
||
// We don't need real data here — the assertion fires before file load.
|
||
// But we DO need a non-empty files list to satisfy the constructor's
|
||
// ensure!. Stub with a path that may or may not exist; if construction
|
||
// fails for IO reasons, that's fine — the test cares only about the
|
||
// mode check.
|
||
let cfg = MultiHorizonLoaderConfig {
|
||
files: vec![std::path::PathBuf::from("/tmp/__nonexistent_foxhunt.dbn.zst")],
|
||
predecoded_dir: std::path::PathBuf::from("/tmp"),
|
||
multi_resolution: crate::data::aggregation::MultiResolutionConfig::from_scales(vec![(1, 1)]).unwrap(),
|
||
horizons: crate::heads::HORIZONS,
|
||
n_max_sequences: 1,
|
||
seed: 0,
|
||
inference_only: false,
|
||
outcome_label_cost: DEFAULT_OUTCOME_LABEL_COST_ES,
|
||
instrument_filter: InstrumentFilter::All,
|
||
};
|
||
if let Ok(mut loader) = MultiHorizonLoader::new(&cfg) {
|
||
let err = loader.next_inference_input();
|
||
assert!(err.is_err(), "must error when inference_only=false");
|
||
}
|
||
// If new() failed (no file), the assertion is moot — covered elsewhere.
|
||
}
|
||
}
|