audit: alert delivery, provider robustness, backtester NaN/cache guards (Batches C+E+F)

2026-08-04 Fable 5 QA audit. Three batches sharing one test file, all
verified by runnable repros before fixing, all gated below.

Batch C — telegram alert delivery (verified by h4 harness):
- Failed breach rescan shrinks the cooldown to 30 s instead of
  silencing the symbol for 5 min; breach during cooldown is PARKED and
  fired on the first tick after expiry (was dropped forever).
- Level cache keyed symbol→TF (M5 scans no longer wipe H1/H4 swing
  levels within minutes); breach labels TF-qualified.
- Markdown alert sends get a plain-text fallback (mirroring _safe_edit);
  fingerprint commits only when a message actually landed.
- Per-key asyncio.Lock closes the dedup TOCTOU between auto-scan and
  breach sends (forced-interleave repro produced duplicate alerts).
- _start_scheduler respawns dead per-TF loops (was all-or-nothing).
- Clean WS close now backs off (was a zero-delay reconnect hot loop);
  backoff resets only after a connection that received ticks; TD
  iter_ticks fires on_connect on the first RECEIVED frame (invalid key
  can no longer pin the backoff at minimum).
- Scan-failure logs carry the real traceback (was "NoneType: None").

Batch E — providers (verified by h5 harness):
- fetch_range: 3 consecutive chunk ERRORS raise instead of returning
  silently truncated history (verified 731-day silent gap repro; also
  covers 429 storms); pagination no-progress guard; head-truncation
  ERROR log.
- Single-flight per (symbol,TF): 4 concurrent scans = 1 API call (was
  4); cache records requested depth so a longer-lookback call refetches.
- Multi-bar forming/future tail fully dropped (was iloc[:-1] once).
- Oanda: malformed candles skipped, never 0.0-defaulted; cleaning drops
  non-positive-price rows BEFORE OHLC repair; inf returns neutralized so
  one corrupt bar can no longer disarm the whole 10-sigma filter.

Batch F — backtester internals (verified by h6 harness):
- NaN bar in a trailing window exits at the last valid bar (was NaN R
  poisoning every metric); static-path TIMEOUT NaN guard;
  compute_metrics excludes non-finite R loudly. Zero real-data exposure
  (0 NaN rows in all 10 Oanda parquets) — defensive.
- fetch_history ERROR-logs pre-2026-07-08 cache files (cleaned-bars era)
  when loaded for raw-path resolution.

Gates: suite 333/333 green; pinned M15 12mo book reproduced identically
(397 trades, +78.7111R) after E and after F; M5 HEAD-config baseline
reproduced identically (556 trades, +97.5604R) after F.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01UnZkzeFcTu9agFg8BDh2tn

providers/base.py

@@ -86,9 +86,15 @@ class BaseDataProvider(ABC):
         self.config = config
         self._cache: Dict[str, pd.DataFrame] = {}
         self._cache_timestamps: Dict[str, float] = {}
+        # Requested fetch depth per key: a cached short frame must not
+        # satisfy a caller asking for a longer lookback.
+        self._cache_outputsize: Dict[str, int] = {}
         # The bot shares one provider across a thread pool; cache eviction
         # is check-then-act and needs locking.
         self._cache_lock = threading.Lock()
+        # Single-flight: one API fetch per (symbol, TF) at a time; other
+        # threads wait and then read the cache.
+        self._inflight: Dict[str, threading.Event] = {}
 
     # ── Abstract: the only provider-specific piece ─────────────────
 
@@ -119,68 +125,97 @@ class BaseDataProvider(ABC):
             lookback_days = self.config.data_lookback_days
 
         cache_key = f"{symbol}_{timeframe.value}"
-        now = time.time()
-
-        cache_ttl = self.CANDLE_SECONDS.get(timeframe, self.config.refresh_interval)
-        with self._cache_lock:
-            cached = self._cache.get(cache_key)
-            cached_at = self._cache_timestamps.get(cache_key)
-        if (
-            cached is not None
-            and cached_at is not None
-            and (now - cached_at) < cache_ttl
-            and not cached.empty
-        ):
-            if (self._is_data_fresh(cached, symbol, timeframe)
-                    and self._has_latest_closed_bar(cached, timeframe)):
-                return cached
-            logger.warning(f"{symbol} {timeframe.value}: cached data is stale, refetching")
-            with self._cache_lock:
-                self._cache.pop(cache_key, None)
-                self._cache_timestamps.pop(cache_key, None)
-
         outputsize = max(
             300,
             min(int(lookback_days * self.BARS_PER_DAY[timeframe]) + 20,
                 self.MAX_BARS_PER_REQUEST),
         )
-        df = self._fetch_raw(symbol, timeframe, outputsize)
-        df = self._drop_forming_candle(df, timeframe)
-        df = self._validate_and_clean(df, symbol, timeframe)
 
-        # The closed-bar guarantee must hold for FRESH responses too, not
-        # just cache hits: the scanner fires seconds after the boundary,
-        # and a lagging feed can answer without the just-closed bar — the
-        # scan would silently analyze (and session-check) the previous
-        # bar. Give the feed one short grace retry, then proceed loudly.
-        if not df.empty and not self._has_latest_closed_bar(df, timeframe):
-            time.sleep(self.FRESH_FETCH_RETRY_SECONDS)
-            try:
-                retry = self._fetch_raw(symbol, timeframe, outputsize)
-                retry = self._drop_forming_candle(retry, timeframe)
-                retry = self._validate_and_clean(retry, symbol, timeframe)
-                if retry is not None and not retry.empty:
-                    df = retry
-            except DataIntegrityError as e:
+        while True:
+            now = time.time()
+            cache_ttl = self.CANDLE_SECONDS.get(
+                timeframe, self.config.refresh_interval)
+            with self._cache_lock:
+                cached = self._cache.get(cache_key)
+                cached_at = self._cache_timestamps.get(cache_key)
+                cached_size = self._cache_outputsize.get(cache_key, 0)
+            if (
+                cached is not None
+                and cached_at is not None
+                and (now - cached_at) < cache_ttl
+                and not cached.empty
+                # A caller asking for MORE history than the cached fetch
+                # requested must not be served the short frame silently.
+                and cached_size >= outputsize
+            ):
+                if (self._is_data_fresh(cached, symbol, timeframe)
+                        and self._has_latest_closed_bar(cached, timeframe)):
+                    return cached
+                logger.warning(f"{symbol} {timeframe.value}: cached data is stale, refetching")
+                with self._cache_lock:
+                    self._cache.pop(cache_key, None)
+                    self._cache_timestamps.pop(cache_key, None)
+                    self._cache_outputsize.pop(cache_key, None)
+
+            # Single-flight: concurrent scans for the same (symbol, TF)
+            # must not each hit the API (the four TF loops fire together
+            # at aligned boundaries and Oanda has no client throttle).
+            # First caller claims the key and fetches; the rest wait,
+            # then re-check the cache.
+            with self._cache_lock:
+                inflight = self._inflight.get(cache_key)
+                if inflight is None:
+                    self._inflight[cache_key] = threading.Event()
+                    break
+            inflight.wait(timeout=90)
+
+        try:
+            df = self._fetch_raw(symbol, timeframe, outputsize)
+            df = self._drop_forming_candle(df, timeframe)
+            df = self._validate_and_clean(df, symbol, timeframe)
+
+            # The closed-bar guarantee must hold for FRESH responses too,
+            # not just cache hits: the scanner fires seconds after the
+            # boundary, and a lagging feed can answer without the
+            # just-closed bar — the scan would silently analyze (and
+            # session-check) the previous bar. Give the feed one short
+            # grace retry, then proceed loudly.
+            if not df.empty and not self._has_latest_closed_bar(df, timeframe):
+                time.sleep(self.FRESH_FETCH_RETRY_SECONDS)
+                try:
+                    retry = self._fetch_raw(symbol, timeframe, outputsize)
+                    retry = self._drop_forming_candle(retry, timeframe)
+                    retry = self._validate_and_clean(retry, symbol, timeframe)
+                    if retry is not None and not retry.empty:
+                        df = retry
+                except DataIntegrityError as e:
+                    logger.warning(
+                        f"{symbol} {timeframe.value}: boundary retry refetch "
+                        f"failed ({e}) — keeping original data")
+                if not self._has_latest_closed_bar(df, timeframe):
+                    logger.warning(
+                        f"{symbol} {timeframe.value}: feed lagging — just-closed bar "
+                        f"still missing after retry; this scan may run one bar behind"
+                    )
+
+            if not self._is_data_fresh(df, symbol, timeframe):
                 logger.warning(
-                    f"{symbol} {timeframe.value}: boundary retry refetch "
-                    f"failed ({e}) — keeping original data")
-            if not self._has_latest_closed_bar(df, timeframe):
-                logger.warning(
-                    f"{symbol} {timeframe.value}: feed lagging — just-closed bar "
-                    f"still missing after retry; this scan may run one bar behind"
+                    f"STALE DATA from API: {symbol} {timeframe.value} — "
+                    f"last bar {df.index[-1]} is too old. Signal quality may be degraded."
                 )
 
-        if not self._is_data_fresh(df, symbol, timeframe):
-            logger.warning(
-                f"STALE DATA from API: {symbol} {timeframe.value} — "
-                f"last bar {df.index[-1]} is too old. Signal quality may be degraded."
-            )
-
-        with self._cache_lock:
-            self._cache[cache_key] = df
-            self._cache_timestamps[cache_key] = now
-        return df
+            with self._cache_lock:
+                self._cache[cache_key] = df
+                self._cache_timestamps[cache_key] = now
+                self._cache_outputsize[cache_key] = outputsize
+            return df
+        finally:
+            # Release waiters even when the fetch raised — they re-check
+            # the cache and (on a miss) claim the key themselves.
+            with self._cache_lock:
+                ev = self._inflight.pop(cache_key, None)
+            if ev is not None:
+                ev.set()
 
     def fetch_range(
         self,
@@ -199,6 +234,7 @@ class BaseDataProvider(ABC):
         chunks = []
         cursor_end = end
         empty_streak = 0
+        error_streak = 0
         while cursor_end > start:
             cursor_start = max(start, cursor_end - timedelta(days=days_per_chunk))
             try:
@@ -207,19 +243,27 @@ class BaseDataProvider(ABC):
                     start=cursor_start, end=cursor_end,
                 )
             except Exception as e:
-                # Providers raise for windows with no data (weekends,
-                # holidays, asset gaps). Skip the window, step back, retry
-                # — but bail after a few consecutive empties.
+                # A raising chunk is a PROVIDER ERROR (auth, 429, 5xx,
+                # malformed payload) — not "no data". Three in a row means
+                # the rest of the history is unreachable: raise instead of
+                # returning a silently truncated frame (the backtester
+                # would treat it as complete).
                 logger.warning(
                     f"Chunk {cursor_start.date()}–{cursor_end.date()} failed: {e}"
                 )
-                empty_streak += 1
-                if empty_streak >= 3:
-                    break
+                error_streak += 1
+                if error_streak >= 3:
+                    raise self.DATA_ERROR(
+                        f"{symbol} {timeframe.value}: {error_streak} "
+                        f"consecutive chunk failures walking back from "
+                        f"{cursor_end.date()} — refusing to return "
+                        f"truncated history (last error: {e})") from e
                 cursor_end = cursor_start - timedelta(seconds=1)
                 continue
 
             if chunk is None or chunk.empty:
+                # Genuinely empty windows (weekends/holidays/asset gaps)
+                # are normal; several in a row = start of history.
                 empty_streak += 1
                 if empty_streak >= 3:
                     break
@@ -227,9 +271,18 @@ class BaseDataProvider(ABC):
                 continue
 
             empty_streak = 0
+            error_streak = 0
             chunks.append(chunk)
             # Step back; dedup on concat.
-            cursor_end = chunk.index[0] - timedelta(seconds=1)
+            new_cursor = chunk.index[0] - timedelta(seconds=1)
+            if new_cursor >= cursor_end:
+                # A provider ignoring start/end would pin the cursor and
+                # loop forever — refuse to spin.
+                logger.warning(
+                    f"{symbol} {timeframe.value}: pagination made no "
+                    f"progress at {cursor_end} — stopping")
+                break
+            cursor_end = new_cursor
             if len(chunk) < 2:
                 break
 
@@ -243,7 +296,17 @@ class BaseDataProvider(ABC):
         df = self._drop_forming_candle(df, timeframe)
         start_ts = pd.Timestamp(start)
         end_ts = pd.Timestamp(end)
-        return df[(df.index >= start_ts) & (df.index <= end_ts)]
+        df = df[(df.index >= start_ts) & (df.index <= end_ts)]
+        # Coverage sanity: a head materially later than the requested
+        # start means history was truncated (empty-streak bail or provider
+        # depth limit). Loud, not fatal — some assets genuinely start late.
+        if not df.empty and (df.index[0] - start_ts) > pd.Timedelta(days=7):
+            logger.error(
+                f"{symbol} {timeframe.value}: fetched history starts "
+                f"{df.index[0]} but {start_ts} was requested — "
+                f"~{(df.index[0] - start_ts).days} days missing from the "
+                f"head; treat downstream results accordingly")
+        return df
 
     def is_data_fresh(
         self, df: pd.DataFrame, symbol: str, timeframe: Timeframe
@@ -323,8 +386,19 @@ class BaseDataProvider(ABC):
             return df
         candle_seconds = self.CANDLE_SECONDS.get(timeframe, 3600)
         now = pd.Timestamp.now(tz="UTC")
-        if df.index[-1] + pd.Timedelta(seconds=candle_seconds) > now:
+        # Loop, not a single slice: a misbehaving feed can return MORE
+        # than one not-yet-closed bar, and a single iloc[:-1] would let
+        # the residual forming bar through cleaning (future-tolerance is
+        # +5 min) straight into the signal engine.
+        dropped = 0
+        while (not df.empty
+               and df.index[-1] + pd.Timedelta(seconds=candle_seconds) > now):
             df = df.iloc[:-1]
+            dropped += 1
+        if dropped > 1:
+            logger.warning(
+                "%s: feed returned %d forming/future bars — all dropped",
+                timeframe.value, dropped)
         return df
 
     def _is_data_fresh(self, df: pd.DataFrame, symbol: str, timeframe: Timeframe) -> bool:
@@ -372,6 +446,20 @@ class BaseDataProvider(ABC):
             logger.warning(f"{symbol}: Dropping {future_mask.sum()} future timestamps")
             df = df[~future_mask]
 
+        # Non-positive prices are corrupt, never real gold quotes. Reject
+        # the row BEFORE the consistency repair below — repairing High/Low
+        # from a corrupt 0.0 Open/Close would propagate the zero into
+        # previously-valid fields (2026-08-04 audit, I45).
+        nonpositive = (df[["Open", "High", "Low", "Close"]] <= 0).any(axis=1)
+        if nonpositive.any():
+            logger.warning(
+                f"{symbol}: Dropping {nonpositive.sum()} bars with "
+                f"non-positive prices (corrupt feed data)")
+            df = df[~nonpositive]
+            if df.empty:
+                raise self.DATA_ERROR(
+                    f"{symbol}: all bars had non-positive prices")
+
         invalid_ohlc = (
             (df["High"] < df["Low"]) |
             (df["High"] < df["Open"]) |
@@ -386,7 +474,10 @@ class BaseDataProvider(ABC):
                 df.loc[idx, "High"] = max(vals + [df.loc[idx, "High"]])
                 df.loc[idx, "Low"]  = min(vals + [df.loc[idx, "Low"]])
 
+        # inf returns (division by a zero/na Close) would NaN the std and
+        # silently disarm the whole 10σ pass — neutralize them first.
         returns = df["Close"].pct_change()
+        returns = returns.replace([float("inf"), float("-inf")], float("nan"))
         std_ret = returns.std()
         if std_ret > 0:
             dev = returns - returns.mean()

providers/oanda.py

@@ -149,23 +149,37 @@ class OandaDataProvider(BaseDataProvider):
         rows = []
         last_bid_close = None
         last_ask_close = None
+        skipped_malformed = 0
         for c in candles:
             if not c.get("complete", False):
                 continue
             mid = c.get("mid", {})
             bid = c.get("bid", {})
             ask = c.get("ask", {})
+            # A candle missing any OHLC field must be SKIPPED, never
+            # defaulted to 0.0: a zero price flows through cleaning as a
+            # "legit" bar and poisons ATR/Keltner/SL math — and a
+            # mid-frame zero Close makes pct_change() emit inf, which
+            # NaNs the std and disarms the 10σ outlier filter for the
+            # whole frame (2026-08-04 audit, I45).
+            if any(mid.get(k) is None for k in ("o", "h", "l", "c")):
+                skipped_malformed += 1
+                continue
             rows.append({
                 "Datetime": pd.Timestamp(c["time"]).tz_convert("UTC"),
-                "Open": float(mid.get("o", 0)),
-                "High": float(mid.get("h", 0)),
-                "Low": float(mid.get("l", 0)),
-                "Close": float(mid.get("c", 0)),
+                "Open": float(mid["o"]),
+                "High": float(mid["h"]),
+                "Low": float(mid["l"]),
+                "Close": float(mid["c"]),
                 "Volume": int(c.get("volume", 0)),
             })
             if bid and ask:
                 last_bid_close = float(bid.get("c", 0))
                 last_ask_close = float(ask.get("c", 0))
+        if skipped_malformed:
+            logger.warning(
+                "Oanda returned %d malformed candle(s) for %s (missing "
+                "OHLC fields) — skipped", skipped_malformed, instrument)
 
         if not rows:
             raise OandaDataError(

providers/twelve_data.py

@@ -210,6 +210,12 @@ class TwelveDataProvider(BaseDataProvider):
         The TD WS speaks display symbols ("XAU/USD"); ticks are mapped back
         to the canonical pair symbols passed in. The caller owns the
         reconnect/backoff loop around this generator.
+
+        `on_connect` fires on the FIRST RECEIVED frame, not after the
+        subscribe send (2026-08-04 audit H38b, mirroring the Tiingo
+        provider) — firing pre-receive let a rejected API key reset the
+        caller's reconnect backoff on every attempt, pinning it at the
+        minimum forever.
         """
         import websockets
 
@@ -222,11 +228,14 @@ class TwelveDataProvider(BaseDataProvider):
             }
             await ws.send(json.dumps(subscribe))
             logger.info(f"WebSocket connected, subscribed to {len(ws_to_pair)} symbols")
-            if on_connect is not None:
-                on_connect()
             hb_task = asyncio.create_task(self._ws_heartbeat(ws))
+            connected = False
             try:
                 async for raw in ws:
+                    if not connected:
+                        connected = True
+                        if on_connect is not None:
+                            on_connect()
                     parsed = self._parse_ws_price(raw)
                     if parsed is None:
                         continue