Audit fixes: cache boundary staleness, 6e session bypass, GPU engine parity

Fixes all high+medium findings from the 2026-07-16 full-codebase audit.

High:
- data_provider: cache now rejects snapshots missing the latest closed
  bar (_has_latest_closed_bar, weekend-exempt). The wall-clock TTL served
  mid-candle snapshots at the next boundary, so live M15/H1 scans ran one
  full bar behind and the bar-close session check evaluated the wrong hour.
- signal_aggregator: gate 6e (SELL quality) only acts on actionable
  grades. It was re-grading session-blocked F reports to D, which
  re-opened the M5 pullback path (entry check: quality != "F") on
  session-blocked bars.
- gpu_backtest: port the live EMA-ribbon compression/expansion logic
  (+10 conf on expansion) into the GPU engine. Its absence diverged
  composite score/conf on 7-9% of bars and flipped signals on ~0.2%.

Medium:
- gpu_backtest: Bollinger precompute uses the exact pandas rolling calls
  (cumsum E[x2]-E[x]2 lost precision at gold price levels; bb_position
  error up to 7.6e-3 crossed signal thresholds, worst during squeezes).
- gpu_backtest: --confirm-tfs defaults to the pair factory's live
  confirmation map instead of a hardcoded 1h,4h.
- gpu_backtest: entries re-anchor to the first RAW bar after the signal
  bar, so >10-sigma spike bars dropped by cleaning are no longer skipped
  at entry (favorable bias).
- gpu_backtest: per-TF warmup pad — D1 confirms get >=210 bars at window
  start (62-day pad gave ~61 bars: no EMA200, unlike live).
- data_provider: thread-safety locks for cache + throttle (shared
  provider across the bot's 4-thread pool); outlier filter keeps the
  snap-back bar after a spike instead of punching a two-bar hole.
- telegram_bot: alert fingerprint gains a risk-relative entry bucket
  (materially-moved setups re-alert) and is committed only after a
  successful send (failed sends no longer consume the alert for 1h).
- engine: disabled_indicators=["macd"] now removes the MACD pillar from
  scoring (histogram still computed for divergence).

Verification: test_all.py 187/187 (new Module 11: 8 audit-fix
regressions incl. a live-vs-GPU parity sweep with ribbon coverage);
real-data parity sweep over 813 slices of 14mo XAU M15: zero divergence
(pre-fix: 7-9% of bars diverged); both original bug repros no longer
reproduce; cached end-to-end backtest run passes with the live confirm
map. Books shift slightly vs pre-fix runs — comparability note added to
CLAUDE.md; pinned tuned numbers need re-validation under the fixed
backtester.

telegram_bot.py

@@ -488,21 +488,41 @@ def _passes_quality(quality: str) -> bool:
     return _QUALITY_RANK.get(quality, 0) >= _QUALITY_RANK.get(_alert_min_quality, 1)
 
 
-def _should_alert(key: str, report: SignalReport) -> bool:
-    """True when this signal is new/changed for the key, or re-armed."""
+def _alert_fingerprint(report: SignalReport) -> tuple:
+    """Alert identity for dedup. Includes a risk-relative entry bucket:
+    a persisting signal whose entry drifts a few ticks between candles
+    keeps the same fingerprint, while a setup that moved by half its risk
+    distance or more (new pullback, new swing anchor) re-alerts instead
+    of being swallowed for the full re-arm hour."""
     setup = report.trade_setup
-    fp = (
+    entry_bucket = None
+    if setup is not None:
+        sl_dist = abs(setup.entry_price - setup.stop_loss)
+        entry_bucket = round(setup.entry_price / max(sl_dist * 0.5, 1e-9))
+    return (
         report.final_signal.name,
         report.signal_quality,
         setup.direction if setup else None,
+        entry_bucket,
     )
-    now = time.time()
-    if (_last_alert_fp.get(key) == fp
-            and now - _last_alert_ts.get(key, 0) < _ALERT_REARM_SECONDS):
-        return False
-    _last_alert_fp[key] = fp
-    _last_alert_ts[key] = now
-    return True
+
+
+def _should_alert(key: str, report: SignalReport) -> bool:
+    """True when this signal is new/changed for the key, or re-armed.
+
+    Pure check — call _commit_alert only after the message actually went
+    out, so a failed Telegram send doesn't consume the alert for an hour.
+    """
+    fp = _alert_fingerprint(report)
+    return not (
+        _last_alert_fp.get(key) == fp
+        and time.time() - _last_alert_ts.get(key, 0) < _ALERT_REARM_SECONDS
+    )
+
+
+def _commit_alert(key: str, report: SignalReport) -> None:
+    _last_alert_fp[key] = _alert_fingerprint(report)
+    _last_alert_ts[key] = time.time()
 
 
 # ── Candle-boundary scheduler ────────────────────────────────────────
@@ -563,13 +583,20 @@ async def _scan_timeframe(
         if _passes_quality(r.signal_quality)
         and _should_alert(f"{r.symbol}_{tf.value}", r)
     ]
+    sent_ok: Set[int] = set()
     for chat_id in _signal_targets():
-        for report in actionable:
+        for i, report in enumerate(actionable):
             detail = format_signal_message(report)
             try:
                 await bot.send_message(chat_id, detail, parse_mode=ParseMode.MARKDOWN)
+                sent_ok.add(i)
             except Exception as exc:
                 logger.warning(f"Failed to send signal for chat {chat_id}: {exc}")
+    # Commit fingerprints only for alerts that reached at least one target;
+    # a fully failed send stays eligible for the next scan.
+    for i, report in enumerate(actionable):
+        if i in sent_ok:
+            _commit_alert(f"{report.symbol}_{tf.value}", report)
 
 
 # ── WebSocket level-breach monitor ──────────────────────────────────
@@ -658,11 +685,15 @@ async def _handle_breach(bot, ws_symbol: str, pair_symbol: str, price: float, le
         f"*{_escape_md(label)}* @ `{_fmt_price(price)}`\n\n"
     )
     detail = header + format_signal_message(report)
+    sent = False
     for chat_id in _signal_targets():
         try:
             await bot.send_message(chat_id, detail, parse_mode=ParseMode.MARKDOWN)
+            sent = True
         except Exception as exc:
             logger.warning(f"Failed to send breach alert to chat {chat_id}: {exc}")
+    if sent:
+        _commit_alert(f"{pair_symbol}_{Timeframe.M15.value}", report)
 
 
 async def _ws_heartbeat(ws) -> None: