0e70bf96fe
BCE and aux losses were hardcoded to 0 in the GPU data path. Now step_batched_from_device returns (bce_loss, aux_loss), stored as last_bce_loss / last_aux_loss and fed into step stats + JSONL. Event-based sync replaces host-blocking stream sync: - perception: raw_event_record after training graph, raw_event_sync in read_deferred_stats (instant — graph completed long ago) - diag_staging: raw_event_record after snapshot_async, raw_event_sync in sync_and_swap (instant — copies completed during GPU work) All JSONL fields preserved. Loss values one-step deferred (same as before for Q/π/V, now also for BCE/aux). Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
1668 lines
74 KiB
Rust
1668 lines
74 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 cudarc::driver::CudaSlice;
|
||
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 rayon::prelude::*;
|
||
|
||
use crate::cfc::snap_features::{Mbp10RawInput, BOOK_LEVELS, ES_TICK_SIZE, REGIME_DIM};
|
||
use crate::heads::N_HORIZONS;
|
||
use crate::multi_horizon_labels::{generate_labels, generate_outcome_labels_ab};
|
||
|
||
/// Cached on-disk shape of per-file label + regime arrays. Sidecar lives
|
||
/// alongside the predecoded MBP-10 snapshot sidecar in `cfg.predecoded_dir`.
|
||
///
|
||
/// Cache key (encoded in the sidecar filename) includes (horizons,
|
||
/// outcome_label_cost, instrument_filter) because labels depend on these.
|
||
/// The existing predecoded-MBP-10 sidecar's `instrument_filter` part of
|
||
/// its filename plus the horizons/cost suffix here give a complete cache
|
||
/// key.
|
||
///
|
||
/// Refilled on schema-mismatch errors (deserialize failure) — the bincode
|
||
/// payload's struct shape is itself the schema check. The load path also
|
||
/// checks `regime_full.len() == snapshots.len()` as a defensive length
|
||
/// guard against silently-stale caches (the underlying MBP-10 sidecar is
|
||
/// invalidated by mtime+size mismatch in `ml_features::predecoded`, but
|
||
/// that doesn't propagate to a label-length check here without this
|
||
/// guard).
|
||
#[derive(serde::Serialize, serde::Deserialize)]
|
||
struct CachedLabels {
|
||
labels_full: [Vec<f32>; N_HORIZONS],
|
||
outcome_prof_long_full: [Vec<f32>; N_HORIZONS],
|
||
outcome_prof_short_full: [Vec<f32>; N_HORIZONS],
|
||
outcome_size_long_full: [Vec<f32>; N_HORIZONS],
|
||
outcome_size_short_full: [Vec<f32>; N_HORIZONS],
|
||
sigma_k_full: [Vec<f32>; N_HORIZONS],
|
||
pos_fraction: Vec<f32>,
|
||
regime_full: Vec<[f32; REGIME_DIM]>,
|
||
}
|
||
|
||
/// Cache-key suffix for the labels sidecar. Encodes (horizons, cost,
|
||
/// instrument_filter) so changes to any of these properties invalidate
|
||
/// the cache. Format: `labels_h<H0>_<H1>_<H2>_cost<HEX>_<FILTER>`.
|
||
/// The cost is encoded as the f32's raw bit pattern in lowercase hex to
|
||
/// keep '.' out of the filename (portable across filesystems) and to
|
||
/// preserve exact-float identity (0.5 != 0.50000001 → different bits →
|
||
/// different cache key).
|
||
fn labels_cache_suffix(
|
||
horizons: &[usize; N_HORIZONS],
|
||
cost_price_units: f32,
|
||
filter: InstrumentFilter,
|
||
) -> String {
|
||
let cost_bits = cost_price_units.to_bits();
|
||
let filter_tag = match filter {
|
||
InstrumentFilter::All => String::from("all"),
|
||
InstrumentFilter::Id(id) => format!("instr{id}"),
|
||
InstrumentFilter::FrontMonth => String::from("front_month"),
|
||
};
|
||
format!(
|
||
"labels_h{}_{}_{}_cost{:08x}_{}",
|
||
horizons[0], horizons[1], horizons[2], cost_bits, filter_tag
|
||
)
|
||
}
|
||
|
||
/// Path to the labels-cache sidecar for `source` under `predecoded_dir`.
|
||
/// Layout: `<predecoded_dir>/<source_stem>.<suffix>.predecoded.bin`.
|
||
fn labels_cache_path(
|
||
source: &std::path::Path,
|
||
predecoded_dir: &std::path::Path,
|
||
horizons: &[usize; N_HORIZONS],
|
||
cost_price_units: f32,
|
||
filter: InstrumentFilter,
|
||
) -> PathBuf {
|
||
let suffix = labels_cache_suffix(horizons, cost_price_units, filter);
|
||
let stem = source
|
||
.file_name()
|
||
.and_then(|s| s.to_str())
|
||
.unwrap_or("unknown");
|
||
predecoded_dir.join(format!("{stem}.{suffix}.predecoded.bin"))
|
||
}
|
||
|
||
/// Try to read cached labels. Returns `Ok(Some(...))` on hit,
|
||
/// `Ok(None)` on miss (file does not exist), and `Err` only on IO /
|
||
/// decode errors. Callers treat `Err` the same as a miss (recompute)
|
||
/// but log a warning.
|
||
fn try_read_labels_cache(path: &std::path::Path) -> Result<Option<CachedLabels>> {
|
||
use std::fs::File;
|
||
use std::io::BufReader;
|
||
if !path.exists() {
|
||
return Ok(None);
|
||
}
|
||
let f = File::open(path).with_context(|| format!("open labels cache {}", path.display()))?;
|
||
let reader = BufReader::new(f);
|
||
let cached: CachedLabels = bincode::deserialize_from(reader)
|
||
.with_context(|| format!("decode labels cache {}", path.display()))?;
|
||
Ok(Some(cached))
|
||
}
|
||
|
||
/// Write labels cache to disk. Best-effort: failures are logged but don't
|
||
/// halt the training run.
|
||
fn write_labels_cache(path: &std::path::Path, cached: &CachedLabels) -> Result<()> {
|
||
use std::fs::File;
|
||
use std::io::BufWriter;
|
||
let f = File::create(path).with_context(|| format!("create labels cache {}", path.display()))?;
|
||
let writer = BufWriter::new(f);
|
||
bincode::serialize_into(writer, cached)
|
||
.with_context(|| format!("encode labels cache {}", path.display()))?;
|
||
Ok(())
|
||
}
|
||
|
||
/// 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,
|
||
|
||
/// SP20 P3 F.5 — FRD head forward-return horizons in tick units.
|
||
/// Determines `compute_frd_labels`'s lookahead distances. Default
|
||
/// is `crate::rl::common::FRD_HORIZON_TICKS = [60, 300, 1800]` —
|
||
/// the spec-recommended microstructure / minute / hour timescales.
|
||
/// The trainer-side ISV slots (`RL_FRD_HORIZON_{1,2,3}_TICKS_INDEX`,
|
||
/// slots 500/501/502) seed the SAME defaults so loader + trainer
|
||
/// agree out of the box. Override here to retune both the
|
||
/// supervised labels AND the ISV seeds (caller is responsible for
|
||
/// the latter via `RL_FRD_HORIZON_*` writes if they want runtime
|
||
/// alignment).
|
||
pub frd_horizon_ticks: [usize; crate::rl::common::FRD_N_HORIZONS],
|
||
|
||
/// SP20 P3 F.5 — FRD bucket-range in σ units. The atom span is
|
||
/// `[-range_σ, +range_σ]` mapped onto `FRD_N_ATOMS=21` atoms.
|
||
/// Default = `crate::rl::common::FRD_BUCKET_RANGE_SIGMA = 3.0`
|
||
/// (matches the ISV seed at `RL_FRD_BUCKET_RANGE_SIGMA_INDEX`
|
||
/// slot 503). Per `pearl_glm_fitter_link_must_match_inference`:
|
||
/// the label generator AND the trainer-side ISV-driven runtime
|
||
/// MUST stay aligned — caller is responsible for matching ISV
|
||
/// slot 503 if overriding here.
|
||
pub frd_bucket_range_sigma: f32,
|
||
}
|
||
|
||
/// 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],
|
||
/// SP20 P3 F.5 — per-(horizon, snapshot) FRD return-bucket label
|
||
/// indices `[0, FRD_N_ATOMS)`. Sentinel `-1` at the rightmost edge
|
||
/// where `snapshot_idx + h_ticks >= snapshots.len()`. Length per
|
||
/// horizon equals `snapshots.len()`. Caller writes the row at
|
||
/// the anchor index into `trainer.frd_labels_d` before each
|
||
/// `step_with_lobsim`. See `compute_frd_labels` for the bucket
|
||
/// math; the trainer-side softmax+CE interpretation MUST match
|
||
/// (same `FRD_BUCKET_RANGE_SIGMA` const).
|
||
pub frd_labels: [Vec<i32>; crate::rl::common::FRD_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],
|
||
/// SP20 P3 F.5 — full-file FRD per-horizon bucket-index labels.
|
||
/// See `LabeledSequence::frd_labels` doc for shape + semantics.
|
||
/// Sliced per anchor on each `next_sequence()` call.
|
||
frd_labels_full: [Vec<i32>; crate::rl::common::FRD_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
|
||
};
|
||
// SPEED-A (2026-05-22): parallel per-file load + label generation.
|
||
// Each file's load_or_predecode + generate_labels + generate_outcome_labels_ab +
|
||
// compute_regime_features is pure CPU work over disjoint inputs (snapshots,
|
||
// cfg.horizons, cfg.outcome_label_cost). par_iter gives ~4-8x speedup on
|
||
// typical 4-8 core machines. Files emitting a "too few snapshots" warning
|
||
// are filtered out post-collection (via Option<LoadedFile>).
|
||
let load_one = |path: &PathBuf| -> Result<Option<LoadedFile>> {
|
||
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)",
|
||
);
|
||
return Ok(None);
|
||
}
|
||
// SPEED-B (2026-05-22): on-disk cache for the per-file label +
|
||
// regime arrays so subsequent training runs skip the multi-second
|
||
// label-generation pass. Cache key encodes (horizons, cost,
|
||
// instrument_filter). Inference-only runs skip the cache WRITE
|
||
// (they don't compute labels) but DO honor a hit if present —
|
||
// the cached arrays would just go unused in that mode.
|
||
let labels_cache = labels_cache_path(
|
||
path,
|
||
&cfg.predecoded_dir,
|
||
&cfg.horizons,
|
||
cfg.outcome_label_cost,
|
||
cfg.instrument_filter,
|
||
);
|
||
let cached = match try_read_labels_cache(&labels_cache) {
|
||
Ok(Some(c)) if c.regime_full.len() == snapshots.len() => {
|
||
tracing::info!(
|
||
path = %labels_cache.display(),
|
||
n_snapshots = snapshots.len(),
|
||
"labels cache HIT",
|
||
);
|
||
Some(c)
|
||
}
|
||
Ok(Some(c)) => {
|
||
// Stale length — the underlying MBP-10 sidecar was
|
||
// refreshed (re-downloaded quarter) but the labels cache
|
||
// wasn't. Treat as miss, recompute, overwrite.
|
||
tracing::warn!(
|
||
path = %labels_cache.display(),
|
||
cached_n = c.regime_full.len(),
|
||
actual_n = snapshots.len(),
|
||
"labels cache length mismatch — recomputing",
|
||
);
|
||
None
|
||
}
|
||
Ok(None) => None,
|
||
Err(e) => {
|
||
tracing::warn!(
|
||
path = %labels_cache.display(),
|
||
err = %e,
|
||
"labels cache read failed — recomputing",
|
||
);
|
||
None
|
||
}
|
||
};
|
||
|
||
let (
|
||
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,
|
||
) = if let Some(c) = cached {
|
||
(
|
||
c.labels_full,
|
||
c.outcome_prof_long_full,
|
||
c.outcome_prof_short_full,
|
||
c.outcome_size_long_full,
|
||
c.outcome_size_short_full,
|
||
c.sigma_k_full,
|
||
c.pos_fraction,
|
||
c.regime_full,
|
||
)
|
||
} else {
|
||
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);
|
||
|
||
// Best-effort write to cache for next run. Skipped when
|
||
// inference_only since labels_full/outcome_* are all-default
|
||
// and a stored cache there would mis-serve a later
|
||
// non-inference run that hit it.
|
||
if !cfg.inference_only {
|
||
let payload = CachedLabels {
|
||
labels_full: labels_full.clone(),
|
||
outcome_prof_long_full: outcome_prof_long_full.clone(),
|
||
outcome_prof_short_full: outcome_prof_short_full.clone(),
|
||
outcome_size_long_full: outcome_size_long_full.clone(),
|
||
outcome_size_short_full: outcome_size_short_full.clone(),
|
||
sigma_k_full: sigma_k_full.clone(),
|
||
pos_fraction: pos_fraction.clone(),
|
||
regime_full: regime_full.clone(),
|
||
};
|
||
if let Err(e) = write_labels_cache(&labels_cache, &payload) {
|
||
tracing::warn!(
|
||
path = %labels_cache.display(),
|
||
err = %e,
|
||
"labels cache write failed (training continues)",
|
||
);
|
||
} else {
|
||
tracing::info!(
|
||
path = %labels_cache.display(),
|
||
n_snapshots = snapshots.len(),
|
||
"labels cache WRITE",
|
||
);
|
||
}
|
||
}
|
||
|
||
(
|
||
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,
|
||
)
|
||
};
|
||
let frd_labels_full = compute_frd_labels(
|
||
&snapshots,
|
||
cfg.frd_horizon_ticks,
|
||
cfg.frd_bucket_range_sigma,
|
||
);
|
||
Ok(Some(LoadedFile {
|
||
snapshots,
|
||
labels_full,
|
||
outcome_prof_long_full,
|
||
outcome_prof_short_full,
|
||
outcome_size_long_full,
|
||
outcome_size_short_full,
|
||
sigma_k_full,
|
||
frd_labels_full,
|
||
pos_fraction,
|
||
regime_full,
|
||
}))
|
||
};
|
||
|
||
let files_loaded_results: Vec<Result<Option<LoadedFile>>> =
|
||
files.par_iter().map(load_one).collect();
|
||
let mut files_loaded: Vec<LoadedFile> = Vec::with_capacity(files.len());
|
||
for result in files_loaded_results {
|
||
match result? {
|
||
Some(lf) => files_loaded.push(lf),
|
||
None => {} // Already warned in load_one.
|
||
}
|
||
}
|
||
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]))
|
||
}
|
||
|
||
/// Upload ALL pre-converted snapshots + FRD labels to GPU as a
|
||
/// single flat buffer. Returns a [`GpuDataset`] that the
|
||
/// [`GpuDataLoader`] kernel reads from each step — zero CPU work
|
||
/// in the training hot path.
|
||
///
|
||
/// This method is the CPU-side half of the GPU-resident data loader:
|
||
/// 1. Pre-converts every `Mbp10Snapshot` → `Mbp10RawInput` (calls
|
||
/// `convert()` once per snapshot at init, not per step).
|
||
/// 2. Concatenates all files into flat arrays.
|
||
/// 3. Uploads via `htod_copy` (one-time bulk transfer).
|
||
/// 4. Builds per-file offset + size tables on device.
|
||
/// 5. Initializes per-batch PRNG seeds on device.
|
||
///
|
||
/// Per `feedback_no_htod_htoh_only_mapped_pinned.md`: the init-time
|
||
/// `htod_copy` is a one-shot outside the captured-graph hot path.
|
||
/// The per-step path is pure device→device (kernel reads dataset
|
||
/// buffers, writes SoA buffers).
|
||
///
|
||
/// [`GpuDataset`]: super::gpu_dataset::GpuDataset
|
||
/// [`GpuDataLoader`]: super::gpu_dataset::GpuDataLoader
|
||
pub fn upload_to_gpu(
|
||
&self,
|
||
dev: &ml_core::device::MlDevice,
|
||
batch_size: usize,
|
||
seed: u64,
|
||
) -> Result<super::gpu_dataset::GpuDataset> {
|
||
use crate::cfc::snap_features::MBP10_RAW_INPUT_BYTES;
|
||
use crate::rl::common::FRD_N_HORIZONS;
|
||
|
||
let stream = dev.cuda_stream().context("upload_to_gpu: CUDA stream")?.clone();
|
||
|
||
// ── 1. Pre-convert ALL snapshots → Mbp10RawInput + build offsets ──
|
||
let n_files = self.files_loaded.len();
|
||
let total_snaps: usize = self.files_loaded.iter().map(|f| f.snapshots.len()).sum();
|
||
tracing::info!(
|
||
n_files,
|
||
total_snaps,
|
||
bytes = total_snaps * MBP10_RAW_INPUT_BYTES,
|
||
"upload_to_gpu: pre-converting snapshots",
|
||
);
|
||
|
||
let mut file_offsets: Vec<i32> = Vec::with_capacity(n_files);
|
||
let mut file_sizes: Vec<i32> = Vec::with_capacity(n_files);
|
||
let mut offset: usize = 0;
|
||
for lf in &self.files_loaded {
|
||
file_offsets.push(offset as i32);
|
||
file_sizes.push(lf.snapshots.len() as i32);
|
||
offset += lf.snapshots.len();
|
||
}
|
||
|
||
// ── 2. Upload snapshots per-file (streaming) ────────────────────
|
||
// Allocate the full GPU buffer upfront, then convert + upload one
|
||
// file at a time. Peak host memory = one file's worth of converted
|
||
// snapshots (~1-2GB) instead of all files (~10GB).
|
||
let total_bytes = total_snaps * MBP10_RAW_INPUT_BYTES;
|
||
let snapshots_d = stream
|
||
.alloc_zeros::<u8>(total_bytes)
|
||
.context("upload_to_gpu: snapshots alloc")?;
|
||
let mut gpu_offset_bytes: usize = 0;
|
||
for (f_idx, lf) in self.files_loaded.iter().enumerate() {
|
||
let n = lf.snapshots.len();
|
||
let mut file_raws: Vec<Mbp10RawInput> = Vec::with_capacity(n);
|
||
for (idx, snap) in lf.snapshots.iter().enumerate() {
|
||
let prev = if idx == 0 { snap } else { &lf.snapshots[idx - 1] };
|
||
file_raws.push(convert(snap, prev, lf.regime_full[idx]));
|
||
}
|
||
let file_bytes = n * MBP10_RAW_INPUT_BYTES;
|
||
unsafe {
|
||
cudarc::driver::sys::cuMemcpyHtoD_v2(
|
||
snapshots_d.raw_ptr() + gpu_offset_bytes as u64,
|
||
file_raws.as_ptr() as *const std::ffi::c_void,
|
||
file_bytes,
|
||
);
|
||
}
|
||
gpu_offset_bytes += file_bytes;
|
||
tracing::info!(
|
||
file = f_idx,
|
||
n_snapshots = n,
|
||
gpu_offset_mb = gpu_offset_bytes / (1024 * 1024),
|
||
"upload_to_gpu: file uploaded",
|
||
);
|
||
drop(file_raws);
|
||
}
|
||
tracing::info!(
|
||
gpu_bytes = total_bytes,
|
||
"upload_to_gpu: snapshots uploaded",
|
||
);
|
||
|
||
// ── 3. Upload file offsets + sizes ───────────────────────────────
|
||
let file_offsets_d = stream
|
||
.alloc_zeros::<i32>(n_files)
|
||
.context("upload_to_gpu: file_offsets alloc")?;
|
||
unsafe {
|
||
cudarc::driver::sys::cuMemcpyHtoD_v2(
|
||
file_offsets_d.raw_ptr(),
|
||
file_offsets.as_ptr() as *const std::ffi::c_void,
|
||
n_files * 4,
|
||
);
|
||
}
|
||
let file_sizes_d = stream
|
||
.alloc_zeros::<i32>(n_files)
|
||
.context("upload_to_gpu: file_sizes alloc")?;
|
||
unsafe {
|
||
cudarc::driver::sys::cuMemcpyHtoD_v2(
|
||
file_sizes_d.raw_ptr(),
|
||
file_sizes.as_ptr() as *const std::ffi::c_void,
|
||
n_files * 4,
|
||
);
|
||
}
|
||
|
||
// ── 4. Upload FRD labels [FRD_N_HORIZONS, total_snaps] row-major ──
|
||
let frd_total = FRD_N_HORIZONS * total_snaps;
|
||
let mut frd_flat: Vec<i32> = vec![-1_i32; frd_total];
|
||
let mut global_offset: usize = 0;
|
||
for lf in &self.files_loaded {
|
||
let n = lf.snapshots.len();
|
||
for h in 0..FRD_N_HORIZONS {
|
||
let src = &lf.frd_labels_full[h];
|
||
let dst_start = h * total_snaps + global_offset;
|
||
frd_flat[dst_start..dst_start + n].copy_from_slice(&src[..n]);
|
||
}
|
||
global_offset += n;
|
||
}
|
||
let frd_labels_d = stream
|
||
.alloc_zeros::<i32>(frd_total)
|
||
.context("upload_to_gpu: frd_labels alloc")?;
|
||
unsafe {
|
||
cudarc::driver::sys::cuMemcpyHtoD_v2(
|
||
frd_labels_d.raw_ptr(),
|
||
frd_flat.as_ptr() as *const std::ffi::c_void,
|
||
frd_total * 4,
|
||
);
|
||
}
|
||
|
||
// ── 5. Upload supervised BCE labels [N_HORIZONS, total_snaps] row-major ──
|
||
//
|
||
// Same row-major layout as FRD labels: `all_labels[h * total_snaps + snap_idx]`.
|
||
// Six float arrays: labels, outcome_prof_{long,short}, outcome_size_{long,short},
|
||
// sigma_k. Plus per-file pos_fraction [n_files, 2 * N_HORIZONS].
|
||
let label_total = N_HORIZONS * total_snaps;
|
||
let upload_f32_labels = |name: &str, accessor: &dyn Fn(&LoadedFile, usize) -> &Vec<f32>| -> Result<CudaSlice<f32>> {
|
||
let mut flat: Vec<f32> = vec![f32::NAN; label_total];
|
||
let mut g_off: usize = 0;
|
||
for lf in &self.files_loaded {
|
||
let n = lf.snapshots.len();
|
||
for h in 0..N_HORIZONS {
|
||
let src = accessor(lf, h);
|
||
let dst_start = h * total_snaps + g_off;
|
||
flat[dst_start..dst_start + n].copy_from_slice(&src[..n]);
|
||
}
|
||
g_off += n;
|
||
}
|
||
let buf = stream
|
||
.alloc_zeros::<f32>(label_total)
|
||
.with_context(|| format!("upload_to_gpu: {name} alloc"))?;
|
||
unsafe {
|
||
cudarc::driver::sys::cuMemcpyHtoD_v2(
|
||
buf.raw_ptr(),
|
||
flat.as_ptr() as *const std::ffi::c_void,
|
||
label_total * 4,
|
||
);
|
||
}
|
||
Ok(buf)
|
||
};
|
||
|
||
let labels_d = upload_f32_labels("labels", &|lf, h| &lf.labels_full[h])?;
|
||
let outcome_prof_long_d = upload_f32_labels("outcome_prof_long", &|lf, h| &lf.outcome_prof_long_full[h])?;
|
||
let outcome_prof_short_d = upload_f32_labels("outcome_prof_short", &|lf, h| &lf.outcome_prof_short_full[h])?;
|
||
let outcome_size_long_d = upload_f32_labels("outcome_size_long", &|lf, h| &lf.outcome_size_long_full[h])?;
|
||
let outcome_size_short_d = upload_f32_labels("outcome_size_short", &|lf, h| &lf.outcome_size_short_full[h])?;
|
||
let sigma_k_d = upload_f32_labels("sigma_k", &|lf, h| &lf.sigma_k_full[h])?;
|
||
|
||
// Per-file pos_fraction: [n_files, 2 * N_HORIZONS].
|
||
let pf_total = n_files * 2 * N_HORIZONS;
|
||
let mut pf_flat: Vec<f32> = vec![0.0_f32; pf_total];
|
||
for (f_idx, lf) in self.files_loaded.iter().enumerate() {
|
||
let base = f_idx * 2 * N_HORIZONS;
|
||
let src = &lf.pos_fraction;
|
||
pf_flat[base..base + src.len()].copy_from_slice(src);
|
||
}
|
||
let pos_fraction_d = stream
|
||
.alloc_zeros::<f32>(pf_total)
|
||
.context("upload_to_gpu: pos_fraction alloc")?;
|
||
unsafe {
|
||
cudarc::driver::sys::cuMemcpyHtoD_v2(
|
||
pos_fraction_d.raw_ptr(),
|
||
pf_flat.as_ptr() as *const std::ffi::c_void,
|
||
pf_total * 4,
|
||
);
|
||
}
|
||
|
||
tracing::info!(
|
||
label_total,
|
||
pf_total,
|
||
"upload_to_gpu: supervised labels uploaded",
|
||
);
|
||
|
||
// ── 6. Initialize per-batch PRNG seeds ──────────────────────────
|
||
let prng_seeds: Vec<u32> = (0..batch_size as u64)
|
||
.map(|b| {
|
||
// Mix seed + batch index to get distinct per-batch seeds.
|
||
// Ensures different batch elements sample different files/anchors.
|
||
let mixed = seed.wrapping_mul(2654435761).wrapping_add(b);
|
||
(mixed & 0xFFFF_FFFF) as u32
|
||
})
|
||
.collect();
|
||
let prng_d = stream
|
||
.alloc_zeros::<u32>(batch_size)
|
||
.context("upload_to_gpu: prng alloc")?;
|
||
unsafe {
|
||
cudarc::driver::sys::cuMemcpyHtoD_v2(
|
||
prng_d.raw_ptr(),
|
||
prng_seeds.as_ptr() as *const std::ffi::c_void,
|
||
batch_size * 4,
|
||
);
|
||
}
|
||
|
||
let max_horizon = *self.cfg.horizons.iter().max().expect("non-empty horizons");
|
||
|
||
tracing::info!(
|
||
n_files,
|
||
total_snaps,
|
||
max_horizon,
|
||
batch_size,
|
||
"upload_to_gpu: dataset ready",
|
||
);
|
||
|
||
Ok(super::gpu_dataset::GpuDataset {
|
||
snapshots_d,
|
||
file_offsets_d,
|
||
file_sizes_d,
|
||
frd_labels_d,
|
||
prng_d,
|
||
labels_d,
|
||
outcome_prof_long_d,
|
||
outcome_prof_short_d,
|
||
outcome_size_long_d,
|
||
outcome_size_short_d,
|
||
sigma_k_d,
|
||
pos_fraction_d,
|
||
n_files,
|
||
total_snapshots: total_snaps,
|
||
max_horizon,
|
||
})
|
||
}
|
||
|
||
/// 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()
|
||
}
|
||
|
||
/// Phase R2 (was G fix): return a `(s_t, s_{t+1})` pair of
|
||
/// consecutive sequences at adjacent anchors in the same file.
|
||
/// The first sequence ends one snapshot earlier than the second —
|
||
/// the canonical RL Bellman-target requirement that
|
||
/// `IntegratedTrainer::step_with_lobsim` consumes as
|
||
/// `(snapshots, next_snapshots)`.
|
||
///
|
||
/// Both yielded sequences come from the SAME random file so the
|
||
/// underlying market regime / time-of-day context is consistent.
|
||
/// The anchor for `s_t` is sampled uniformly from `[min_anchor,
|
||
/// max_anchor − 1)` so `anchor + 1` also satisfies the
|
||
/// upper-bound constraint. Counts as ONE yielded sequence against
|
||
/// `n_max_sequences` (the loader views the pair as one training
|
||
/// step's worth of input).
|
||
pub fn next_sequence_pair(
|
||
&mut self,
|
||
) -> Result<Option<(LabeledSequence, LabeledSequence)>> {
|
||
if self.yielded >= self.cfg.n_max_sequences {
|
||
return Ok(None);
|
||
}
|
||
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 max_horizon = *self.cfg.horizons.iter().max().expect("non-empty horizons");
|
||
let total = self.cfg.multi_resolution.total_positions();
|
||
let lookback = self.cfg.multi_resolution.required_lookback_ticks();
|
||
let min_anchor = lookback.saturating_sub(total);
|
||
// Reserve one extra event past the standard max_anchor so
|
||
// `anchor + 1`'s window also fits within the file.
|
||
let max_anchor = lf
|
||
.snapshots
|
||
.len()
|
||
.saturating_sub(total + max_horizon + 1);
|
||
anyhow::ensure!(
|
||
max_anchor > min_anchor,
|
||
"file too short for paired-window sampling (total={total}, lookback={lookback}, max_horizon={max_horizon}): \
|
||
snapshots.len()={} cannot span [{min_anchor}, {max_anchor})",
|
||
lf.snapshots.len()
|
||
);
|
||
let anchor_t: usize = self.rng.gen_range(min_anchor..max_anchor);
|
||
|
||
let s_t = self.build_sequence_at(lf, anchor_t)?;
|
||
let s_tp1 = self.build_sequence_at(lf, anchor_t + 1)?;
|
||
|
||
self.yielded += 1;
|
||
Ok(Some((s_t, s_tp1)))
|
||
}
|
||
|
||
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];
|
||
// Multi-resolution anchor sampling has TWO bounds:
|
||
// * Upper: anchor + total + max_horizon < snapshots.len()
|
||
// (room for the forward-label window after the input edge)
|
||
// * Lower: anchor + total >= lookback
|
||
// (room to look back across coarsest aggregation scale)
|
||
// For the default 3-scale config (total=32, lookback=1510) the lower
|
||
// bound is binding: anchor must be ≥ lookback - total = 1478.
|
||
let total = self.cfg.multi_resolution.total_positions();
|
||
let lookback = self.cfg.multi_resolution.required_lookback_ticks();
|
||
let min_anchor = lookback.saturating_sub(total);
|
||
let max_anchor = lf.snapshots.len().saturating_sub(total + max_horizon);
|
||
anyhow::ensure!(
|
||
max_anchor > min_anchor,
|
||
"file too short for multi-resolution(total={total}, lookback={lookback}) max_horizon={max_horizon}: \
|
||
snapshots.len()={} cannot span [{min_anchor}, {max_anchor})",
|
||
lf.snapshots.len()
|
||
);
|
||
let anchor: usize = self.rng.gen_range(min_anchor..max_anchor);
|
||
|
||
let seq = self.build_sequence_at(lf, anchor)?;
|
||
self.yielded += 1;
|
||
Ok(Some(seq))
|
||
}
|
||
|
||
/// Construct a `LabeledSequence` at a specific `(file, anchor)`
|
||
/// pair. Pure function of the loader config + file contents; does
|
||
/// not advance the loader's `yielded` counter (callers update it).
|
||
///
|
||
/// Phase R2: this helper exists so `next_sequence_random` (random
|
||
/// anchor) and `next_sequence_pair` (paired adjacent anchors)
|
||
/// share the multi-resolution windowing logic verbatim per
|
||
/// `feedback_single_source_of_truth_no_duplicates`.
|
||
fn build_sequence_at(
|
||
&self,
|
||
lf: &LoadedFile,
|
||
anchor: usize,
|
||
) -> Result<LabeledSequence> {
|
||
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);
|
||
|
||
// SP20 P3 F.5 — FRD labels for THIS sequence. The encoder
|
||
// forward emits h_t aligned to the newest snapshot (= the
|
||
// last position in `sequence` = file index `edge - 1`).
|
||
// That single per-horizon bucket index becomes the
|
||
// supervised target the trainer writes into frd_labels_d
|
||
// for the current batch slot.
|
||
let newest_idx = edge - 1;
|
||
let mut frd_labels: [Vec<i32>; crate::rl::common::FRD_N_HORIZONS] = Default::default();
|
||
for h in 0..crate::rl::common::FRD_N_HORIZONS {
|
||
// Yield a length-1 vec per horizon (the trainer
|
||
// consumes labels[K-1] only — FRD supervises h_t at
|
||
// the rightmost K position, not the full K window).
|
||
let v = lf
|
||
.frd_labels_full
|
||
.get(h)
|
||
.and_then(|col| col.get(newest_idx).copied())
|
||
.unwrap_or(-1);
|
||
frd_labels[h] = vec![v];
|
||
}
|
||
|
||
Ok(LabeledSequence {
|
||
snapshots: sequence,
|
||
labels,
|
||
outcome_prof_long,
|
||
outcome_prof_short,
|
||
outcome_size_long,
|
||
outcome_size_short,
|
||
sigma_k,
|
||
frd_labels,
|
||
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
|
||
}
|
||
|
||
/// SP20 P3 F.5 — per-file FRD label generation.
|
||
///
|
||
/// For each snapshot index `i` and each FRD horizon `h_ticks ∈
|
||
/// FRD_HORIZON_TICKS = [60, 300, 1800]`, compute the σ-normalized
|
||
/// forward return:
|
||
///
|
||
/// ```text
|
||
/// r = (mid[i + h_ticks] - mid[i]) / sigma_per_step
|
||
/// ```
|
||
///
|
||
/// where `sigma_per_step` is the file-level sample std of single-
|
||
/// tick mid increments. Bucket `r` into `FRD_N_ATOMS = 21` atoms
|
||
/// uniform over `[-FRD_BUCKET_RANGE_SIGMA, +FRD_BUCKET_RANGE_SIGMA]`
|
||
/// (default ±3σ) — clip to `[0, FRD_N_ATOMS-1]` for tail returns.
|
||
///
|
||
/// Sentinel `-1` marks `i + h_ticks >= snapshots.len()` (the
|
||
/// forward return isn't realized at the rightmost edge of the
|
||
/// file) — the `rl_frd_softmax_ce_grad` kernel zeros loss + grad
|
||
/// for those rows.
|
||
///
|
||
/// Output shape: `[FRD_N_HORIZONS][snapshots.len()]` row-major
|
||
/// per horizon. The caller slices into this at the anchor index
|
||
/// when yielding a sequence (same pattern as the BCE labels).
|
||
///
|
||
/// Per `pearl_glm_fitter_link_must_match_inference`: the bucket
|
||
/// edges here MUST match the trainer-side FRD head's softmax + CE
|
||
/// interpretation (atom `a` = bucket center `(a / (N-1) × 2 - 1) ×
|
||
/// range_σ`). The `range_sigma` argument and trainer-side ISV slot
|
||
/// 503 MUST stay aligned — caller is responsible for keeping them
|
||
/// in sync (default path: both come from `FRD_BUCKET_RANGE_SIGMA`).
|
||
///
|
||
/// Per `feedback_isv_for_adaptive_bounds`: horizons + range come in
|
||
/// as parameters (sourced from `MultiHorizonLoaderConfig`, defaults
|
||
/// from the canonical consts) — NOT hardcoded literals in the
|
||
/// function body. Single source of truth: `rl::common`.
|
||
fn compute_frd_labels(
|
||
snapshots: &[Mbp10Snapshot],
|
||
horizon_ticks: [usize; crate::rl::common::FRD_N_HORIZONS],
|
||
range_sigma: f32,
|
||
) -> [Vec<i32>; crate::rl::common::FRD_N_HORIZONS] {
|
||
use crate::rl::common::{FRD_N_ATOMS, FRD_N_HORIZONS};
|
||
|
||
let n = snapshots.len();
|
||
let mids: Vec<f32> = snapshots.iter().map(mid_price_f32).collect();
|
||
|
||
// File-level sample std of single-tick mid increments. Bootstraps
|
||
// off the loaded file alone (no warmup blackout — for the very
|
||
// first file this is the only signal). At ~3M ticks per file the
|
||
// sample is statistically reliable.
|
||
let sigma_per_step: f32 = {
|
||
if mids.len() < 2 {
|
||
// Pathological 1-snapshot file — sentinel everything.
|
||
1.0
|
||
} else {
|
||
let mut sum_sq = 0.0_f64;
|
||
let mut cnt = 0_usize;
|
||
for i in 1..mids.len() {
|
||
let d = (mids[i] - mids[i - 1]) as f64;
|
||
if d.is_finite() {
|
||
sum_sq += d * d;
|
||
cnt += 1;
|
||
}
|
||
}
|
||
if cnt == 0 {
|
||
1.0
|
||
} else {
|
||
((sum_sq / cnt as f64).sqrt() as f32).max(1e-6)
|
||
}
|
||
}
|
||
};
|
||
|
||
let mut out: [Vec<i32>; FRD_N_HORIZONS] = Default::default();
|
||
let n_atoms_f = FRD_N_ATOMS as f32;
|
||
let half = (FRD_N_ATOMS - 1) as f32 / 2.0;
|
||
let scale = half / range_sigma.max(1e-6);
|
||
|
||
for (h_idx, &h_ticks) in horizon_ticks.iter().enumerate() {
|
||
// sqrt(h_ticks) Brownian scaling — variance of an h-step
|
||
// sum of iid increments is h × per-step variance, so std
|
||
// scales with sqrt(h).
|
||
let sigma_h = sigma_per_step * (h_ticks as f32).sqrt();
|
||
let mut labels = vec![-1_i32; n];
|
||
if h_ticks < n {
|
||
for i in 0..(n - h_ticks) {
|
||
let r = (mids[i + h_ticks] - mids[i]) / sigma_h.max(1e-6);
|
||
// Bucket center mapping: atom `a` → center `(a - half) / scale`.
|
||
// Inverse: `a = round(r × scale + half)`.
|
||
let a_raw = (r * scale + half).round() as i32;
|
||
let a_clamped = a_raw.clamp(0, (FRD_N_ATOMS - 1) as i32);
|
||
labels[i] = a_clamped;
|
||
}
|
||
}
|
||
// Final safety — guard against pathological NaN propagation.
|
||
for v in labels.iter_mut() {
|
||
if *v < -1 || *v >= n_atoms_f as i32 {
|
||
*v = -1;
|
||
}
|
||
}
|
||
out[h_idx] = labels;
|
||
}
|
||
out
|
||
}
|
||
|
||
fn convert(
|
||
cur: &Mbp10Snapshot,
|
||
prev: &Mbp10Snapshot,
|
||
regime: [f32; REGIME_DIM],
|
||
) -> Mbp10RawInput {
|
||
let mut bid_px = [0.0f32; BOOK_LEVELS];
|
||
let mut bid_sz = [0.0f32; BOOK_LEVELS];
|
||
let mut ask_px = [0.0f32; BOOK_LEVELS];
|
||
let mut ask_sz = [0.0f32; BOOK_LEVELS];
|
||
for (i, lvl) in cur.levels.iter().take(BOOK_LEVELS).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(); BOOK_LEVELS];
|
||
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 frd_label_tests {
|
||
use super::*;
|
||
use crate::rl::common::{FRD_BUCKET_RANGE_SIGMA, FRD_HORIZON_TICKS, FRD_N_ATOMS, FRD_N_HORIZONS};
|
||
|
||
/// Construct a synthetic snapshot stream with a deterministic
|
||
/// mid-price function. Only L1 bid/ask matter — `compute_frd_labels`
|
||
/// reads `mid_price_f32` which uses only the first level.
|
||
fn snap_with_mid(mid: f64) -> Mbp10Snapshot {
|
||
let half_spread = 0.005;
|
||
let bid = mid - half_spread;
|
||
let ask = mid + half_spread;
|
||
let mut levels = vec![BidAskPair::empty(); BOOK_LEVELS];
|
||
levels[0] = BidAskPair {
|
||
bid_px: BidAskPair::price_from_f64(bid),
|
||
bid_sz: 1,
|
||
bid_ct: 1,
|
||
ask_px: BidAskPair::price_from_f64(ask),
|
||
ask_sz: 1,
|
||
ask_ct: 1,
|
||
};
|
||
Mbp10Snapshot::new("ES.FUT".into(), 0, levels, 0, 0)
|
||
}
|
||
|
||
#[test]
|
||
fn frd_labels_flat_price_maps_to_mid_bucket() {
|
||
// Constant mid → all forward returns 0 → bucket = round(half) = 10.
|
||
let snaps: Vec<Mbp10Snapshot> = (0..2500).map(|_| snap_with_mid(100.0)).collect();
|
||
let labels = compute_frd_labels(&snaps, FRD_HORIZON_TICKS, FRD_BUCKET_RANGE_SIGMA);
|
||
let mid_bucket = (FRD_N_ATOMS as i32 - 1) / 2;
|
||
for h in 0..FRD_N_HORIZONS {
|
||
let h_ticks = FRD_HORIZON_TICKS[h];
|
||
// Non-sentinel range: [0, n - h_ticks)
|
||
for i in 0..(snaps.len() - h_ticks) {
|
||
assert_eq!(
|
||
labels[h][i], mid_bucket,
|
||
"flat-price horizon {} idx {} should be mid bucket {}; got {}",
|
||
h, i, mid_bucket, labels[h][i]
|
||
);
|
||
}
|
||
// Sentinel range: [n - h_ticks, n)
|
||
for i in (snaps.len() - h_ticks)..snaps.len() {
|
||
assert_eq!(
|
||
labels[h][i], -1,
|
||
"right-edge sentinel at h={} idx={} should be -1; got {}",
|
||
h, i, labels[h][i]
|
||
);
|
||
}
|
||
}
|
||
}
|
||
|
||
#[test]
|
||
fn frd_labels_monotonic_ramp_lands_in_upper_buckets() {
|
||
// Monotone ramp: mid[i] = 100 + i × 0.01. Forward return at
|
||
// each horizon is positive → bucket > mid (= 10). Exact bucket
|
||
// depends on σ_per_step (= 0.01 here, std of constant
|
||
// increments), so a 60-tick forward return is +0.6 ticks.
|
||
// σ_h = 0.01 × sqrt(60) ≈ 0.0775 → r = 0.6 / 0.0775 ≈ 7.75.
|
||
// 7.75 is way > range_σ=3, so bucket clamps to FRD_N_ATOMS-1 = 20.
|
||
let snaps: Vec<Mbp10Snapshot> =
|
||
(0..2500).map(|i| snap_with_mid(100.0 + 0.01 * i as f64)).collect();
|
||
let labels = compute_frd_labels(&snaps, FRD_HORIZON_TICKS, FRD_BUCKET_RANGE_SIGMA);
|
||
let max_bucket = (FRD_N_ATOMS - 1) as i32;
|
||
for h in 0..FRD_N_HORIZONS {
|
||
let h_ticks = FRD_HORIZON_TICKS[h];
|
||
for i in 0..(snaps.len() - h_ticks) {
|
||
assert_eq!(
|
||
labels[h][i], max_bucket,
|
||
"monotonic-up horizon {} idx {} should clamp to max bucket {}; got {}",
|
||
h, i, max_bucket, labels[h][i]
|
||
);
|
||
}
|
||
}
|
||
}
|
||
|
||
#[test]
|
||
fn frd_labels_short_input_below_h_ticks_all_sentinel() {
|
||
// n=10 < h_ticks[0]=60 → every label sentinel for h0; same for
|
||
// h1 (=300) and h2 (=1800).
|
||
let snaps: Vec<Mbp10Snapshot> = (0..10).map(|_| snap_with_mid(100.0)).collect();
|
||
let labels = compute_frd_labels(&snaps, FRD_HORIZON_TICKS, FRD_BUCKET_RANGE_SIGMA);
|
||
for h in 0..FRD_N_HORIZONS {
|
||
for (i, v) in labels[h].iter().enumerate() {
|
||
assert_eq!(*v, -1, "sub-h_ticks input must be all -1; got {} at h={} i={}", v, h, i);
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
#[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,
|
||
frd_horizon_ticks: crate::rl::common::FRD_HORIZON_TICKS,
|
||
frd_bucket_range_sigma: crate::rl::common::FRD_BUCKET_RANGE_SIGMA,
|
||
})
|
||
}
|
||
|
||
/// 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,
|
||
frd_horizon_ticks: crate::rl::common::FRD_HORIZON_TICKS,
|
||
frd_bucket_range_sigma: crate::rl::common::FRD_BUCKET_RANGE_SIGMA,
|
||
};
|
||
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.
|
||
}
|
||
}
|
||
|
||
#[cfg(test)]
|
||
mod labels_cache_tests {
|
||
use super::*;
|
||
|
||
#[test]
|
||
fn labels_cache_suffix_is_deterministic_and_keys_on_inputs() {
|
||
let h1 = [100, 300, 1000];
|
||
let h2 = [100, 300, 1001]; // different last horizon
|
||
let c1 = 0.5_f32;
|
||
let c2 = 0.25_f32;
|
||
let f1 = InstrumentFilter::All;
|
||
let f2 = InstrumentFilter::FrontMonth;
|
||
|
||
// Same inputs → same suffix
|
||
assert_eq!(
|
||
labels_cache_suffix(&h1, c1, f1),
|
||
labels_cache_suffix(&h1, c1, f1)
|
||
);
|
||
// Different horizon → different suffix
|
||
assert_ne!(
|
||
labels_cache_suffix(&h1, c1, f1),
|
||
labels_cache_suffix(&h2, c1, f1)
|
||
);
|
||
// Different cost → different suffix
|
||
assert_ne!(
|
||
labels_cache_suffix(&h1, c1, f1),
|
||
labels_cache_suffix(&h1, c2, f1)
|
||
);
|
||
// Different filter → different suffix
|
||
assert_ne!(
|
||
labels_cache_suffix(&h1, c1, f1),
|
||
labels_cache_suffix(&h1, c1, f2)
|
||
);
|
||
}
|
||
}
|