diff --git a/app/services/backtest_service.py b/app/services/backtest_service.py index 317f3e7..45aa35a 100644 --- a/app/services/backtest_service.py +++ b/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, "") diff --git a/tests/unit/test_backtest_service.py b/tests/unit/test_backtest_service.py index aff0eb6..9742582 100644 --- a/tests/unit/test_backtest_service.py +++ b/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