audit: GPU-engine cache-miss observability + dead-guard alignment + full parity test

2026-08-04 audit, Batch B. Verified first (h1 harness): the two engines
are byte-identical on every reachable input — 370-prefix flat-bar deck,
warmup band, 322 real-data M15 positions incl. all 7 zero-range bars,
zero diffs. The claimed ADX NaN desync / warmup-guard divergences are
unreachable below the shared 30-bar analyze() floor. Changes here are
therefore observability + hygiene, proven behavior-neutral by exact
reproduction of the pinned M15 12mo book (397 trades, +78.7111R,
byte-identical trade rows).

- _resolve_pos: cache-miss fallback to live-engine recompute now warns
  on first miss and counts every miss (was fully silent — a run could
  switch math with no trace).
- ROC/Keltner warmup guards textually aligned with the live engine so a
  future period/floor change can't open a silent divergence window.
- New regression: full-sweep parity test comparing ALL indicator
  signals/confidences, composite, and regimes on every prefix incl. a
  flat zero-DM/zero-TR run (the old parity test skipped
  ADX/MACD/ROC/Keltner/RSI entirely); plus a loud-fallback test.

Suite: 314 green. Book gate: oanda_m15_12mo_030 reproduced identically.

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

backtest/gpu_backtest.py

@@ -402,6 +402,11 @@ class GPUTechnicalEngine(TechnicalEngine):
         # cache key = timeframe.value (e.g. "15min", "1h", "4h")
         self.caches = caches
         self._current_tf: Optional[str] = None
+        # Cache-miss fallbacks recompute with the live engine's pandas
+        # math (different EWM seeding than the precomputed cache). That
+        # is a parity hazard, so it must never happen silently.
+        self.fallback_count = 0
+        self._fallback_warned = False
 
     def analyze(self, df: pd.DataFrame, symbol: str, timeframe: str) -> TechnicalAnalysis:
         # Latch the timeframe so per-indicator overrides know which cache to use.
@@ -419,9 +424,23 @@ class GPUTechnicalEngine(TechnicalEngine):
             return None, -1
         cache = self.caches.get(self._current_tf)
         if cache is None:
+            self._note_fallback(f"no cache for TF {self._current_tf!r}")
             return None, -1
         pos = cache.index_pos.get(df.index[-1])
-        return (cache, pos) if pos is not None else (None, -1)
+        if pos is None:
+            self._note_fallback(
+                f"{self._current_tf}: slice end {df.index[-1]} not in cache")
+            return None, -1
+        return cache, pos
+
+    def _note_fallback(self, why: str) -> None:
+        self.fallback_count += 1
+        if not self._fallback_warned:
+            self._fallback_warned = True
+            logger.warning(
+                "GPU cache miss → falling back to live-engine recompute "
+                "(%s). Further misses counted silently; check "
+                "fallback_count at end of run.", why)
 
     # ── EMA overrides ──────────────────────────────────────────────
 
@@ -826,7 +845,12 @@ class GPUTechnicalEngine(TechnicalEngine):
         c = self.config
         n = c.roc_period
 
-        if pos < n + 1:
+        # Mirror the live guard (engine._compute_roc: len < n + 5 → skip,
+        # i.e. emit only from pos >= n + 4). Unreachable in practice —
+        # analyze() floors at 30 bars and n+4 < 30 for shipped periods —
+        # but keep the engines textually aligned so a future period/floor
+        # change can't open a silent divergence window.
+        if pos < n + 4:
             return
 
         roc_val = float(cache.roc[pos])
@@ -876,6 +900,11 @@ class GPUTechnicalEngine(TechnicalEngine):
             return super()._compute_keltner(df, result)
 
         c = self.config
+        # Mirror the live guard (engine._compute_keltner: len < max(kc_period,
+        # kc_atr_period) + 5 → skip). Unreachable below the 30-bar analyze()
+        # floor with shipped periods; kept aligned for the same reason as ROC.
+        if pos < max(c.kc_period, c.kc_atr_period) + 4:
+            return
         latest = float(df["Close"].iloc[-1])
         kc_upper_val = float(cache.kc_upper[pos])
         kc_lower_val = float(cache.kc_lower[pos])

test_all.py

@@ -5525,6 +5525,84 @@ def _():
         os.environ.pop("OANDA_LOT_SIZE", None)
 
 
+@test("GPU parity: ALL indicators + composite identical on every prefix")
+def _():
+    # Closes the 2026-08-04 audit gap: the Module 11 parity test never
+    # compared ADX/MACD/ROC/Keltner/RSI outputs or the warmup band. This
+    # sweep diffs EVERY indicator signal/conf/value, the composite, and
+    # both regimes, on every analyzable prefix incl. a flat (zero-DM,
+    # zero-TR) run that exercises the divisor edge cases.
+    import gpu_backtest as _gb
+    from engine import TechnicalEngine as _LiveEng
+
+    rng = np.random.default_rng(11)
+    n = 160
+    steps = rng.normal(0.3, 2.0, n)
+    close = 2300 + np.cumsum(steps)
+    high = close + np.abs(rng.normal(1.0, 0.4, n))
+    low = close - np.abs(rng.normal(1.0, 0.4, n))
+    open_ = np.roll(close, 1); open_[0] = close[0]
+    p = close[79]
+    for arr in (close, high, low, open_):
+        arr[80:84] = p                       # flat run: H=L=O=C
+    idx = pd.date_range("2026-01-05", periods=n, freq="15min", tz="UTC")
+    deck = pd.DataFrame({"Open": open_, "High": high, "Low": low,
+                         "Close": close, "Volume": 1.0}, index=idx)
+
+    tech = TechnicalConfig()
+    caches = {"15min": _gb.precompute_indicators(deck, tech)}
+    live_eng = _LiveEng(tech)
+    gpu_eng = _gb.GPUTechnicalEngine(tech, caches)
+
+    for pos in range(30, n, 1):
+        sl = deck.iloc[:pos + 1]
+        ra = live_eng.analyze(sl, "XAU_USD", "15min")
+        rb = gpu_eng.analyze(sl, "XAU_USD", "15min")
+        assert set(ra.indicators) == set(rb.indicators), \
+            f"pos {pos}: indicator sets differ"
+        for name, a in ra.indicators.items():
+            b = rb.indicators[name]
+            assert str(a.signal) == str(b.signal), \
+                f"pos {pos} {name}: {a.signal} vs {b.signal}"
+            assert abs(a.confidence - b.confidence) < 1e-6, \
+                f"pos {pos} {name} conf: {a.confidence} vs {b.confidence}"
+        assert abs(ra.composite_score - rb.composite_score) < 1e-6, \
+            f"pos {pos} composite: {ra.composite_score} vs {rb.composite_score}"
+        assert ra.market_regime == rb.market_regime, f"pos {pos} regime"
+        assert ra.volatility_regime == rb.volatility_regime, f"pos {pos} vol"
+    assert gpu_eng.fallback_count == 0, "sweep must never miss the cache"
+
+
+@test("GPU parity: cache miss falls back LOUDLY (counter + warning)")
+def _():
+    import gpu_backtest as _gb
+    tech = TechnicalConfig()
+    idx = pd.date_range("2026-01-05", periods=60, freq="15min", tz="UTC")
+    rng = np.random.default_rng(3)
+    close = 2300 + np.cumsum(rng.normal(0, 1.5, 60))
+    deck = pd.DataFrame({"Open": close, "High": close + 1, "Low": close - 1,
+                         "Close": close, "Volume": 1.0}, index=idx)
+    caches = {"15min": _gb.precompute_indicators(deck, tech)}
+    gpu_eng = _gb.GPUTechnicalEngine(tech, caches)
+
+    alien = deck.copy()
+    alien.index = alien.index + pd.Timedelta(minutes=7)   # not in cache
+    import logging as _lg
+    records = []
+    h = _lg.Handler(); h.emit = lambda rec: records.append(rec)
+    _gb.logger.addHandler(h)
+    old_level = _gb.logger.level
+    _gb.logger.setLevel(_lg.WARNING)
+    try:
+        gpu_eng.analyze(alien, "XAU_USD", "15min")
+    finally:
+        _gb.logger.removeHandler(h)
+        _gb.logger.setLevel(old_level)
+    assert gpu_eng.fallback_count > 0, "cache miss must increment counter"
+    assert any("cache miss" in rec.getMessage() for rec in records), \
+        "first cache miss must emit a WARNING"
+
+
 # ══════════════════════════════════════════════════════════════════════
 #  SUMMARY
 # ══════════════════════════════════════════════════════════════════════