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.

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

advisory_outcome.py

@@ -348,6 +348,39 @@ def advance_outcome(
         return state
 
 
+def resolve_locked_floor(state: Dict[str, Any]) -> Dict[str, Any]:
+    """If an UNRESOLVABLE gap outcome had already trailed its stop past the
+    entry, finish it at that stop as a FLOOR (the true outcome is at least
+    this good) instead of discarding it. Returns the state (mutated)."""
+    if state.get("status") != "UNRESOLVABLE":
+        return state
+    reason = str(state.get("reason", ""))
+    # Both gap reasons ("lost continuity" / "contain an internal gap")
+    # share this prefix; an invalid seed or evaluator error never floors.
+    if not reason.startswith("stored candle snapshots"):
+        return state
+    progress = state.get("progress")
+    setup = state.get("setup")
+    if not isinstance(progress, dict) or not isinstance(setup, dict):
+        return state
+    entry = _finite(progress.get("entry_price"))
+    stop = _finite(progress.get("current_stop"))
+    risk = _finite(setup.get("stop_distance"))
+    exit_at = progress.get("last_processed_bar")
+    if (entry is None or stop is None or risk is None or risk <= 0
+            or not exit_at):
+        return state
+    buy = setup.get("direction") == "BUY"
+    locked = stop >= entry if buy else stop <= entry
+    if not locked:
+        return state
+    _finish(state, "GAP_LOCKED", str(exit_at), stop)
+    state["result"]["floor"] = True
+    state["reason"] = (
+        f"{reason} — resolved at the locked trailing stop (floor)")
+    return state
+
+
 def canonical_r(state: Dict[str, Any]) -> Optional[float]:
     result = state.get("result")
     value = result.get("net_r_065") if isinstance(result, dict) else None

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)
+    return (skipped, candidates,
-                                 if record_bars else None)
+            (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),

dashboard_publisher.py

@@ -26,6 +26,11 @@ whichever timeframes just closed. Due-classification uses the RAW wake
 time — flooring 12:00:10 to the 5-minute grid yields the 12:00 boundary
 that just closed. Do NOT subtract the buffer first: 12:00:10 − 15 s
 floors to 11:55 and silently skips every H1/H4 scan (v1 bug, 2026-08-02).
+A cycle can overrun its 5-minute slot (or the process can be down), so
+each wake also catches up the timeframes whose latest closed boundary is
+newer than their last scanned one (`catch_up_timeframes`) — without it the
+page served an hours-old H1/H4 report behind a fresh countdown
+(2026-09-13 audit).
 
 Scans run sequentially M5→M15→H1→H4 through one shared data provider so
 later timeframes hit the warm boundary cache — that ordering IS the
@@ -46,10 +51,12 @@ import os
 import shutil
 import sys
 import time
-from datetime import datetime, timezone
+from datetime import datetime, timedelta, timezone
 from logging.handlers import TimedRotatingFileHandler
 from typing import Dict, List, Optional
 
+import pandas as pd
+
 from agent_panel import attach_panel
 from app_config import get_app_config, violent_watch_enabled
 from config import PlatformConfig, Timeframe, XAUUSD
@@ -230,10 +237,60 @@ def due_timeframes(wake_dt: datetime) -> List[Timeframe]:
     """Timeframes whose candle just closed at the boundary this wake
     belongs to. Takes the RAW wake time (boundary + buffer): flooring to
     the 5-minute grid yields the boundary that just passed. Never
-    subtract the buffer first — see module docstring (v1 bug)."""
+    subtract the buffer first — see module docstring (v1 bug).
+
+    A timeframe whose just-closed bar would carry a label inside gold's
+    daily settlement pause is skipped: no such bar exists, so the scan
+    would only re-read the previous one (2026-09-13 audit)."""
     minutes = wake_dt.hour * 60 + wake_dt.minute
     floored = minutes - (minutes % 5)
-    return [tf for tf in DASHBOARD_TFS if floored % TF_INTERVAL_MINUTES[tf] == 0]
+    boundary = wake_dt.replace(
+        hour=floored // 60, minute=floored % 60, second=0, microsecond=0)
+    return [
+        tf for tf in DASHBOARD_TFS
+        if floored % TF_INTERVAL_MINUTES[tf] == 0
+        and not _providers_base._is_daily_maintenance(
+            boundary - timedelta(minutes=TF_INTERVAL_MINUTES[tf]))
+    ]
+
+
+def catch_up_timeframes(wake_dt: datetime,
+                        last_boundary: Dict[str, float]) -> List[Timeframe]:
+    """Timeframes whose most recent closed boundary (<= wake) is newer than
+    the boundary of their last successful scan — i.e. boundaries that were
+    missed while a cycle overran or the process was down. A TF never scanned
+    is due. Excludes bars whose label sits in the daily settlement pause
+    (same rule as due_timeframes).
+
+    `due_timeframes` only ever names the timeframes whose boundary coincides
+    with THIS wake, so an overrunning cycle (scan + advisory panel + news can
+    take minutes) or a restart left the page serving the previous H1/H4
+    report for up to 1 h / 4 h while advertising a fresh next_close_epoch
+    (2026-09-13 audit)."""
+    minutes = wake_dt.hour * 60 + wake_dt.minute
+    out = []
+    for tf in DASHBOARD_TFS:
+        interval = TF_INTERVAL_MINUTES[tf]
+        floored = minutes - (minutes % interval)
+        last_closed = wake_dt.replace(
+            hour=floored // 60, minute=floored % 60, second=0, microsecond=0)
+        # No bar carries a label inside a scheduled closure (settlement
+        # pause or weekend), so nothing can have been "missed" there — and a
+        # weekend-blocked scan never advances the ledger, so without the
+        # weekend half of this test every weekend wake would re-scan
+        # M15/H1/H4 for nothing (~1,800 blocked rows per weekend).
+        # (Whole-span rule, as in providers.base._has_latest_closed_bar: an
+        # H4 bar labelled Sunday 20:00 UTC starts in the weekend but runs
+        # through the reopen, so it exists.)
+        label = last_closed - timedelta(minutes=interval)
+        if (_providers_base._is_scheduled_closure(label)
+                and _providers_base._is_scheduled_closure(
+                    last_closed - timedelta(seconds=1))):
+            continue
+        scanned = last_boundary.get(tf.value)
+        if scanned is None or scanned < last_closed.timestamp():
+            out.append(tf)
+    return out
 
 
 def seconds_until_boundary(interval_minutes: int, buffer: int = 10) -> float:
@@ -324,6 +381,9 @@ class DashboardPublisher:
         self.out_dir = out_dir
         self.snapshots: Dict[str, dict] = {}
         self._reports = {}
+        # tf.value → epoch seconds of the boundary the last successful scan
+        # belonged to; feeds catch_up_timeframes (2026-09-13 audit).
+        self._last_boundary: Dict[str, float] = {}
         self.news_payload: Optional[dict] = None
         self._dp = None
         self._news_engine: Optional[NewsEngine] = None
@@ -355,7 +415,14 @@ class DashboardPublisher:
             shutil.copyfile(src, dst + ".tmp")
             os.replace(dst + ".tmp", dst)
 
-    def scan_tf(self, tf: Timeframe) -> None:
+    def scan_tf(self, tf: Timeframe,
+                boundary_epoch: Optional[float] = None) -> None:
+        """Scan one timeframe and store its snapshot entry.
+
+        `boundary_epoch` is the fallback boundary recorded when the report
+        carries no analysis (weekend/blocked or failed scan) or its bar
+        label will not parse; normally the boundary comes from the report
+        itself (bar label + interval)."""
         started = time.time()
         try:
             report = produce_signal(XAUUSD, tf, self._provider(),
@@ -365,6 +432,19 @@ class DashboardPublisher:
             if violent_watch_enabled():
                 attach_watch(report, tf,
                              get_app_config().violent_watch_settings)
+            # Which boundary this report belongs to: the bar LABEL it closed
+            # on plus one interval. Recorded so a boundary missed by an
+            # overrunning cycle or a restart is caught up on the next wake
+            # instead of sitting stale (2026-09-13 audit).
+            boundary = boundary_epoch
+            if report.primary_analysis is not None:
+                try:
+                    boundary = (pd.Timestamp(report.timestamp).timestamp()
+                                + TF_INTERVAL_MINUTES[tf] * 60)
+                except Exception:
+                    boundary = boundary_epoch
+            if boundary is not None:
+                self._last_boundary[tf.value] = boundary
             self._reports[tf.value] = report
             self.snapshots[tf.value] = {
                 "status": "ok",
@@ -449,9 +529,15 @@ class DashboardPublisher:
             logger.info(f"next M5 boundary in {delay:.0f}s")
             time.sleep(delay)
             try:
-                due = due_timeframes(datetime.now(timezone.utc))
+                now = datetime.now(timezone.utc)
+                due = due_timeframes(now)
                 logger.info(f"boundary wake — scanning {[t.value for t in due]}")
-                self.run_cycle(due)
+                missed = catch_up_timeframes(now, self._last_boundary)
+                extra = [tf for tf in missed if tf not in due]
+                if extra:
+                    logger.info(f"catch-up — scanning "
+                                f"{[t.value for t in extra]} (missed boundary)")
+                self.run_cycle(due + extra)
             except Exception:
                 logger.exception("cycle failed")
 

engine.py

@@ -12,6 +12,7 @@ Design philosophy:
 """
 
 import logging
+import math
 from dataclasses import dataclass, field
 from typing import Dict, List, Optional, Tuple
 
@@ -84,15 +85,65 @@ _DEFAULT_REGIME_WEIGHTS = {
 }
 
 
+def _validate_regime_override(override: Dict[str, Dict[str, float]]) -> None:
+    """Reject a PARTIAL ``regime_weights_override`` (raises ValueError).
+
+    An override REPLACES the whole table, it does not patch it: whatever
+    it omits is multiplied by 1.0 in _compute_composite, not by the
+    shipped default. A half-filled research table (``--set
+    technical.regime_weights_override=<JSON>``) would therefore silently
+    run 1.0 where the default said 0.3 or 2.5, with no warning — so every
+    regime and all nine scored indicators must be present, with a finite
+    multiplier >= 0. The shipped M15 override is derived from the default
+    table (pair_signals/xauusd._m15_regime_weights) and passes."""
+    problems: List[str] = []
+    want_regimes = set(_DEFAULT_REGIME_WEIGHTS)
+    want_indicators = set(_TRENDING_WEIGHTS)
+    got_regimes = set(override)
+    if got_regimes != want_regimes:
+        problems.append(
+            f"missing regimes {sorted(want_regimes - got_regimes)}, "
+            f"unexpected {sorted(got_regimes - want_regimes)}")
+    for regime in sorted(got_regimes & want_regimes):
+        weights = override[regime]
+        if not isinstance(weights, dict):
+            problems.append(f"regime '{regime}' is not an indicator -> "
+                            f"multiplier mapping: {weights!r}")
+            continue
+        got = set(weights)
+        if got != want_indicators:
+            problems.append(
+                f"regime '{regime}' missing {sorted(want_indicators - got)}, "
+                f"unexpected {sorted(got - want_indicators)}")
+        for name in sorted(got & want_indicators):
+            mult = weights[name]
+            if (isinstance(mult, bool) or not isinstance(mult, (int, float))
+                    or not math.isfinite(mult) or mult < 0):
+                problems.append(
+                    f"regime '{regime}' indicator '{name}' multiplier "
+                    f"{mult!r} is not a finite number >= 0")
+    if problems:
+        raise ValueError(
+            "regime_weights_override is partial: " + "; ".join(problems)
+            + "; an override replaces the WHOLE table "
+              "(see engine._DEFAULT_REGIME_WEIGHTS)")
+
+
 def regime_weight_table(config: TechnicalConfig) -> Dict[str, Dict[str, float]]:
     """The regime -> indicator weight multipliers _compute_composite runs
     with (``regime_weights_override`` wins, else the gold-optimized
     defaults above).
 
+    A non-empty override must cover the WHOLE table (every regime, all
+    nine scored indicators) — a partial one raises ValueError instead of
+    silently multiplying the cells it omits by 1.0.
+
     Pure and side-effect free: ONE source of truth shared by the composite
     and by signal_aggregator.engine_config_fingerprint. Returns a copy, so
     a caller can never mutate the shipped table."""
     override = getattr(config, "regime_weights_override", None)
+    if override:
+        _validate_regime_override(override)
     table = override if override else _DEFAULT_REGIME_WEIGHTS
     return {regime: dict(weights) for regime, weights in table.items()}
 
@@ -119,7 +170,7 @@ class TechnicalEngine:
             return TechnicalAnalysis(
                 symbol=symbol,
                 timeframe=timeframe,
-                timestamp=str(pd.Timestamp.utcnow()),
+                timestamp=str(pd.Timestamp.now(tz="UTC")),
                 confidence=0.0,
             )
 

providers/base.py

@@ -65,6 +65,46 @@ def _is_forex_weekend(now=None) -> bool:
     return False
 
 
+def _is_daily_maintenance(now=None) -> bool:
+    """Gold's daily settlement pause: Monday-Thursday 17:00-17:59 New York
+    (21:00 UTC in US summer, 22:00 in winter). No bar carries a label
+    inside this hour. Friday 17:00 is the weekend (see _is_forex_weekend).
+    Same rule as advisory_outcome._has_continuity's daily_maintenance."""
+    if now is None:
+        now = pd.Timestamp.now(tz="UTC")
+    ny = now.astimezone(_NY_TZ)
+    return ny.weekday() < 4 and ny.hour == 17
+
+
+def _is_scheduled_closure(now=None) -> bool:
+    """The feed is legitimately silent: weekend or daily settlement pause."""
+    return _is_forex_weekend(now) or _is_daily_maintenance(now)
+
+
+def _scheduled_closed_seconds(start, end) -> float:
+    """Seconds of the [start, end) span that fall inside a scheduled
+    closure (weekend or the daily settlement pause).
+
+    Bar age is measured from a bar LABEL, so at 21:30 UTC in summer the
+    20:55 M5 bar looks 35 minutes old even though the feed could not have
+    printed anything since 21:00. Subtracting the closed span keeps the
+    3-candle freshness limit meaningful instead of failing it — and
+    logging a false "feed lagging"/"STALE DATA" — every settlement hour
+    (2026-09-13 audit: 61 suppressed scans on three nights).
+    """
+    hour = pd.Timedelta(hours=1)
+    closed = 0.0
+    t = start.floor("h")
+    steps = 0
+    while t < end and steps < 24 * 10:
+        if _is_scheduled_closure(t):
+            closed += max(
+                0.0, (min(t + hour, end) - max(t, start)).total_seconds())
+        t += hour
+        steps += 1
+    return closed
+
+
 class DataIntegrityError(Exception):
     """Raised when data fails validation checks."""
     pass
@@ -276,7 +316,7 @@ class BaseDataProvider(ABC):
         chunks = []
         cursor_end = end
         empty_streak = 0
-        error_streak = 0
+        window_errors = 0
         while cursor_end > start:
             cursor_start = max(start, cursor_end - timedelta(days=days_per_chunk))
             try:
@@ -286,23 +326,31 @@ class BaseDataProvider(ABC):
                 )
             except Exception as e:
                 # A raising chunk is a PROVIDER ERROR (auth, 429, 5xx,
-                # malformed payload) — not "no data". Three in a row means
+                # malformed payload) — not "no data". RETRY THE SAME
-                # the rest of the history is unreachable: raise instead of
+                # WINDOW: until 2026-09-13 a failure advanced the cursor
-                # returning a silently truncated frame (the backtester
+                # past the window, so one or two NON-consecutive page
-                # would treat it as complete).
+                # failures (Oanda's sporadic 401/5xx, even after the
+                # provider's own retry ladder) silently dropped up to
+                # days_per_chunk of bars — 15 days of M5 — out of the
+                # middle of the frame, and only the HEAD was
+                # coverage-checked below, so the backtester cached the hole
+                # as complete history forever. Three failures on the SAME
+                # window means it is unreachable: raise instead of
+                # returning a frame with a hole in it.
                 logger.warning(
                     f"Chunk {cursor_start.date()}–{cursor_end.date()} failed: {e}"
                 )
-                error_streak += 1
+                window_errors += 1
-                if error_streak >= 3:
+                if window_errors >= 3:
                     raise self.DATA_ERROR(
-                        f"{symbol} {timeframe.value}: {error_streak} "
+                        f"{symbol} {timeframe.value}: {window_errors} "
-                        f"consecutive chunk failures walking back from "
+                        f"consecutive failures fetching window "
-                        f"{cursor_end.date()} — refusing to return "
+                        f"{cursor_start.date()}–{cursor_end.date()} — "
-                        f"truncated history (last error: {e})") from e
+                        f"refusing to return truncated history "
-                cursor_end = cursor_start - timedelta(seconds=1)
+                        f"(last error: {e})") from e
-                continue
+                continue        # re-request the SAME window
 
+            window_errors = 0
             if chunk is None or chunk.empty:
                 # Genuinely empty windows (weekends/holidays/asset gaps)
                 # are normal; several in a row = start of history.
@@ -313,7 +361,6 @@ class BaseDataProvider(ABC):
                 continue
 
             empty_streak = 0
-            error_streak = 0
             chunks.append(chunk)
             # Step back; dedup on concat.
             new_cursor = chunk.index[0] - timedelta(seconds=1)
@@ -348,6 +395,38 @@ class BaseDataProvider(ABC):
                 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")
+        # INTERIOR-gap audit (2026-09-13): the head check above is blind to
+        # a hole in the MIDDLE of the frame, which is exactly what a
+        # dropped page used to leave behind. Missing labels inside a
+        # scheduled closure (weekend, daily settlement pause) are normal,
+        # so only unscheduled runs are reported — and a real market holiday
+        # (Christmas, Good Friday, July 4) looks identical to a lost page,
+        # so this is LOUD BUT NOT FATAL and the frame is returned unchanged.
+        if len(df) > 400_000:
+            logger.info(
+                f"{symbol} {timeframe.value}: interior-gap audit skipped "
+                f"({len(df)} rows)")
+        elif len(df) > 1:
+            step = pd.Timedelta(seconds=self.CANDLE_SECONDS[timeframe])
+            labels = df.index
+            # Missing labels between each consecutive pair; >3 is the floor
+            # worth a look (weekends/settlement hours leave 1-2 labels that
+            # the NY-anchored closure rule does not cover).
+            spans = ((labels[1:] - labels[:-1]) / step).astype("int64") - 1
+            gaps = []
+            for i in (spans > 3).nonzero()[0]:
+                missing = sum(
+                    1 for k in range(1, int(spans[i]) + 1)
+                    if not _is_scheduled_closure(labels[i] + k * step))
+                if missing > 3:
+                    gaps.append((labels[i], labels[i + 1], missing))
+            if gaps:
+                logger.error(
+                    f"{symbol} {timeframe.value}: {len(gaps)} interior "
+                    f"gap(s) of >3 unscheduled bars in fetched history — "
+                    f"first: {gaps[0]}; a page fetch may have been dropped "
+                    f"or the market was closed (holiday); verify before "
+                    f"trusting this window")
         return df
 
     def is_data_fresh(
@@ -398,7 +477,8 @@ class BaseDataProvider(ABC):
 
     # ── Shared internals ───────────────────────────────────────────
 
-    def _has_latest_closed_bar(self, df: pd.DataFrame, timeframe: Timeframe) -> bool:
+    def _has_latest_closed_bar(self, df: pd.DataFrame, timeframe: Timeframe,
+                               now=None) -> bool:
         """Cached data must include the most recent fully closed bar.
 
         The TTL alone is wall-clock based: a snapshot fetched mid-candle is
@@ -408,12 +488,29 @@ class BaseDataProvider(ABC):
         """
         if df is None or df.empty:
             return False
+        if now is None:
             now = pd.Timestamp.now(tz="UTC")
         if _is_forex_weekend(now):
             return True
         candle_seconds = self.CANDLE_SECONDS.get(timeframe, 3600)
         dur = pd.Timedelta(seconds=candle_seconds)
         expected_label = now.floor(dur) - dur
+        # No bar ever carries a label inside a scheduled closure, so the
+        # newest bar that MUST be present is the last label before it.
+        # Without this walk-back every scan from 21:05 to 22:00 UTC in
+        # summer evicted the cache, refetched twice and logged a false
+        # "feed lagging" (2026-09-13 audit).
+        # A label counts as "inside" only when the closure covers the bar's
+        # whole span: a D1 bar labelled Sunday 00:00 UTC starts in the
+        # weekend but runs through the Sunday reopen, so it exists and
+        # must be present (68 such bars in the 3-year Oanda D1 cache).
+        steps = 0
+        last_second = dur - pd.Timedelta(seconds=1)
+        while (_is_scheduled_closure(expected_label)
+               and _is_scheduled_closure(expected_label + last_second)
+               and steps < 2000):
+            expected_label -= dur
+            steps += 1
         return df.index[-1] >= expected_label
 
     def _drop_forming_candle(self, df: pd.DataFrame, timeframe: Timeframe) -> pd.DataFrame:
@@ -443,11 +540,13 @@ class BaseDataProvider(ABC):
                 timeframe.value, dropped)
         return df
 
-    def _is_data_fresh(self, df: pd.DataFrame, symbol: str, timeframe: Timeframe) -> bool:
+    def _is_data_fresh(self, df: pd.DataFrame, symbol: str, timeframe: Timeframe,
+                       now=None) -> bool:
         """Check if the last candle is within acceptable age for this timeframe."""
         if df is None or df.empty:
             return False
 
+        if now is None:
             now = pd.Timestamp.now(tz="UTC")
 
         if _is_forex_weekend(now):
@@ -456,7 +555,10 @@ class BaseDataProvider(ABC):
         last_bar = df.index[-1]
         candle_seconds = self.CANDLE_SECONDS.get(timeframe, 3600)
         max_age_seconds = candle_seconds * 3
-        age_seconds = (now - last_bar).total_seconds()
+        # Scheduled closures (weekend, daily settlement pause) are dead
+        # time in which no bar can print — they must not count as age.
+        age_seconds = ((now - last_bar).total_seconds()
+                       - _scheduled_closed_seconds(last_bar, now))
 
         if age_seconds > max_age_seconds:
             logger.warning(
@@ -482,7 +584,7 @@ class BaseDataProvider(ABC):
         df = df.sort_index()
         df = df[~df.index.duplicated(keep="last")]
 
-        now = pd.Timestamp.utcnow()
+        now = pd.Timestamp.now(tz="UTC")
         future_mask = df.index > now + pd.Timedelta(minutes=5)
         if future_mask.any():
             logger.warning(f"{symbol}: Dropping {future_mask.sum()} future timestamps")

requirements.txt

@@ -1,5 +1,5 @@
-numpy>=1.24
+numpy>=1.24,<3
-pandas>=2.0
+pandas>=2.0,<4
 requests>=2.28
 twelvedata>=1.2.0
 python-telegram-bot>=20.0

signal_aggregator.py

@@ -242,6 +242,36 @@ class SignalAggregator:
             except Exception as e:
                 logger.warning(f"Fetch failed for {symbol} {tf.value}: {e}")
 
+        # A missing frame is a FAILED scan, never a degraded one. Losing a
+        # CONFIRMATION frame used to let the scan continue with fewer
+        # frames: _check_mtf_agreement needs >= 2 signals, so it returned
+        # False and the composite took a 15-point mtf_disagreement penalty
+        # the market never earned (A/A+ became impossible), while on
+        # M15/H1/H4 a B/C setup could still ship on one confirmation with
+        # no warning at all. The backtester never sees that state
+        # (run_backtest raises on an empty frame), so it was a live-only
+        # divergence. 2026-09-13 audit.
+        missing = [tf for tf in timeframes if tf not in dfs]
+        if missing:
+            report = SignalReport(
+                symbol=symbol,
+                display_name=pair_info["display"],
+                timestamp=str(now_utc),
+            )
+            report.final_signal = SignalStrength.NEUTRAL
+            report.signal_quality = "F"
+            report.trade_setup = None
+            if self.config.primary_timeframe in missing:
+                # Legacy text signal_log.gate_reason already recognises.
+                report.warnings.append("Primary timeframe data unavailable")
+            report.warnings.append(
+                f"DATA UNAVAILABLE: {', '.join(tf.value for tf in missing)}"
+                " — scan failed"
+            )
+            report.action_recommendation = "NO TRADE — market data unavailable"
+            self._log_signal(report, log_source, dfs, failed=True)
+            return report
+
         report = self.analyze_from_dataframes(
             symbol=symbol,
             dfs=dfs,
@@ -303,7 +333,7 @@ class SignalAggregator:
     def _log_signal(
         self, report: SignalReport, source: Optional[str],
         dataframes: Optional[Dict[Timeframe, pd.DataFrame]] = None,
-        blocked: bool = False,
+        blocked: bool = False, failed: bool = False,
     ) -> None:
         """Persist the scan to the signal-history DB (signal_log.py).
 
@@ -314,9 +344,11 @@ class SignalAggregator:
         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 scan whose PRIMARY fetch failed is logged as ``scan_kind``
+        A scan missing ANY required frame (primary or confirmation, or
-        ``failed`` (price 0, no signal): liveness evidence that readers
+        the regime-filter frame when that filter is on) is logged as
-        skip like ``blocked``, so it never pollutes NEUTRAL statistics."""
+        ``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:
@@ -325,9 +357,9 @@ class SignalAggregator:
                 logger.debug("signal log: news context unavailable",
                              exc_info=True)
         kind = "blocked" if blocked else "scan"
-        if not blocked and dataframes is not None:
+        primary = (dataframes or {}).get(self.config.primary_timeframe)
-            primary = dataframes.get(self.config.primary_timeframe)
+        if failed or (not blocked and dataframes is not None
-            if primary is None or primary.empty:
+                      and (primary is None or primary.empty)):
             kind = "failed"
         extra = {"scan_kind": kind}
         for key, fingerprint in (
@@ -1297,6 +1329,28 @@ class SignalAggregator:
                 report.trade_setup = self._generate_xau_atr_trade_setup(report, atr)
         if report.trade_setup is None and report.signal_quality in ("A+", "A", "B", "C"):
             report.trade_setup = self._generate_trade_setup(report, pip_value)
+        # A shipped setup always names a direction. The signal thresholds
+        # (BUY/SELL need confidence >= 50) sit above the C grade floor
+        # (>= 45), and gate 6c grants B at >= 45 in "transitioning", so a
+        # C/B report could leave here as NEUTRAL with a full setup: the
+        # backtester traded it (the replay never reads final_signal) while
+        # the Telegram alert gate rejects NEUTRAL first (2026-09-13 audit:
+        # 5/91 H1 12-mo book trades, +3.7R of +22.2R, were never alerted).
+        # The generators derived the direction from the score sign — the
+        # same sign every direction gate above used — so this changes
+        # display and alert eligibility only; the trade books are unchanged.
+        # A report no generator served stays NEUTRAL, as before.
+        if (report.trade_setup is not None
+                and report.final_signal == SignalStrength.NEUTRAL):
+            report.final_signal = (SignalStrength.BUY
+                                   if report.trade_setup.direction == "BUY"
+                                   else SignalStrength.SELL)
+            # The action text was written before the gate chain with the
+            # old signal name; keep it consistent with the header.
+            if report.action_recommendation.startswith("NEUTRAL"):
+                report.action_recommendation = (
+                    report.final_signal.name.replace("_", " ")
+                    + report.action_recommendation[len("NEUTRAL"):])
 
     @staticmethod
     def _drop_stale_spread_gate_vetoes(report: SignalReport) -> None:

signal_log.py

@@ -104,6 +104,7 @@ from advisory_outcome import (
     canonical_r,
     initial_outcome,
     normalize_timeframe,
+    resolve_locked_floor,
     timeframe_value,
 )
 # Private on purpose — the scan-outcome labeller must apply EXACTLY the
@@ -258,6 +259,7 @@ _GATE_PREFIXES = (
     "SESSION FILTER",
     "Primary timeframe stale",
     "Primary timeframe data unavailable",
+    "DATA UNAVAILABLE:",
     "REGIME FILTER",
     "EXHAUSTION BLOCK",
     "EXHAUSTED TREND FILTER",
@@ -278,6 +280,10 @@ SCAN_OUTCOME_WINDOW_H = 24
 # A 24 h window can only be judged once it has elapsed; the extra hour
 # leaves room for the last bar of the window to be stored.
 SCAN_LABEL_MIN_AGE_H = 25
+# An incomplete row older than this is FINAL: its window had a real hole
+# (a feed outage), which no later bar can fill. It is neither re-examined
+# nor allowed to anchor the shared bar stream for the rest of the batch.
+SCAN_LABEL_MAX_AGE_H = 72
 # Missing bars tolerated inside a window ON TOP of the scheduled closures
 # advisory_outcome._has_continuity already allows.
 SCAN_LABEL_MAX_GAP_BARS = 3
@@ -399,6 +405,20 @@ def setup_key(setup: Optional[Dict[str, Any]], timeframe: Any) -> Optional[str]:
     ))
 
 
+def _is_gap_reason(reason: Any) -> bool:
+    """True when an outcome was left UNRESOLVABLE by a candle gap.
+
+    advance_outcome writes TWO such reasons — "stored candle snapshots
+    lost continuity" (no overlap with the last processed bar) and "stored
+    candle snapshots contain an internal gap" — and each call site used to
+    match only one of them, so the stream rebuild and the locked-floor
+    rescue skipped half the gap rows. The substring clauses keep states
+    written by older code inside the same net."""
+    text = str(reason or "")
+    return (text.startswith("stored candle snapshots")
+            or "continuity" in text or "gap" in text)
+
+
 def gate_reason(
     warnings: Iterable[str],
     cleaned_warnings: Iterable[str],
@@ -1150,12 +1170,19 @@ class SignalLogger:
                     before = _dumps(json.loads(hit[0]))
                     advanced = advance_outcome(json.loads(before), bars)
                     if (advanced.get("status") == "UNRESOLVABLE"
-                            and "continuity" in str(advanced.get("reason", ""))):
+                            and _is_gap_reason(advanced.get("reason"))):
                         union = self._stream_since(
                             symbol, timeframe,
                             json.loads(before)["progress"]["last_processed_bar"])
                         if union:
                             advanced = advance_outcome(json.loads(before), union)
+                        # The database could not fill the hole. A trade
+                        # whose trail had already locked profit still has a
+                        # real answer — its floor — so record that instead
+                        # of dropping the row out of the scorecard.
+                        if (advanced.get("status") == "UNRESOLVABLE"
+                                and _is_gap_reason(advanced.get("reason"))):
+                            advanced = resolve_locked_floor(advanced)
                     after = _dumps(advanced)
                     if after == before:
                         self._conn.rollback()
@@ -1194,6 +1221,11 @@ class SignalLogger:
                 " WHERE symbol=? AND candles_json IS NOT NULL"
                 " AND logged_at >= ? ORDER BY id LIMIT ?",
                 (f'$."{key}"', symbol, since, int(max_rows))).fetchall()
+            if len(rows) == int(max_rows):
+                logger.warning(
+                    "signal_log: bar stream truncated for %s %s (key %s)"
+                    " since %s — hit the %s-row cap, later bars are missing",
+                    symbol, tf, key, since, int(max_rows))
             for (blob,) in rows:
                 if not blob:
                     continue
@@ -1212,42 +1244,82 @@ class SignalLogger:
 
         Candidates are rows that carry indicator snapshots (a real scan)
         and a price, whose signal bar closed more than 25 h ago and which
-        have no COMPLETE label yet; the least recently labelled go first,
+        have no COMPLETE label yet. An incomplete row is re-examined only
-        so a permanently short stream can never starve the rest. Returns
+        while it is younger than SCAN_LABEL_MAX_AGE_H (counted from the
-        the number of rows actually WRITTEN — a row whose numbers did not
+        newest logged row when the database is behind the clock) — beyond
-        change is skipped, which makes repeated passes idempotent and lets
+        that its window had a real hole that no later bar can fill, so the
-        a batch driver stop at 0.
+        label it already carries is final — and a stale row never anchors
+        the shared bar stream either (a key that has nothing else due
+        starts at the age bound plus one window). Both bounds are
+        load-bearing: a handful of permanently incomplete rows (a feed
+        outage) sort to the front of every pass, and anchoring the stream
+        at their bar pushes its row cap past today, which silently labels
+        every genuinely due row NULL and incomplete — the scan_outcomes
+        table then goes blind. Returns the number of rows actually WRITTEN
+        — a row whose numbers did not change is not rewritten (only its
+        labeled_at is bumped, so it rotates to the back of the queue),
+        which makes repeated passes idempotent and lets a batch driver
+        stop at 0.
 
         Bounded on purpose: at most ``budget`` candidates are examined and
         the pass stops after ``time_budget`` seconds (default 2 s) so the
         opportunistic call inside log() stays cheap. Per-row failures are
         logged and skipped."""
         moment = now or datetime.now(timezone.utc)
-        cutoff = _utc(moment) - timedelta(hours=SCAN_LABEL_MIN_AGE_H)
+        moment_utc = _utc(moment)
+        cutoff = moment_utc - timedelta(hours=SCAN_LABEL_MIN_AGE_H)
         deadline = time.monotonic() + float(
             SCAN_LABEL_TIME_BUDGET if time_budget is None else time_budget)
         written = 0
         with self._lock:
+            # Age is measured from the newest row when the database itself
+            # is behind the clock (an imported snapshot on the dev box),
+            # otherwise every row of a 3-day-old import would be declared
+            # final on the first pass. On a live logger the two coincide.
+            horizon = moment_utc
+            newest = self._conn.execute(
+                "SELECT logged_at FROM signals ORDER BY id DESC"
+                " LIMIT 1").fetchone()
+            try:
+                horizon = min(horizon, _utc(datetime.fromisoformat(
+                    str(newest[0]).replace(" ", "T"))))
+            except (TypeError, ValueError, IndexError):
+                pass
+            stale_before = horizon - timedelta(hours=SCAN_LABEL_MAX_AGE_H)
+            # Deepest a stream may start when every row due for that key is
+            # stale: one window before the age bound, so such a row still
+            # gets its single final pass without dragging the stream back.
+            anchor_floor = stale_before - timedelta(hours=SCAN_LABEL_MIN_AGE_H)
+            # Row ids are monotonic in logged_at (a uniform ISO string), so
+            # an id bound is the cheap, index-friendly way to leave stale
+            # incomplete rows out of the re-examination set. bar_ts mixes
+            # separators and must never be compared as a string.
+            min_id = self._conn.execute(
+                "SELECT MIN(id) FROM signals WHERE logged_at >= ?",
+                (stale_before.isoformat(),)).fetchone()[0]
+            if min_id is None:
+                min_id = 2 ** 62          # nothing recent: re-examine none
             rows = self._conn.execute(
                 "SELECT s.id, s.symbol, s.timeframe, s.price,"
                 " COALESCE(s.bar_ts, json_extract(s.indicators_json,"
                 " '$.\"' || s.timeframe || '\".timestamp')) AS bar,"
                 " json_extract(s.indicators_json,"
-                " '$.\"' || s.timeframe || '\".key_levels.atr') AS atr"
+                " '$.\"' || s.timeframe || '\".key_levels.atr') AS atr,"
+                " (o.signal_id IS NULL) AS unlabelled"
                 " FROM signals s"
                 " LEFT JOIN scan_outcomes o ON o.signal_id = s.id"
                 " WHERE s.indicators_json IS NOT NULL AND s.price > 0"
-                " AND (o.signal_id IS NULL OR o.complete = 0)"
+                " AND (o.signal_id IS NULL OR (o.complete = 0 AND s.id >= ?))"
                 " AND bar IS NOT NULL"
                 " ORDER BY COALESCE(o.labeled_at, '') ASC, s.id ASC"
-                " LIMIT ?", (int(budget),)).fetchall()
+                " LIMIT ?", (int(min_id), int(budget))).fetchall()
             # One bar stream per (symbol, timeframe) for the whole batch,
             # built from the OLDEST bar in it — _stream_since is a scan
             # over every row logged since that bar, far too expensive to
             # repeat per candidate.
             due: List[Tuple[Any, ...]] = []
             oldest: Dict[Tuple[str, str], Tuple[datetime, str]] = {}
-            for rowid, symbol, tf, price, bar, atr in rows:
+            for rowid, symbol, tf, price, bar, atr, unlabelled in rows:
                 try:
                     stamp = _utc(datetime.fromisoformat(
                         str(bar).replace(" ", "T")))
@@ -1255,8 +1327,16 @@ class SignalLogger:
                     continue
                 if stamp > cutoff:
                     continue              # window has not elapsed yet
+                # Stale = old by the bar it closed on AND by when it was
+                # written (on a live database the two agree within minutes;
+                # requiring both keeps a freshly imported or backfilled DB,
+                # whose rows all carry today's logged_at, out of it).
+                stale = stamp < stale_before and rowid < min_id
+                if stale and not unlabelled:
+                    continue   # final: a hole no later bar can ever fill
                 key = (str(symbol), str(tf))
-                if key not in oldest or stamp < oldest[key][0]:
+                if not stale and (key not in oldest
+                                  or stamp < oldest[key][0]):
                     oldest[key] = (stamp, str(bar))
                 due.append((rowid, symbol, tf, price, bar, atr, key))
             streams: Dict[Tuple[str, str], List[Dict[str, Any]]] = {}
@@ -1265,8 +1345,13 @@ class SignalLogger:
                     break
                 try:
                     if key not in streams:
+                        # 3 days of anchor is ~2,100 rows on the live DB;
+                        # the cap is a safety net, not the working limit.
+                        anchor = oldest.get(key)
                         streams[key] = self._stream_since(
-                            symbol, tf, oldest[key][1])
+                            symbol, tf,
+                            anchor[1] if anchor else anchor_floor.isoformat(),
+                            max_rows=10000)
                     label = scan_outcome(bar, tf, streams[key],
                                          _finite(atr), price)
                     if label is None:
@@ -1277,6 +1362,14 @@ class SignalLogger:
                         + " FROM scan_outcomes WHERE signal_id=?",
                         (rowid,)).fetchone()
                     if prior is not None and tuple(prior) == values:
+                        # Unchanged, so NOT counted as written — but its
+                        # queue position moves, or the same short-stream
+                        # rows would head every pass for ever.
+                        now_iso = datetime.now(timezone.utc).isoformat()
+                        self._conn.execute(
+                            "UPDATE scan_outcomes SET labeled_at=?"
+                            " WHERE signal_id=?", (now_iso, rowid))
+                        self._conn.commit()
                         continue
                     self._conn.execute(
                         _SCAN_OUTCOME_INSERT,
@@ -1376,6 +1469,9 @@ class SignalLogger:
                     continue
                 before = _dumps(state)
                 advanced = advance_outcome(json.loads(before), union)
+                if (advanced.get("status") == "UNRESOLVABLE"
+                        and _is_gap_reason(advanced.get("reason"))):
+                    advanced = resolve_locked_floor(advanced)
                 after = _dumps(advanced)
                 if after == before:
                     continue
@@ -1393,7 +1489,9 @@ class SignalLogger:
         """Bounded retry for continuity-broken outcomes: re-freeze the
         seed from the row's own setup/indicators (at most ``max_reseeds``
         times per row) so the next resolver pass can walk the now-filled
-        stream. Invalid-seed rows stay terminal."""
+        stream. Invalid-seed rows stay terminal. A gap row that had already
+        trailed its stop past the entry is finished at that locked floor
+        instead of re-seeded; the returned list holds re-seeded ids only."""
         reseeded: List[int] = []
         with self._lock:
             rows = self._conn.execute(
@@ -1403,8 +1501,21 @@ class SignalLogger:
                 " AND advisory_event_id = id").fetchall()
             for rowid, tf, setup_json, ind_json, state_json in rows:
                 old = json.loads(state_json) if state_json else {}
-                if "continuity" not in str(old.get("reason", "")) and \
+                if not _is_gap_reason(old.get("reason")):
-                        "gap" not in str(old.get("reason", "")):
+                    continue
+                # The locked trailing stop lives ONLY in this progress
+                # block, and a re-seed rebuilds the walk from scratch over
+                # a stream whose hole is still there (the continuity check
+                # is all-or-nothing), so the floor would be lost for good.
+                # Finish such a row here; re-seed only what has nothing.
+                floored = resolve_locked_floor(json.loads(_dumps(old)))
+                if floored.get("status") == "RESOLVED":
+                    self._conn.execute(
+                        "UPDATE signals SET advisory_outcome_status=?,"
+                        " advisory_r_multiple=?, advisory_outcome_json=?"
+                        " WHERE id=?",
+                        ("RESOLVED", canonical_r(floored),
+                         _dumps(floored), rowid))
                     continue
                 count = int((old.get("reseed") or {}).get("count", 0))
                 if count >= max_reseeds or not setup_json:

telegram_bot.py

@@ -49,9 +49,10 @@ import sys
 import threading
 import time
 from concurrent.futures import ThreadPoolExecutor
-from datetime import datetime, timezone
+from datetime import datetime, timedelta, timezone
 from typing import TYPE_CHECKING, Any, Dict, List, Optional, Set
 
+import pandas as pd
 from telegram import InlineKeyboardButton, InlineKeyboardMarkup, Update
 from telegram.constants import ParseMode
 from telegram.error import BadRequest
@@ -71,6 +72,7 @@ from app_config import (
     telegram_channel_id, violent_watch_enabled,
 )
 from providers import BaseDataProvider, make_data_provider
+from providers.base import _is_daily_maintenance
 from agent_panel import attach_panel, panel_health
 from llm_client import get_llm_client
 from market_outlook import maybe_run_outlook
@@ -110,8 +112,12 @@ _file_handler = logging.handlers.TimedRotatingFileHandler(
     when="midnight", backupCount=30, encoding="utf-8",
 )
 _file_handler.setFormatter(_log_formatter)
+# Console only at IMPORT: `python test_all.py` imports this module, and an
+# import-time file handler wrote fixture alerts (2027 dates, 3300.00 prices)
+# into logs/telegram.log on whatever box ran the suite. main() attaches
+# _file_handler for real bot runs.
 logging.basicConfig(level=getattr(logging, _APP.log_level, logging.INFO),
-                    handlers=[_console_handler, _file_handler],
+                    handlers=[_console_handler],
                     force=True)
 logging.getLogger("httpx").setLevel(logging.WARNING)
 logging.getLogger("httpcore").setLevel(logging.WARNING)
@@ -545,7 +551,21 @@ def _bar_epoch(tf: Timeframe, now: Optional[float] = None) -> int:
 
 
 def _remember_scan(symbol: str, tf: Timeframe, report: SignalReport) -> None:
-    _last_scan[f"{symbol}_{tf.value}"] = (_bar_epoch(tf), report)
+    # File the report under the bar it actually ANALYSED, not the wall
+    # clock at completion: a scan that straddles a boundary would land
+    # under the new bar, so the boundary loop's own scan of that bar
+    # found no cache entry and re-fetched/re-scored/re-logged the same
+    # closed bar (15 duplicate M15 rows in five days — 2026-09-13 audit).
+    epoch = _bar_epoch(tf)
+    if report.primary_analysis is not None:
+        try:
+            # report.timestamp is the primary bar's OPEN label; the bar
+            # closes (= the boundary this scan belongs to) one TF later.
+            epoch = (int(pd.Timestamp(report.timestamp).timestamp())
+                     + _TF_SECONDS_CACHE.get(tf, 900))
+        except (ValueError, TypeError):
+            pass   # unparseable label → wall-clock boundary, as before
+    _last_scan[f"{symbol}_{tf.value}"] = (epoch, report)
 
 
 async def _cached_or_scan(
@@ -767,6 +787,16 @@ def _seconds_until_boundary(interval_minutes: int, buffer: int = 10) -> float:
     return remaining
 
 
+def _closed_bar_in_settlement(interval_minutes: int,
+                              now: Optional[datetime] = None) -> bool:
+    """True when the bar that just closed (label = now - interval) falls in
+    gold's daily settlement pause — no such bar exists, so a scan would only
+    re-read the previous bar (61 stale rows in the 2026-09 audit DB)."""
+    label_time = ((now or datetime.now(timezone.utc))
+                  - timedelta(minutes=interval_minutes))
+    return _is_daily_maintenance(label_time)
+
+
 # ── Violent-move watch (display-only) ────────────────────────────────
 # Per-process heads-up budget: {"day": date, "count": int, "sent": {...}}.
 _watch_state: Dict[str, Any] = {}
@@ -800,7 +830,11 @@ async def _scan_timeframe(
     confirmation_tfs: List[Timeframe],
 ) -> None:
     """Scan all pairs for one timeframe and broadcast results."""
-    tasks = [run_signal(symbol, tf, confirmation_tfs) for symbol in SCAN_PAIRS]
+    # _cached_or_scan, not run_signal: the WS breach handler may already
+    # have scanned this closed bar (and vice versa) — one fetch/score/log
+    # per bar per TF (2026-09-13 audit).
+    tasks = [_cached_or_scan(symbol, tf, confirmation_tfs)
+             for symbol in SCAN_PAIRS]
     results = await asyncio.gather(*tasks, return_exceptions=True)
     reports: List[SignalReport] = []
     for i, result in enumerate(results):
@@ -930,6 +964,9 @@ _pending_breach: Dict[str, tuple] = {}
 # A failed breach rescan shrinks the just-set cooldown to this retry
 # window instead of silencing the symbol for the full cooldown (H33a).
 _BREACH_RETRY_SECONDS = 30
+# A breach landing inside this many seconds after an M15 boundary waits for
+# the candle loop's scan (boundary+10s) and reuses it — 2026-09-13 audit.
+_BREACH_DEFER_SECONDS = 20.0
 
 _cfg = PlatformConfig()
 _WS_COOLDOWN = _cfg.ws_cooldown_seconds
@@ -1003,6 +1040,13 @@ async def _handle_breach(bot, pair_symbol: str, price: float, level_name: str) -
     logger.info(f"Level breach: {display} crossed {label} at {price} — rescanning M15")
 
     try:
+        # A tick a second after an M15 boundary would scan a bar the feed
+        # may still be lagging on, and the candle loop (boundary+10s) would
+        # then scan the same bar again. Wait for the loop and reuse it.
+        elapsed = time.time() - _bar_epoch(Timeframe.M15)
+        if elapsed < _BREACH_DEFER_SECONDS:
+            logger.debug("deferring breach rescan to the boundary scan")
+            await asyncio.sleep(_BREACH_DEFER_SECONDS - elapsed)
         report = await _cached_or_scan(pair_symbol, Timeframe.M15)
         _update_level_cache([report])
     except Exception:
@@ -1152,6 +1196,11 @@ async def _candle_loop(bot, tf: Timeframe, interval_minutes: int) -> None:
             await asyncio.sleep(wait)
             if not _feed_active():
                 break
+            if _closed_bar_in_settlement(interval_minutes):
+                logger.info(
+                    f"{label} auto-scan: skipped — bar label inside the "
+                    f"daily settlement pause")
+                continue
             logger.info(f"{label} auto-scan: firing")
             try:
                 await _scan_timeframe(
@@ -1824,6 +1873,13 @@ def main() -> None:
             "and telegram.chat_id in config.json, or the TELEGRAM_TOKEN and "
             "TELEGRAM_CHAT_ID environment variables.")
 
+    # Only a real bot run writes the rotating log file: tests import this
+    # module (an import-time handler logged their fixtures into
+    # logs/telegram.log) and one of them calls main() to assert the refusal
+    # above, so the attach sits after that guard — 2026-09-13 audit.
+    if _file_handler not in logging.getLogger().handlers:
+        logging.getLogger().addHandler(_file_handler)
+
     _install_excepthooks()
     logger.info("Starting ForexBot2 Telegram bot…")