Scan survives score-refresh failures; fix expired-ORM crash after rollback
A scoring error no longer skips setup detection for the ticker: the rollback already restores a clean transaction, and qualification re-gates on live scores at alert time, so a stale score is recoverable but a skipped scan is not. Iterating symbol strings instead of Ticker instances fixes a latent crash the new regression test caught: rollback() expires ORM objects regardless of expire_on_commit, so touching ticker.symbol in the except handler triggered sync lazy-loading, which raises on an AsyncSession and killed the whole scan on the first per-ticker error. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
@@ -553,9 +553,12 @@ async def scan_all_tickers(
|
|||||||
``progress_callback(processed, total, current_symbol)`` is invoked as each
|
``progress_callback(processed, total, current_symbol)`` is invoked as each
|
||||||
ticker is scanned so callers (e.g. the scheduler) can surface live progress.
|
ticker is scanned so callers (e.g. the scheduler) can surface live progress.
|
||||||
"""
|
"""
|
||||||
result = await db.execute(select(Ticker).order_by(Ticker.symbol))
|
# Plain strings, not Ticker instances: the rollbacks below expire any ORM
|
||||||
tickers = list(result.scalars().all())
|
# objects held across them, and touching an expired attribute afterwards
|
||||||
total = len(tickers)
|
# 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 the universe up front so each new setup carries both the residual
|
# Rank the universe up front so each new setup carries both the residual
|
||||||
# activation gate percentile and the promoted production ordering score.
|
# activation gate percentile and the promoted production ordering score.
|
||||||
@@ -571,34 +574,35 @@ async def scan_all_tickers(
|
|||||||
ranks = {}
|
ranks = {}
|
||||||
|
|
||||||
all_setups: list[TradeSetup] = []
|
all_setups: list[TradeSetup] = []
|
||||||
for index, ticker in enumerate(tickers):
|
for index, symbol in enumerate(symbols):
|
||||||
if progress_callback is not None:
|
if progress_callback is not None:
|
||||||
progress_callback(index, total, ticker.symbol)
|
progress_callback(index, total, symbol)
|
||||||
# Refresh scores first so the scheduled scan works off current data.
|
# Refresh scores first so the scheduled scan works off current data.
|
||||||
# Nothing else marks scores stale, so without this they'd never update
|
# Nothing else marks scores stale, so without this they'd never update
|
||||||
# for tickers the user doesn't manually fetch.
|
# for tickers the user doesn't manually fetch. A refresh failure still
|
||||||
|
# scans the ticker: qualification re-gates on live scores at alert
|
||||||
|
# time, so a stale score is recoverable but a skipped scan is not.
|
||||||
try:
|
try:
|
||||||
from app.services import scoring_service
|
from app.services import scoring_service
|
||||||
|
|
||||||
await scoring_service.compute_all_dimensions(db, ticker.symbol)
|
await scoring_service.compute_all_dimensions(db, symbol)
|
||||||
await scoring_service.compute_composite_score(db, ticker.symbol)
|
await scoring_service.compute_composite_score(db, symbol)
|
||||||
await db.commit()
|
await db.commit()
|
||||||
except Exception:
|
except Exception:
|
||||||
await db.rollback()
|
await db.rollback()
|
||||||
logger.exception("Error refreshing scores for %s", ticker.symbol)
|
logger.exception("Error refreshing scores for %s", symbol)
|
||||||
continue
|
|
||||||
|
|
||||||
try:
|
try:
|
||||||
setups = await scan_ticker(
|
setups = await scan_ticker(
|
||||||
db, ticker.symbol, rr_threshold, atr_multiplier,
|
db, symbol, rr_threshold, atr_multiplier,
|
||||||
momentum_percentile=(ranks.get(ticker.symbol) or {}).get("momentum_percentile"),
|
momentum_percentile=(ranks.get(symbol) or {}).get("momentum_percentile"),
|
||||||
strategy_rank=(ranks.get(ticker.symbol) or {}).get("strategy_rank"),
|
strategy_rank=(ranks.get(symbol) or {}).get("strategy_rank"),
|
||||||
volatility_percentile=(ranks.get(ticker.symbol) or {}).get("volatility_percentile"),
|
volatility_percentile=(ranks.get(symbol) or {}).get("volatility_percentile"),
|
||||||
)
|
)
|
||||||
all_setups.extend(setups)
|
all_setups.extend(setups)
|
||||||
except Exception:
|
except Exception:
|
||||||
await db.rollback()
|
await db.rollback()
|
||||||
logger.exception("Error scanning ticker %s", ticker.symbol)
|
logger.exception("Error scanning ticker %s", symbol)
|
||||||
|
|
||||||
if progress_callback is not None and total:
|
if progress_callback is not None and total:
|
||||||
progress_callback(total, total, "")
|
progress_callback(total, total, "")
|
||||||
|
|||||||
@@ -0,0 +1,61 @@
|
|||||||
|
"""Tests for scan_all_tickers orchestration: error isolation between phases."""
|
||||||
|
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import pytest
|
||||||
|
|
||||||
|
from app.models.ticker import Ticker
|
||||||
|
from app.services import rr_scanner_service, scoring_service
|
||||||
|
from tests.conftest import _test_session_factory # type: ignore
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.fixture
|
||||||
|
async def session():
|
||||||
|
async with _test_session_factory() as s:
|
||||||
|
yield s
|
||||||
|
|
||||||
|
|
||||||
|
async def test_scan_proceeds_when_score_refresh_fails(session, monkeypatch):
|
||||||
|
"""A scoring failure must not skip setup detection for the ticker.
|
||||||
|
|
||||||
|
Qualification re-gates on live scores at alert time, so a stale score is
|
||||||
|
recoverable — a skipped scan is not.
|
||||||
|
"""
|
||||||
|
session.add(Ticker(symbol="AAA"))
|
||||||
|
await session.commit()
|
||||||
|
|
||||||
|
async def _boom(db, symbol):
|
||||||
|
raise RuntimeError("scoring unavailable")
|
||||||
|
|
||||||
|
scanned: list[str] = []
|
||||||
|
|
||||||
|
async def _fake_scan_ticker(db, symbol, *args, **kwargs):
|
||||||
|
scanned.append(symbol)
|
||||||
|
return []
|
||||||
|
|
||||||
|
monkeypatch.setattr(scoring_service, "compute_all_dimensions", _boom)
|
||||||
|
monkeypatch.setattr(rr_scanner_service, "scan_ticker", _fake_scan_ticker)
|
||||||
|
|
||||||
|
setups = await rr_scanner_service.scan_all_tickers(session)
|
||||||
|
|
||||||
|
assert scanned == ["AAA"]
|
||||||
|
assert setups == []
|
||||||
|
|
||||||
|
|
||||||
|
async def test_scan_error_does_not_stop_later_tickers(session, monkeypatch):
|
||||||
|
session.add_all([Ticker(symbol="AAA"), Ticker(symbol="BBB")])
|
||||||
|
await session.commit()
|
||||||
|
|
||||||
|
scanned: list[str] = []
|
||||||
|
|
||||||
|
async def _fake_scan_ticker(db, symbol, *args, **kwargs):
|
||||||
|
scanned.append(symbol)
|
||||||
|
if symbol == "AAA":
|
||||||
|
raise RuntimeError("scan blew up")
|
||||||
|
return []
|
||||||
|
|
||||||
|
monkeypatch.setattr(rr_scanner_service, "scan_ticker", _fake_scan_ticker)
|
||||||
|
|
||||||
|
await rr_scanner_service.scan_all_tickers(session)
|
||||||
|
|
||||||
|
assert scanned == ["AAA", "BBB"]
|
||||||
Reference in New Issue
Block a user