providers: keep real news bars (snap-back bad-print rule), Oanda retry ladder, scan_kind=failed

2026-09-07 signal-log inspection follow-up — three fixes that do not depend on sample size.

- providers/base.py: the signal-side 10σ outlier filter is now a bad-print filter. A >10σ
  close-to-close bar is dropped only when the NEXT close snaps back ≥80% of the way
  (OUTLIER_SIGMA / OUTLIER_REVERT_FRACTION); persisting moves and the newest bar are kept.
  The unconditional rule deleted ~36 real M5 news bars/yr in the backtester (whole-frame σ)
  and a different ~24/yr live (fetch-window σ) — e.g. the 2026-09-04 12:30 UTC NFP bar
  vanished from M5+M15 and the 12:35 scan was a byte-identical repeat of 12:30. 3-yr Oanda
  drops are now M5 2 / M15 1 / H1 0.
- providers/oanda.py: transient 401/5xx/network faults retry along RETRY_DELAYS_S=(2,5,10)s
  under RETRY_BUDGET_S=45s. 11 boundary scans in two days had lost their primary frame at
  xx:00:22–:26 UTC; one 2 s retry could not outlast the top-of-hour blip.
- signal_aggregator/_log_signal: a scan whose primary fetch failed logs scan_kind='failed'
  (price 0, no signal); analyze_signal_log.parity_report skips it like 'blocked'.
- test_all.py: 466 tests (+6): retry ladder walks all delays, 5xx blip recovers on the third
  attempt, retry budget stops the ladder, persisting/newest/partially-retraced 10σ bars kept,
  failed-scan row kind, parity skip.
- CLAUDE.md: cleaning rule, retry ladder, failed scan kind, dated current-state note.

Book impact at $0.65 (rule 1 still passes everywhere, both halves positive):
  12mo  M5 557/+112.21/PF1.56 → 560/+108.81/1.54   M15 397/+73.93/1.53 → 398/+74.61/1.53   H1 identical
  3yr   M5 857/+161.3/1.53 → 870/+153.6/1.50   M15 1086/+148.7/1.37 → 1092/+151.9/1.38
        H1 255/+70.5/1.98 → 253/+70.0/1.98   H4 113/+22.7/1.63 → 114/+21.7/1.59
Only trades on the previously-deleted event bars move; the pinned M5/M15 CSVs are no longer
byte-reproducible and should be regenerated at the next re-pin.

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

analyze_signal_log.py

@@ -545,7 +545,7 @@ def parity_report(rows: Sequence[LogRow]) -> dict:
     source (over-scanning symptom), and the divergence list."""
     groups: Dict[Tuple[str, str], List[LogRow]] = {}
     for r in rows:
-        if not r.bar_ts or r.scan_kind == "blocked":
+        if not r.bar_ts or r.scan_kind in ("blocked", "failed"):
             continue
         groups.setdefault((r.timeframe, str(r.bar_ts)), []).append(r)
     per_bar: Dict[Tuple[str, str], List[int]] = {}

providers/base.py

@@ -100,6 +100,14 @@ class BaseDataProvider(ABC):
     # boundary is missing the just-closed bar (feed latency).
     FRESH_FETCH_RETRY_SECONDS = 3.0
 
+    # Signal-side bad-print filter (see _validate_and_clean): a bar is a
+    # candidate when its close-to-close return sits more than
+    # OUTLIER_SIGMA standard deviations from the frame mean, and is
+    # dropped only when the NEXT close snaps back at least
+    # OUTLIER_REVERT_FRACTION of the way to the pre-spike close.
+    OUTLIER_SIGMA = 10.0
+    OUTLIER_REVERT_FRACTION = 0.8
+
     def __init__(self, config: PlatformConfig):
         self.config = config
         self._cache: Dict[str, pd.DataFrame] = {}
@@ -498,20 +506,32 @@ class BaseDataProvider(ABC):
         returns = returns.replace([float("inf"), float("-inf")], float("nan"))
         std_ret = returns.std()
         if std_ret > 0:
+            # A 10σ close-to-close move is NOT bad data by default. On
+            # three years of Oanda XAU bars every such bar was a real
+            # news/reopen move (NFP, CPI, Sunday gap — 2026-09-07 audit:
+            # 155 flagged M5 bars, 7 snapped back), and dropping them hid
+            # the most violent bar of the day from ATR/exhaustion math —
+            # while the backtester (whole-frame σ) dropped a DIFFERENT
+            # set than live (fetch-window σ). A bad print has the one
+            # signature a real move lacks: the NEXT close returns to the
+            # pre-spike level. Drop a candidate only on that observed
+            # snap-back; the newest bar (no successor yet) is kept as
+            # real, and the snap-back bar itself is never a candidate
+            # (its residual against the pre-spike close is small).
             dev = returns - returns.mean()
-            outlier_mask = dev.abs() > 10 * std_ret
+            candidate = dev.abs() > self.OUTLIER_SIGMA * std_ret
-            # A spike bar makes the NEXT bar's return look extreme too (it is
+            close = df["Close"]
-            # measured against the spike). If the previous bar is flagged and
+            spike = close - close.shift(1)
-            # this bar reverses direction, it is the legitimate snap-back —
+            residual = close.shift(-1) - close.shift(1)
-            # keep it, or every spike would punch a two-bar hole in the data.
+            snapped_back = residual.notna() & (
-            recovery = (
+                residual.abs()
-                outlier_mask
+                <= (1.0 - self.OUTLIER_REVERT_FRACTION) * spike.abs()
-                & outlier_mask.shift(1, fill_value=False)
-                & (dev * dev.shift(1) < 0)
             )
-            outlier_mask = outlier_mask & ~recovery
+            outlier_mask = candidate & snapped_back
             if outlier_mask.any():
-                logger.warning(f"{symbol}: Removing {outlier_mask.sum()} extreme outlier bars")
+                logger.warning(
+                    f"{symbol}: Removing {int(outlier_mask.sum())} bad-print "
+                    f"bar(s) (>{self.OUTLIER_SIGMA:.0f}σ move that snapped back)")
                 df = df[~outlier_mask]
 
         min_required = max(

providers/oanda.py

@@ -32,7 +32,11 @@ class OandaDataProvider(BaseDataProvider):
 
     name = "oanda"
     DATA_ERROR = OandaDataError
-    RETRY_DELAY_S = 2.0
+    # Transient-fault retry ladder (see _fetch_oanda_candles): sleeps
+    # between attempts, plus a wall-clock cap on the whole sequence.
+    RETRY_DELAYS_S = (2.0, 5.0, 10.0)
+    RETRY_BUDGET_S = 45.0
+    RETRY_DELAY_S = RETRY_DELAYS_S[0]  # first delay; kept for callers/tests
 
     OANDA_GRANULARITY = {
         Timeframe.M1: "M1",
@@ -128,34 +132,53 @@ class OandaDataProvider(BaseDataProvider):
         else:
             params["count"] = min(count, self.MAX_BARS_PER_REQUEST)
 
-        # One retry absorbs Oanda's sporadic spurious 401s (~15-20/day on
+        # Retry ladder for transient faults: Oanda's sporadic spurious
-        # the practice API — the same token succeeds on adjacent requests)
+        # 401s (~15-20/day on the practice API — the same token succeeds
-        # plus transient 5xx/network faults. A real credential problem
+        # on adjacent requests), 5xx and network errors. Other 4xx (bad
-        # fails both attempts and raises exactly as before. Other 4xx
+        # request, 429) are never retried; a real credential problem
-        # (bad request, 429) are never retried.
+        # fails every attempt and raises as before. The 2026-09-07
+        # signal-log audit found 11 boundary scans in two days that lost
+        # their primary frame at xx:00:22-:26 UTC: a single 2 s retry
+        # could not outlast a 10-15 s provider blip at the top of the
+        # hour, and a scan without its primary frame logs a price-0
+        # NEUTRAL row. RETRY_BUDGET_S bounds the whole sequence so a scan
+        # (three fetches) still finishes well inside an M5 bar.
         last_err: Optional[OandaDataError] = None
-        for attempt in range(2):
+        resp = None
+        attempts = 1 + len(self.RETRY_DELAYS_S)
+        started = time.monotonic()
+        for attempt in range(attempts):
             if attempt:
-                time.sleep(self.RETRY_DELAY_S)
+                delay = self.RETRY_DELAYS_S[attempt - 1]
+                if time.monotonic() - started + delay > self.RETRY_BUDGET_S:
+                    logger.warning(
+                        f"Oanda fetch for {instrument}: retry budget "
+                        f"({self.RETRY_BUDGET_S:.0f}s) exhausted after "
+                        f"{attempt} attempt(s) — giving up")
+                    break
+                time.sleep(delay)
             try:
                 resp = self._session.get(url, params=params, timeout=30)
             except requests.RequestException as e:
                 last_err = OandaDataError(
                     f"Oanda candle fetch failed for {instrument}: {e}")
                 last_err.__cause__ = e
+                resp = None
+                logger.warning(
+                    f"Oanda fetch for {instrument} attempt "
+                    f"{attempt + 1}/{attempts} failed: {e}")
                 continue
             if resp.status_code == 401 or resp.status_code >= 500:
                 last_err = OandaDataError(
                     f"Oanda candle fetch failed for {instrument}: "
                     f"HTTP {resp.status_code}: {resp.text[:300]}")
-                if attempt == 0:
                 logger.warning(
-                        f"Oanda fetch for {instrument} got HTTP "
+                    f"Oanda fetch for {instrument} attempt "
-                        f"{resp.status_code}; retrying in "
+                    f"{attempt + 1}/{attempts} got HTTP {resp.status_code}")
-                        f"{self.RETRY_DELAY_S}s")
+                resp = None
                 continue
             break
-        else:
+        if resp is None:
             raise last_err
 
         if resp.status_code != 200:

signal_aggregator.py

@@ -305,7 +305,10 @@ class SignalAggregator:
         matches the context the signal just used; if it fails, the row
         is still written, just without the news JSON blobs. ``blocked``
         marks weekend/day early returns: no news fetch, no snapshots —
-        a ~1 KB liveness row (or nothing, per signal_log.log_blocked_scans)."""
+        a ~1 KB liveness row (or nothing, per signal_log.log_blocked_scans).
+        A scan whose PRIMARY fetch failed is logged as ``scan_kind``
+        ``failed`` (price 0, no signal): liveness evidence that readers
+        skip like ``blocked``, so it never pollutes NEUTRAL statistics."""
         context = None
         if not blocked:
             try:
@@ -313,7 +316,12 @@ class SignalAggregator:
             except Exception:
                 logger.debug("signal log: news context unavailable",
                              exc_info=True)
-        extra = {"scan_kind": "blocked" if blocked else "scan"}
+        kind = "blocked" if blocked else "scan"
+        if not blocked and dataframes is not None:
+            primary = dataframes.get(self.config.primary_timeframe)
+            if primary is None or primary.empty:
+                kind = "failed"
+        extra = {"scan_kind": kind}
         for key, fingerprint in (
                 ("config_fingerprint", self.gate_config_fingerprint),
                 ("engine_fingerprint", self.engine_config_fingerprint)):

signal_log.py

@@ -53,8 +53,10 @@ or news_engine) so this module stays at the bottom of the import graph.
   XAU_CP acceptance removed from ``warnings_json``), ``bar_range_median``
   (the value historically stored as ``spread`` — it is the median
   High-Low of the primary frame, NOT a broker spread), ``scan_kind``
-  (``scan`` / ``blocked`` for weekend/day early returns, which carry no
+  (``scan``; ``blocked`` for weekend/day early returns, which carry no
-  news or candles) and ``config_fingerprint`` (hash of the gate config the
+  news or candles; ``failed`` when the primary frame could not be
+  fetched — price 0, no signal — readers skip it like ``blocked``) and
+  ``config_fingerprint`` (hash of the gate config the
   producing process ran with — surfaces config drift between processes).
 
 2026-09 weight-research additions:

test_all.py

@@ -2384,6 +2384,55 @@ def _():
         "snap-back bar after the spike must be kept (two-bar hole regression)"
 
 
+@test("Outlier filter keeps a >10σ bar whose move PERSISTS (real news bar)")
+def _():
+    # 2026-09-07 audit: the 12:30 UTC NFP bar (-1.7%, ~23σ) was deleted
+    # from the signal-side M5/M15 data, hiding the day's most violent bar
+    # from ATR/exhaustion math. Three years of Oanda XAU bars: 155 such
+    # M5 bars, 7 snapped back. A level shift that holds is real price.
+    provider = TwelveDataProvider(PlatformConfig())
+    df = make_ohlcv(n=300, seed=3)
+    k = 150
+    jump_ts = df.index[k]
+    shift = df["Close"].iloc[k - 1] * 0.05  # ~+5% step on a ~0.1%-σ series
+    df.loc[df.index[k]:, ["Open", "High", "Low", "Close"]] += shift
+    cleaned = provider._validate_and_clean(df, "XAU/USD", Timeframe.H1)
+    assert jump_ts in cleaned.index, \
+        "a 10σ move that the next close confirms must be KEPT"
+    assert len(cleaned) == len(df), "no bar may be dropped for a real move"
+
+
+@test("Outlier filter keeps the NEWEST >10σ bar (no successor to judge it)")
+def _():
+    # Live at 12:35 on a release day the crash bar IS the last closed
+    # bar. Dropping it would make the scan run on pre-release data (the
+    # 2026-09-04 12:35 scan was a byte-identical repeat of 12:30). With
+    # no next close available the bar is treated as real; a true bad
+    # print on the newest bar is caught on the following scan.
+    provider = TwelveDataProvider(PlatformConfig())
+    df = make_ohlcv(n=300, seed=3)
+    last_ts = df.index[-1]
+    df.loc[last_ts, ["Open", "High", "Low", "Close"]] = df["Close"].iloc[-2] * 1.05
+    cleaned = provider._validate_and_clean(df, "XAU/USD", Timeframe.H1)
+    assert last_ts in cleaned.index, "newest 10σ bar must be kept"
+
+
+@test("Outlier filter: partial snap-back (below the revert fraction) is a real move")
+def _():
+    provider = TwelveDataProvider(PlatformConfig())
+    df = make_ohlcv(n=300, seed=3)
+    k = 150
+    base = df["Close"].iloc[k - 1]
+    spike_ts = df.index[k]
+    # +5% spike, next close gives back only half — a violent real move,
+    # not a bad print (revert fraction 0.5 < OUTLIER_REVERT_FRACTION 0.8).
+    df.loc[spike_ts, ["Open", "High", "Low", "Close"]] = base * 1.05
+    df.loc[df.index[k + 1]:, ["Open", "High", "Low", "Close"]] += base * 0.025
+    cleaned = provider._validate_and_clean(df, "XAU/USD", Timeframe.H1)
+    assert spike_ts in cleaned.index, "half-retraced spike is real price"
+    assert provider.OUTLIER_REVERT_FRACTION == 0.8
+
+
 @test("Cache: pre-boundary snapshot refetched; boundary-complete snapshot served")
 def _():
     def bars_ending(last):
@@ -3558,20 +3607,26 @@ def _():
         f"Expected one retry sleep, got {slept}"
 
 
-@test("providers.oanda: persistent 401 raises after retry; 4xx never retried")
+@test("providers.oanda: persistent 401 walks the whole retry ladder then raises; 4xx never retried")
 def _():
     orig_sleep = providers.oanda.time.sleep
-    providers.oanda.time.sleep = lambda s: None
+    slept = []
+    providers.oanda.time.sleep = lambda s: slept.append(s)
     try:
+        ladder = providers.oanda.OandaDataProvider.RETRY_DELAYS_S
         provider = _make_oanda_provider([
-            _oanda_resp(401, text='{"errorMessage":"Insufficient authorization"}'),
+            _oanda_resp(401, text='{"errorMessage":"Insufficient authorization"}')
-            _oanda_resp(401, text='{"errorMessage":"Insufficient authorization"}'),
+            for _ in range(1 + len(ladder))
         ])
         try:
             provider._fetch_oanda_candles("XAU/USD", Timeframe.H1, 300)
             assert False, "Persistent 401 should raise OandaDataError"
         except providers.oanda.OandaDataError as e:
             assert "401" in str(e)
+        assert slept == list(ladder), \
+            f"every ladder delay must be slept before giving up, got {slept}"
+        assert ladder == (2.0, 5.0, 10.0), \
+            "ladder is sized to outlast a 10-15 s top-of-hour provider blip"
 
         # Non-auth 4xx must fail immediately — only one response queued,
         # so a retry attempt would IndexError on the empty queue.
@@ -3585,6 +3640,55 @@ def _():
         providers.oanda.time.sleep = orig_sleep
 
 
+@test("providers.oanda: a 5xx blip that clears on the third attempt recovers")
+def _():
+    # The 2026-09-07 signal-log audit: 11 boundary scans lost their
+    # primary frame because one 2 s retry could not outlast a ~12 s
+    # provider blip. Two failures then success must recover, having
+    # slept exactly the first two ladder steps.
+    orig_sleep = providers.oanda.time.sleep
+    slept = []
+    providers.oanda.time.sleep = lambda s: slept.append(s)
+    try:
+        provider = _make_oanda_provider([
+            _oanda_resp(502, text="bad gateway"),
+            _oanda_resp(503, text="unavailable"),
+            _oanda_resp(200, candles=[_OANDA_OK_CANDLE]),
+        ])
+        df = provider._fetch_oanda_candles("XAU/USD", Timeframe.H1, 300)
+    finally:
+        providers.oanda.time.sleep = orig_sleep
+    assert len(df) == 1, "third attempt must return the frame"
+    assert slept == list(
+        providers.oanda.OandaDataProvider.RETRY_DELAYS_S[:2]), slept
+
+
+@test("providers.oanda: retry budget stops the ladder before a delay would overrun it")
+def _():
+    # A scan fetches three frames; the whole ladder must stay bounded so
+    # an M5 scan finishes inside its bar. With the clock jumped past the
+    # budget after the first failure, no further attempt (and no sleep)
+    # may happen — the queue holds only ONE response, so a second
+    # attempt would IndexError.
+    orig_sleep = providers.oanda.time.sleep
+    orig_mono = providers.oanda.time.monotonic
+    slept = []
+    ticks = iter([0.0, 10_000.0, 10_000.0, 10_000.0])
+    providers.oanda.time.sleep = lambda s: slept.append(s)
+    providers.oanda.time.monotonic = lambda: next(ticks)
+    try:
+        provider = _make_oanda_provider([_oanda_resp(503, text="unavailable")])
+        try:
+            provider._fetch_oanda_candles("XAU/USD", Timeframe.H1, 300)
+            assert False, "budget exhaustion must raise the last error"
+        except providers.oanda.OandaDataError as e:
+            assert "503" in str(e)
+    finally:
+        providers.oanda.time.sleep = orig_sleep
+        providers.oanda.time.monotonic = orig_mono
+    assert slept == [], "no sleep once the budget cannot fit the next delay"
+
+
 @test("providers.oanda: instrument name mapping")
 def _():
     provider = providers.oanda.OandaDataProvider.__new__(
@@ -7772,6 +7876,48 @@ def _():
             os.remove(path)
 
 
+@test("signal_log: a scan whose PRIMARY fetch fails is logged as scan_kind 'failed'")
+def _():
+    # 2026-09-07 audit: 11 rows in two days carried price 0 / score 0 /
+    # NEUTRAL F because the top-of-hour Oanda fetch failed; logged as
+    # ordinary scans they polluted NEUTRAL statistics. They are liveness
+    # evidence and readers skip them like ``blocked``.
+    import signal_log as slog
+    fd, path = tempfile.mkstemp(suffix=".db")
+    os.close(fd)
+    os.remove(path)
+    try:
+        os.environ["SIGNAL_LOG_PATH"] = path
+        os.environ.pop("SIGNAL_LOG_ENABLED", None)
+        cfg = PlatformConfig()
+        cfg.xau_strategy.weekend_filter = False   # deterministic on any day
+        cfg.xau_strategy.allowed_weekdays = []
+        agg = SignalAggregator(cfg)
+        agg._get_news_engine = lambda: _NoNews()
+
+        def _boom(*a, **k):
+            raise RuntimeError("HTTP 503 at the boundary")
+        agg.data_provider.fetch_ohlcv = _boom
+        report = agg.analyze_pair(symbol="XAU/USD", log_source="test")
+        assert "Primary timeframe data unavailable" in report.warnings
+        import sqlite3
+        conn = sqlite3.connect(path)
+        row = conn.execute(
+            "SELECT scan_kind, gate_reason, signal, quality FROM signals"
+        ).fetchone()
+        conn.close()
+        assert row is not None, "the failed scan must still leave a liveness row"
+        assert row[0] == "failed", f"scan_kind must be 'failed', got {row[0]!r}"
+        assert row[1] == "Primary timeframe data unavailable"
+        assert row[2] == "NEUTRAL" and row[3] == "F"
+    finally:
+        os.environ.pop("SIGNAL_LOG_PATH", None)
+        os.environ.pop("SIGNAL_LOG_ENABLED", None)
+        slog.reset_loggers_for_tests()
+        if os.path.exists(path):
+            os.remove(path)
+
+
 @test("signal_log parity: analyze_from_dataframes never logs")
 def _():
     import signal_log as slog
@@ -7985,6 +8131,18 @@ def _():
     assert "config fingerprint" in d["diffs"] and "action" in d["diffs"]
     assert rep["rows_per_bar"]["telegram/15min"] == 1.5
     assert rep["rows_per_bar"]["dashboard/15min"] == 1.0
+    # A ``failed`` row (primary fetch lost, price 0) is liveness evidence,
+    # not a scan: it must not count as a source on its bar.
+    import dataclasses
+    failed = dataclasses.replace(
+        _asl_row(6, t0, bar_ts="2026-08-19T12:30:00+00:00", source="cli",
+                 quality="F", setup=False, price=0.0,
+                 action="NO TRADE — No clear signal"),
+        scan_kind="failed")
+    rep2 = asl.parity_report(rows + [failed])
+    assert rep2["bars_with_2_sources"] == 2
+    assert "cli/15min" not in rep2["rows_per_bar"]
+    assert len(rep2["divergences"]) == 1
 
 
 @test("analyze_signal_log: load_rows sees WAL-resident rows and never writes")