gitadmin/foxhunt
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(ml-alpha): Phase 0 — per-step authoritative USD pnl in diag + fix mislabeled fields

Browse Source 
The diag's `trading.*_pnl_*_usd` fields were systematically mislabeled, making
the eval verdict signal untrustworthy. Three quantities shared the `_usd`
suffix but were not in USD:

  - `eval_summary.total_pnl_usd` (USD ground truth, $50/pt ES applied)
  - `trading.realized_pnl_cum_usd` (cumulative SHAPED reward at close events;
    pts × lots × Phase-5 shaping; NOT USD despite the suffix)
  - `trading.pnl_cum_usd` (mathematically nonsense — reads rewards_d after
    apply_reward_scale AND rl_popart_normalize whiten it in-place; dividing
    by current_scale un-does only the first transform)

At baseline mid-smoke (b=128, 2000+500, seed=42): eval_summary.total_pnl_usd
= -$4,457,625 while diag.trading.realized_pnl_cum_usd = +$469k and
diag.trading.pnl_cum_usd = +$12.9k. 346x ratio, sign-opposite. Same
mechanism as `pearl_grwwh_eval_catastrophic_collapse` ($234M cluster
discrepancy). Every Pearson(reward, Δpnl_usd) computation against these
fields was shaped-vs-shaped tautology, not pnl alignment.

This commit:

  DELETE `trading.pnl_cum_usd` (mathematically nonsense post-popart).

  RENAME `trading.realized_pnl_cum_usd` -> `trading.shaped_reward_close_event_cum`.
  Truthful name; the underlying values are post-Phase-5 shaped reward
  summed at close events, useful for gradient-signal diagnostics but never
  to be confused with USD pnl.

  ADD `trading.realized_pnl_usd_delta` and `trading.realized_pnl_usd_cum` in
  diag.jsonl / eval_diag.jsonl. Source: new per-batch float buffer
  pnl_step_close_usd_d on LobSimCuda, zeroed via raw_memset_d8_zero before
  each pnl_track_step launch, written by pnl_track.cu's close branch as
  realised_pnl_usd_fp / 100.0 (same arithmetic as eval_summary's
  TradeRecord aggregator, applying ES $50/pt). Single-writer per block —
  no atomicAdd per feedback_no_atomicadd.

  Direct readback via read_slice_d_into<f32> from sim.pnl_step_close_usd_d()
  after step_with_lobsim_gpu returns (main stream already synchronized via
  self.stream.synchronize() at pnl_track_step exit). NOT routed through
  diag_staging double-buffer — a prior implementation tried this and
  broke determinism (cross-stream race between main-stream
  raw_memset_d8_zero + pnl_track_step write and diag-stream
  raw_memcpy_dtod_async read). Direct same-stream read avoids the race
  entirely and stays mega-graph compatible (the readback runs AFTER
  per-step pipeline finishes, outside any captured CUDA graph).

  Cumulative tracked Rust-side in alpha_rl_train.rs; reset at train->eval
  boundary so realized_pnl_usd_cum tracks the eval window only.

  scripts/tier1_5_verdict.py upgrade:
  - signal_reward_alignment reads trading.realized_pnl_usd_cum
  - signal_eval_pnl falls back to realized_pnl_usd_cum
  - signal_consistency upgraded from WARN-at-5% to KILL-at-$50 with a new
    consistency_kill_usd threshold key; cross-source disagreement is now
    a hard kill not a warning

  tests/eval_diag_emission.rs: bumped EXPECTED_LEAVES 711 -> 712 (-1 deleted,
  -0 renamed, +2 added); added invariant that cum == running_sum(delta) and
  regression check that legacy pnl_cum_usd / realized_pnl_cum_usd fields are
  absent from the schema; allowed n_eval_steps + 1 row count for the
  drain-row pattern.

Falsification gate (load-bearing, must hold every smoke from this commit
forward): `diag.last_eval_row.trading.realized_pnl_usd_cum` matches
`eval_summary.total_pnl_usd` within $50.

Validation (mid-smoke b=128, 2000 train + 500 eval, seed=42, RTX 3050):
  - cargo build --release --example alpha_rl_train -p ml-alpha: exit 0
  - cargo test -p ml-alpha --test eval_diag_emission: PASS (712 leaves)
  - local-mid-smoke.sh: exit 0, drain row at step 2500
  - cross-source consistency: eval_summary=-$4,457,625.00 vs
    diag.realized_pnl_usd_cum=-$4,457,625.18 -> delta=$0.18 (exactly fp/100
    f32 rounding noise; well within $50 tolerance)
  - ./scripts/determinism-check.sh --quick: PASS — all checksums.* leaves
    match across all 200 rows (rel-tol 1e-5, abs-tol 1e-7)
  - Pre-commit hook on staged diff: PASS

Memory pearls created in this session document the diagnostic chain:
  - pearl_diag_pnl_fields_are_shaped_reward_not_usd (the mislabeling root)
  - pearl_reward_signal_anti_aligned_with_pnl ADDENDUM 2026-06-02b
    (correction to the 2026-06-02 ADDENDUM — the "+0.93 Pearson at HEAD"
    finding was a measurement artifact because both sides were in shaped
    pts x lots units, not USD)
  - pearl_advantage_kernel_is_done_gated_td_not_gae (related signal-density
    gap — Phase 1.5 follow-up)
  - pearl_phase5_term_4_is_almost_potential (Ng-Harada-Russell analysis of
    Phase 5 shaping terms — Phase 1 follow-up)

Unlocks: TRUE Pearson(rewards.sum, Δpnl_usd) can now be measured per-step
at HEAD. The eval-collapse investigation (next phases: Ng-Harada-Russell
shaping cleanup + GAE upgrade) is now falsifiable with bit-deterministic
verdicts grounded in authoritative USD pnl.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
jgrusewski
2026-06-02 19:55:31 +02:00
parent 629ebd667c
commit cfaa420bd4
6 changed files with 436 additions and 94 deletions
Download Patch File Download Diff File Expand all files Collapse all files

250
crates/ml-alpha/examples/alpha_rl_train.rs
View File

@@ -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<f32> = 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<i32> = 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

53
crates/ml-alpha/src/trainer/integrated.rs
View File

@@ -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<T: Copy>(
pub fn read_slice_d_into<T: Copy>(
stream: &Arc<CudaStream>,
src: &CudaSlice<T>,
dst: &mut [T],

92
crates/ml-alpha/tests/eval_diag_emission.rs
View File

@@ -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<Value> = 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!(

20
crates/ml-backtesting/cuda/pnl_track.cu
View File

@@ -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;

56
crates/ml-backtesting/src/sim/mod.rs
View File

@@ -349,6 +349,18 @@ pub struct LobSimCuda {
pub(crate) diag_smoothed_current_run_length_d: CudaSlice<u32>, // [n_backtests * N_HORIZONS]
pub(crate) diag_smoothed_sum_run_length_d: CudaSlice<u64>, // [n_backtests * N_HORIZONS]
pub(crate) diag_smoothed_prev_dir_d: CudaSlice<i8>, // [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<f32>` 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<f32>, // [n_backtests]
}
impl LobSimCuda {
@@ -679,6 +691,12 @@ impl LobSimCuda {
.alloc_zeros::<i8>(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::<f32>(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<f32> {
&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()?;

59
scripts/tier1_5_verdict.py
View File

@@ -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 ────────────────────────────────────────────────────────────