feat(backtest): make an incomplete run visibly incomplete

A snapshot whose schema predated a migration made every _fetch_columns call
raise, both dispatch paths caught the per-ticker exception and continued, and
the run wrote a report with zero candidates that was indistinguishable from an
honest run that found no setups. Nothing said the run had lost its universe.

run_backtest now builds a completeness ledger and emits it as
report["data_quality"]: expected / loaded / evaluated counts, per-symbol
failures with stage and reason, bar-less symbols named, and benchmark
availability. Every universe symbol lands in exactly one terminal bucket and
the arithmetic is checked, so a counter that stops being maintained reports
itself as a failure rather than under-counting quietly.

Benchmark availability is a separate axis: without SPY, residual 12-1 momentum
silently degrades to raw 12-1, and the report now says so instead of leaving it
in a log line.

Three statuses, because severity differs. "failed" (nothing evaluated, empty
universe, unbalanced ledger, missing benchmark) replaces the performance tiles
with the callout rather than warning above them — a banner over ordinary-looking
numbers is exactly how the zero-candidate run would still have read as a quiet
week. "degraded" keeps the tiles with a note. A report with no data_quality
block reads as unknown, not clean; defaulting old reports to complete would
reproduce the bug being fixed.

The same gate applies to the two other consumers of backtest numbers:
productionBaseRates returns null for a failed run, and EvaluationPanel will not
report drift against a broken baseline.

"complete" means the processing and coverage checks passed. It does not mean
the prices are correct — a complete run over silently wrong bars is still
complete, and the card says so.

Also threads an explicit terminal-events option through the replay, validated
before dispatch so a stale events file aborts the run instead of quietly
degrading it. Verified against a healthy snapshot (degraded, 505/506) and one
with tickers.cik dropped (failed, 0/506, 506 fetch failures).

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

app/services/backtest_service.py

@@ -50,14 +50,16 @@ import statistics
 from collections import defaultdict
 from collections.abc import Callable
 from concurrent.futures import ProcessPoolExecutor
+from dataclasses import dataclass, field
 from datetime import date, datetime, timezone
 from types import SimpleNamespace
 from typing import Any
 
-from sqlalchemy import select
+from sqlalchemy import func, select
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.config import settings
+from app.models.ohlcv import OHLCVRecord
 from app.models.ticker import Ticker
 from app.services import settings_store
 from app.services.admin_service import get_activation_config, update_setting
@@ -497,16 +499,40 @@ def _replay_ticker(
     benchmark_closes: dict[date, float] | None = None,
     target_model: str = PRODUCTION_GTL_TARGET_MODEL,
     cadence: str = DEFAULT_BACKTEST_CADENCE,
+    terminal_date: date | None = None,
 ) -> list[dict]:
-    """Walk one ticker at the selected cadence and resolve each setup outcome."""
+    """Walk one ticker at the selected cadence and resolve each setup outcome.
+
+    ``terminal_date`` marks a symbol whose final bar is a research terminal mark
+    (a confirmed near-total loss written by ``prep_survivorship_snapshot.py``).
+    The normal loop stops ``HORIZON`` bars early so every setup has a full
+    resolution window, which would hide every signal in the last 29 real bars —
+    exactly the ones a collapse resolves. For a marked symbol the walk instead
+    runs through the last *real* bar and never proposes a setup on the synthetic
+    one. Nothing else changes: outcomes still come from the real forward bars,
+    and a setup the shortened history cannot resolve is skipped rather than
+    counted as an expiry.
+    """
     cadence = validate_backtest_cadence(cadence)
     step_sessions = backtest_step_sessions(cadence)
     candidates: list[dict] = []
     n = len(records)
-    if n < MIN_LOOKBACK + HORIZON:
-        return candidates
 
-    for i in range(MIN_LOOKBACK - 1, n - HORIZON, step_sessions):
+    if terminal_date is not None:
+        if not records or records[-1].date != terminal_date:
+            raise ValueError(
+                f"{symbol}: terminal event {terminal_date} does not match the final bar "
+                f"{records[-1].date if records else None}"
+            )
+        if n < MIN_LOOKBACK + 1:
+            return candidates
+        stop_index = n - 1  # exclusive: the last signal date is the last real bar
+    else:
+        if n < MIN_LOOKBACK + HORIZON:
+            return candidates
+        stop_index = n - HORIZON
+
+    for i in range(MIN_LOOKBACK - 1, stop_index, step_sessions):
         window = records[: i + 1]
         forward = records[i + 1 :]
         forward_bars = [Bar(date=r.date, high=r.high, low=r.low) for r in forward]
@@ -1278,6 +1304,7 @@ def _replay_and_signals(
     target_model: str = PRODUCTION_GTL_TARGET_MODEL,
     cadence: str = DEFAULT_BACKTEST_CADENCE,
     signal_only: bool = False,
+    terminal_date: date | None = None,
 ) -> tuple[list[dict], dict]:
     """The CPU-bound per-ticker work, as a top-level (picklable) function so it can
     run in a worker process. Takes primitive column arrays (cheap to pickle),
@@ -1303,6 +1330,7 @@ def _replay_and_signals(
             benchmark_closes,
             target_model,
             cadence,
+            terminal_date,
         )
     return (
         candidates,
@@ -1475,6 +1503,217 @@ async def _rollback_quietly(db: AsyncSession, context: str) -> None:
         logger.exception("Session rollback after %s also failed", context)
 
 
+# Failures are listed individually up to this many; beyond it the report keeps
+# the counts and says the list was truncated.
+DATA_QUALITY_FAILURE_SAMPLE = 25
+# Bar-less universe rows are named rather than counted while the list is short.
+NO_DATA_SAMPLE = 5
+
+
+def _has_short_history(columns: tuple) -> int:
+    """1 when a loaded ticker has too few bars for ``_replay_ticker`` to resolve
+    a single outcome — a recent listing, not an error, but worth counting since
+    it looks identical to "found no setups" in the report."""
+    return int(len(columns[0]) < MIN_LOOKBACK + HORIZON)
+
+
+@dataclass
+class _DataQuality:
+    """Completeness ledger for one backtest run.
+
+    Both dispatch paths swallow per-ticker exceptions so a single unreadable
+    name can't abort a 500-symbol replay. That is right for a long run and wrong
+    for its report: a snapshot whose schema predates a migration fails *every*
+    ``_fetch_columns`` call and still writes a report that reads like an
+    ordinary one that simply found no setups. This records what actually
+    happened so the report — and the UI — can say so.
+
+    Every universe symbol ends in exactly one of ``no_data``, ``failures`` or
+    ``evaluated``. ``short_history`` is a subset of ``evaluated`` (loaded, but
+    too few bars to produce a setup), not a fourth bucket. :meth:`as_report`
+    checks the three against ``expected`` and treats an unbalanced ledger as a
+    failure in its own right — a counter that stopped being maintained should
+    surface as loudly as a missing ticker.
+
+    **What ``complete`` does and does not claim.** It means the run's processing
+    and coverage checks passed: every expected symbol was loaded and evaluated,
+    and the benchmark was there. It says nothing about whether the underlying
+    prices are *correct* — a complete run over silently wrong bars is still
+    complete. Splits, adjustment drift and bad vendor prints are a different
+    problem, and this ledger must not be read as clearing them.
+    """
+
+    expected: int = 0
+    evaluated: int = 0
+    short_history: int = 0
+    no_data_symbols: list[str] = field(default_factory=list)
+    failures: list[dict] = field(default_factory=list)
+    benchmark_sessions: int = 0
+    benchmark_error: str | None = None
+    benchmark_first: date | None = None
+    benchmark_last: date | None = None
+
+    def record_failure(self, symbol: str, stage: str, exc: BaseException) -> None:
+        """Stage is ``fetch`` (reading bars) or ``replay`` (evaluating them).
+
+        Only the first line is kept: a SQLAlchemy error appends the full SQL and
+        its parameters, which buries the one useful clause ("no such column:
+        tickers.cik") under 300 characters of SELECT. The full traceback is
+        already in the run log.
+        """
+        first_line = str(exc).strip().splitlines()[0] if str(exc).strip() else ""
+        self.failures.append({
+            "symbol": symbol,
+            "stage": stage,
+            "error": f"{type(exc).__name__}: {first_line}"[:300],
+        })
+
+    def record_benchmark(self, closes: dict[date, float], error: BaseException | None) -> None:
+        self.benchmark_sessions = len(closes)
+        self.benchmark_error = f"{type(error).__name__}: {error}"[:300] if error else None
+        if closes:
+            self.benchmark_first = min(closes)
+            self.benchmark_last = max(closes)
+
+    def as_report(self) -> dict:
+        # Lazily imported like ``_load_benchmark_closes_for_backtest`` does, so
+        # building a report never drags in the provider stack.
+        from app.services.benchmark_service import BENCHMARK_SYMBOL
+
+        by_stage: dict[str, int] = defaultdict(int)
+        for entry in self.failures:
+            by_stage[entry["stage"]] += 1
+        no_data = len(self.no_data_symbols)
+        accounted = no_data + len(self.failures) + self.evaluated
+        balanced = accounted == self.expected
+
+        # A run is "failed" when its headline numbers can't be read as the
+        # strategy the report claims to describe; "degraded" when the strategy
+        # is intact but ran on less than the whole universe.
+        reasons: list[str] = []
+        if self.expected == 0:
+            reasons.append("no symbols in the backtest universe")
+        elif self.evaluated == 0:
+            reasons.append("no symbol was evaluated successfully")
+        if not balanced:
+            reasons.append(
+                f"ledger does not balance: {accounted} symbols accounted for, "
+                f"{self.expected} expected"
+            )
+        if self.benchmark_sessions == 0:
+            # Residual 12-1 momentum silently degrades to raw 12-1 without a
+            # benchmark stream, so the gate is not the production gate.
+            reasons.append(
+                "benchmark unavailable — residual momentum fell back to raw momentum"
+                + (f" ({self.benchmark_error})" if self.benchmark_error else "")
+            )
+        failed = bool(reasons)
+
+        # These strings are shown verbatim in the UI, so they read as sentences.
+        def _symbols(n: int) -> str:
+            return f"{n} symbol" if n == 1 else f"{n} symbols"
+
+        if by_stage.get("fetch"):
+            reasons.append(f"{_symbols(by_stage['fetch'])} failed to load")
+        if by_stage.get("replay"):
+            reasons.append(f"{_symbols(by_stage['replay'])} failed during replay")
+        if no_data:
+            # Named while the list is short, because a universe row with no bars
+            # is a standing wart that recurs on every run until someone clears
+            # it — an unnamed count would just nag.
+            named = ", ".join(sorted(self.no_data_symbols)[:NO_DATA_SAMPLE])
+            if no_data > NO_DATA_SAMPLE:
+                named += ", …"
+            reasons.append(f"{_symbols(no_data)} had no stored bars ({named})")
+
+        return {
+            "status": "failed" if failed else ("degraded" if reasons else "ok"),
+            "complete": not reasons,
+            "reasons": reasons,
+            "expected": self.expected,
+            "loaded": self.expected - no_data - by_stage.get("fetch", 0),
+            "evaluated": self.evaluated,
+            "no_data": no_data,
+            "no_data_symbols": sorted(self.no_data_symbols)[:NO_DATA_SAMPLE],
+            "short_history": self.short_history,
+            "failed": len(self.failures),
+            "failures_by_stage": dict(by_stage),
+            "failures": self.failures[:DATA_QUALITY_FAILURE_SAMPLE],
+            "failures_truncated": len(self.failures) > DATA_QUALITY_FAILURE_SAMPLE,
+            "benchmark": {
+                "symbol": BENCHMARK_SYMBOL,
+                "sessions": self.benchmark_sessions,
+                "first": self.benchmark_first.isoformat() if self.benchmark_first else None,
+                "last": self.benchmark_last.isoformat() if self.benchmark_last else None,
+                "momentum_basis": "residual" if self.benchmark_sessions else "raw_fallback",
+                "error": self.benchmark_error,
+            },
+        }
+
+
+async def _load_backtest_symbols(db: AsyncSession) -> list[str]:
+    """Every ticker with stored history — deliberately *not* ``active_only``.
+
+    A replay grades what the strategy could have selected at each past date, so
+    it must see symbols that have since been delisted or dropped out of the
+    index. Filtering on today's membership here is precisely how a universe
+    becomes the survivors: the audit in ``docs/research/survivorship-audit.md``
+    measured 2.09 Sharpe against 1.57 for the same window on that difference
+    alone. Guarded by
+    ``tests/unit/test_universe_membership.py::test_todays_membership_does_not_erase_yesterdays_candidates``.
+
+    Plain strings, not Ticker instances: the rollbacks in the replay loop expire
+    any ORM objects held across them, and touching an expired attribute
+    afterwards triggers sync lazy-loading, which raises on an AsyncSession.
+    """
+    result = await db.execute(select(Ticker.symbol).order_by(Ticker.symbol))
+    return list(result.scalars().all())
+
+
+class TerminalEventError(RuntimeError):
+    """A terminal-event map does not describe the snapshot it was given."""
+
+
+async def _validate_terminal_events(
+    db: AsyncSession, terminal_events: dict[str, date], symbols: list[str]
+) -> None:
+    """Check every mapped symbol before any replay is dispatched.
+
+    ``_replay_ticker`` also rejects a mismatch, but both dispatch paths swallow
+    per-ticker exceptions so the replay survives one unreadable name — which
+    would turn a stale events file into a quietly incomplete report instead of a
+    failure. Validating up front makes the whole run fail, before any work.
+    """
+    tracked = set(symbols)
+    missing = sorted(set(terminal_events) - tracked)
+
+    rows = await db.execute(
+        select(Ticker.symbol, func.max(OHLCVRecord.date))
+        .join(OHLCVRecord, OHLCVRecord.ticker_id == Ticker.id)
+        .where(Ticker.symbol.in_(list(terminal_events)))
+        .group_by(Ticker.symbol)
+    )
+    final_bar = dict(rows.all())
+
+    mismatched = []
+    for symbol, expected in sorted(terminal_events.items()):
+        if symbol in missing:
+            continue
+        actual = final_bar.get(symbol)
+        if actual != expected:
+            mismatched.append(f"{symbol}: events file says {expected}, final bar is {actual}")
+
+    problems = []
+    if missing:
+        problems.append(f"symbols not in this snapshot: {', '.join(missing)}")
+    if mismatched:
+        problems.append("; ".join(mismatched))
+    if problems:
+        raise TerminalEventError(
+            "Terminal-event map does not match the snapshot — " + " | ".join(problems)
+        )
+
+
 async def _fetch_columns(db: AsyncSession, symbol: str) -> tuple | None:
     """Read one ticker's OHLCV and detach it to primitive column arrays in the
     event loop (safe ORM access), ready to hand to a worker. None if no data."""
@@ -4066,19 +4305,25 @@ async def run_backtest(
     *,
     target_model: str = PRODUCTION_GTL_TARGET_MODEL,
     cadence: str = DEFAULT_BACKTEST_CADENCE,
+    terminal_events: dict[str, date] | None = None,
 ) -> dict:
-    """Replay every ticker and aggregate the Phase-1 reports for the current config."""
+    """Replay every ticker and aggregate the Phase-1 reports for the current config.
+
+    ``terminal_events`` maps a symbol to the date of a research terminal mark on
+    its final bar (see ``scripts/prep_survivorship_snapshot.py``). Unmapped
+    symbols keep the ordinary replay bounds, so an empty or omitted map leaves
+    every existing run byte-identical.
+    """
     target_model = validate_backtest_target_model(target_model)
     cadence = validate_backtest_cadence(cadence)
+    terminal_events = terminal_events or {}
     config = await get_recommendation_config(db)
     activation = await get_activation_config(db)
 
-    # Plain strings, not Ticker instances: the rollbacks below expire any ORM
-    # objects held across them, and touching an expired attribute afterwards
-    # triggers sync lazy-loading, which raises on an AsyncSession.
-    result = await db.execute(select(Ticker.symbol).order_by(Ticker.symbol))
-    symbols = list(result.scalars().all())
+    symbols = await _load_backtest_symbols(db)
     total = len(symbols)
+    if terminal_events:
+        await _validate_terminal_events(db, terminal_events, symbols)
     rank_only_symbols = await _load_research_rank_only_symbols(db)
     if rank_only_symbols:
         logger.info(json.dumps({
@@ -4086,6 +4331,7 @@ async def run_backtest(
             "count": len(rank_only_symbols),
         }))
 
+    quality = _DataQuality(expected=total)
     candidates: list[dict] = []
     # Signal IC remains a weekly, non-overlapping diagnostic regardless of the
     # entry cadence. Production activation ranks are assigned from candidates
@@ -4096,13 +4342,16 @@ async def run_backtest(
     # if SPY benchmark data is unavailable, the residual signal simply won't be
     # emitted and the rest of the report remains valid.
     benchmark_closes: dict[date, float] = {}
+    benchmark_error: Exception | None = None
     try:
         benchmark_closes = await _load_benchmark_closes_for_backtest(
             db, days=settings.ohlcv_history_days + 365
         )
-    except Exception:
+    except Exception as exc:
+        benchmark_error = exc
         logger.exception("Benchmark load for residual momentum failed")
         await _rollback_quietly(db, "benchmark load")
+    quality.record_benchmark(benchmark_closes, benchmark_error)
 
     def _merge(result: tuple[list[dict], dict]) -> None:
         cands, series = result
@@ -4135,31 +4384,48 @@ async def run_backtest(
         with pool:
             for start in range(0, total, chunk):
                 batch = symbols[start : start + chunk]
-                futures = []
+                # Symbol travels with its future: ``gather`` returns bare
+                # exceptions, so without the pairing a worker failure can't name
+                # the ticker it lost.
+                dispatched: list[tuple[str, Any]] = []
                 for symbol in batch:
                     try:
                         columns = await _fetch_columns(db, symbol)
-                    except Exception:
+                    except Exception as exc:
+                        quality.record_failure(symbol, "fetch", exc)
                         logger.exception("Backtest fetch failed for %s", symbol)
                         await _rollback_quietly(db, f"fetch for {symbol}")
                         continue
-                    if columns is not None:
-                        futures.append(loop.run_in_executor(
-                            pool,
-                            _replay_and_signals,
-                            symbol,
-                            columns,
-                            config,
-                            activation,
-                            benchmark_closes,
-                            target_model,
-                            cadence,
-                            symbol in rank_only_symbols,
-                        ))
-                for result in await asyncio.gather(*futures, return_exceptions=True):
-                    if isinstance(result, Exception):
-                        logger.error(json.dumps({"event": "backtest_worker_error", "message": str(result)}))
+                    if columns is None:
+                        quality.no_data_symbols.append(symbol)
+                        continue
+                    quality.short_history += _has_short_history(columns)
+                    dispatched.append((symbol, loop.run_in_executor(
+                        pool,
+                        _replay_and_signals,
+                        symbol,
+                        columns,
+                        config,
+                        activation,
+                        benchmark_closes,
+                        target_model,
+                        cadence,
+                        symbol in rank_only_symbols,
+                        terminal_events.get(symbol),
+                    )))
+                results = await asyncio.gather(
+                    *(fut for _, fut in dispatched), return_exceptions=True
+                )
+                for (symbol, _), result in zip(dispatched, results):
+                    if isinstance(result, BaseException):
+                        quality.record_failure(symbol, "replay", result)
+                        logger.error(json.dumps({
+                            "event": "backtest_worker_error",
+                            "symbol": symbol,
+                            "message": str(result),
+                        }))
                     else:
+                        quality.evaluated += 1
                         _merge(result)
                 done += len(batch)
                 if progress_cb is not None:
@@ -4172,21 +4438,34 @@ async def run_backtest(
                 progress_cb(index, total, symbol)
             try:
                 columns = await _fetch_columns(db, symbol)
-                if columns is not None:
-                    _merge(await asyncio.to_thread(
-                        _replay_and_signals,
-                        symbol,
-                        columns,
-                        config,
-                        activation,
-                        benchmark_closes,
-                        target_model,
-                        cadence,
-                        symbol in rank_only_symbols,
-                    ))
-            except Exception:
+            except Exception as exc:
+                quality.record_failure(symbol, "fetch", exc)
+                logger.exception("Backtest fetch failed for %s", symbol)
+                await _rollback_quietly(db, f"fetch for {symbol}")
+                continue
+            if columns is None:
+                quality.no_data_symbols.append(symbol)
+                continue
+            quality.short_history += _has_short_history(columns)
+            try:
+                _merge(await asyncio.to_thread(
+                    _replay_and_signals,
+                    symbol,
+                    columns,
+                    config,
+                    activation,
+                    benchmark_closes,
+                    target_model,
+                    cadence,
+                    symbol in rank_only_symbols,
+                    terminal_events.get(symbol),
+                ))
+            except Exception as exc:
+                quality.record_failure(symbol, "replay", exc)
                 logger.exception("Backtest replay failed for %s", symbol)
                 await _rollback_quietly(db, f"replay for {symbol}")
+                continue
+            quality.evaluated += 1
 
     if progress_cb is not None and total:
         progress_cb(total, total, "")
@@ -4298,12 +4577,21 @@ async def run_backtest(
             # backstop for whichever DB call actually failed.
             await _rollback_quietly(db, "portfolio simulation")
 
+    data_quality = quality.as_report()
+    if not data_quality["complete"]:
+        logger.warning(json.dumps({
+            "event": "backtest_incomplete",
+            "status": data_quality["status"],
+            "reasons": data_quality["reasons"],
+        }))
+
     report = {
         "generated_at": datetime.now(timezone.utc).isoformat(),
         "tickers": total,
         "rank_only_tickers": len(rank_only_symbols),
         "candidates": len(candidates),
         "qualified": len(qualified),
+        "data_quality": data_quality,
         "params": {
             # Keep step_days for old report consumers; the value counts stored
             # market sessions rather than calendar days.