19986c8d96
Replaces the placeholder `trade_signed_vol = trade_count_delta` (always
non-negative, sign-neutral) with a proper Databento-standard tick-rule
inference applied to L1 size + price deltas across consecutive MBP-10
snapshots:
• 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 hit
the bid; subtract (prev.bid_sz − cur.bid_sz).
• ask_px[0] moved up → previous best ask cleared; add prev.ask_sz.
• bid_px[0] moved down → previous best bid hit; subtract prev.bid_sz.
• Pure cancellations (size shrank but price moved AWAY from us) =
ambiguous; ignore.
Convention matches `Mbp10RawInput::trade_signed_vol`: positive =
buyer-initiated, negative = seller-initiated. This is a LOWER-BOUND
estimator — won't catch trades that cleared multiple levels (those
manifest only via deeper-level deltas) or trades against hidden /
off-book liquidity. Acceptable for v1 queue-decay signal; production
deployments can layer the trades-stream loader for ground-truth flow.
Wired into BacktestHarness::run() → sim.step_resting_orders(ts, vol)
so the queue-decay branch of resting_orders.cu finally fires with
non-zero input. Previously the harness passed 0.0 unconditionally,
which meant resting limits could only fill via the price-cross
marketability branch — same-price queue-decay was inert.
Six new unit tests cover each branch of the inference (pure cancel,
ask-shrank, bid-shrank, ask-px-up, bid-px-down, mixed both-sides).
34 ml-alpha lib tests + 33 ml-backtesting lib + 12 GPU fixtures + 3
fuzz still green.
Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
659 lines
29 KiB
Rust
659 lines
29 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 `seq_len`-sized windows of `Mbp10RawInput` plus 5-horizon 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;
|
||
use rand::{Rng, SeedableRng};
|
||
use rand_chacha::ChaCha8Rng;
|
||
|
||
use crate::cfc::snap_features::{Mbp10RawInput, ES_TICK_SIZE, REGIME_DIM};
|
||
use crate::multi_horizon_labels::generate_labels;
|
||
|
||
/// EMA smoothing factors. Half-life ≈ ln(2)/α snapshots.
|
||
const ALPHA_MED: f32 = 0.02; // half-life ≈ 35 snapshots ≈ 9s @ 250ms
|
||
const ALPHA_SLOW: f32 = 0.0005; // half-life ≈ 1400 snapshots ≈ 6 min
|
||
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_med = f32::NAN;
|
||
let mut ema_mid_slow = f32::NAN;
|
||
let mut ema_mid2_med = f32::NAN; // E[mid²] for var estimate
|
||
let mut ema_mid2_slow = 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_med.is_nan() {
|
||
ema_mid_med = mid; ema_mid_slow = mid;
|
||
ema_mid2_med = mid * mid; ema_mid2_slow = mid * mid;
|
||
ema_spread = spread_t; ema_trades = trades as f32;
|
||
} else {
|
||
ema_mid_med = (1.0 - ALPHA_MED) * ema_mid_med + ALPHA_MED * mid;
|
||
ema_mid_slow = (1.0 - ALPHA_SLOW) * ema_mid_slow + ALPHA_SLOW * mid;
|
||
ema_mid2_med = (1.0 - ALPHA_MED) * ema_mid2_med + ALPHA_MED * mid * mid;
|
||
ema_mid2_slow = (1.0 - ALPHA_SLOW) * ema_mid2_slow + ALPHA_SLOW * mid * mid;
|
||
ema_spread = (1.0 - ALPHA_MED) * ema_spread + ALPHA_MED * spread_t;
|
||
ema_trades = (1.0 - ALPHA_MED) * ema_trades + ALPHA_MED * trades as f32;
|
||
}
|
||
|
||
let var_med = (ema_mid2_med - ema_mid_med * ema_mid_med ).max(0.0);
|
||
let var_slow = (ema_mid2_slow - ema_mid_slow * ema_mid_slow).max(0.0);
|
||
let std_med = (var_med + REGIME_EPS).sqrt();
|
||
let std_slow = (var_slow + REGIME_EPS).sqrt();
|
||
|
||
out[k][0] = (mid - ema_mid_med) / std_med; // z-score med
|
||
out[k][1] = (mid - ema_mid_slow) / std_slow; // z-score slow
|
||
out[k][2] = (ema_mid_med - ema_mid_slow) / std_slow; // trend signal
|
||
out[k][3] = (1.0 + var_slow.sqrt() * 1e4).ln(); // log-compressed slow 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,
|
||
/// Number of consecutive snapshots per training sequence.
|
||
pub seq_len: usize,
|
||
/// Forward-horizon offsets in snapshots; spec mandates 5 of these.
|
||
pub horizons: [usize; 5],
|
||
/// 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,
|
||
/// Decision-stride sampling. With `decision_stride = S`, position k
|
||
/// of the yielded sequence reads snapshot `anchor + k * S` from the
|
||
/// source file. `S = 1` (default) preserves the original
|
||
/// consecutive-snapshot behaviour. `S > 1` expands the effective
|
||
/// time-window covered by each sequence (window = (seq_len - 1) * S
|
||
/// + 1 snapshots) at the same K-positions compute cost. The
|
||
/// per-sequence `Δt = ts_ns - prev_ts_ns` then reflects the actual
|
||
/// elapsed time between consecutive K-positions, which is what
|
||
/// downstream Mamba2 SSM dt_s + Phase 2C TGN Fourier features
|
||
/// consume. Labels stay in absolute-snapshot horizons (h=6000 is
|
||
/// always "predict 6000 snapshots forward" regardless of stride).
|
||
pub decision_stride: usize,
|
||
/// 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,
|
||
}
|
||
|
||
/// 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 {
|
||
/// `seq_len` consecutive raw snapshot inputs.
|
||
pub snapshots: Vec<Mbp10RawInput>,
|
||
/// `[5][seq_len]` 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>; 5],
|
||
}
|
||
|
||
/// 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 label vector for horizon
|
||
/// `cfg.horizons[h]`, indexed the same as `snapshots`. Sliced per
|
||
/// anchor on each `next_sequence()` call.
|
||
labels_full: [Vec<f32>; 5],
|
||
/// 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");
|
||
// Training requires `seq_len + max_horizon + 1` snapshots/file to compute
|
||
// forward labels; inference only needs `seq_len` (no forward window).
|
||
let min_size = if cfg.inference_only {
|
||
cfg.seq_len.max(1)
|
||
} else {
|
||
cfg.seq_len + 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)
|
||
.with_context(|| format!("load mbp10 {}", path.display()))?;
|
||
if snapshots.len() < min_size {
|
||
tracing::warn!(
|
||
path = %path.display(),
|
||
snapshots = snapshots.len(),
|
||
min_size,
|
||
"skipping file (too few snapshots for seq_len + max_horizon)",
|
||
);
|
||
continue;
|
||
}
|
||
let mut labels_full: [Vec<f32>; 5] = Default::default();
|
||
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;
|
||
}
|
||
}
|
||
let regime_full = compute_regime_features(&snapshots);
|
||
files_loaded.push(LoadedFile { snapshots, labels_full, 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 `CfcTrunk::capture_graph_a`. 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 and does
|
||
/// NOT apply `decision_stride` (caller's concern — see
|
||
/// `docs/superpowers/specs/2026-05-18-real-lob-integration-design.md` §7).
|
||
/// 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>> {
|
||
if self.yielded >= self.cfg.n_max_sequences {
|
||
return Ok(None);
|
||
}
|
||
let max_horizon = *self.cfg.horizons.iter().max().expect("non-empty horizons");
|
||
let stride = self.cfg.decision_stride.max(1);
|
||
|
||
// 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];
|
||
// With stride S, position k reads snapshot `anchor + k * S`; the
|
||
// last sequence index is `anchor + (seq_len-1) * S`, and the
|
||
// label there looks `max_horizon` forward in raw snapshots
|
||
// (labels stay in absolute-snapshot units regardless of stride).
|
||
let last_seq_offset = (self.cfg.seq_len - 1) * stride;
|
||
let needed = last_seq_offset + 1 + max_horizon;
|
||
anyhow::ensure!(
|
||
lf.snapshots.len() > needed,
|
||
"file too short for seq_len={} stride={} max_horizon={}: {} <= {}",
|
||
self.cfg.seq_len, stride, 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>; 5] = Default::default();
|
||
for h in 0..5 {
|
||
let mut row = Vec::with_capacity(self.cfg.seq_len);
|
||
for k in 0..self.cfg.seq_len {
|
||
row.push(lf.labels_full[h][anchor + k * stride]);
|
||
}
|
||
labels[h] = row;
|
||
}
|
||
let mut sequence = Vec::with_capacity(self.cfg.seq_len);
|
||
for k in 0..self.cfg.seq_len {
|
||
let idx = anchor + k * stride;
|
||
let cur = &lf.snapshots[idx];
|
||
// `prev` for microstructure features is the snapshot one
|
||
// K-position back in the strided sequence, NOT the
|
||
// consecutive-snapshot prior. This makes `Δt = ts_ns -
|
||
// prev_ts_ns` carry the actual elapsed time between
|
||
// consecutive K-positions — Mamba2's dt_s and the Phase 2C
|
||
// TGN Fourier features both consume this.
|
||
let prev_idx = if k == 0 {
|
||
// At k=0 there's no prior K-position; fall back to the
|
||
// immediately-prior snapshot (or self if anchor=0). Δt
|
||
// here will be ~tick-scale, but only at position 0.
|
||
if idx == 0 { 0 } else { idx - 1 }
|
||
} else {
|
||
anchor + (k - 1) * stride
|
||
};
|
||
let prev = &lf.snapshots[prev_idx];
|
||
sequence.push(convert(cur, prev, lf.regime_full[idx]));
|
||
}
|
||
|
||
self.yielded += 1;
|
||
Ok(Some(LabeledSequence { snapshots: sequence, labels }))
|
||
}
|
||
}
|
||
|
||
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(),
|
||
seq_len: 1,
|
||
horizons: [30, 100, 300, 1000, 6000],
|
||
n_max_sequences: 1,
|
||
seed: 0xCAFEF00D,
|
||
decision_stride: 1,
|
||
inference_only,
|
||
})
|
||
}
|
||
|
||
/// 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"),
|
||
seq_len: 1,
|
||
horizons: [30, 100, 300, 1000, 6000],
|
||
n_max_sequences: 1,
|
||
seed: 0,
|
||
decision_stride: 1,
|
||
inference_only: false,
|
||
};
|
||
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.
|
||
}
|
||
}
|