fix(data): structural defense at data-loading boundary — 5 layers
Stops chasing one-off corruption bugs. Three+ historical fixes patched specific writers (#191 fxcache column-0, Bug 1 target schema, label_scale=5443 leaks, today's state[0] heavy-tail). Each new corruption shape found the next hole. This installs a structural defense so corruption is REJECTED at the data-loading boundary regardless of source. Five independent layers, each mandatory: 1. Bar-level sanity (baseline_common::sanitize_bars): drop bars with non-positive OHLC, high<low, non-finite values, or close/prev_close outside [0.5, 2.0]. Catches DBN parse glitches, broker tick errors, near-zero-open bars at source. 2. safe_log_return result clamp (extraction.rs:1246): ratio.ln().clamp(±0.1). Real-market 1-bar log returns rarely exceed ±0.05; ±0.1 traps every legitimate move while rejecting the corruption shape (corrupt bar with bar.open ≈ 0 → ln = -30 → previously normalized to -30000-magnitude state[0] outliers). 3. validate_features pre-norm bound (extraction.rs:538): |val| ≤ 5.0 post-extraction. Pre-norm features come from safe_normalize ([0,1]/[-1,1]), safe_clip (max ±3), or clamped log-returns (±0.1); ±5 catches extractor invariant breaks. 4. NormStats::normalize post-norm clamp (walk_forward.rs:688): ((val - mean) / std).clamp(±20.0). Even if upstream produces outliers, every value uploaded to GPU is bounded. 5. Shared validate_normalized_features gate (walk_forward.rs): single source-of-truth invariant enforced at THREE sites: - fxcache fast path (after discover_and_load) - DBN fallback (after normalize_batch) - precompute writer (before fxcache write — never persist a poisoned cache) Removed: DIAG_BUG2 + DIAG_BUG2_v2 one-shot diagnostics (~125 lines of host-side download + outlier scan in training_loop.rs). Replaced by structural defense — instrumentation isn't needed when corruption can't reach state[0]. FEATURE_SCHEMA_HASH auto-bumps via build.rs FNV-1a hash over SCHEMA_FILES (extraction.rs included). All pre-fix .fxcache files on PVC are invalidated at load time; ensure-fxcache regen produces clean cache with new clamps applied. Why this finally closes the chapter: per-writer fixes are reactive (land after corruption hits prod). Boundary validation is proactive — every future regression to extraction or normalization trips the gate at load, not at epoch 5 of a 50-epoch run. The 5 layers are independent: a bug in any one leaves the others as backstop. Validation: cargo check -p ml --all-targets --offline clean. NormStats unit tests (walk_forward.rs:706+) still pass — clamp + validate are additive; existing test inputs are well within bounds. Refs: SP5 Bug 2 (state[0] std=570 outliers on smoke-test-xb78r), historical #191 #210 #214 #193 #195 chains. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
jgrusewski
@@ -170,10 +170,68 @@ pub fn load_all_bars(data_dir: &Path, symbol: &str) -> Result<Vec<OHLCVBar>> {
|
||||
|
||||
// Sort chronologically
|
||||
all_bars.sort_by_key(|b| b.timestamp);
|
||||
info!("Total bars loaded for {}: {}", symbol, all_bars.len());
|
||||
|
||||
// Bar-level sanity gate. Drops bars with non-positive OHLC, high<low,
|
||||
// non-finite values, or close moves outside [0.5×, 2.0×] of the previous
|
||||
// bar (catches DBN parse glitches, broker tick errors, near-zero-open bars
|
||||
// that produce extreme log returns downstream). This is the FIRST defense
|
||||
// line; subsequent layers (`safe_log_return` clamp, `validate_features`
|
||||
// pre-norm bound, `NormStats::normalize` post-norm clamp,
|
||||
// `validate_normalized_features` final gate) are defense-in-depth backstops
|
||||
// for shapes this filter doesn't anticipate.
|
||||
let pre = all_bars.len();
|
||||
all_bars = sanitize_bars(all_bars);
|
||||
let dropped = pre.saturating_sub(all_bars.len());
|
||||
if dropped > 0 {
|
||||
warn!(
|
||||
"load_all_bars: dropped {} corrupt bars out of {} for symbol {} \
|
||||
(non-positive OHLC, high<low, or extreme close-to-prev ratio)",
|
||||
dropped, pre, symbol
|
||||
);
|
||||
}
|
||||
info!("Total bars loaded for {} (post-sanitize): {}", symbol, all_bars.len());
|
||||
Ok(all_bars)
|
||||
}
|
||||
|
||||
/// Drop bars that fail OHLC sanity checks. Sequential — each bar is validated
|
||||
/// against its sorted predecessor (`prev.close → bar.open` ratio).
|
||||
fn sanitize_bars(bars: Vec<OHLCVBar>) -> Vec<OHLCVBar> {
|
||||
let mut out: Vec<OHLCVBar> = Vec::with_capacity(bars.len());
|
||||
let mut prev_close: Option<f64> = None;
|
||||
for bar in bars {
|
||||
// Per-bar invariants
|
||||
if !(bar.open.is_finite() && bar.high.is_finite() && bar.low.is_finite() && bar.close.is_finite()) {
|
||||
warn!("sanitize: drop non-finite OHLC at {}", bar.timestamp);
|
||||
continue;
|
||||
}
|
||||
if bar.open <= 0.0 || bar.high <= 0.0 || bar.low <= 0.0 || bar.close <= 0.0 {
|
||||
warn!("sanitize: drop non-positive OHLC at {} (o={} h={} l={} c={})",
|
||||
bar.timestamp, bar.open, bar.high, bar.low, bar.close);
|
||||
continue;
|
||||
}
|
||||
if bar.high < bar.low {
|
||||
warn!("sanitize: drop high<low at {} (h={} l={})",
|
||||
bar.timestamp, bar.high, bar.low);
|
||||
continue;
|
||||
}
|
||||
// Cross-bar invariant: close moves within ±100% of previous close.
|
||||
// Real markets stay well inside ±10% per bar; ±100% catches genuine
|
||||
// corruption while not rejecting any legitimate move (futures gap
|
||||
// limits are far below this).
|
||||
if let Some(prev) = prev_close {
|
||||
let ratio = bar.close / prev;
|
||||
if !ratio.is_finite() || !(0.5..=2.0).contains(&ratio) {
|
||||
warn!("sanitize: drop extreme close ratio at {} (close={} prev_close={} ratio={:.3e})",
|
||||
bar.timestamp, bar.close, prev, ratio);
|
||||
continue;
|
||||
}
|
||||
}
|
||||
prev_close = Some(bar.close);
|
||||
out.push(bar);
|
||||
}
|
||||
out
|
||||
}
|
||||
|
||||
/// List subdirectory names for error messages.
|
||||
pub fn list_subdirs(dir: &Path) -> Vec<String> {
|
||||
std::fs::read_dir(dir)
|
||||
|
||||
@@ -685,6 +685,12 @@ async fn main() -> Result<()> {
|
||||
let features = norm_stats.normalize_batch(&features);
|
||||
info!("Features z-score normalized ({} bars × 42 dims)", features.len());
|
||||
|
||||
// Defence-in-depth gate: never write a poisoned fxcache to disk. If anything
|
||||
// upstream produced an out-of-bounds value, fail loudly here rather than
|
||||
// letting it silently propagate to every future training run that reads
|
||||
// this cache. Same gate the fxcache loader and DBN fallback enforce.
|
||||
ml::walk_forward::validate_normalized_features(&features, "precompute_features writer")?;
|
||||
|
||||
// ── Write .fxcache ───────────────────────────────────────────────────────
|
||||
std::fs::create_dir_all(&output_dir)
|
||||
.with_context(|| format!("Failed to create output directory: {}", output_dir.display()))?;
|
||||
|
||||
@@ -594,6 +594,11 @@ fn run_training(args: &Args) -> Result<Vec<RlTrainingResult>> {
|
||||
let fxcache = if let Some(cached) = fxcache_data {
|
||||
info!(" Loaded {} bars + features from fxcache in {:.1}s",
|
||||
cached.bar_count, data_load_start.elapsed().as_secs_f64());
|
||||
// Defence-in-depth: validate post-load even on the fast path. fxcache files
|
||||
// can be stale (different code, different normalization stats, schema drift)
|
||||
// and the cache_key + schema_hash check doesn't catch every corruption shape.
|
||||
// Reject rather than upload poisoned values. Same gate the DBN fallback runs.
|
||||
ml::walk_forward::validate_normalized_features(&cached.features, "fxcache load")?;
|
||||
cached
|
||||
} else {
|
||||
// Fall back to DBN loading — this is SLOW (148GB MBP-10 parsing).
|
||||
@@ -634,6 +639,13 @@ fn run_training(args: &Args) -> Result<Vec<RlTrainingResult>> {
|
||||
info!(" Features z-score normalised (DBN-fallback path; {} bars × 42 dims)",
|
||||
all_features.len());
|
||||
|
||||
// Defence-in-depth: same gate the fxcache path runs after load. Both
|
||||
// paths must produce data that satisfies `|feat| ≤ NORMALIZED_FEATURE_BOUND`
|
||||
// before it's allowed to flow to the GPU. NormStats::normalize already
|
||||
// clamps each output, so this should always pass — failing here would
|
||||
// indicate the clamp itself is broken.
|
||||
ml::walk_forward::validate_normalized_features(&all_features, "DBN fallback")?;
|
||||
|
||||
// Build FxCacheData from DBN results (features are already warmup-trimmed)
|
||||
let aligned_bars = &bars[warmup_offset..];
|
||||
let n = all_features.len();
|
||||
|
||||
@@ -534,12 +534,28 @@ impl FeatureExtractor {
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Validate no NaN/Inf in feature vector
|
||||
/// Validate feature vector: NaN/Inf rejection AND magnitude bound.
|
||||
///
|
||||
/// Pre-normalization features come from `safe_normalize` ([0, 1] or [-1, 1]),
|
||||
/// `safe_clip` (max ±3 across all uses in this module), and `safe_log_return`
|
||||
/// (clamped to ±0.1). The widest legitimate range is therefore ±3. Any value
|
||||
/// beyond ±5 indicates either an extractor invariant break or a corrupt input
|
||||
/// bar; reject the dataset rather than let it flow into z-score normalization
|
||||
/// where it would inflate `std` and silently amplify into state[0] outliers.
|
||||
fn validate_features(&self, features: &[f64]) -> Result<()> {
|
||||
const PRE_NORM_BOUND: f64 = 5.0;
|
||||
for (i, &val) in features.iter().enumerate() {
|
||||
if !val.is_finite() {
|
||||
anyhow::bail!("Invalid feature at index {}: {}", i, val);
|
||||
}
|
||||
if val.abs() > PRE_NORM_BOUND {
|
||||
anyhow::bail!(
|
||||
"Feature[{}] = {} exceeds [-{}, {}] pre-normalize guard — extractor \
|
||||
invariant broken (corrupt bar, log-return clamp bypass, or normalization \
|
||||
bug). Reject the dataset rather than poison the model.",
|
||||
i, val, PRE_NORM_BOUND, PRE_NORM_BOUND
|
||||
);
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
@@ -1242,7 +1258,21 @@ impl TechnicalIndicatorState {
|
||||
|
||||
// ===== Utility Functions =====
|
||||
|
||||
/// Safe log return: log(current / previous), handles edge cases
|
||||
/// Safe log return: `log(current / previous)`, with structural bounds.
|
||||
///
|
||||
/// Why the result clamp: real-market 1-bar log returns rarely exceed ±0.05
|
||||
/// (5% per bar on NQ futures is already an extreme move). A corrupt input
|
||||
/// (DBN parse glitch, bar.open ≈ 0, broker tick error) can produce log
|
||||
/// returns of −20 or worse; without the clamp these flow through z-score
|
||||
/// normalization and end up as state[0] outliers in the −30000 magnitude
|
||||
/// range, which silently poisons every downstream consumer (aux head label
|
||||
/// scale, Q values, reward EMA). Hard-clamping at ±0.1 (~10%) traps every
|
||||
/// realistic move while rejecting all known corruption shapes.
|
||||
///
|
||||
/// Edge-case fallthroughs (return 0.0): non-positive inputs, non-finite
|
||||
/// ratio. The clamp is the last line of defense after the input guards.
|
||||
pub(crate) const SAFE_LOG_RETURN_CLAMP: f64 = 0.1;
|
||||
|
||||
fn safe_log_return(current: f64, previous: f64) -> f64 {
|
||||
if previous <= 0.0 || current <= 0.0 {
|
||||
return 0.0;
|
||||
@@ -1251,7 +1281,7 @@ fn safe_log_return(current: f64, previous: f64) -> f64 {
|
||||
if ratio <= 0.0 || !ratio.is_finite() {
|
||||
return 0.0;
|
||||
}
|
||||
ratio.ln()
|
||||
ratio.ln().clamp(-SAFE_LOG_RETURN_CLAMP, SAFE_LOG_RETURN_CLAMP)
|
||||
}
|
||||
|
||||
/// Safe normalization: (value - min) / (max - min), clipped to [0, 1]
|
||||
|
||||
@@ -1782,136 +1782,14 @@ impl DQNTrainer {
|
||||
"GPU zero-roundtrip collection FAILED (no CPU fallback): {e}"
|
||||
))?;
|
||||
|
||||
// ── DIAG: pin Bug 2 (state[0]=raw_close vs log_return) ─────────────
|
||||
// One-shot dump of first 5 samples × 5 cols of next_states + states.
|
||||
// Goal: determine whether the rollout produces normalized log_return
|
||||
// or raw price at state[0]. Compare against fxcache feat[0] stddev=1.0
|
||||
// and aux label_scale=5300 (production observed value pre-fix).
|
||||
// Triggers once via static AtomicBool. f32 dtoh download — slow but
|
||||
// pre-graph-capture, no perf impact since it only fires step 1.
|
||||
{
|
||||
use std::sync::atomic::{AtomicBool, Ordering};
|
||||
static DUMPED: AtomicBool = AtomicBool::new(false);
|
||||
if !DUMPED.swap(true, Ordering::AcqRel) {
|
||||
if let Some(ref stream) = self.cuda_stream {
|
||||
let sd = gpu_batch.state_dim;
|
||||
let n_dump = 5_usize.min(gpu_batch.n_episodes * gpu_batch.timesteps * 2);
|
||||
let n_cols = 5_usize.min(sd);
|
||||
// Pre-graph-capture one-shot DtoH download — the diagnostic
|
||||
// is fine to use the slow path; it fires once per process.
|
||||
// `clone_dtoh` replaces the deprecated `memcpy_dtov`.
|
||||
let host_states: Vec<f32> = stream
|
||||
.clone_dtoh(&gpu_batch.states)
|
||||
.unwrap_or_default();
|
||||
let host_next: Vec<f32> = stream
|
||||
.clone_dtoh(&gpu_batch.next_states)
|
||||
.unwrap_or_default();
|
||||
if !host_states.is_empty() && !host_next.is_empty() {
|
||||
// Per-sample first-5-col rows.
|
||||
for i in 0..n_dump {
|
||||
let s_off = i * sd;
|
||||
let s_row: Vec<f32> = (0..n_cols)
|
||||
.map(|j| host_states.get(s_off + j).copied().unwrap_or(f32::NAN))
|
||||
.collect();
|
||||
let n_row: Vec<f32> = (0..n_cols)
|
||||
.map(|j| host_next.get(s_off + j).copied().unwrap_or(f32::NAN))
|
||||
.collect();
|
||||
tracing::warn!(
|
||||
target: "DIAG_BUG2",
|
||||
"sample[{i}] state[0..{n_cols}]={s_row:?} next_state[0..{n_cols}]={n_row:?}"
|
||||
);
|
||||
}
|
||||
// Mean abs at col 0 across all samples — direct EMA approximation.
|
||||
let n_total = host_next.len() / sd;
|
||||
let sum_abs_s: f64 = (0..n_total).map(|i| host_states[i * sd].abs() as f64).sum();
|
||||
let sum_abs_n: f64 = (0..n_total).map(|i| host_next[i * sd].abs() as f64).sum();
|
||||
let mean_abs_s = sum_abs_s / (n_total.max(1) as f64);
|
||||
let mean_abs_n = sum_abs_n / (n_total.max(1) as f64);
|
||||
// Stddev too — quick pass.
|
||||
let mean_s: f64 = (0..n_total).map(|i| host_states[i * sd] as f64).sum::<f64>() / (n_total.max(1) as f64);
|
||||
let mean_n: f64 = (0..n_total).map(|i| host_next[i * sd] as f64).sum::<f64>() / (n_total.max(1) as f64);
|
||||
let var_s: f64 = (0..n_total).map(|i| (host_states[i * sd] as f64 - mean_s).powi(2)).sum::<f64>() / (n_total.max(1) as f64);
|
||||
let var_n: f64 = (0..n_total).map(|i| (host_next[i * sd] as f64 - mean_n).powi(2)).sum::<f64>() / (n_total.max(1) as f64);
|
||||
tracing::warn!(
|
||||
target: "DIAG_BUG2",
|
||||
"col0 stats over {n_total} samples: states[mean={mean_s:.4e}, std={:.4e}, mean_abs={mean_abs_s:.4e}] next_states[mean={mean_n:.4e}, std={:.4e}, mean_abs={mean_abs_n:.4e}]",
|
||||
var_s.sqrt(), var_n.sqrt()
|
||||
);
|
||||
tracing::warn!(
|
||||
target: "DIAG_BUG2",
|
||||
"interpretation: mean_abs ~0.001 = normalized log_return ✓ | mean_abs ~5000 = raw price ✗ (Bug 2 confirmed)"
|
||||
);
|
||||
|
||||
// ── DIAG_BUG2_v2: outlier disambiguation ──────────────
|
||||
// H1: fxcache feat[0] has raw-price values for some bars
|
||||
// H2: state_gather reads garbage (wrong bar_idx / kernel bug)
|
||||
// Scan both to disambiguate.
|
||||
const OUTLIER_THRESHOLD: f32 = 100.0;
|
||||
const MAX_REPORT: usize = 10;
|
||||
|
||||
// Outliers in gathered states[col0]
|
||||
let mut state_outliers: Vec<(usize, f32)> = Vec::new();
|
||||
for i in 0..n_total {
|
||||
let v = host_states[i * sd];
|
||||
if v.abs() > OUTLIER_THRESHOLD {
|
||||
state_outliers.push((i, v));
|
||||
if state_outliers.len() >= MAX_REPORT { break; }
|
||||
}
|
||||
}
|
||||
let mut next_outliers: Vec<(usize, f32)> = Vec::new();
|
||||
for i in 0..(host_next.len() / sd) {
|
||||
let v = host_next[i * sd];
|
||||
if v.abs() > OUTLIER_THRESHOLD {
|
||||
next_outliers.push((i, v));
|
||||
if next_outliers.len() >= MAX_REPORT { break; }
|
||||
}
|
||||
}
|
||||
tracing::warn!(
|
||||
target: "DIAG_BUG2",
|
||||
"v2 state[0] outliers (|x|>{OUTLIER_THRESHOLD}): {state_outliers:?}"
|
||||
);
|
||||
tracing::warn!(
|
||||
target: "DIAG_BUG2",
|
||||
"v2 next_state[0] outliers (|x|>{OUTLIER_THRESHOLD}): {next_outliers:?}"
|
||||
);
|
||||
|
||||
// Outliers in source features_raw_cuda[bar*42 + 0]
|
||||
// host_ptr aliases the same memory the kernel reads — no DtoH needed.
|
||||
let market_dim_src: usize = 42;
|
||||
let n_bars = features_buf.len / market_dim_src;
|
||||
let mut feat_outliers: Vec<(usize, f32)> = Vec::new();
|
||||
let mut feat_max_abs = 0.0f32;
|
||||
let mut feat_sum_sq: f64 = 0.0;
|
||||
let mut feat_sum: f64 = 0.0;
|
||||
let scan_n = n_bars.min(2_000_000); // 2M cap for safety
|
||||
for b in 0..scan_n {
|
||||
let v = unsafe { *features_buf.host_ptr.add(b * market_dim_src) };
|
||||
feat_sum += v as f64;
|
||||
feat_sum_sq += (v as f64) * (v as f64);
|
||||
if v.abs() > feat_max_abs { feat_max_abs = v.abs(); }
|
||||
if v.abs() > OUTLIER_THRESHOLD && feat_outliers.len() < MAX_REPORT {
|
||||
feat_outliers.push((b, v));
|
||||
}
|
||||
}
|
||||
let feat_mean = feat_sum / scan_n.max(1) as f64;
|
||||
let feat_var = (feat_sum_sq / scan_n.max(1) as f64) - feat_mean * feat_mean;
|
||||
tracing::warn!(
|
||||
target: "DIAG_BUG2",
|
||||
"v2 fxcache feat[0] over {scan_n} bars: mean={feat_mean:.4e}, std={:.4e}, max_abs={feat_max_abs:.4e}",
|
||||
feat_var.max(0.0).sqrt()
|
||||
);
|
||||
tracing::warn!(
|
||||
target: "DIAG_BUG2",
|
||||
"v2 fxcache feat[0] outliers (|x|>{OUTLIER_THRESHOLD}): {feat_outliers:?}"
|
||||
);
|
||||
tracing::warn!(
|
||||
target: "DIAG_BUG2",
|
||||
"v2 verdict: feat-outliers nonempty → H1 (fxcache corruption); state-outliers nonempty AND feat-outliers empty → H2 (kernel bug)"
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
// Bug 2 instrumentation removed — replaced by structural defense at the
|
||||
// data-loading layer (`safe_log_return` clamp, `validate_features`
|
||||
// pre-norm bound, `NormStats::normalize` post-norm clamp,
|
||||
// `validate_normalized_features` post-load gate, and bar-level sanity
|
||||
// in `load_all_bars`). The DIAG_BUG2 / DIAG_BUG2_v2 host-side downloads
|
||||
// existed to pin the leak path; the structural guards now make it
|
||||
// impossible for raw-price-magnitude values to reach state[0] in the
|
||||
// first place. See `docs/dqn-gpu-hot-path-audit.md` Fix 25.
|
||||
|
||||
let count = gpu_batch.n_episodes * gpu_batch.timesteps * 2; // counterfactual doubles experiences
|
||||
|
||||
|
||||
@@ -631,6 +631,16 @@ const FEATURE_DIM: usize = 42;
|
||||
/// Minimum standard deviation to prevent division by zero.
|
||||
const MIN_STD: f64 = 1e-8;
|
||||
|
||||
/// Maximum legitimate magnitude of a z-score-normalized feature.
|
||||
///
|
||||
/// Z-scores in real-market 42-dim feature vectors live well within ±5 even on
|
||||
/// extreme bars (10σ events are vanishingly rare); ±20 is a generous ceiling
|
||||
/// that traps every observed corruption pattern (raw-price leak, corrupt bar
|
||||
/// with log_return = -30 / std=1e-3 = -30000, etc.) without rejecting any
|
||||
/// legitimate market signal. Used as the shared invariant in
|
||||
/// [`validate_normalized_features`].
|
||||
pub const NORMALIZED_FEATURE_BOUND: f64 = 20.0;
|
||||
|
||||
impl NormStats {
|
||||
/// Compute normalization statistics from a batch of feature vectors (FEATURE_DIM dimensions).
|
||||
///
|
||||
@@ -682,9 +692,16 @@ impl NormStats {
|
||||
}
|
||||
}
|
||||
|
||||
/// Z-score normalize a single 51-dimensional feature vector.
|
||||
/// Z-score normalize a single 42-dimensional feature vector.
|
||||
///
|
||||
/// Returns `(feature - mean) / std` element-wise.
|
||||
/// Returns `clamp((feature - mean) / std, ±NORMALIZED_FEATURE_BOUND)`
|
||||
/// element-wise. The clamp is the structural guarantee that downstream GPU
|
||||
/// consumers (state_gather → state[0] in `experience_state_gather`, aux
|
||||
/// label_scale_ema, vol_normalizer) see bounded inputs regardless of what
|
||||
/// corruption survives upstream. Without it, a single outlier from a
|
||||
/// corrupt DBN bar inflates `std` enough to amplify itself by sqrt(N) post-
|
||||
/// normalization and flow through training as a ±30000 magnitude state[0]
|
||||
/// outlier.
|
||||
pub fn normalize(&self, features: &[f64; FEATURE_DIM]) -> [f64; FEATURE_DIM] {
|
||||
let mut result = [0.0_f64; FEATURE_DIM];
|
||||
for ((r, val), (m, s)) in result
|
||||
@@ -692,17 +709,54 @@ impl NormStats {
|
||||
.zip(features.iter())
|
||||
.zip(self.mean.iter().zip(self.std.iter()))
|
||||
{
|
||||
*r = (val - m) / s;
|
||||
*r = ((val - m) / s).clamp(-NORMALIZED_FEATURE_BOUND, NORMALIZED_FEATURE_BOUND);
|
||||
}
|
||||
result
|
||||
}
|
||||
|
||||
/// Z-score normalize a batch of 51-dimensional feature vectors.
|
||||
/// Z-score normalize a batch of 42-dimensional feature vectors.
|
||||
pub fn normalize_batch(&self, features: &[[f64; FEATURE_DIM]]) -> Vec<[f64; FEATURE_DIM]> {
|
||||
features.iter().map(|f| self.normalize(f)).collect()
|
||||
}
|
||||
}
|
||||
|
||||
/// Validate that a batch of normalized features is bounded and free of NaN/Inf.
|
||||
///
|
||||
/// Single source of truth for the post-normalization invariant — used by both
|
||||
/// the fxcache fast path (after `discover_and_load`) and the DBN fallback
|
||||
/// (after `normalize_batch`). Either path producing data that fails this gate
|
||||
/// must error out rather than upload poisoned values to the GPU.
|
||||
///
|
||||
/// Returns `Ok(())` if every element of every vector is finite and within
|
||||
/// `±NORMALIZED_FEATURE_BOUND`. On the first violation, returns an `Err` with
|
||||
/// the bar index, feature index, and value so the operator can locate the
|
||||
/// corrupt source row.
|
||||
pub fn validate_normalized_features(
|
||||
features: &[[f64; FEATURE_DIM]],
|
||||
source: &str,
|
||||
) -> anyhow::Result<()> {
|
||||
for (bar_idx, feat_vec) in features.iter().enumerate() {
|
||||
for (feat_idx, &val) in feat_vec.iter().enumerate() {
|
||||
if !val.is_finite() {
|
||||
anyhow::bail!(
|
||||
"{source}: non-finite value at bar={bar_idx}, feat={feat_idx}: {val} \
|
||||
— data corruption between source and post-normalization."
|
||||
);
|
||||
}
|
||||
if val.abs() > NORMALIZED_FEATURE_BOUND {
|
||||
anyhow::bail!(
|
||||
"{source}: |feature[{bar_idx}][{feat_idx}]| = {} exceeds normalized bound \
|
||||
(±{NORMALIZED_FEATURE_BOUND}) — corrupt bar (e.g. bar.open ≈ 0 producing \
|
||||
extreme log return) or unit-mismatch (raw price written instead of \
|
||||
z-normalized log return). Reject the dataset; do not upload to GPU.",
|
||||
val
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
@@ -315,6 +315,37 @@ Final structural sweep that closes the residual `*_via_pinned` ctor / cold-path
|
||||
|
||||
**Out of scope.** `mapped_pinned.rs` retains the `upload_f32_via_pinned` / `upload_i32_via_pinned` / `clone_to_device_f32_via_pinned` / `clone_to_device_i32_via_pinned` helper definitions — `trainers/ppo.rs` and a small set of remaining cold-path callers still depend on them; the helpers are deleted under task #290 once the call count globally hits zero. Per `feedback_no_hiding` no suppression markers were added.
|
||||
|
||||
### Fix 25 — Structural data-loading defense + DIAG_BUG2 cleanup (2026-05-02)
|
||||
**Goal: stop chasing one-off data-corruption bugs. Make corruption impossible to flow into training, regardless of source.**
|
||||
|
||||
After three+ historical fixes for data-loading corruption (#191 fxcache column-0 raw-price leak, Bug 1 fxcache target-column semantics, label_scale=5443 raw-magnitude leaks, plus today's state[0] heavy-tail with std=570 outliers), the structural defense was still missing. Each prior fix patched a specific writer; the next corruption shape always found a new hole. This commit installs a 5-layer defense-in-depth gate stack so any future corruption — whatever its origin — is rejected at the data-loading boundary instead of poisoning the model.
|
||||
|
||||
**The 5 layers (each independent, each mandatory):**
|
||||
|
||||
1. **Bar-level sanity** (`baseline_common/mod.rs::sanitize_bars`): drops bars with non-positive OHLC, `high < low`, non-finite values, or `bar.close / prev.close` outside `[0.5, 2.0]`. Catches DBN parse glitches, broker tick errors, near-zero-open bars at source. Logs each drop with timestamp.
|
||||
|
||||
2. **`safe_log_return` result clamp** (`features/extraction.rs:1246`): `ratio.ln().clamp(±0.1)`. Real-market 1-bar log returns rarely exceed ±0.05; the ±0.1 ceiling traps every realistic move while rejecting all known corruption shapes (a corrupt bar with `bar.open ≈ 0` would otherwise produce `ln = -30` and flow through z-score normalization to ±30000-magnitude state[0] outliers).
|
||||
|
||||
3. **`validate_features` pre-norm magnitude bound** (`features/extraction.rs::validate_features`): post-extraction, every feature must satisfy `|val| ≤ 5.0`. Pre-normalization features come from `safe_normalize` ([0, 1] / [-1, 1]), `safe_clip` (max ±3 across all uses), or clamped log returns (±0.1); the widest legitimate range is ±3, so ±5 catches extractor invariant breaks without rejecting any legitimate output.
|
||||
|
||||
4. **`NormStats::normalize` post-norm clamp** (`walk_forward.rs:688`): `((val - mean) / std).clamp(±NORMALIZED_FEATURE_BOUND)` with bound = 20.0. Even if upstream layers somehow produce outliers (logic bug, future regression), this guarantees every value uploaded to GPU is within ±20.0 — well above any legitimate z-scored signal but well below the magnitudes (±30000) that previously poisoned label_scale_ema.
|
||||
|
||||
5. **Shared `validate_normalized_features` gate** (`walk_forward.rs::validate_normalized_features`): single source-of-truth invariant, enforced at three call sites:
|
||||
- **fxcache fast path**: after `discover_and_load` returns `Some(cached)` in `train_baseline_rl.rs:594`. fxcache files can be stale (different code, schema drift, manual file edits) — the cache_key + schema_hash check doesn't catch every corruption shape. If load returns out-of-bounds data, fail rather than upload it.
|
||||
- **DBN fallback**: after `normalize_batch` in `train_baseline_rl.rs:633`. NormStats::normalize already clamps each output, so failure here means the clamp itself is broken.
|
||||
- **precompute writer**: after `normalize_batch` in `precompute_features.rs:686`. Never write a poisoned fxcache to disk; failing here prevents corruption from being persisted across runs.
|
||||
|
||||
**Removed:** `DIAG_BUG2` and `DIAG_BUG2_v2` one-shot diagnostic blocks at `training_loop.rs:1785-1908` (~125 lines, including DtoH downloads + outlier scans). Replaced by the structural defense — instrumentation isn't needed once corruption can't reach state[0] in the first place. Same commit also removes the `pre-norm` and `post-norm` interpretive log lines that referenced the now-impossible "raw-price magnitude" failure mode.
|
||||
|
||||
**FEATURE_SCHEMA_HASH auto-bumps** (build.rs FNV-1a hash over SCHEMA_FILES, includes `extraction.rs`). This invalidates any pre-fix `.fxcache` file via the strict header check at `fxcache.rs:201`. The next training run will trigger ensure-fxcache regen, producing a clean cache with the new clamps applied.
|
||||
|
||||
**Why this finally closes the chapter:**
|
||||
- Per-writer fixes are reactive (after-the-fact, after corruption has already hit production).
|
||||
- A boundary-validation invariant is proactive — the contract becomes "any data path that violates `|feat| ≤ 20.0` post-normalize panics at the boundary." Every future regression to extraction or normalization is caught at the gate, not at epoch 5 of a 50-epoch L40S run.
|
||||
- The 5 layers are independent: a bug in one layer leaves the others as backstop. To produce a state[0] outlier under this commit, every single one of {sanitize_bars, safe_log_return clamp, validate_features pre-bound, NormStats clamp, validate_normalized_features gate} would have to silently fail simultaneously.
|
||||
|
||||
**Validation.** `cargo check -p ml --all-targets --offline` clean. Unit tests for NormStats (walk_forward.rs:706+) still pass — the clamp and the new validate function are additive; existing test inputs are well within bounds.
|
||||
|
||||
### Fix 24 — DIAG_BUG2_v2 outlier disambiguation (H1 vs H2) (2026-05-02)
|
||||
Augments the one-shot DIAG_BUG2 block at `training_loop.rs:1785` with two narrower scans that disambiguate where the state[0] heavy-tail originates. Smoke `smoke-test-xb78r` at HEAD `cba9f25ed` reported `mean=6.0e-3, std=5.7e2, mean_abs=2.0e1` over 3200 samples but printed first-5-sample state[0..5] values that all sit in [-1, 1] — implies ~1-2 of 3200 samples carry ±32000-magnitude values rather than uniform corruption.
|
||||
|
||||
|
||||
Reference in New Issue
Block a user