From 6a10c8ff09bcfd81339b97d11c1e382fd9bc1f87 Mon Sep 17 00:00:00 2001 From: Dennis Thiessen Date: Tue, 21 Jul 2026 09:31:24 +0200 Subject: [PATCH] fix: guarantee shadow scan freshness, long-only, user-scoped setup list Second review round on the shadow book; all three findings were real. - Scan freshness is now proven, not assumed. Pipeline steps run and fail independently, so a disabled or failed scan step still let the shadow step run on the newest *stored* setups -- a prior session's picks at stale prices. scan_all_tickers now records a run boundary (last_scan_run_started_at / _completed_at) only on successful completion; the shadow book refuses to trade unless COMPLETED is fresh and selects only setups with detected_at >= the run start. Deduplication to the latest row per ticker now happens BEFORE qualification, so a newer unqualified row suppresses an older qualified one rather than the reverse. - Shadow selection is hard long-only. setup_qualifies only enforces long-only when min_momentum_percentile > 0, but 0 is a legal admin setting, and the cash accounting assumes long positions -- so the constraint is enforced in shadow selection regardless of gate config. - The personal setup list excludes only the caller's own open positions. get_trade_setups gained exclude_open_trade_user_id; the trades route passes the authenticated user, while the Telegram broadcast stays global since it has no single owner. New tests cover stale/absent scan markers, prior-run exclusion, newer unqualified suppressing older qualified, long-only under a disabled gate, and both sides of the user-scoped exclusion. Co-Authored-By: Claude Fable 5 --- app/routers/trades.py | 3 +- app/services/rr_scanner_service.py | 32 ++++- app/services/shadow_book_service.py | 85 +++++++++---- tests/unit/test_rr_scanner_preservation.py | 18 +++ tests/unit/test_shadow_book_service.py | 132 ++++++++++++++++++--- 5 files changed, 230 insertions(+), 40 deletions(-) diff --git a/app/routers/trades.py b/app/routers/trades.py index 1059089..5ae705b 100644 --- a/app/routers/trades.py +++ b/app/routers/trades.py @@ -25,7 +25,7 @@ async def list_trade_setups( None, description="Filter by action: LONG_HIGH, LONG_MODERATE, SHORT_HIGH, SHORT_MODERATE, NEUTRAL", ), - _user=Depends(require_access), + user=Depends(require_access), db: AsyncSession = Depends(get_db), ) -> APIEnvelope: """Get latest trade setups with recommendation data.""" @@ -36,6 +36,7 @@ async def list_trade_setups( recommended_action=recommended_action, live_recommendation=True, exclude_open_trade_tickers=True, + exclude_open_trade_user_id=user.id, exclude_reentry_gate_locked_tickers=True, ) diff --git a/app/services/rr_scanner_service.py b/app/services/rr_scanner_service.py index 7fbe2cf..2139fc9 100644 --- a/app/services/rr_scanner_service.py +++ b/app/services/rr_scanner_service.py @@ -30,6 +30,7 @@ from app.services.indicator_service import _extract_ohlcv, compute_atr from app.services.price_service import query_ohlcv from app.services.qualification import setup_qualifies from app.services.sr_service import detect_gate_target_ladder +from app.services import settings_store from app.services.trade_policy import ( MANUAL_BOOK, SHADOW_BOOK, @@ -46,6 +47,13 @@ from app.services.recommendation_service import ( logger = logging.getLogger(__name__) +# Boundary of the most recent *successful* scan. Written only when +# scan_all_tickers completes, so a consumer can tell a scan actually ran this +# pipeline pass (freshness of COMPLETED) and which setups belong to it +# (detected_at >= STARTED). The shadow book relies on both. +KEY_LAST_SCAN_STARTED = "last_scan_run_started_at" +KEY_LAST_SCAN_COMPLETED = "last_scan_run_completed_at" + STRATEGY_VERSION = "residual_highvol_80_20_atr_trail3_v1" # A setup counts as live only while the daily scan keeps re-emitting it. The @@ -812,6 +820,19 @@ async def scan_all_tickers( if progress_callback is not None and total: progress_callback(total, total, "") + # Record the run boundary only now that the scan has completed. The shadow + # book refuses to trade unless COMPLETED is fresh (proving a scan ran in this + # pipeline pass, not a prior session) and selects only setups from this run + # (detected_at >= STARTED). Written after the loop so a hard failure above + # leaves the previous, now-stale, marker in place. + await settings_store.upsert_setting( + db, KEY_LAST_SCAN_STARTED, gate_observation_started_at.isoformat() + ) + await settings_store.upsert_setting( + db, KEY_LAST_SCAN_COMPLETED, datetime.now(timezone.utc).isoformat() + ) + await db.commit() + return all_setups @@ -823,6 +844,7 @@ async def get_trade_setups( symbol: str | None = None, live_recommendation: bool = False, exclude_open_trade_tickers: bool = False, + exclude_open_trade_user_id: int | None = None, exclude_reentry_gate_locked_tickers: bool = False, include_reentry_gate_lock: bool = False, ) -> list[dict]: @@ -855,11 +877,19 @@ async def get_trade_setups( # construction, so letting its positions hide setups would leave the # discretionary list picking over leftovers — and would bias the very # shadow-vs-manual comparison the shadow book exists to measure. - open_trade_result = await db.execute( + open_trade_stmt = ( select(PaperTrade.ticker_id) .where(PaperTrade.status == "open", PaperTrade.book == MANUAL_BOOK) .distinct() ) + # Scope to one user for the personal setup list (don't hide a name just + # because someone else holds it); leave it global for the Telegram + # broadcast, which has no single owner. + if exclude_open_trade_user_id is not None: + open_trade_stmt = open_trade_stmt.where( + PaperTrade.user_id == exclude_open_trade_user_id + ) + open_trade_result = await db.execute(open_trade_stmt) excluded_ticker_ids.update( ticker_id for ticker_id, in open_trade_result.all() ) diff --git a/app/services/shadow_book_service.py b/app/services/shadow_book_service.py index 23eeb1e..705b3a1 100644 --- a/app/services/shadow_book_service.py +++ b/app/services/shadow_book_service.py @@ -53,8 +53,11 @@ DEFAULT_START_EQUITY = 100_000.0 # equity — a leveraged trade the validated strategy would never have taken. NOTIONAL_CAP = 0.20 -# Setups older than this mean the scan did not run in this pipeline pass. -MAX_SETUP_AGE = timedelta(hours=6) +# If the last successful scan completed longer ago than this, no scan ran in the +# current pipeline pass (scans are daily, ~24h apart), so there is nothing fresh +# to trade. Comfortably longer than a scan's own duration, far shorter than the +# gap between scans. +MAX_SCAN_AGE = timedelta(hours=6) async def get_config(db: AsyncSession) -> dict: @@ -166,37 +169,77 @@ async def _shadow_user_id(db: AsyncSession) -> int | None: return int(row[0]) if row else None +async def _last_scan_start(db: AsyncSession, *, now: datetime) -> datetime | None: + """Start of the last successful scan, if it ran in this pipeline pass. + + Returns None — meaning "no scan to act on" — unless the scanner's COMPLETED + marker is fresh. Pipeline steps fail independently, so a scan that was + disabled, errored, or produced nothing leaves a stale marker; trading on the + newest stored setups then would enter a previous session's picks at stale + prices. Freshness is proven by the marker, not by setup age. + """ + from app.services import rr_scanner_service as rr + + started = _parse_dt(await settings_store.get_value(db, rr.KEY_LAST_SCAN_STARTED)) + completed = _parse_dt( + await settings_store.get_value(db, rr.KEY_LAST_SCAN_COMPLETED) + ) + if started is None or completed is None: + return None + if now - completed > MAX_SCAN_AGE: + return None + return started + + +def _parse_dt(raw: str | None) -> datetime | None: + if not raw: + return None + try: + return datetime.fromisoformat(raw) + except ValueError: + return None + + async def _todays_qualified_setups( db: AsyncSession, config: dict, *, now: datetime ) -> list[TradeSetup]: - """Latest qualified setup per ticker from the scan that just ran. + """Long-only qualified setups from the scan that just ran, best rank first. - Freshness is a hard requirement, not a nicety: pipeline steps are allowed to - fail independently, so if the scan is disabled or errors, the newest stored - setups belong to a previous session. Trading those would enter yesterday's - picks at yesterday's prices and quietly corrupt the record. Anything older - than ``MAX_SETUP_AGE`` is treated as "no scan happened". + Order matters here, and matches the review's requirement: - Ordered by ``strategy_rank`` descending — the ordering the backtest selects - on. Setups without a rank sort last; they cannot be compared to ranked ones. + 1. Take only rows from the current run (``detected_at >= scan start``). The + previous run's setups sit ~24h earlier and are excluded, so a stale row + can never be traded even if it once qualified. + 2. Keep long only. The validated strategy is long-only, but the gate permits + shorts when ``min_momentum_percentile`` is 0 (a legal admin setting), and + the cash accounting assumes longs — so this is enforced here, not left to + the gate. + 3. Deduplicate to the latest row per ticker *before* qualifying, so a newer + unqualified row correctly suppresses an older qualified one rather than + the reverse. + 4. Qualify, then rank by ``strategy_rank`` (unranked sort last). """ - cutoff = now - MAX_SETUP_AGE - result = await db.execute( - select(TradeSetup).where(TradeSetup.detected_at >= cutoff) - ) - qualified = [s for s in result.scalars() if setup_qualifies(s, config)] + run_start = await _last_scan_start(db, now=now) + if run_start is None: + return [] + + result = await db.execute( + select(TradeSetup).where(TradeSetup.detected_at >= run_start) + ) + rows = [s for s in result.scalars() if (s.direction or "long") == "long"] - # One setup per ticker — the most recent wins. A ticker can have several - # rows in a scan (e.g. both directions); ranking over duplicates would let - # one name occupy more than its share of the ordering. latest: dict[int, TradeSetup] = {} - for setup in qualified: + for setup in rows: held = latest.get(setup.ticker_id) - if held is None or setup.detected_at > held.detected_at: + if held is None or (setup.detected_at, setup.id) > ( + held.detected_at, + held.id, + ): latest[setup.ticker_id] = setup + qualified = [s for s in latest.values() if setup_qualifies(s, config)] return sorted( - latest.values(), + qualified, key=lambda s: ( s.strategy_rank if s.strategy_rank is not None else float("-inf") ), diff --git a/tests/unit/test_rr_scanner_preservation.py b/tests/unit/test_rr_scanner_preservation.py index 50921f7..c5f6cd5 100644 --- a/tests/unit/test_rr_scanner_preservation.py +++ b/tests/unit/test_rr_scanner_preservation.py @@ -603,6 +603,24 @@ async def test_get_trade_setups_can_exclude_tickers_with_open_paper_trades( assert "OPENQ" not in discovery_symbols assert {"CLOSEDQ", "FREEQ"}.issubset(discovery_symbols) + # Scoped to a *different* user: OPENQ is user 1's position, so user 2's + # personal list must still show it (their list is not filtered by someone + # else's holdings). + scoped_rows = await get_trade_setups( + db_session, + exclude_open_trade_tickers=True, + exclude_open_trade_user_id=2, + ) + assert "OPENQ" in {row["symbol"] for row in scoped_rows} + + # Scoped to the owning user: their own open position is excluded. + owner_rows = await get_trade_setups( + db_session, + exclude_open_trade_tickers=True, + exclude_open_trade_user_id=1, + ) + assert "OPENQ" not in {row["symbol"] for row in owner_rows} + ticker_rows = await get_trade_setups(db_session, symbol="OPENQ") assert [row["symbol"] for row in ticker_rows] == ["OPENQ"] diff --git a/tests/unit/test_shadow_book_service.py b/tests/unit/test_shadow_book_service.py index ba0f823..33043d2 100644 --- a/tests/unit/test_shadow_book_service.py +++ b/tests/unit/test_shadow_book_service.py @@ -49,11 +49,20 @@ async def _seed(session, symbols: list[str]) -> dict[str, int]: return ids -def _setup(ticker_id: int, *, rank: float, detected: datetime, entry=100.0, stop=95.0): - target = entry + 3 * (entry - stop) +def _setup( + ticker_id: int, + *, + rank: float, + detected: datetime, + entry=100.0, + stop=95.0, + direction="long", +): + reward = abs(entry - stop) * 3 + target = entry + reward if direction == "long" else entry - reward return TradeSetup( ticker_id=ticker_id, - direction="long", + direction=direction, entry_price=entry, stop_loss=stop, target=target, @@ -70,6 +79,20 @@ def _setup(ticker_id: int, *, rank: float, detected: datetime, entry=100.0, stop ) +async def _mark_scan(session, *, started: datetime, completed: datetime | None = None): + """Record a successful scan run so the shadow book has something to act on.""" + from app.services import rr_scanner_service as rr + + completed = completed or started + await shadow_book_service.settings_store.upsert_setting( + session, rr.KEY_LAST_SCAN_STARTED, started.isoformat() + ) + await shadow_book_service.settings_store.upsert_setting( + session, rr.KEY_LAST_SCAN_COMPLETED, completed.isoformat() + ) + await session.commit() + + class TestSizing: def test_risks_one_percent_down_to_the_stop(self): shares = shadow_book_service.position_shares(100_000, 1.0, 100.0, 95.0) @@ -105,6 +128,7 @@ class TestSelection: async def test_takes_top_ranked_up_to_capacity(self, session): ids = await _seed(session, ["AAA", "BBB", "CCC"]) now = datetime.now(timezone.utc) + scan_start = now - timedelta(minutes=5) session.add_all( [ _setup(ids["AAA"], rank=0.10, detected=now), @@ -113,9 +137,11 @@ class TestSelection: ] ) await session.commit() + await _mark_scan(session, started=scan_start, completed=now) await shadow_book_service.settings_store.upsert_setting( session, shadow_book_service.KEY_CAPACITY, "2" ) + await session.commit() summary = await shadow_book_service.open_shadow_positions( session, activation_config=_CONFIG @@ -140,6 +166,7 @@ class TestSelection: ) ) await session.commit() + await _mark_scan(session, started=now - timedelta(minutes=5), completed=now) summary = await shadow_book_service.open_shadow_positions( session, activation_config=_CONFIG @@ -163,6 +190,7 @@ class TestSelection: ) ) await session.commit() + await _mark_scan(session, started=now - timedelta(minutes=5), completed=now) summary = await shadow_book_service.open_shadow_positions( session, activation_config=_CONFIG @@ -172,14 +200,12 @@ class TestSelection: assert summary["skipped_locked"] == 1 -class TestSetupFreshness: +class TestScanFreshness: @pytest.mark.asyncio - async def test_ignores_setups_from_a_previous_session(self, session): - """If the scan step failed or was disabled, the newest stored setups are - yesterday's — trading them would enter stale picks at stale prices.""" + async def test_no_scan_marker_means_no_trades(self, session): + """A fresh DB / never-run scan must not trade anything.""" ids = await _seed(session, ["AAA"]) - stale = datetime.now(timezone.utc) - timedelta(days=1) - session.add(_setup(ids["AAA"], rank=0.9, detected=stale)) + session.add(_setup(ids["AAA"], rank=0.9, detected=datetime.now(timezone.utc))) await session.commit() summary = await shadow_book_service.open_shadow_positions( @@ -189,22 +215,94 @@ class TestSetupFreshness: assert summary["opened"] == 0 @pytest.mark.asyncio - async def test_one_position_per_ticker_from_duplicate_setups(self, session): + async def test_stale_scan_marker_refuses_even_fresh_looking_setups(self, session): + """If this pipeline's scan failed/was disabled, the completion marker is + from a prior session — refuse, no matter how recent the setup rows look.""" ids = await _seed(session, ["AAA"]) now = datetime.now(timezone.utc) - session.add_all( - [ - _setup(ids["AAA"], rank=0.5, detected=now - timedelta(minutes=30)), - _setup(ids["AAA"], rank=0.9, detected=now), - ] - ) + session.add(_setup(ids["AAA"], rank=0.9, detected=now)) await session.commit() + # Marker is a day old → no scan ran in this pass. + await _mark_scan(session, started=now - timedelta(days=1, minutes=5), + completed=now - timedelta(days=1)) summary = await shadow_book_service.open_shadow_positions( session, activation_config=_CONFIG ) - assert summary["opened"] == 1 + assert summary["opened"] == 0 + + @pytest.mark.asyncio + async def test_setups_before_this_run_are_excluded(self, session): + """A qualified row from a previous run (before the current scan start) + must not be traded even though the current scan completed.""" + ids = await _seed(session, ["AAA", "BBB"]) + now = datetime.now(timezone.utc) + scan_start = now - timedelta(minutes=5) + session.add_all( + [ + _setup(ids["AAA"], rank=0.9, detected=now), # this run + _setup(ids["BBB"], rank=0.8, detected=now - timedelta(hours=20)), # prior run + ] + ) + await session.commit() + await _mark_scan(session, started=scan_start, completed=now) + + summary = await shadow_book_service.open_shadow_positions( + session, activation_config=_CONFIG + ) + + assert summary["symbols"] == [ids["AAA"]] + + @pytest.mark.asyncio + async def test_newer_unqualified_row_suppresses_older_qualified(self, session): + """Dedup happens before qualification: a fresh unqualified row for a + ticker must beat an earlier qualified row, not the other way round.""" + ids = await _seed(session, ["AAA"]) + now = datetime.now(timezone.utc) + scan_start = now - timedelta(minutes=10) + # Earlier row qualifies; later row fails the R:R floor (rr 1.0 < 2.0). + older = _setup(ids["AAA"], rank=0.9, detected=now - timedelta(minutes=8)) + newer = TradeSetup( + ticker_id=ids["AAA"], direction="long", entry_price=100.0, + stop_loss=95.0, target=105.0, rr_ratio=1.0, composite_score=70.0, + confidence_score=70.0, detected_at=now, strategy_rank=0.9, + momentum_percentile=90.0, recommended_action="buy", + targets_json=json.dumps( + [{"price": 105.0, "probability": 45.0, "is_primary": True, "rr": 1.0}] + ), + ) + session.add_all([older, newer]) + await session.commit() + await _mark_scan(session, started=scan_start, completed=now) + + summary = await shadow_book_service.open_shadow_positions( + session, activation_config=_CONFIG + ) + + assert summary["opened"] == 0 + + +class TestLongOnly: + @pytest.mark.asyncio + async def test_shorts_are_never_taken_even_with_gate_disabled(self, session): + """min_momentum_percentile=0 lets shorts pass the gate; shadow is always + long-only regardless, and its cash accounting assumes longs.""" + ids = await _seed(session, ["AAA"]) + now = datetime.now(timezone.utc) + session.add( + _setup(ids["AAA"], rank=0.9, detected=now, direction="short", + entry=100.0, stop=105.0) + ) + await session.commit() + await _mark_scan(session, started=now - timedelta(minutes=5), completed=now) + + gate_off = {**_CONFIG, "min_momentum_percentile": 0.0} + summary = await shadow_book_service.open_shadow_positions( + session, activation_config=gate_off + ) + + assert summary["opened"] == 0 class TestBookIsolation: