diff --git a/app/services/backtest_service.py b/app/services/backtest_service.py index eaf7b2e..93ce5a8 100644 --- a/app/services/backtest_service.py +++ b/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. diff --git a/frontend/src/components/signals/BacktestDataQualityCard.tsx b/frontend/src/components/signals/BacktestDataQualityCard.tsx new file mode 100644 index 0000000..91e42b7 --- /dev/null +++ b/frontend/src/components/signals/BacktestDataQualityCard.tsx @@ -0,0 +1,77 @@ +import { Callout } from '../ui/Callout'; +import { Disclosure } from '../ui/Disclosure'; +import type { BacktestDataQuality } from '../../lib/types'; + +/** How many stocks were expected, loaded and actually evaluated. + * + * A run that lost every ticker still writes a report — the replay loop catches + * per-ticker errors so one bad name can't abort a 500-symbol run — and that + * report used to be indistinguishable from an honest one that found no setups. + * When the status is `failed` the panel shows this instead of the performance + * tiles, because a headline computed from a broken run is worse than none. + */ +export function BacktestDataQualityCard({ quality }: { quality: BacktestDataQuality }) { + const failed = quality.status === 'failed'; + const { benchmark } = quality; + + return ( + +

+ {failed ? 'This run is not usable' : 'This run is incomplete'} +

+ +

+ {quality.evaluated} of {quality.expected} tickers evaluated + {quality.loaded !== quality.evaluated && <> · {quality.loaded} loaded} + {quality.short_history > 0 && ( + <> · {quality.short_history} too short to trade + )} + {' · '} + {benchmark.momentum_basis === 'residual' ? ( + <> + {benchmark.symbol} {benchmark.sessions} sessions + {benchmark.first && benchmark.last && ( + <> ({benchmark.first} → {benchmark.last}) + )} + + ) : ( + <>no {benchmark.symbol} benchmark — momentum ran raw, not residual + )} +

+ {failed && ( +

+ Performance figures are hidden for this run. Re-run the backtest once the + underlying data is readable. +

+ )} + {/* Says what the check covers, so a clean run isn't read as a price audit. */} +

+ Processing and coverage checks only — this does not verify that the underlying + prices are correct. +

+ {quality.failures.length > 0 && ( +
+ +
    + {quality.failures.map((f) => ( +
  • + {f.symbol} · {f.stage} · {f.error} +
    • + ))} +
+ {quality.failures_truncated && ( +

+ Showing the first {quality.failures.length} of {quality.failed} — the + rest are in the run log. +

+ )} + +
+ )} + + ); +} diff --git a/frontend/src/components/signals/BacktestPanel.tsx b/frontend/src/components/signals/BacktestPanel.tsx index 8155ea2..3b6e41d 100644 --- a/frontend/src/components/signals/BacktestPanel.tsx +++ b/frontend/src/components/signals/BacktestPanel.tsx @@ -9,6 +9,7 @@ import { Disclosure } from '../ui/Disclosure'; import { Dropdown } from '../ui/Dropdown'; import { Section } from '../ui/Section'; import { useToast } from '../ui/Toast'; +import { BacktestDataQualityCard } from './BacktestDataQualityCard'; import { BacktestRecommendationCard } from './BacktestRecommendationCard'; import { PortfolioMonitorPanel } from './PortfolioMonitorPanel'; @@ -39,6 +40,8 @@ export function BacktestPanel() { const [targetModel, setTargetModel] = useState('production_gtl'); const [cadence, setCadence] = useState('weekly'); + const quality = report?.data_quality ?? null; + const runIsUnusable = quality?.status === 'failed'; const monitor = report?.portfolio_monitor ?? null; const activeStrategy = selectedStrategy || monitor?.production_strategy || monitor?.strategies[0]?.strategy || ''; @@ -89,8 +92,15 @@ export function BacktestPanel() {

Last run

{report ? (

- {timeAgo(report.generated_at)} · {report.tickers} tickers ·{' '} - {report.candidates} setups ({report.qualified} qualified) ·{' '} + {timeAgo(report.generated_at)} ·{' '} + {/* Evaluated over expected, not the universe size: "506 tickers" + on a run that evaluated none was the misleading part. */} + + {quality + ? `${quality.evaluated}/${quality.expected} tickers` + : `${report.tickers} tickers`} + {' '} + · {report.candidates} setups ({report.qualified} qualified) ·{' '} {report.params.entry_cadence ?? 'weekly'},{' '} {report.params.horizon_days}d horizon {report.params.cost_per_side_pct != null && ( @@ -177,7 +187,20 @@ export function BacktestPanel() { )} - {report && ( + {/* Completeness gates the headline. `failed` replaces the tiles rather + than warning above them — a banner over ordinary-looking numbers is + exactly how a zero-candidate run reads as a quiet week. */} + {report && quality && quality.status !== 'ok' && ( + + )} + + {report && !quality && ( +

+ Completeness was not recorded for this run — it predates the check. +

+ )} + + {report && !runIsUnusable && ( <> = MIN_MATURED) { status = btAvgR == null ? 'no-backtest' : liveAvgR >= btAvgR - DRIFT_TOLERANCE_R ? 'tracking' : 'drift'; } const statusNote: Record = { building: `Fewer than ~${MIN_MATURED} matured setups so far — too few to compare.`, - 'no-backtest': 'Run the backtest to get a target/stop baseline to check against.', + 'no-backtest': + 'No usable backtest baseline to check against — run the backtest, or fix the data the last run reported as incomplete.', tracking: "Live setup outcomes are resolving in line with the backtest's target/stop model — the outcome-evaluation pipeline shows no look-ahead, config or data drift. (Checks the setup-grading pipeline, not the ATR-trail production book above.)", drift: diff --git a/frontend/src/lib/baseRates.ts b/frontend/src/lib/baseRates.ts index d93391f..94b0f94 100644 --- a/frontend/src/lib/baseRates.ts +++ b/frontend/src/lib/baseRates.ts @@ -44,9 +44,15 @@ const EXIT_LABELS: Record = { /** * Pull the production strategy's full-history row out of a backtest report. - * Returns null when the report hasn't run or has no production row. + * Returns null when the report hasn't run, failed its completeness check, or + * has no production row. */ export function productionBaseRates(report: unknown): BaseRates | null { + // Base rates quoted next to a live setup carry no caveat, so a run that lost + // its universe must yield none at all rather than plausible-looking ones. + if ((report as { data_quality?: { status?: string } })?.data_quality?.status === 'failed') { + return null; + } const monitor = (report as { portfolio_monitor?: { production_strategy?: string; runs?: MonitorRun[] } }) ?.portfolio_monitor; if (!monitor?.runs?.length) return null; diff --git a/frontend/src/lib/types.ts b/frontend/src/lib/types.ts index f52dde3..7754fb2 100644 --- a/frontend/src/lib/types.ts +++ b/frontend/src/lib/types.ts @@ -427,11 +427,39 @@ export interface BacktestSignalEvalRow { reliable: boolean; } +export interface BacktestDataQuality { + /** `failed` = the headline can't be read as the strategy it claims to be. */ + status: 'ok' | 'degraded' | 'failed'; + complete: boolean; + reasons: string[]; + expected: number; + loaded: number; + evaluated: number; + no_data: number; + no_data_symbols: string[]; + short_history: number; + failed: number; + failures_by_stage: Record; + failures: { symbol: string; stage: string; error: string }[]; + failures_truncated: boolean; + benchmark: { + symbol: string; + sessions: number; + first: string | null; + last: string | null; + momentum_basis: 'residual' | 'raw_fallback'; + error: string | null; + }; +} + export interface BacktestReport { generated_at: string; tickers: number; candidates: number; qualified: number; + /** Absent on reports generated before completeness tracking — treat as + * unknown rather than clean; that assumption is the bug this block fixes. */ + data_quality?: BacktestDataQuality; params: { step_days: number; step_sessions?: number; @@ -1010,9 +1038,23 @@ export interface TickerUniverseBootstrapResult { total_universe_symbols: number; added: number; already_tracked: number; + /** Always 0 — pruning marks symbols untracked and never deletes rows. */ deleted: number; + /** Symbols dropped from the scanner; their bars and linked records remain. */ + untracked?: number; + untracked_symbols?: string[]; + /** Symbols that rejoined the universe and were returned to the live path. */ + retracked?: number; + /** Absent from the universe but manually owned, so a prune left them alone. */ + kept_manual?: string[]; + kept_manual_count?: number; + /** False when the live fetch failed (cache/seed fallback) — nothing was recorded. */ + membership_observed?: boolean; + membership_opened?: number; + membership_closed?: number; /** Sample of newly added symbols (capped server-side). */ added_symbols?: string[]; + prune_note?: string; } export interface PipelineReadiness { diff --git a/scripts/run_backtest_snapshot.py b/scripts/run_backtest_snapshot.py index b78ecf8..9ede515 100644 --- a/scripts/run_backtest_snapshot.py +++ b/scripts/run_backtest_snapshot.py @@ -72,9 +72,29 @@ def _parse_args() -> argparse.Namespace: default=None, help="Add a disjoint train/test portfolio report split at YYYY-MM-DD.", ) + parser.add_argument( + "--terminal-events", + default=None, + help=( + "JSON map of {symbol: YYYY-MM-DD} marking symbols whose final bar is a " + "research terminal mark, written by prep_survivorship_snapshot.py. Those " + "symbols are replayed through their last real bar instead of stopping 30 " + "bars early; every other symbol is unaffected." + ), + ) return parser.parse_args() +def _load_terminal_events(path: str | None) -> dict[str, date]: + """Parse the terminal-event map. Explicit input only — never inferred.""" + if not path: + return {} + raw = json.loads(Path(path).read_text(encoding="utf-8")) + if not isinstance(raw, dict): + raise SystemExit(f"Terminal events file must be a JSON object: {path}") + return {str(sym).upper(): date.fromisoformat(str(value)) for sym, value in raw.items()} + + def _default_output_path(cadence: str) -> Path: stamp = datetime.now().strftime("%Y%m%d-%H%M%S") return Path("reports") / f"backtest-{cadence}-{stamp}.json" @@ -190,6 +210,10 @@ async def _main() -> None: raise SystemExit("--holdout-split must use YYYY-MM-DD") from exc os.environ["BACKTEST_HOLDOUT_SPLIT"] = args.holdout_split + terminal_events = _load_terminal_events(args.terminal_events) + if terminal_events: + print(f"terminal events: {len(terminal_events)} symbols ({', '.join(sorted(terminal_events))})") + from app.config import settings from app.services.backtest_service import run_backtest @@ -222,6 +246,7 @@ async def _main() -> None: progress_cb=progress, target_model=args.target_model, cadence=args.cadence, + terminal_events=terminal_events, ) finally: await engine.dispose() diff --git a/tests/unit/test_backtest_data_quality.py b/tests/unit/test_backtest_data_quality.py new file mode 100644 index 0000000..0060485 --- /dev/null +++ b/tests/unit/test_backtest_data_quality.py @@ -0,0 +1,141 @@ +"""Guards for the backtest completeness ledger. + +The failure this exists to prevent is concrete: a snapshot whose schema predates +a migration makes every ``_fetch_columns`` call raise, both dispatch paths catch +per-ticker exceptions and continue, and the run writes a report with zero +candidates that is indistinguishable from an honest run that found no setups. +The ledger has to turn that into a report that says so. +""" + +from __future__ import annotations + +from datetime import date + +from app.services.backtest_service import ( + DATA_QUALITY_FAILURE_SAMPLE, + HORIZON, + MIN_LOOKBACK, + _DataQuality, + _has_short_history, +) + + +def _healthy(expected: int = 3, evaluated: int = 3) -> _DataQuality: + q = _DataQuality(expected=expected, evaluated=evaluated) + q.record_benchmark({date(2024, 1, 2): 470.0, date(2024, 1, 3): 471.0}, None) + return q + + +def test_a_complete_run_is_ok(): + report = _healthy().as_report() + assert report["status"] == "ok" + assert report["complete"] is True + assert report["reasons"] == [] + assert report["loaded"] == report["evaluated"] == 3 + assert report["benchmark"]["momentum_basis"] == "residual" + assert report["benchmark"]["first"] == "2024-01-02" + assert report["benchmark"]["last"] == "2024-01-03" + + +def test_every_symbol_failing_to_load_reads_as_failed_not_as_no_setups(): + """The unsynced-snapshot case: 3 expected, 3 fetch errors, 0 evaluated.""" + q = _DataQuality(expected=3) + q.record_benchmark({date(2024, 1, 2): 470.0}, None) + for symbol in ("AAPL", "MSFT", "NVDA"): + q.record_failure(symbol, "fetch", RuntimeError("no such column: tickers.cik")) + + report = q.as_report() + assert report["status"] == "failed" + assert report["complete"] is False + assert report["evaluated"] == 0 + assert report["loaded"] == 0 + assert report["failed"] == 3 + assert report["failures_by_stage"] == {"fetch": 3} + assert "no symbol was evaluated successfully" in report["reasons"] + assert report["failures"][0]["symbol"] == "AAPL" + assert "tickers.cik" in report["failures"][0]["error"] + + +def test_a_few_failures_degrade_but_do_not_fail_the_run(): + q = _DataQuality(expected=10, evaluated=8, no_data_symbols=["RHM"]) + q.record_benchmark({date(2024, 1, 2): 470.0}, None) + q.record_failure("HBI", "replay", ValueError("bad bar")) + + report = q.as_report() + assert report["status"] == "degraded" + assert report["complete"] is False + assert report["loaded"] == 9 # no_data is not a load failure + assert report["reasons"] == [ + "1 symbol failed during replay", + "1 symbol had no stored bars (RHM)", + ] + + +def test_bar_less_symbols_are_named_so_the_reason_is_actionable(): + """A universe row with no bars recurs on every run until someone clears it. + An unnamed count would just nag; the symbol makes it a one-line fix.""" + q = _DataQuality(expected=9, evaluated=2, no_data_symbols=list("ABCDEFG")) + q.record_benchmark({date(2024, 1, 2): 470.0}, None) + q.evaluated = 2 + + report = q.as_report() + assert report["no_data"] == 7 + assert report["no_data_symbols"] == ["A", "B", "C", "D", "E"] + assert report["reasons"] == ["7 symbols had no stored bars (A, B, C, D, E, …)"] + + +def test_a_missing_benchmark_fails_the_run_because_the_signal_changed(): + """Without SPY the gate ranks on raw 12-1 momentum, so the headline no + longer describes the production strategy even though every ticker loaded.""" + q = _DataQuality(expected=3, evaluated=3) + q.record_benchmark({}, ConnectionError("alpaca unreachable")) + + report = q.as_report() + assert report["status"] == "failed" + assert report["benchmark"]["momentum_basis"] == "raw_fallback" + assert report["benchmark"]["sessions"] == 0 + assert "alpaca unreachable" in report["benchmark"]["error"] + assert any("fell back to raw momentum" in r for r in report["reasons"]) + + +def test_an_unbalanced_ledger_is_itself_a_failure(): + """If a counter stops being maintained the block must not quietly report a + clean run — the arithmetic is the self-check.""" + q = _healthy(expected=10, evaluated=3) # 7 symbols unaccounted for + report = q.as_report() + assert report["status"] == "failed" + assert any("ledger does not balance" in r for r in report["reasons"]) + + +def test_an_empty_universe_is_a_failure(): + q = _DataQuality(expected=0) + q.record_benchmark({date(2024, 1, 2): 470.0}, None) + assert q.as_report()["status"] == "failed" + + +def test_the_failure_list_is_capped_but_the_count_is_not(): + q = _DataQuality(expected=100) + q.record_benchmark({date(2024, 1, 2): 470.0}, None) + for i in range(60): + q.record_failure(f"SYM{i}", "fetch", RuntimeError("boom")) + q.evaluated = 40 + + report = q.as_report() + assert report["failed"] == 60 + assert len(report["failures"]) == DATA_QUALITY_FAILURE_SAMPLE + assert report["failures_truncated"] is True + + +def test_short_history_counts_a_recent_listing_without_failing_the_run(): + q = _healthy() + q.short_history = 1 + report = q.as_report() + assert report["short_history"] == 1 + assert report["status"] == "ok", "a recent listing is not a data-quality failure" + + +def test_has_short_history_uses_the_replay_minimum(): + enough = ([0] * (MIN_LOOKBACK + HORIZON),) + ((),) * 5 + one_short = ([0] * (MIN_LOOKBACK + HORIZON - 1),) + ((),) * 5 + assert _has_short_history(enough) == 0 + assert _has_short_history(one_short) == 1 diff --git a/tests/unit/test_backtest_service.py b/tests/unit/test_backtest_service.py index b16303a..1eec5b5 100644 --- a/tests/unit/test_backtest_service.py +++ b/tests/unit/test_backtest_service.py @@ -1689,6 +1689,74 @@ async def test_run_backtest_rolls_back_a_failed_ticker_fetch(session, monkeypatc assert report["tickers"] == 2 assert report["candidates"] >= 1 + # ...and the report says one ticker was lost rather than presenting the + # remaining one as the whole universe. + quality = report["data_quality"] + assert quality["expected"] == 2 + assert quality["evaluated"] == 1 + assert quality["failed"] == 1 + assert quality["failures"][0]["symbol"] == "AAA" + assert quality["failures"][0]["stage"] == "fetch" + assert "simulated OHLCV read failure" in quality["failures"][0]["error"] + assert quality["complete"] is False + assert quality["status"] == "degraded" + + +async def test_run_backtest_reports_its_own_completeness(session): + """The healthy case: every ticker accounted for, so the report says so.""" + await _seed_oscillating_ticker(session, "OSC") + report = await bt.run_backtest(session) + + quality = report["data_quality"] + assert quality["expected"] == report["tickers"] == 1 + assert quality["evaluated"] == 1 + assert quality["failed"] == 0 + assert quality["no_data"] == 0 + # Ledger identity — the three terminal buckets must cover the universe. + assert quality["no_data"] + quality["failed"] + quality["evaluated"] == quality["expected"] + + +async def test_run_backtest_marks_a_run_failed_when_every_ticker_fails(session, monkeypatch): + """The demonstrated failure: an unsynced snapshot raises on every read, the + replay loop swallows each one, and the run writes a report with no + candidates. It must come back "failed", not look like a quiet run. + """ + await _seed_oscillating_ticker(session, "AAA") + await _seed_oscillating_ticker(session, "OSC") + + async def always_failing_fetch(db, symbol): + raise RuntimeError("no such column: tickers.cik") + + monkeypatch.setattr(bt, "_fetch_columns", always_failing_fetch) + + report = await bt.run_backtest(session) + + assert report["candidates"] == 0 + quality = report["data_quality"] + assert quality["status"] == "failed" + assert quality["evaluated"] == 0 + assert quality["loaded"] == 0 + assert quality["failed"] == 2 + assert "no symbol was evaluated successfully" in quality["reasons"] + + +async def test_run_backtest_flags_a_missing_benchmark(session, monkeypatch): + """Residual momentum degrades to raw without SPY. The numbers then describe + a different gate than the one the report claims, so the run is not clean.""" + await _seed_oscillating_ticker(session, "OSC") + + async def no_benchmark(db, **kwargs): + return {} + + monkeypatch.setattr(bt, "_load_benchmark_closes_for_backtest", no_benchmark) + + report = await bt.run_backtest(session) + benchmark = report["data_quality"]["benchmark"] + + assert benchmark["sessions"] == 0 + assert benchmark["momentum_basis"] == "raw_fallback" + assert report["data_quality"]["status"] == "failed" + async def test_run_backtest_rolls_back_a_failed_portfolio_sim_load(session, monkeypatch): """The portfolio-sim block loads the benchmark and the live exit policy from diff --git a/tests/unit/test_backtest_terminal_events.py b/tests/unit/test_backtest_terminal_events.py new file mode 100644 index 0000000..a4c8449 --- /dev/null +++ b/tests/unit/test_backtest_terminal_events.py @@ -0,0 +1,167 @@ +"""Replay bounds for a symbol carrying a research terminal mark. + +``_replay_ticker`` normally stops ``HORIZON`` bars before the end so every setup +has a full resolution window. For a delisted name that is exactly wrong: the +collapse happens in those final bars, so the signals that would have been caught +by it are the ones the bound hides. ``terminal_date`` opts a single symbol into +walking through its last real bar. + +The fixtures build a controlled setup rather than replaying a real ticker, so +the tests do not depend on whether SIVB happened to qualify on a given day. +""" + +from __future__ import annotations + +from datetime import date, timedelta +from types import SimpleNamespace + +import pytest + +from app.services import backtest_service as bt + + +def _bar(day: date, close: float, *, high: float | None = None, low: float | None = None): + return SimpleNamespace( + date=day, + open=close, + high=close * 1.01 if high is None else high, + low=close * 0.99 if low is None else low, + close=close, + volume=1_000_000, + ) + + +def _series(n: int, *, start: date = date(2021, 1, 4)) -> list: + """A steadily rising series long enough to clear MIN_LOOKBACK.""" + bars = [] + day = start + for i in range(n): + while day.weekday() >= 5: + day += timedelta(days=1) + bars.append(_bar(day, 100.0 + i * 0.25)) + day += timedelta(days=1) + return bars + + +def _terminal_bar(previous: date) -> SimpleNamespace: + day = previous + timedelta(days=1) + while day.weekday() >= 5: + day += timedelta(days=1) + return SimpleNamespace( + date=day, open=0.01, high=0.01, low=0.01, close=0.01, volume=0 + ) + + +@pytest.fixture() +def replay(monkeypatch): + """Force one long setup on every evaluated bar so signal dates are countable.""" + + def _fake_setups(window, config, activation, *, target_model): + close = float(window[-1].close) + return [{ + "direction": "long", + "entry": close, + "stop": close * 0.95, + # Reachable within HORIZON on the rising fixture, but not within the + # four bars left before a terminal event. + "target": close * 1.02, + "rr": 2.0, + "confidence": 80, + "momentum": 0.1, + "best_prob": 0.5, + "primary_prob": 0.5, + "meets_core": True, + "action": "BUY", + "risk_level": "medium", + "target_model": target_model, + "primary_sources": [], + "primary_strength": 1.0, + "primary_rejection_count": 0, + "primary_last_rejection_age": None, + "primary_distance_atr": 1.0, + "raw_level_count": 1, + "gate_level_count": 1, + }] + + monkeypatch.setattr(bt, "_window_setups", _fake_setups) + + def _run(records, terminal_date=None): + return bt._replay_ticker( + "TEST", records, {}, {}, None, + bt.PRODUCTION_GTL_TARGET_MODEL, "daily", terminal_date, + ) + + return _run + + +def test_setup_near_a_terminal_event_is_emitted_and_resolves_at_the_mark(replay): + """A setup five sessions before the halt is replayed, and if it is still open + at the terminal bar the near-zero price resolves it as a loss.""" + real = _series(bt.MIN_LOOKBACK + 40) + records = real + [_terminal_bar(real[-1].date)] + + without = replay(records) + with_mark = replay(records, terminal_date=records[-1].date) + + last_real = real[-1].date + fifth_last = real[-5].date + + without_dates = {c["date"] for c in without} + with_dates = {c["date"] for c in with_mark} + + assert fifth_last.isoformat() not in without_dates, ( + "baseline should hide signals inside the final HORIZON bars" + ) + assert fifth_last.isoformat() in with_dates + assert last_real.isoformat() in with_dates + + late = next(c for c in with_mark if c["date"] == fifth_last.isoformat()) + assert late["outcome"] == bt.OUTCOME_STOP_HIT + assert late["realized_r"] < -1.0, "a gap to 0.01 must fill below the stop, not at it" + + +def test_no_setup_is_generated_on_the_terminal_bar(replay): + real = _series(bt.MIN_LOOKBACK + 40) + terminal = _terminal_bar(real[-1].date) + records = real + [terminal] + + candidates = replay(records, terminal_date=terminal.date) + + assert terminal.date.isoformat() not in {c["date"] for c in candidates} + assert max(c["date"] for c in candidates) == real[-1].date.isoformat() + + +def test_a_position_that_exits_before_the_event_keeps_its_own_exit(replay): + """An early setup resolves on its own bars; the terminal mark must not + retroactively turn it into a total loss.""" + real = _series(bt.MIN_LOOKBACK + 40) + records = real + [_terminal_bar(real[-1].date)] + + with_mark = replay(records, terminal_date=records[-1].date) + early_date = real[bt.MIN_LOOKBACK - 1].date.isoformat() + + early = next(c for c in with_mark if c["date"] == early_date) + assert early["outcome"] == bt.OUTCOME_TARGET_HIT + assert early["realized_r"] == pytest.approx(2.0) + + +def test_unmapped_symbols_and_runs_without_the_option_are_unchanged(replay): + """The ordinary replay bounds are untouched when no terminal date is given.""" + records = _series(bt.MIN_LOOKBACK + 40) + + baseline = replay(records) + again = replay(records, terminal_date=None) + + assert [c["date"] for c in baseline] == [c["date"] for c in again] + assert max(c["date"] for c in baseline) < records[-1].date.isoformat(), ( + "without a terminal mark the walk must still stop HORIZON bars early" + ) + + +def test_a_mismatched_terminal_date_is_rejected(replay): + """The date is validated against the final bar rather than trusted, so a + stale events file fails loudly instead of shifting the replay window.""" + records = _series(bt.MIN_LOOKBACK + 40) + + with pytest.raises(ValueError, match="does not match the final bar"): + replay(records, terminal_date=date(1999, 1, 1)) diff --git a/tests/unit/test_run_backtest_snapshot_terminal_events.py b/tests/unit/test_run_backtest_snapshot_terminal_events.py new file mode 100644 index 0000000..a916ceb --- /dev/null +++ b/tests/unit/test_run_backtest_snapshot_terminal_events.py @@ -0,0 +1,123 @@ +"""Runner-level guard: a stale terminal-event map must abort the whole run. + +``_replay_ticker`` rejects a mismatched date, but both dispatch paths in +``run_backtest`` swallow per-ticker exceptions so one unreadable symbol cannot +kill a long replay. That safety net would turn a wrong events file into a +quietly incomplete report, so the map is validated before any replay starts — +and the runner must therefore write no report at all. +""" + +from __future__ import annotations + +import json +from datetime import date, datetime, timedelta +from pathlib import Path + +import pytest +from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine + +from app.database import Base +from app.models.ohlcv import OHLCVRecord +from app.models.ticker import Ticker +from app.services.backtest_service import TerminalEventError, run_backtest + + +async def _make_snapshot(path: Path, *, symbol: str = "DEAD", bars: int = 40) -> date: + """A minimal snapshot with one ticker. Returns its final bar date.""" + engine = create_async_engine(f"sqlite+aiosqlite:///{path.as_posix()}") + async with engine.begin() as conn: + await conn.run_sync(Base.metadata.create_all) + + Session = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False) + day = date(2022, 1, 3) + last = day + async with Session() as db: + ticker = Ticker(symbol=symbol, name=symbol, created_at=datetime.utcnow()) + db.add(ticker) + await db.flush() + for i in range(bars): + while day.weekday() >= 5: + day += timedelta(days=1) + price = 100.0 + i + db.add(OHLCVRecord( + ticker_id=ticker.id, date=day, open=price, high=price * 1.01, + low=price * 0.99, close=price, volume=1_000_000, + created_at=datetime.utcnow(), + )) + last = day + day += timedelta(days=1) + await db.commit() + await engine.dispose() + return last + + +@pytest.mark.asyncio +async def test_a_mismatched_terminal_date_aborts_the_run(tmp_path): + snapshot = tmp_path / "snap.sqlite" + final_bar = await _make_snapshot(snapshot) + + engine = create_async_engine(f"sqlite+aiosqlite:///{snapshot.as_posix()}") + Session = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False) + try: + async with Session() as db: + with pytest.raises(TerminalEventError, match="final bar"): + await run_backtest(db, terminal_events={"DEAD": final_bar + timedelta(days=5)}) + finally: + await engine.dispose() + + +@pytest.mark.asyncio +async def test_an_unknown_symbol_aborts_the_run(tmp_path): + snapshot = tmp_path / "snap.sqlite" + await _make_snapshot(snapshot) + + engine = create_async_engine(f"sqlite+aiosqlite:///{snapshot.as_posix()}") + Session = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False) + try: + async with Session() as db: + with pytest.raises(TerminalEventError, match="not in this snapshot"): + await run_backtest(db, terminal_events={"NOPE": date(2022, 2, 1)}) + finally: + await engine.dispose() + + +@pytest.mark.asyncio +async def test_the_runner_writes_no_report_when_validation_fails(tmp_path, monkeypatch): + """The report is written only after run_backtest returns, so an abort must + leave the output path untouched rather than a partial or empty report.""" + snapshot = tmp_path / "snap.sqlite" + final_bar = await _make_snapshot(snapshot) + + events = tmp_path / "events.json" + events.write_text(json.dumps({"DEAD": (final_bar + timedelta(days=5)).isoformat()})) + out = tmp_path / "report.json" + + import scripts.run_backtest_snapshot as runner + + monkeypatch.setattr( + runner.sys, "argv", + ["run_backtest_snapshot.py", str(snapshot), "--out", str(out), + "--terminal-events", str(events), "--quiet", "--workers", "1"], + ) + + with pytest.raises(TerminalEventError): + await runner._main() + + assert not out.exists(), "a failed run must not leave a report behind" + + +@pytest.mark.asyncio +async def test_a_matching_map_is_accepted(tmp_path): + """The guard must not reject a correct map — the run proceeds normally.""" + snapshot = tmp_path / "snap.sqlite" + final_bar = await _make_snapshot(snapshot) + + engine = create_async_engine(f"sqlite+aiosqlite:///{snapshot.as_posix()}") + Session = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False) + try: + async with Session() as db: + report = await run_backtest(db, terminal_events={"DEAD": final_bar}) + finally: + await engine.dispose() + + assert isinstance(report, dict)