fix(backtest): roll back the session after a swallowed DB failure

Every DB call in run_backtest is best-effort so one unreadable ticker cannot
abort the whole replay, but the handlers swallowed the exception without
clearing the transaction. asyncpg then reports "current transaction is
aborted" for every later statement, and the first unguarded one — the report
write — surfaced it as the job error, long after the real cause.

Add _rollback_quietly at the three swallowing sites (benchmark load, parallel
fetch, sequential replay), matching the guard price_service already uses.

Load plain symbols instead of Ticker instances: a rollback expires ORM objects
held across it, and touching an expired attribute afterwards triggers sync
lazy-loading, which raises on an AsyncSession. rr_scanner_service hit this
same trap. Only .symbol was ever used.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

app/services/backtest_service.py

@@ -1460,6 +1460,21 @@ def _mp_context():
     return None
 
 
+async def _rollback_quietly(db: AsyncSession, context: str) -> None:
+    """Discard a failed unit of work so later statements on this session survive.
+
+    Every DB call in ``run_backtest`` is best-effort — one unreadable ticker must
+    not abort the whole replay. But swallowing the exception alone leaves asyncpg
+    in "current transaction is aborted": every later statement then fails the same
+    way until the first unguarded one (the report write) surfaces it as the job
+    error, long after the real cause. Same guard as ``price_service``.
+    """
+    try:
+        await db.rollback()
+    except Exception:
+        logger.exception("Session rollback after %s also failed", context)
+
+
 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."""
@@ -4037,9 +4052,12 @@ async def run_backtest(
     config = await get_recommendation_config(db)
     activation = await get_activation_config(db)
 
-    result = await db.execute(select(Ticker).order_by(Ticker.symbol))
-    tickers = list(result.scalars().all())
-    total = len(tickers)
+    # 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())
+    total = len(symbols)
     rank_only_symbols = await _load_research_rank_only_symbols(db)
     if rank_only_symbols:
         logger.info(json.dumps({
@@ -4063,6 +4081,7 @@ async def run_backtest(
         )
     except Exception:
         logger.exception("Benchmark load for residual momentum failed")
+        await _rollback_quietly(db, "benchmark load")
 
     def _merge(result: tuple[list[dict], dict]) -> None:
         cands, series = result
@@ -4094,26 +4113,27 @@ async def run_backtest(
         done = 0
         with pool:
             for start in range(0, total, chunk):
-                batch = tickers[start : start + chunk]
+                batch = symbols[start : start + chunk]
                 futures = []
-                for ticker in batch:
+                for symbol in batch:
                     try:
-                        columns = await _fetch_columns(db, ticker.symbol)
+                        columns = await _fetch_columns(db, symbol)
                     except Exception:
-                        logger.exception("Backtest fetch failed for %s", ticker.symbol)
+                        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,
-                            ticker.symbol,
+                            symbol,
                             columns,
                             config,
                             activation,
                             benchmark_closes,
                             target_model,
                             cadence,
-                            ticker.symbol in rank_only_symbols,
+                            symbol in rank_only_symbols,
                         ))
                 for result in await asyncio.gather(*futures, return_exceptions=True):
                     if isinstance(result, Exception):
@@ -4126,25 +4146,26 @@ async def run_backtest(
     else:
         # Sequential fallback (Windows / 1 worker): run each replay in a worker
         # thread so the event loop — and the API server — stays responsive.
-        for index, ticker in enumerate(tickers):
+        for index, symbol in enumerate(symbols):
             if progress_cb is not None:
-                progress_cb(index, total, ticker.symbol)
+                progress_cb(index, total, symbol)
             try:
-                columns = await _fetch_columns(db, ticker.symbol)
+                columns = await _fetch_columns(db, symbol)
                 if columns is not None:
                     _merge(await asyncio.to_thread(
                         _replay_and_signals,
-                        ticker.symbol,
+                        symbol,
                         columns,
                         config,
                         activation,
                         benchmark_closes,
                         target_model,
                         cadence,
-                        ticker.symbol in rank_only_symbols,
+                        symbol in rank_only_symbols,
                     ))
             except Exception:
-                logger.exception("Backtest replay failed for %s", ticker.symbol)
+                logger.exception("Backtest replay failed for %s", symbol)
+                await _rollback_quietly(db, f"replay for {symbol}")
 
     if progress_cb is not None and total:
         progress_cb(total, total, "")

tests/unit/test_backtest_service.py

@@ -1620,3 +1620,39 @@ async def test_run_backtest_smoke(session):
     sweep = sorted(report["sweep"], key=lambda r: r["min_momentum_percentile"], reverse=True)
     counts = [r["total"] for r in sweep]
     assert counts == sorted(counts)  # ascending as threshold descends
+
+
+async def test_run_backtest_rolls_back_a_failed_ticker_fetch(session, monkeypatch):
+    """A failed per-ticker read must not leave the session mid-failed-transaction.
+
+    Every DB call in the replay loop is best-effort, but swallowing the error
+    without a rollback leaves asyncpg in "current transaction is aborted": every
+    later statement fails the same way until the first unguarded one — the report
+    write — surfaces it as the job error, long after the real cause.
+    """
+    await _seed_oscillating_ticker(session, "AAA")
+    await _seed_oscillating_ticker(session, "OSC")
+
+    real_fetch = bt._fetch_columns
+    rolled_back: list[str] = []
+
+    async def failing_fetch(db, symbol):
+        if symbol == "AAA":
+            raise RuntimeError("simulated OHLCV read failure")
+        return await real_fetch(db, symbol)
+
+    real_rollback = session.rollback
+
+    async def tracking_rollback():
+        rolled_back.append("x")
+        await real_rollback()
+
+    monkeypatch.setattr(bt, "_fetch_columns", failing_fetch)
+    monkeypatch.setattr(session, "rollback", tracking_rollback)
+
+    report = await bt.run_backtest(session)
+
+    assert rolled_back, "a failed ticker fetch left the session un-rolled-back"
+    # the surviving ticker is still replayed after the rollback
+    assert report["tickers"] == 2
+    assert report["candidates"] >= 1