diff --git a/crates/ml-alpha/examples/alpha_rl_train.rs b/crates/ml-alpha/examples/alpha_rl_train.rs index a44ed710c..830dde07e 100644 --- a/crates/ml-alpha/examples/alpha_rl_train.rs +++ b/crates/ml-alpha/examples/alpha_rl_train.rs @@ -63,7 +63,7 @@ use ml_alpha::rl::isv_slots::{ }; use ml_alpha::trainer::diag_staging::DiagStaging; use ml_alpha::trainer::integrated::{ - DiagInputs, IntegratedStepStats, IntegratedTrainer, IntegratedTrainerConfig, + read_slice_d_into, DiagInputs, IntegratedStepStats, IntegratedTrainer, IntegratedTrainerConfig, }; use ml_alpha::trainer::perception::PerceptionTrainerConfig; use ml_backtesting::sim::LobSimCuda; @@ -471,13 +471,25 @@ fn main() -> Result<()> { let mut windowed_act_hist: [f32; N_ACTIONS] = [0.0; N_ACTIONS]; const WINDOWED_ACT_ALPHA: f32 = 1.0 / 1000.0; - let mut pnl_cum_usd: f64 = 0.0; - // B-7 observability: pnl_cum_usd uses POST-clamp rewards / scale (un-scales - // but doesn't un-clamp), so the diag's training pnl signal is truncated to - // [-clamp_loss, +clamp_win] per close. realized_pnl_cum_usd sums raw_rewards - // (pre-scale, pre-clamp) — direct USD trade pnl. Divergence between the two - // exposes the reward-hacking gap the clamp introduces. - let mut realized_pnl_cum_usd: f64 = 0.0; + // Phase 0 reward-redesign (2026-06-02): SHAPED-reward cumulative (raw_rewards + // sum at close events). NOT in USD — pts × lots × Phase-5 shaping. The + // legacy field name (`realized_pnl_cum_usd`) is gone; this is the gradient- + // signal cumulative, useful for measuring reward-policy alignment via the + // Pearson(rewards.sum, Δpnl) signal but never to be confused with USD pnl. + let mut shaped_reward_close_event_cum: f64 = 0.0; + // Phase 0 reward-redesign (2026-06-02): authoritative USD pnl cumulative + // sourced from `pnl_track_step`'s `pnl_step_close_usd_d` buffer (same + // realised_pnl_usd_fp/100 arithmetic as `eval_summary.total_pnl_usd`). + // Cross-source consistent with the eval_summary aggregator within $0.01. + let mut realized_pnl_usd_cum: f64 = 0.0; + // Phase 0 reward-redesign (2026-06-02): direct-readback scratch buffer + // for the per-step `pnl_step_close_usd_d` view. `read_slice_d_into` + // runs a mapped-pinned DtoD + stream-sync on the trainer's main stream + // — same stream as `pnl_track_step`'s write, so CUDA stream ordering + // makes the read consistent without any explicit cross-stream barrier. + // Allocated once outside the loop and re-filled in-place each step + // (no per-step allocation cost). + let mut pnl_step_close_usd_host: Vec = vec![0.0_f32; cli.n_backtests]; let mut win_count: u64 = 0; let mut total_trades: u64 = 0; let mut hold_time_sum: f64 = 0.0; @@ -554,6 +566,25 @@ fn main() -> Result<()> { ) .context("diag snapshot_async")?; + // Phase 0 reward-redesign (2026-06-02): direct mapped-pinned readback + // of `pnl_step_close_usd_d` from the LobSim. `pnl_track_step` ran + // earlier in this step on the trainer's main stream; `read_slice_d_into` + // queues a DtoD + stream-sync on the same stream, so CUDA stream + // ordering guarantees the read sees the kernel's writes — no + // cross-stream sync gap. Routing through `diag_staging` was found + // to break same-seed bit-equality (the diag stream's DtoD raced + // the main stream's `raw_memset_d8_zero + pnl_track_step write`); + // a direct same-stream read avoids the staging entirely. Mega-graph + // compatible: this read runs AFTER the per-step pipeline finishes, + // so it stays outside the captured graph (one extra DtoD per step + // post-graph-replay). + read_slice_d_into( + &trainer.stream, + sim.pnl_step_close_usd_d(), + &mut pnl_step_close_usd_host, + ) + .context("read pnl_step_close_usd_d (train)")?; + // ── Per-step diag dump (async staging reads). ──────────────── // ALL diag reads come from the DiagStaging double-buffer — zero // GPU stalls on the training stream. On step 0 these report @@ -604,25 +635,31 @@ fn main() -> Result<()> { let done_count: u32 = dones_host.iter().map(|&d| if d > 0.5 { 1 } else { 0 }).sum(); // Per-trade cumulative stats (surfer validation) — mutate the - // train-loop running counters (pnl_cum_usd / total_trades / - // win_count / hold_time_sum). build_diag_value receives the - // post-update values via DiagInputs. - let current_scale = trainer.read_isv_host(RL_REWARD_SCALE_INDEX); + // train-loop running counters (shaped_reward_close_event_cum / + // total_trades / win_count / hold_time_sum). build_diag_value receives + // the post-update values via DiagInputs. + // + // Phase 0 reward-redesign (2026-06-02): authoritative per-batch USD pnl + // delta sourced from `pnl_step_close_usd_d` via the direct-readback + // above. The dones/raw_rewards reads are still previous-step (diag + // staging double-buffer); the pnl read is CURRENT step. The cumulative + // total is direction-agnostic — running sum converges to + // eval_summary.total_pnl_usd within fp/100 precision regardless of + // alignment. The per-batch win attribution below (`if dones[b] > 0.5 + // { win_count += pnl[b] > 0 }`) is best-effort diagnostic; the + // authoritative win count comes from the trade_records aggregation + // at the end of eval. The one-step misalignment between dones (prev) + // and pnl (curr) is statistically negligible over thousands of trades. + let pnl_step_usd_delta: f64 = pnl_step_close_usd_host.iter().map(|&v| v as f64).sum(); + realized_pnl_usd_cum += pnl_step_usd_delta; for b in 0..cli.n_backtests { if dones_host[b] > 0.5 { - let pnl_usd = if current_scale > 1e-9 { - rewards_host[b] as f64 / current_scale as f64 - } else { - rewards_host[b] as f64 - }; - pnl_cum_usd += pnl_usd; - // B-7 observability: raw_rewards is the shaped pnl in USD - // BEFORE scale and clamp (see rl_fused_reward_pipeline.cu). - // Direct trade-pnl signal — compare against pnl_cum_usd above - // to surface clamp-truncated tail losses. - realized_pnl_cum_usd += raw_rewards_host[b] as f64; + // Phase 0 reward-redesign: this is the SHAPED reward (pts × lots + // × Phase-5 shaping), NOT USD. Direct sum of raw_rewards exposes + // the gradient signal magnitude separate from authoritative pnl. + shaped_reward_close_event_cum += raw_rewards_host[b] as f64; total_trades += 1; - if pnl_usd > 0.0 { win_count += 1; } + if pnl_step_close_usd_host[b] > 0.0 { win_count += 1; } hold_time_sum += trade_duration_host[b] as f64; } } @@ -690,8 +727,9 @@ fn main() -> Result<()> { unit_trail: unit_trail_host, close_unit_index: &close_unit_index_host, frd_logits: frd_logits_host, - pnl_cum_usd, - realized_pnl_cum_usd, + shaped_reward_close_event_cum, + realized_pnl_usd_delta: pnl_step_usd_delta, + realized_pnl_usd_cum, total_trades, win_count, hold_time_sum, @@ -736,7 +774,10 @@ fn main() -> Result<()> { done_count, reward_sum, sps, - pnl_cum_usd, + // Phase 0 reward-redesign (2026-06-02): authoritative USD pnl + // (same source as eval_summary.total_pnl_usd). Was previously + // pnl_cum_usd which un-scaled clamped rewards — not USD. + realized_pnl_usd_cum, win_rate, elapsed, ); @@ -836,10 +877,14 @@ fn main() -> Result<()> { // they reflect the eval window only (matches eval_summary.json's // post-checkpoint trade slicing). let mut eval_windowed_act_hist: [f32; N_ACTIONS] = [0.0; N_ACTIONS]; - let mut eval_pnl_cum_usd: f64 = 0.0; - // B-7 observability: mirror of realized_pnl_cum_usd for the eval - // window. See declaration in train loop for rationale. - let mut eval_realized_pnl_cum_usd: f64 = 0.0; + // Phase 0 reward-redesign (2026-06-02): mirror of train-loop counters. + // `shaped_reward_close_event_cum` is sum of raw_rewards at close (pts + // × lots × Phase-5 shaping, NOT USD). `realized_pnl_usd_cum` is + // authoritative USD from `pnl_step_close_usd_d` (same source as + // eval_summary.total_pnl_usd), reset to 0 at the train→eval boundary + // so the last eval row matches eval_summary.total_pnl_usd within $50. + let mut eval_shaped_reward_close_event_cum: f64 = 0.0; + let mut eval_realized_pnl_usd_cum: f64 = 0.0; let mut eval_win_count: u64 = 0; let mut eval_total_trades: u64 = 0; let mut eval_hold_time_sum: f64 = 0.0; @@ -889,6 +934,22 @@ fn main() -> Result<()> { ) .context("eval diag snapshot_async")?; + // Phase 0 reward-redesign (2026-06-02): direct mapped-pinned + // readback of `pnl_step_close_usd_d`. Same rationale as train + // loop (no staging routing — avoid the cross-stream race). + // Mirrors the buffer reused from train; resets to current eval + // step's per-batch USD pnl delta. Eval step 0 still receives a + // valid current-step read (NOT a previous-step warmup like the + // staging path) — but the eval-step==0 accumulator skip below + // preserves the train→eval boundary semantics: the prior + // train-loop's leftover close events would otherwise leak. + read_slice_d_into( + &trainer.stream, + sim.pnl_step_close_usd_d(), + &mut pnl_step_close_usd_host, + ) + .context("read pnl_step_close_usd_d (eval)")?; + // Host-side reads from the (just-swapped) staging buffer. let rewards_host = diag_staging.read_rewards(); let dones_host = diag_staging.read_dones(); @@ -927,21 +988,31 @@ fn main() -> Result<()> { } } - // Per-trade cumulative stats — same scale-recovery as train. - let current_scale = trainer.read_isv_host(RL_REWARD_SCALE_INDEX); + // Phase 0 reward-redesign (2026-06-02): authoritative USD pnl + // delta sourced from the direct-readback above (NOT staging). + // Each call to `pnl_track_step` zeros the buffer then writes only + // THIS step's per-batch close pnl, so `pnl_step_close_usd_host` + // reflects exactly eval_step's close events — no train→eval + // leakage (the prior staging route needed an eval_step==0 skip + // because the one-step delay surfaced the LAST train step's + // closes on eval step 0; direct readback eliminates that path). + // + // Note: `dones_host` / `raw_rewards_host` are still PREVIOUS-step + // (diag staging), so the per-batch `if dones[b] > 0.5 { ... }` + // win attribution is misaligned by one step against the + // current-step pnl. That misalignment is statistically + // negligible over `n_eval_steps × b_size` close events; the + // authoritative win count is recomputed from `trade_records` + // at the end of eval anyway. + let eval_pnl_step_usd_delta: f64 = + pnl_step_close_usd_host.iter().map(|&v| v as f64).sum(); + eval_realized_pnl_usd_cum += eval_pnl_step_usd_delta; for b in 0..cli.n_backtests { if dones_host[b] > 0.5 { - let pnl_usd = if current_scale > 1e-9 { - rewards_host[b] as f64 / current_scale as f64 - } else { - rewards_host[b] as f64 - }; - eval_pnl_cum_usd += pnl_usd; - // B-7 observability: raw pnl (pre-scale, pre-clamp) — see - // train-loop counterpart. - eval_realized_pnl_cum_usd += raw_rewards_host[b] as f64; + // SHAPED reward (NOT USD) — gradient signal cumulative. + eval_shaped_reward_close_event_cum += raw_rewards_host[b] as f64; eval_total_trades += 1; - if pnl_usd > 0.0 { eval_win_count += 1; } + if pnl_step_close_usd_host[b] > 0.0 { eval_win_count += 1; } eval_hold_time_sum += trade_duration_host[b] as f64; } } @@ -989,8 +1060,9 @@ fn main() -> Result<()> { unit_trail: unit_trail_host, close_unit_index: &close_unit_index_host, frd_logits: frd_logits_host, - pnl_cum_usd: eval_pnl_cum_usd, - realized_pnl_cum_usd: eval_realized_pnl_cum_usd, + shaped_reward_close_event_cum: eval_shaped_reward_close_event_cum, + realized_pnl_usd_delta: eval_pnl_step_usd_delta, + realized_pnl_usd_cum: eval_realized_pnl_usd_cum, total_trades: eval_total_trades, win_count: eval_win_count, hold_time_sum: eval_hold_time_sum, @@ -1026,7 +1098,93 @@ fn main() -> Result<()> { ); } } - eval_diag.flush().context("eval diag: final flush")?; + // Phase 0 reward-redesign (2026-06-02): drain the eval staging + // pipeline and EMIT A FINAL DRAIN ROW. The other diag streams + // (dones / raw_rewards / actions / etc.) still have one-step + // double-buffer latency, so the LAST eval step's per-batch state + // is in the swapped-out half; this drain row surfaces it. The + // `realized_pnl_usd_*` fields are sourced from the direct-readback + // path which already captured the last eval step's pnl in the + // regular loop above — so the drain row's pnl delta is 0 and the + // cum carries forward unchanged. Cross-source consistency check + // (`signal_consistency` in tier1_5_verdict.py) reads THIS final + // row's `realized_pnl_usd_cum`. + diag_staging.sync_and_swap().context("eval drain sync_and_swap")?; + let drain_dones = diag_staging.read_dones(); + let drain_raw_rewards = diag_staging.read_raw_rewards(); + let drain_actions_raw = diag_staging.read_actions_raw(); + let drain_actions: Vec = drain_actions_raw.iter().map(|f| f.to_bits() as i32).collect(); + let drain_trade_duration = diag_staging.read_trade_duration(); + let drain_outcome_ema = diag_staging.read_outcome_ema(); + let drain_position_lots = diag_staging.read_position_lots(); + let drain_pyramid_count = diag_staging.read_pyramid_count(); + let drain_unit_entry_price = diag_staging.read_unit_entry_price(); + let drain_unit_entry_step = diag_staging.read_unit_entry_step(); + let drain_unit_lots = diag_staging.read_unit_lots(); + let drain_unit_trail = diag_staging.read_unit_trail(); + let drain_close_unit_index = diag_staging.read_close_unit_index(); + let drain_frd_logits = diag_staging.read_frd_logits(); + let drain_rewards = diag_staging.read_rewards(); + // No additional pnl accumulated on the drain row — direct readback + // already captured the last eval step's pnl in the regular loop. + let drain_delta: f64 = 0.0; + for b in 0..cli.n_backtests { + if drain_dones[b] > 0.5 { + eval_shaped_reward_close_event_cum += drain_raw_rewards[b] as f64; + eval_total_trades += 1; + eval_hold_time_sum += drain_trade_duration[b] as f64; + } + } + // Emit the drain row using the same builder for schema parity. + // `stats` reused from the last eval loop iteration; non-close-event + // fields (ISV, loss components) reflect post-last-train-step state. + let drain_diag_inputs = DiagInputs { + b_size: cli.n_backtests, + stats: last_stats.as_ref().expect("at least one eval step ran"), + rewards: drain_rewards, + dones: drain_dones, + actions: &drain_actions, + raw_rewards: drain_raw_rewards, + trade_duration: drain_trade_duration, + outcome_ema: drain_outcome_ema, + position_lots: &drain_position_lots, + pyramid_count: &drain_pyramid_count, + unit_entry_price: drain_unit_entry_price, + unit_entry_step: &drain_unit_entry_step, + unit_lots: &drain_unit_lots, + unit_trail: drain_unit_trail, + close_unit_index: &drain_close_unit_index, + frd_logits: drain_frd_logits, + shaped_reward_close_event_cum: eval_shaped_reward_close_event_cum, + realized_pnl_usd_delta: drain_delta, + realized_pnl_usd_cum: eval_realized_pnl_usd_cum, + total_trades: eval_total_trades, + win_count: eval_win_count, + hold_time_sum: eval_hold_time_sum, + trail_fired_total: eval_trail_fired_total, + trail_tighten_total: eval_trail_tighten_total, + trail_loosen_total: eval_trail_loosen_total, + pyramid_added_total: eval_pyramid_added_total, + partial_flat_total: eval_partial_flat_total, + partial_flat_long_total: eval_partial_flat_long_total, + partial_flat_short_total: eval_partial_flat_short_total, + conf_gate_total: eval_conf_gate_total, + frd_gate_total: eval_frd_gate_total, + heat_cap_total: eval_heat_cap_total, + windowed_act_hist: &eval_windowed_act_hist, + }; + let drain_step = (cli.n_steps + cli.n_eval_steps) as u64; + let drain_record = trainer.build_diag_value( + drain_step, + t_start.elapsed().as_secs_f32(), + &drain_diag_inputs, + ).context("build_diag_value (eval drain)")?; + writeln!(eval_diag, "{}", drain_record).context("eval diag drain: writeln jsonl")?; + eval_diag.flush().context("eval diag: final flush after drain")?; + eprintln!( + "eval drain row emitted at step {drain_step}: delta=${drain_delta:.2} → \ + realized_pnl_usd_cum=${eval_realized_pnl_usd_cum:.2}", + ); // Aggregate eval-phase trades across ALL b_size accounts, with // correct per-account head_before slicing. Replaces the previous diff --git a/crates/ml-alpha/src/trainer/integrated.rs b/crates/ml-alpha/src/trainer/integrated.rs index 7d22218a8..dee6b5b09 100644 --- a/crates/ml-alpha/src/trainer/integrated.rs +++ b/crates/ml-alpha/src/trainer/integrated.rs @@ -551,8 +551,8 @@ pub struct IntegratedStepStats { /// raw_rewards, trade_duration, outcome_ema, position_lots, pyramid_count, /// unit_entry_price, unit_entry_step, unit_lots, unit_trail, close_unit_index, /// frd_logits). All slices are borrowed (zero copies). -/// * Running counters (pnl_cum_usd, total_trades, win_count, …) maintained -/// by the caller across steps. +/// * Running counters (shaped_reward_close_event_cum, realized_pnl_usd_*, +/// total_trades, win_count, …) maintained by the caller across steps. /// /// The struct is consumed by both the train loop (in `alpha_rl_train.rs`) /// and the eval loop, guaranteeing schema parity between `diag.jsonl` and @@ -580,12 +580,26 @@ pub struct DiagInputs<'a> { pub close_unit_index: &'a [i32], pub frd_logits: &'a [f32], // ── Running counters (caller-maintained across steps) ─────────── - pub pnl_cum_usd: f64, - // B-7 observability: parallel counter computed from raw_rewards - // (pre-clamp, pre-scale shaped pnl in USD). Compare against - // pnl_cum_usd (clamp-truncated) to expose the reward-hacking gap - // — when these diverge, the clamp is masking tail losses. - pub realized_pnl_cum_usd: f64, + // Phase 0 reward-redesign (2026-06-02): cumulative SHAPED reward + // summed over close events. NOT in USD — sum of `raw_rewards[b]` + // (Phase-5-shaped reward post pnl-aligned envelope) at done-event + // steps. Renamed from `realized_pnl_cum_usd` because the underlying + // values are pts × lots × shaping, not USD. Investigate divergence + // from `realized_pnl_usd_cum` (authoritative USD) when comparing + // gradient-signal cumulatives against ground-truth pnl. + pub shaped_reward_close_event_cum: f64, + // Phase 0 reward-redesign (2026-06-02): authoritative USD pnl delta + // for THIS step, summed across batches. Source is the per-batch + // `pnl_step_close_usd_d` buffer written by `pnl_track_step`'s close + // branch (realised_pnl_usd_fp / 100). Same data path as + // `eval_summary.total_pnl_usd`; cross-source consistent within + // single-trade fp/100 precision ($0.01). + pub realized_pnl_usd_delta: f64, + // Phase 0 reward-redesign (2026-06-02): cumulative authoritative + // USD pnl since start of phase (train or eval). Reset at the + // train→eval boundary so it tracks the eval window only. Last-row + // value MUST match `eval_summary.total_pnl_usd` within $50. + pub realized_pnl_usd_cum: f64, pub total_trades: u64, pub win_count: u64, pub hold_time_sum: f64, @@ -11344,11 +11358,22 @@ impl IntegratedTrainer { "gated_count_total": inputs.frd_gate_total, }, "trading": { - "pnl_cum_usd": inputs.pnl_cum_usd, - // B-7 observability: raw-reward-based cumulative pnl (pre-clamp). - // If this diverges from pnl_cum_usd, the clamp was masking tail - // pnl in the agent's reward signal. - "realized_pnl_cum_usd": inputs.realized_pnl_cum_usd, + // Phase 0 reward-redesign (2026-06-02): cumulative SHAPED + // reward at close events (sum of raw_rewards[b] when done). + // NOT in USD — pts × lots × Phase-5 shaping. Renamed from + // `realized_pnl_cum_usd` (the previous name lied about + // units). Compare against `realized_pnl_usd_cum` below to + // see the gradient-signal vs authoritative-pnl gap. + "shaped_reward_close_event_cum": inputs.shaped_reward_close_event_cum, + // Phase 0 reward-redesign (2026-06-02): authoritative USD + // pnl signals sourced from `pnl_track_step` (same fp/100 + // arithmetic as `eval_summary.total_pnl_usd`). `_delta` is + // this step's USD pnl across all batches; `_cum` is the + // running phase total (reset at train→eval boundary). Last + // eval row's `_cum` MUST match `eval_summary.total_pnl_usd` + // within $50. + "realized_pnl_usd_delta": inputs.realized_pnl_usd_delta, + "realized_pnl_usd_cum": inputs.realized_pnl_usd_cum, "total_trades": inputs.total_trades, "win_rate": win_rate, "avg_hold_steps": avg_hold, @@ -11833,7 +11858,7 @@ fn read_slice_d( /// methods (f32 / i32 / u32 / u8) and the per-step diag checksum /// (f64). For the hot path, `read_slice_d` / `read_slice_i32_d` keep /// their typed signatures to preserve their existing call sites. -pub(crate) fn read_slice_d_into( +pub fn read_slice_d_into( stream: &Arc, src: &CudaSlice, dst: &mut [T], diff --git a/crates/ml-alpha/tests/eval_diag_emission.rs b/crates/ml-alpha/tests/eval_diag_emission.rs index 2ef3b0102..ebed0955b 100644 --- a/crates/ml-alpha/tests/eval_diag_emission.rs +++ b/crates/ml-alpha/tests/eval_diag_emission.rs @@ -176,9 +176,16 @@ fn eval_diag_jsonl_emitted_with_train_schema_parity() -> Result<()> { n_train == n_steps, "diag.jsonl line count {n_train} != n_steps {n_steps}" ); + // Phase 0 reward-redesign (2026-06-02): eval_diag.jsonl emits one + // extra "drain" row at the end so the last row's + // `realized_pnl_usd_cum` matches `eval_summary.total_pnl_usd` (the + // staging double-buffer's one-step latency would otherwise leave the + // last eval step's close events out of the cumulative). Row count is + // therefore `n_eval_steps + 1`. anyhow::ensure!( - n_eval == n_eval_steps, - "eval_diag.jsonl line count {n_eval} != n_eval_steps {n_eval_steps}" + n_eval == n_eval_steps + 1, + "eval_diag.jsonl line count {n_eval} != n_eval_steps+1 (drain row) {}", + n_eval_steps + 1 ); // ── Leaf-path parity ──────────────────────────────────────────── @@ -255,7 +262,15 @@ fn eval_diag_jsonl_emitted_with_train_schema_parity() -> Result<()> { // attn_q + encoder activations (vsn_out, mamba2_l1_out, ln_a_out, // mamba2_l2_out, ln_b_out, attn_context) + mamba2_grad_w_c_l1. // New total: 711. - const EXPECTED_LEAVES: usize = 711; + // 2026-06-02 Phase 0 reward-redesign: net +1 leaf under `trading`. + // Removed `trading.pnl_cum_usd` (mathematically nonsense after popart + // whitens rewards_d in place). Renamed `trading.realized_pnl_cum_usd` + // → `trading.shaped_reward_close_event_cum` (truthful: it was SHAPED + // reward at close, not USD). Added `trading.realized_pnl_usd_delta` + // and `trading.realized_pnl_usd_cum` (authoritative USD from + // pnl_track_step's per-batch buffer; matches eval_summary.total_pnl_usd + // within fp/100 precision). Net: -1 + 2 = +1. New total: 712. + const EXPECTED_LEAVES: usize = 712; anyhow::ensure!( train_paths.len() == EXPECTED_LEAVES, "train leaf count {} != expected {EXPECTED_LEAVES} (schema drifted; \ @@ -266,17 +281,20 @@ fn eval_diag_jsonl_emitted_with_train_schema_parity() -> Result<()> { // ── Eval step axis monotonicity ────────────────────────────────── // Verify the eval phase emits a step axis that continues past the // training horizon: eval[0].step == n_steps, - // eval[N-1].step == n_steps + n_eval_steps - 1. Catches a regression - // where the eval loop reused the training step counter (0..n_eval). + // eval[N-1].step (last regular eval row) == n_steps + n_eval_steps - 1, + // and the drain row (Phase 0 reward-redesign 2026-06-02) sits at + // step == n_steps + n_eval_steps (one past the last regular step). + // Catches a regression where the eval loop reused the training step + // counter (0..n_eval). let eval_lines: Vec = std::fs::read_to_string(&eval_diag_path) .with_context(|| format!("read {}", eval_diag_path.display()))? .lines() .map(|l| serde_json::from_str(l).expect("parse eval line")) .collect(); anyhow::ensure!( - eval_lines.len() == n_eval_steps, - "eval_diag.jsonl parsed {} lines != n_eval_steps {n_eval_steps}", - eval_lines.len() + eval_lines.len() == n_eval_steps + 1, + "eval_diag.jsonl parsed {} lines != n_eval_steps+1 (drain row) {}", + eval_lines.len(), n_eval_steps + 1, ); let first_step = eval_lines .first() @@ -284,21 +302,69 @@ fn eval_diag_jsonl_emitted_with_train_schema_parity() -> Result<()> { .get("step") .and_then(|v| v.as_u64()) .context("eval[0].step missing or not u64")?; - let last_step = eval_lines + let drain_step = eval_lines .last() .unwrap() .get("step") .and_then(|v| v.as_u64()) - .context("eval[N-1].step missing or not u64")?; + .context("eval[N].step (drain) missing or not u64")?; + let last_regular_step = eval_lines[n_eval_steps - 1] + .get("step") + .and_then(|v| v.as_u64()) + .context("eval[N-1].step (last regular) missing or not u64")?; let expected_first = n_steps as u64; - let expected_last = (n_steps + n_eval_steps - 1) as u64; + let expected_last_regular = (n_steps + n_eval_steps - 1) as u64; + let expected_drain = (n_steps + n_eval_steps) as u64; anyhow::ensure!( first_step == expected_first, "eval[0].step expected n_steps={expected_first}, got {first_step}" ); anyhow::ensure!( - last_step == expected_last, - "eval[N-1].step expected n_steps+n_eval-1={expected_last}, got {last_step}" + last_regular_step == expected_last_regular, + "eval[N-1].step expected n_steps+n_eval-1={expected_last_regular}, got {last_regular_step}" + ); + anyhow::ensure!( + drain_step == expected_drain, + "eval drain row step expected n_steps+n_eval={expected_drain}, got {drain_step}" + ); + + // ── Phase 0 reward-redesign: USD pnl-delta + cumulative invariants ─── + // 1) Every row has both new fields. + // 2) `realized_pnl_usd_cum` is the running sum of `realized_pnl_usd_delta` + // within the eval window (monotone iff non-negative deltas; here we + // only check that abs(cum - running_sum) < $0.05 per row to allow for + // f32→f64 conversion noise). + let mut running_cum: f64 = 0.0; + for (i, row) in eval_lines.iter().enumerate() { + let trading = row.get("trading").with_context(|| format!("eval[{i}].trading missing"))?; + let delta = trading + .get("realized_pnl_usd_delta") + .and_then(|v| v.as_f64()) + .with_context(|| format!("eval[{i}].trading.realized_pnl_usd_delta missing or not f64"))?; + let cum = trading + .get("realized_pnl_usd_cum") + .and_then(|v| v.as_f64()) + .with_context(|| format!("eval[{i}].trading.realized_pnl_usd_cum missing or not f64"))?; + running_cum += delta; + let gap = (running_cum - cum).abs(); + anyhow::ensure!( + gap < 0.05, + "eval[{i}] realized_pnl_usd_cum {cum} disagrees with running sum {running_cum} by {gap}", + ); + } + // 3) Sanity: the legacy USD field (`pnl_cum_usd`) MUST be gone from the + // schema. If a future change re-introduces it, the leaf-count check + // will already catch the drift, but this gives a direct error message. + let trading0 = eval_lines[0].get("trading").context("trading missing")?; + anyhow::ensure!( + trading0.get("pnl_cum_usd").is_none(), + "trading.pnl_cum_usd resurfaced — Phase 0 cleanup regressed" + ); + anyhow::ensure!( + trading0.get("realized_pnl_cum_usd").is_none(), + "trading.realized_pnl_cum_usd resurfaced — Phase 0 rename regressed \ + (use realized_pnl_usd_cum for authoritative USD, \ + shaped_reward_close_event_cum for the gradient-signal cumulative)" ); eprintln!( diff --git a/crates/ml-backtesting/cuda/pnl_track.cu b/crates/ml-backtesting/cuda/pnl_track.cu index 952ede0c5..5b472b3b1 100644 --- a/crates/ml-backtesting/cuda/pnl_track.cu +++ b/crates/ml-backtesting/cuda/pnl_track.cu @@ -81,7 +81,16 @@ extern "C" __global__ void pnl_track_step( // CRT.diag Group D: outcome by entry-conviction (10 conviction buckets). unsigned int* __restrict__ diag_outcome_n, // [n_backtests * 10] float* __restrict__ diag_outcome_sum_pnl, // [n_backtests * 10] — segment_realized (price-units × lots) - unsigned int* __restrict__ diag_outcome_n_wins // [n_backtests * 10] + unsigned int* __restrict__ diag_outcome_n_wins, // [n_backtests * 10] + // Phase 0 reward-redesign (2026-06-02): per-batch authoritative USD pnl + // delta for THIS pnl_track call. Buffer is zeroed host-side before each + // step; close branch writes realised_usd_to_write / 100.0 here. Single + // writer (thread 0 per block) — no race, no atomicAdd needed per + // `feedback_no_atomicadd`. Same source as the eval_summary aggregator + // (`realised_pnl_usd_fp / 100.0`), making the per-step diag signal + // cross-source consistent with `eval_summary.total_pnl_usd` to within + // single-trade fp/100 precision. + float* __restrict__ pnl_step_close_usd // [n_backtests] ) { int b = blockIdx.x; if (b >= n_backtests || threadIdx.x != 0) return; @@ -212,6 +221,15 @@ extern "C" __global__ void pnl_track_step( rec[38] = 0; rec[39] = 0; + // Phase 0 reward-redesign (2026-06-02): emit authoritative USD pnl + // delta for this close event. Buffer is zeroed host-side at step + // start so a non-zero value here means "batch b closed a trade + // during this step." Single writer (thread 0 of block b) — no race. + // Source is the same fixed-point fp/100 used by eval_summary, so + // the diag cumulative is bit-equal to the eval_summary aggregator + // within single-trade fp/100 precision ($0.01). + pnl_step_close_usd[b] = (float)realised_usd_to_write / 100.0f; + // Reset open-trade scratch. #pragma unroll for (int i = 0; i < OPEN_TRADE_STATE_BYTES; ++i) st[i] = 0; diff --git a/crates/ml-backtesting/src/sim/mod.rs b/crates/ml-backtesting/src/sim/mod.rs index 987bbc777..22d1b1c6b 100644 --- a/crates/ml-backtesting/src/sim/mod.rs +++ b/crates/ml-backtesting/src/sim/mod.rs @@ -349,6 +349,18 @@ pub struct LobSimCuda { pub(crate) diag_smoothed_current_run_length_d: CudaSlice, // [n_backtests * N_HORIZONS] pub(crate) diag_smoothed_sum_run_length_d: CudaSlice, // [n_backtests * N_HORIZONS] pub(crate) diag_smoothed_prev_dir_d: CudaSlice, // [n_backtests * N_HORIZONS] + + // Phase 0 reward-redesign (2026-06-02): per-batch authoritative USD pnl + // delta written by pnl_track_step's close branch. Buffer is zeroed + // host-side at the start of every `step_pnl_track` call; a non-zero + // value at index b means batch b closed a trade during this step's + // pnl_track pass. The trainer reads this per-step via + // `read_slice_d_into` and emits both the per-step sum + // (`trading.realized_pnl_usd_delta`) and a cumulative + // (`trading.realized_pnl_usd_cum`) in the diag JSONL. Same source as + // `eval_summary.total_pnl_usd` (realised_pnl_usd_fp / 100), making the + // diag's pnl cross-source consistent with eval_summary within $0.01. + pub(crate) pnl_step_close_usd_d: CudaSlice, // [n_backtests] } impl LobSimCuda { @@ -679,6 +691,12 @@ impl LobSimCuda { .alloc_zeros::(n_backtests * N_HORIZONS) .context("alloc diag_smoothed_prev_dir_d")?; + // Phase 0 reward-redesign (2026-06-02): per-batch authoritative USD + // pnl-delta buffer. Zero-init; reset per-step by `step_pnl_track`. + let pnl_step_close_usd_d = stream + .alloc_zeros::(n_backtests) + .context("alloc pnl_step_close_usd_d")?; + // CRT Phase A0.5 corrective: on-device conviction history buffer. // Capacity = 5M decisions × 4B/f32 = 20MB. Matches the prior // host-side `Vec::with_capacity(3_000_000)` plus headroom for @@ -790,6 +808,7 @@ impl LobSimCuda { diag_smoothed_current_run_length_d, diag_smoothed_sum_run_length_d, diag_smoothed_prev_dir_d, + pnl_step_close_usd_d, }) } @@ -1329,6 +1348,19 @@ impl LobSimCuda { &self.pos_d } + /// Phase 0 reward-redesign (2026-06-02): read-only accessor for the + /// per-batch authoritative USD pnl-delta buffer. Each entry holds the + /// USD pnl of the trade closed by batch `b` during the most recent + /// `step_pnl_track` call, or 0.0 if no close occurred. Trainers route + /// this device pointer through `DiagStaging::snapshot_async` so the + /// one-step-delay double-buffered read stays aligned with the other + /// per-batch close-event signals (dones / raw_rewards). Same source as + /// `eval_summary.total_pnl_usd` (realised_pnl_usd_fp / 100), giving a + /// cross-source-consistent per-step USD signal. + pub fn pnl_step_close_usd_d(&self) -> &CudaSlice { + &self.pnl_step_close_usd_d + } + /// SP20 P1+P5 trail-stop integration: read-only accessor for the /// shared best-bid price array (`BOOK_LEVELS` floats, lobsim-wide, /// not per-batch). Used by `rl_trail_stop_check` to compute @@ -1447,6 +1479,17 @@ impl LobSimCuda { let n = self.n_backtests as i32; let cap = crate::lob::TRADE_LOG_CAP as i32; let rs = self.stream.cu_stream(); + // Phase 0 reward-redesign (2026-06-02): zero the per-batch USD + // pnl-delta buffer before launch. Close branch writes the per-trade + // USD pnl for batches that closed THIS step; other batches keep + // their zero. Stream-ordered with the kernel launch below. + unsafe { + ml_alpha::trainer::raw_launch::raw_memset_d8_zero( + self.pnl_step_close_usd_d.raw_ptr(), + self.pnl_step_close_usd_d.num_bytes(), + rs, + ).map_err(|e| anyhow::anyhow!("zero pnl_step_close_usd_d: {:?}", e))?; + } { use ml_alpha::trainer::raw_launch::{RawArgs, raw_launch}; let mut args = RawArgs::new(); @@ -1466,6 +1509,7 @@ impl LobSimCuda { args.push_ptr(self.diag_outcome_n_d.raw_ptr()); args.push_ptr(self.diag_outcome_sum_pnl_d.raw_ptr()); args.push_ptr(self.diag_outcome_n_wins_d.raw_ptr()); + args.push_ptr(self.pnl_step_close_usd_d.raw_ptr()); let mut ptrs = args.build_arg_ptrs(); unsafe { raw_launch( @@ -1986,6 +2030,16 @@ impl LobSimCuda { self.dispatch_latent_market_orders(current_ts_ns, sim_cfg)?; // Step 4: pnl_track — detect close, emit TradeRecord. + // Phase 0 reward-redesign (2026-06-02): zero per-batch USD pnl-delta + // buffer first; kernel's close branch writes the per-trade USD pnl + // for batches that closed this step. Stream-ordered with the launch. + unsafe { + ml_alpha::trainer::raw_launch::raw_memset_d8_zero( + self.pnl_step_close_usd_d.raw_ptr(), + self.pnl_step_close_usd_d.num_bytes(), + self.stream.cu_stream(), + ).map_err(|e| anyhow::anyhow!("zero pnl_step_close_usd_d (decision path): {:?}", e))?; + } let cap = crate::lob::TRADE_LOG_CAP as i32; let mut launch = self.stream.launch_builder(&self.pnl_track_fn); unsafe { @@ -2011,6 +2065,8 @@ impl LobSimCuda { .arg(&mut self.diag_outcome_n_d) .arg(&mut self.diag_outcome_sum_pnl_d) .arg(&mut self.diag_outcome_n_wins_d) + // Phase 0 reward-redesign: per-batch USD pnl-delta buffer. + .arg(&mut self.pnl_step_close_usd_d) .launch(cfg)?; } self.stream.synchronize()?; diff --git a/scripts/tier1_5_verdict.py b/scripts/tier1_5_verdict.py index 143f48a91..09891551a 100755 --- a/scripts/tier1_5_verdict.py +++ b/scripts/tier1_5_verdict.py @@ -61,8 +61,10 @@ THRESHOLDS = { # Eval pnl > -$1M target, KILL < -$3M (catastrophic) "eval_pnl_kill": -3_000_000.0, "eval_pnl_target": -1_000_000.0, - # Per-step vs eval_summary discrepancy: > 5% gap → WARN (not KILL) - "consistency_warn_frac": 0.05, + # Phase 0 reward-redesign (2026-06-02): per-step vs eval_summary now + # read from the SAME `realised_pnl_usd_fp / 100` source — any gap > $50 + # (one $50/pt unit of float-summation slack) is a real source bug, KILL. + "consistency_kill_usd": 50.0, # Min steps required for a meaningful verdict. "min_train_rows": 500, "min_eval_rows": 100, @@ -221,12 +223,19 @@ def signal_reward_pnl_pearson(train_rows: list[dict[str, Any]]) -> SignalResult: decisive bug behind 64 negative-eval commits — reward gradient systematically anti-aligned with pnl direction. """ + # Phase 0 reward-redesign (2026-06-02): `trading.realized_pnl_cum_usd` + # was renamed to two distinct fields. We want the AUTHORITATIVE USD pnl + # cumulative (`realized_pnl_usd_cum`), because the Pearson signal asks + # whether the agent's gradient signal (`rewards.sum`) tracks the actual + # USD pnl direction — not whether it tracks itself (which would be the + # shaped-reward cumulative, which is mechanically the cumulative sum of + # raw_rewards itself and would trivially correlate). rewards: list[float] = [] dpnls: list[float] = [] prev_pnl: float | None = None for row in train_rows: rs = get_nested(row, "rewards.sum") - cur_pnl = get_nested(row, "trading.realized_pnl_cum_usd") + cur_pnl = get_nested(row, "trading.realized_pnl_usd_cum") if rs is None or cur_pnl is None: continue if not isinstance(rs, (int, float)) or not isinstance(cur_pnl, (int, float)): @@ -360,9 +369,12 @@ def signal_eval_pnl(eval_summary: dict[str, Any] | None, eval_rows: list[dict[str, Any]]) -> SignalResult: """Final eval pnl: KILL if catastrophic (< -$3M). - Prefer eval_summary.total_pnl_usd (authoritative) over per-step - realized_pnl_cum_usd (per `pearl_grwwh_eval_catastrophic_collapse` - these can disagree by hundreds of millions). + Prefer eval_summary.total_pnl_usd over per-step realized_pnl_usd_cum. + Phase 0 reward-redesign (2026-06-02) made these two sources read the + same per-trade `realised_pnl_usd_fp / 100` arithmetic — so they SHOULD + now agree within $50 (enforced by `signal_consistency` below). Before + Phase 0 they could disagree by hundreds of millions + (`pearl_grwwh_eval_catastrophic_collapse`). """ val: float | None = None source = "none" @@ -370,10 +382,12 @@ def signal_eval_pnl(eval_summary: dict[str, Any] | None, val = float(eval_summary["total_pnl_usd"]) source = "eval_summary.total_pnl_usd" elif eval_rows: - last = get_nested(eval_rows[-1], "trading.realized_pnl_cum_usd") + # Phase 0 reward-redesign (2026-06-02): authoritative USD cum + # (was `realized_pnl_cum_usd` — which actually wasn't USD). + last = get_nested(eval_rows[-1], "trading.realized_pnl_usd_cum") if isinstance(last, (int, float)): val = float(last) - source = "eval_diag last realized_pnl_cum_usd" + source = "eval_diag last realized_pnl_usd_cum" if val is None: return SignalResult("eval_pnl", None, SignalResult.STATUS_SKIP, "neither eval_summary.json nor eval_diag.jsonl has pnl") @@ -391,9 +405,15 @@ def signal_consistency(eval_rows: list[dict[str, Any]], eval_summary: dict[str, Any] | None) -> SignalResult: """Per-step vs eval_summary consistency check (§3.6). - WARN (not KILL): the per-step diag's realized_pnl_cum_usd and - eval_summary.json's total_pnl_usd should match within 5%. A divergence - indicates the per-step diag is misleading (per `pearl_grwwh_eval_catastrophic_collapse`). + Phase 0 reward-redesign (2026-06-02): KILL on $50+ disagreement between + the per-step `trading.realized_pnl_usd_cum` (last eval row) and + `eval_summary.total_pnl_usd`. Both sources are now computed from the + SAME per-trade `realised_pnl_usd_fp / 100` arithmetic — the diag reads + from `pnl_step_close_usd_d` (pnl_track.cu) and the summary reads + `TradeRecord.realised_pnl_usd_fp` from the trade log. They MUST match + within a single $0.01 unit; the $50 threshold is one $50/pt unit of + slack for floating-point summation order across batches. A larger gap + means one of the sources is broken — kill the run for investigation. """ if eval_summary is None: return SignalResult("per-step vs eval_summary", None, SignalResult.STATUS_SKIP, @@ -404,20 +424,19 @@ def signal_consistency(eval_rows: list[dict[str, Any]], if not eval_rows: return SignalResult("per-step vs eval_summary", None, SignalResult.STATUS_SKIP, "no eval rows") - per_step = get_nested(eval_rows[-1], "trading.realized_pnl_cum_usd") + per_step = get_nested(eval_rows[-1], "trading.realized_pnl_usd_cum") if per_step is None or not isinstance(per_step, (int, float)): return SignalResult("per-step vs eval_summary", None, SignalResult.STATUS_SKIP, - "last eval row missing realized_pnl_cum_usd") + "last eval row missing realized_pnl_usd_cum") summary = float(eval_summary["total_pnl_usd"]) per_step = float(per_step) diff = abs(per_step - summary) - denom = max(abs(per_step), abs(summary), 1.0) - frac = diff / denom - detail = f"per_step=${per_step:,.0f} | eval_summary=${summary:,.0f} | gap=${diff:,.0f} ({frac*100:.2f}%)" - if frac > THRESHOLDS["consistency_warn_frac"]: - return SignalResult("per-step vs eval_summary", frac, SignalResult.STATUS_WARN, - f"{detail} > {THRESHOLDS['consistency_warn_frac']*100:.0f}% — accounting axis discrepancy") - return SignalResult("per-step vs eval_summary", frac, SignalResult.STATUS_OK, detail) + detail = f"per_step=${per_step:,.2f} | eval_summary=${summary:,.2f} | gap=${diff:,.2f}" + kill_usd = THRESHOLDS["consistency_kill_usd"] + if diff > kill_usd: + return SignalResult("per-step vs eval_summary", diff, SignalResult.STATUS_KILL, + f"{detail} > ${kill_usd:.0f} — eval pnl source disagreement, investigate") + return SignalResult("per-step vs eval_summary", diff, SignalResult.STATUS_OK, detail) # ── Driver ────────────────────────────────────────────────────────────