audit 2026-09-13: eleven delivery/measurement fixes; books byte-identical

Signal path
- A shipped setup always names a direction: grade C needs confidence
  >= 45 (gate 6c grants B at >= 45) while BUY/SELL needs >= 50, so C/B
  reports reached the alert gate as NEUTRAL with a full setup. The
  backtester traded them (the replay never reads final_signal) while
  Telegram rejected NEUTRAL first — 5/91 trades, +3.7R of +22.2R, in
  the H1 12-month book. _attach_core_setup stamps the setup's direction
  and repairs the pre-gate action text; no-setup reports stay NEUTRAL.
- analyze_pair fails the scan when ANY required frame is missing
  ("DATA UNAVAILABLE:", scan_kind='failed'); a lost confirmation frame
  used to run with an unearned -15 MTF penalty.

Outcomes and labels
- advisory_outcome.resolve_locked_floor: a gap the database cannot fill
  resolves at the locked trailing stop (GAP_LOCKED, result.floor) when
  the stop had crossed the entry; the rebuild fires on both gap reasons
  (rows 915/920: +1.14R / +0.84R floors; resolved sum -2.46R -> -0.48R).
- label_scan_outcomes: SCAN_LABEL_MAX_AGE_H = 72 h, stale rows never
  anchor the shared bar stream, labeled_at bumped on re-examination,
  _stream_since warns at its row cap (the Labor-Day pile would have
  blinded scan_outcomes around 2026-09-14).

Providers and scheduling
- The daily settlement pause (Mon-Thu 17:00 New York) is a scheduled
  closure for _has_latest_closed_bar (whole-span label walk-back) and
  _is_data_fresh (closed seconds are not age); bot and dashboard skip
  scans whose just-closed label sits in the pause (61 false stale rows
  in three nights).
- fetch_range retries a failed page in place (three failures on one
  window raise) and logs an ERROR for interior gaps of >3 unscheduled
  bars.
- pd.Timestamp.utcnow() replaced in providers/base.py and engine.py;
  requirements pin pandas<4, numpy<3.
- telegram_bot: boundary loops share one scan per bar with the breach
  handler (_remember_scan keys on the analysed bar; breach defers 20 s
  after an M15 boundary); the rotating file handler attaches in main()
  only (the test suite used to write into logs/telegram.log).
- dashboard_publisher: catch_up_timeframes re-scans boundaries missed
  while a cycle overran or the process was down.

Backtester and engine guards
- _replay_span returns the per-span GPU cache-miss count; run_backtest
  sums, logs it (must be 0) and writes cache_fallbacks to the manifest.
- regime_weight_table raises on a partial regime_weights_override.

Verification: 519 tests; M5/M15 2-yr, M15 OOS, H1/H4 3-yr books
byte-identical before and after; live signals.db untouched (checked on
a copy). CLAUDE.md carries the dated entry and the blocked-hours union
semantics that every book was validated under.

Claude-Session: https://claude.ai/code/session_0139xz1J6YccndA2be6yFnEo

backtest/gpu_backtest.py

@@ -1540,6 +1540,10 @@ def _replay_span(span: Tuple[int, int]):
     is applied by the caller in a sequential pass over the returned
     candidates, so parallel and sequential runs produce identical books.
     Must stay behaviorally identical to the original sequential loop.
+
+    Returns (skipped, candidates, bar_rows, cache_fallbacks) — the last
+    element is this span's GPU-cache-miss count, which the caller sums so
+    a parity hazard inside a worker process cannot stay invisible.
     """
     ctx = _REPLAY_CTX
     primary_df = ctx["primary_df"]
@@ -1550,6 +1554,12 @@ def _replay_span(span: Tuple[int, int]):
     bt_cfg = ctx["bt_cfg"]
     primary_dur = ctx["primary_dur"]
 
+    # One worker process handles several spans, so the engine's cache-miss
+    # counter must be per span, not cumulative across them.
+    engine = aggregator.tech_engine
+    if hasattr(engine, "fallback_count"):
+        engine.fallback_count = 0
+
     skipped: Dict[str, int] = {}
     candidates: List[Tuple[int, Trade]] = []
     is_xau = PAIR_INFO[bt_cfg.pair]["display"] == "XAU/USD"
@@ -1719,8 +1729,9 @@ def _replay_span(span: Tuple[int, int]):
                           if tag == "XAU_PB" else 0.0),
         )))
 
-    return skipped, candidates, (replay_record.finish_bar_rows(bar_buf)
-                                 if record_bars else None)
+    return (skipped, candidates,
+            (replay_record.finish_bar_rows(bar_buf) if record_bars else None),
+            int(getattr(engine, "fallback_count", 0)))
 
 
 def _reanchor_entries_to_raw(
@@ -1927,14 +1938,28 @@ def run_backtest(
     skipped_reasons: Dict[str, int] = {}
     candidates: List[Tuple[int, Trade]] = []
     bar_chunks: List = []
-    for skips, cands, bars in results:
+    cache_fallbacks = 0
+    for skips, cands, bars, fallbacks in results:
         for reason, count in skips.items():
             skipped_reasons[reason] = skipped_reasons.get(reason, 0) + count
         candidates.extend(cands)
         if bars is not None:
             bar_chunks.append(bars)
+        cache_fallbacks += fallbacks
     candidates.sort(key=lambda pc: pc[0])
 
+    # The workers' counters die with their processes, so surface the run
+    # total here and leave it on the parent engine for callers to read.
+    aggregator.tech_engine.fallback_count = cache_fallbacks
+    if cache_fallbacks > 0:
+        logger.warning(
+            f"GPU cache fallbacks: {cache_fallbacks} indicator recomputes "
+            f"(one per _compute_* override per missed bar) fell back to the "
+            f"live engine's pandas math (different EWM seeding) — parity "
+            f"hazard; see GPUTechnicalEngine._note_fallback")
+    else:
+        logger.info("GPU cache fallbacks: 0")
+
     # ── Phase 2: entry cooldown — the only sequential rule.
     accepted = _apply_cooldown(candidates, bt_cfg.cooldown_bars)
     trades: List[Trade] = [t for _, t in accepted]
@@ -1985,7 +2010,8 @@ def run_backtest(
                           skipped=skipped_reasons, n_candidates=len(candidates),
                           workers=workers,
                           backend="CuPy GPU" if _GPU_AVAILABLE else "NumPy CPU fallback",
-                          cache_files=cache_names, replay_seconds=replay_seconds),
+                          cache_files=cache_names, replay_seconds=replay_seconds,
+                          cache_fallbacks=cache_fallbacks),
             sim_df=sim_df, bar_chunks=bar_chunks or None)
         logger.info(f"--record: wrote run to {bt_cfg.record_dir}")
 

backtest/replay_record.py

@@ -1055,6 +1055,9 @@ def write_run(record_dir, *, trades, book_df, bt_cfg, platform_cfg, aggregator,
         "max_hold_bars": bt_cfg.max_hold_bars,
         "fixed_rr_tp": bt_cfg.fixed_rr_tp,
         "workers": run_info.get("workers"),
+        # >0 means some bar-analyses fell back to the live engine's pandas
+        # math instead of the pre-computed cache (a parity hazard).
+        "cache_fallbacks": run_info.get("cache_fallbacks"),
         "engine_fingerprint": engine_fp,
         "gate_fingerprint": gate_fp,
         "technical_config": dataclasses.asdict(platform_cfg.technical),