From 565484de879388793eaebd2de765cad5cc0627c7 Mon Sep 17 00:00:00 2001 From: Dennis Thiessen Date: Tue, 21 Jul 2026 11:25:03 +0200 Subject: [PATCH] fix: select shadow book setups by scan run id, not a time window The run-id marker proved which scan wrote last, but the shadow book still selected setups by detected_at >= scan_start. An overlapping manual scan could insert rows in that same window; if the pipeline's scan wrote the marker last its id matched and the shadow book proceeded, then swept in -- or ranked highest -- a manual-scan row. The identity check gated entry but selection did not. Carry the run id onto the rows. Migration 025 adds an indexed trade_setups.scan_run_id. scan_all_tickers computes one id per run (pipeline's when a step, else fresh), passes it to scan_ticker which stamps every row after enhancement, and writes the same id to the completion marker. The shadow book selects WHERE scan_run_id == the matched id, so a concurrent scan's rows are excluded by identity regardless of their detected_at. The now-unused STARTED marker is dropped; COMPLETED (freshness) and RUN_ID (identity) remain. Decisive test: the pipeline's id matches, but a same-window manual row with a higher rank is present and is excluded -- only the pipeline's own row is traded. A time-window select would have swept it in and ranked it first. Co-Authored-By: Claude Fable 5 --- .../versions/025_trade_setup_scan_run_id.py | 38 +++++++++ app/models/trade_setup.py | 5 ++ app/services/rr_scanner_service.py | 42 +++++----- app/services/shadow_book_service.py | 42 +++++----- tests/unit/test_shadow_book_service.py | 80 ++++++++++++------- 5 files changed, 136 insertions(+), 71 deletions(-) create mode 100644 alembic/versions/025_trade_setup_scan_run_id.py diff --git a/alembic/versions/025_trade_setup_scan_run_id.py b/alembic/versions/025_trade_setup_scan_run_id.py new file mode 100644 index 0000000..dfda761 --- /dev/null +++ b/alembic/versions/025_trade_setup_scan_run_id.py @@ -0,0 +1,38 @@ +"""trade_setup scan_run_id — identity of the producing scan run + +Revision ID: 025 +Revises: 024 +Create Date: 2026-07-21 00:00:00.000000 + +The shadow book must select the exact batch produced by its pipeline's scan. +Matching the scan-completion marker's run id proves which scan wrote last, but +setup selection was still a detected_at window that a concurrent manual scan +could write rows into. Stamping each row with its scan's run id lets the shadow +book select by identity instead. Existing rows are null (they predate the +column and are never traded by the shadow book). +""" +from typing import Sequence, Union + +from alembic import op +import sqlalchemy as sa + + +revision: str = "025" +down_revision: Union[str, None] = "024" +branch_labels: Union[str, Sequence[str], None] = None +depends_on: Union[str, Sequence[str], None] = None + + +def upgrade() -> None: + op.add_column( + "trade_setups", + sa.Column("scan_run_id", sa.String(length=32), nullable=True), + ) + op.create_index( + "ix_trade_setups_scan_run_id", "trade_setups", ["scan_run_id"] + ) + + +def downgrade() -> None: + op.drop_index("ix_trade_setups_scan_run_id", table_name="trade_setups") + op.drop_column("trade_setups", "scan_run_id") diff --git a/app/models/trade_setup.py b/app/models/trade_setup.py index 0375364..f60bfcf 100644 --- a/app/models/trade_setup.py +++ b/app/models/trade_setup.py @@ -45,6 +45,11 @@ class TradeSetup(Base): DateTime(timezone=True), nullable=True ) outcome_date: Mapped[date | None] = mapped_column(Date, nullable=True) + # Identity of the scan run that produced this row. The shadow book selects + # its batch by this id, not by a detected_at window, so a concurrent manual + # scan writing rows in the same time window is excluded by identity. Null on + # rows predating the column and on any non-scan creator. + scan_run_id: Mapped[str | None] = mapped_column(String(32), nullable=True) ticker = relationship("Ticker", back_populates="trade_setups") diff --git a/app/services/rr_scanner_service.py b/app/services/rr_scanner_service.py index cc56ce3..7c1503b 100644 --- a/app/services/rr_scanner_service.py +++ b/app/services/rr_scanner_service.py @@ -47,13 +47,11 @@ from app.services.recommendation_service import ( logger = logging.getLogger(__name__) -# Boundary of the most recent *successful* scan. Written together, only when -# scan_all_tickers completes: STARTED bounds which setups belong to the run -# (detected_at >= STARTED), COMPLETED gives its freshness, and RUN_ID identifies -# the pipeline invocation that produced it (or a fresh id for a manual scan). -# The shadow book matches RUN_ID exactly rather than trusting timestamps, so a -# concurrent manual scan cannot be mistaken for the pipeline's own. -KEY_LAST_SCAN_STARTED = "last_scan_run_started_at" +# Markers of the most recent *successful* scan, written together only when +# scan_all_tickers completes. COMPLETED gives its freshness; RUN_ID identifies +# the run — the same id stamped on every setup row it produced. The shadow book +# matches RUN_ID exactly and then selects setups by that id, so neither a +# concurrent manual scan nor a stale prior run can be mistaken for it. KEY_LAST_SCAN_COMPLETED = "last_scan_run_completed_at" KEY_LAST_SCAN_RUN_ID = "last_scan_run_id" @@ -527,6 +525,7 @@ async def scan_ticker( volatility_percentile: float | None = None, primary_min_rr: float | None = None, gate_levels_override: list[Any] | None = None, + scan_run_id: str | None = None, ) -> list[TradeSetup]: """Scan a single ticker for trade setups meeting the R:R threshold. @@ -693,6 +692,9 @@ async def scan_ticker( enhanced_setups.append(setup) for setup in enhanced_setups: + # Stamp identity after enhancement so it survives regardless of how the + # enhancer rebuilds the row; the shadow book selects its batch by this. + setup.scan_run_id = scan_run_id db.add(setup) await db.commit() @@ -753,6 +755,13 @@ async def scan_all_tickers( evaluated_ticker_ids: set[int] = set() qualified_ticker_ids: set[int] = set() gate_observation_started_at = datetime.now(timezone.utc) + # One id for the whole run: stamped on every setup row and written to the + # completion marker, so the shadow book can select this run's batch by + # identity. From the pipeline when run as its scan step; a fresh id (never + # matching any pipeline's) when triggered standalone. + from app.services import pipeline_run + + scan_run_id = pipeline_run.current() or pipeline_run.new_run_id() for index, (ticker_id, symbol) in enumerate(ticker_rows): if progress_callback is not None: progress_callback(index, total, symbol) @@ -785,6 +794,7 @@ async def scan_all_tickers( strategy_rank=(ranks.get(symbol) or {}).get("strategy_rank"), volatility_percentile=(ranks.get(symbol) or {}).get("volatility_percentile"), primary_min_rr=PRIMARY_TARGET_MIN_RR, + scan_run_id=scan_run_id, ) all_setups.extend(setups) if activation is not None: @@ -823,22 +833,14 @@ 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, stamped with - # the run id of the pipeline this scan ran inside (or a fresh id when run - # standalone — a manual scan then can never match a pipeline's expected id). - # All three markers are written in one commit so a reader sees a consistent - # (started, completed, run_id) triple, and a hard failure above leaves the - # previous, now-superseded, markers in place. - from app.services import pipeline_run - - run_id = pipeline_run.current() or pipeline_run.new_run_id() - await settings_store.upsert_setting( - db, KEY_LAST_SCAN_STARTED, gate_observation_started_at.isoformat() - ) + # Publish the run markers only now that the scan has completed: COMPLETED for + # freshness and RUN_ID (the same id stamped on this run's setup rows) for + # identity, in one commit. A hard failure above leaves the previous, + # now-superseded, markers in place — so the shadow book will not match. await settings_store.upsert_setting( db, KEY_LAST_SCAN_COMPLETED, datetime.now(timezone.utc).isoformat() ) - await settings_store.upsert_setting(db, KEY_LAST_SCAN_RUN_ID, run_id) + await settings_store.upsert_setting(db, KEY_LAST_SCAN_RUN_ID, scan_run_id) await db.commit() return all_setups diff --git a/app/services/shadow_book_service.py b/app/services/shadow_book_service.py index 9ff00d2..87c992f 100644 --- a/app/services/shadow_book_service.py +++ b/app/services/shadow_book_service.py @@ -169,13 +169,13 @@ async def _shadow_user_id(db: AsyncSession) -> int | None: return int(row[0]) if row else None -async def _last_scan_start( +async def _scan_run_to_trade( db: AsyncSession, *, now: datetime, expected_run_id: str | None = None, -) -> datetime | None: - """Start of the scan we may act on, or None if there is none. +) -> str | None: + """The run id whose setups the shadow book may act on, or None. * ``expected_run_id`` set (pipeline step): the stored run id must match it exactly. This is the airtight guarantee — a scan that was disabled or @@ -184,27 +184,25 @@ async def _last_scan_start( stamps its own id even when it finishes last, so neither can be mistaken for the pipeline's own scan. Timestamp order alone cannot tell them apart. * ``expected_run_id`` None (direct Admin trigger): fall back to the freshness - window. There is no pipeline scan to bind to, so acting on a recent scan is - the operator's explicit choice. + window on the last scan's own id. There is no pipeline scan to bind to, so + acting on a recent scan is the operator's explicit choice. - The three markers are written in one commit, so the returned STARTED belongs - to the same run as the matched RUN_ID and correctly bounds its setups. + Setups are then selected by ``scan_run_id`` equal to the returned id, so a + concurrent scan's rows in the same time window are excluded by identity. """ 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) ) run_id = await settings_store.get_value(db, rr.KEY_LAST_SCAN_RUN_ID) - if started is None or completed is None: + if completed is None or not run_id: return None if expected_run_id is not None: - if not run_id or run_id != expected_run_id: - return None - elif now - completed > MAX_SCAN_AGE: + return run_id if run_id == expected_run_id else None + if now - completed > MAX_SCAN_AGE: return None - return started + return run_id def _parse_dt(raw: str | None) -> datetime | None: @@ -223,13 +221,13 @@ async def _todays_qualified_setups( now: datetime, expected_run_id: str | None = None, ) -> list[TradeSetup]: - """Long-only qualified setups from the scan that just ran, best rank first. + """Long-only qualified setups from the scan we may act on, best rank first. Order matters here, and matches the review's requirement: - 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. + 1. Take only rows the matched scan produced (``scan_run_id == run id``). A + previous run, or a manual scan overlapping in time, carries a different + id and is excluded by identity — not by a time window it could write into. 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 @@ -239,14 +237,12 @@ async def _todays_qualified_setups( the reverse. 4. Qualify, then rank by ``strategy_rank`` (unranked sort last). """ - run_start = await _last_scan_start( - db, now=now, expected_run_id=expected_run_id - ) - if run_start is None: + run_id = await _scan_run_to_trade(db, now=now, expected_run_id=expected_run_id) + if run_id is None: return [] result = await db.execute( - select(TradeSetup).where(TradeSetup.detected_at >= run_start) + select(TradeSetup).where(TradeSetup.scan_run_id == run_id) ) rows = [s for s in result.scalars() if (s.direction or "long") == "long"] @@ -285,7 +281,7 @@ async def open_shadow_positions( ``expected_run_id`` binds this run to the scan that stamped that exact id (the pipeline's own scan), so a scan that failed in this pipeline — or a concurrent manual scan that finished last — cannot substitute for it. See - ``_last_scan_start``. + ``_scan_run_to_trade``. """ summary = { "opened": 0, diff --git a/tests/unit/test_shadow_book_service.py b/tests/unit/test_shadow_book_service.py index 41edcb5..4138b08 100644 --- a/tests/unit/test_shadow_book_service.py +++ b/tests/unit/test_shadow_book_service.py @@ -57,6 +57,7 @@ def _setup( entry=100.0, stop=95.0, direction="long", + scan_run_id: str = "scan-run", ): reward = abs(entry - stop) * 3 target = entry + reward if direction == "long" else entry - reward @@ -73,6 +74,7 @@ def _setup( strategy_rank=rank, momentum_percentile=90.0, recommended_action="buy", + scan_run_id=scan_run_id, targets_json=json.dumps( [{"price": target, "probability": 45.0, "is_primary": True, "rr": 3.0}] ), @@ -82,17 +84,18 @@ def _setup( async def _mark_scan( session, *, - started: datetime, + started: datetime | None = None, completed: datetime | None = None, run_id: str = "scan-run", ): - """Record a successful scan run so the shadow book has something to act on.""" + """Record a successful scan run so the shadow book has something to act on. + + ``started`` is accepted for readability at call sites but only ``completed`` + (freshness) and ``run_id`` (identity) are persisted. + """ 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() - ) + completed = completed or started or datetime.now(timezone.utc) await shadow_book_service.settings_store.upsert_setting( session, rr.KEY_LAST_SCAN_COMPLETED, completed.isoformat() ) @@ -242,20 +245,20 @@ class TestScanFreshness: 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.""" + async def test_setups_from_another_run_are_excluded_by_identity(self, session): + """A row from a different scan run must not be traded even if its + detected_at falls in the same window — selection is by scan_run_id.""" 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 + _setup(ids["AAA"], rank=0.9, detected=now, scan_run_id="scan-run"), + # Same time window, different run — e.g. an overlapping manual scan. + _setup(ids["BBB"], rank=0.8, detected=now, scan_run_id="other-run"), ] ) await session.commit() - await _mark_scan(session, started=scan_start, completed=now) + await _mark_scan(session, completed=now, run_id="scan-run") summary = await shadow_book_service.open_shadow_positions( session, activation_config=_CONFIG @@ -269,21 +272,22 @@ class TestScanFreshness: 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). + # Both belong to the same scan run. 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", + scan_run_id="scan-run", 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) + await _mark_scan(session, completed=now, run_id="scan-run") summary = await shadow_book_service.open_shadow_positions( session, activation_config=_CONFIG @@ -298,10 +302,9 @@ class TestPipelineScanBinding: """The scan that stamped this pipeline's run id is the one to act on.""" ids = await _seed(session, ["AAA"]) now = datetime.now(timezone.utc) - session.add(_setup(ids["AAA"], rank=0.9, detected=now)) + session.add(_setup(ids["AAA"], rank=0.9, detected=now, scan_run_id="pipeline-A")) await session.commit() - await _mark_scan(session, started=now - timedelta(minutes=1), - completed=now, run_id="pipeline-A") + await _mark_scan(session, completed=now, run_id="pipeline-A") summary = await shadow_book_service.open_shadow_positions( session, activation_config=_CONFIG, expected_run_id="pipeline-A" @@ -309,19 +312,41 @@ class TestPipelineScanBinding: assert summary["opened"] == 1 + @pytest.mark.asyncio + async def test_concurrent_manual_row_excluded_even_when_pipeline_matches(self, session): + """The reported P2: the pipeline's scan matches (it wrote the marker + last), but an overlapping manual scan inserted a row in the same time + window. Identity selection must exclude that manual row — a detected_at + window would have swept it in.""" + ids = await _seed(session, ["AAA", "BBB"]) + now = datetime.now(timezone.utc) + session.add_all( + [ + _setup(ids["AAA"], rank=0.9, detected=now, scan_run_id="pipeline-A"), + # Overlapping manual scan, same window, higher rank — must NOT win. + _setup(ids["BBB"], rank=0.99, detected=now, scan_run_id="manual-X"), + ] + ) + await session.commit() + await _mark_scan(session, completed=now, run_id="pipeline-A") + + summary = await shadow_book_service.open_shadow_positions( + session, activation_config=_CONFIG, expected_run_id="pipeline-A" + ) + + assert summary["symbols"] == [ids["AAA"]] + @pytest.mark.asyncio async def test_concurrent_manual_scan_finishing_last_is_refused(self, session): - """The reported race: a manual rr_scanner overlaps the near-close - pipeline and writes the markers last. Its completion timestamp is later - than the pipeline start, but its run id is not the pipeline's — so the - shadow book must refuse, even though a timestamp check would accept.""" + """A manual rr_scanner overlaps the pipeline and writes the markers last. + Its completion timestamp is later than the pipeline start, but its run id + is not the pipeline's — refuse, though a timestamp check would accept.""" ids = await _seed(session, ["AAA"]) now = datetime.now(timezone.utc) - session.add(_setup(ids["AAA"], rank=0.9, detected=now)) + session.add(_setup(ids["AAA"], rank=0.9, detected=now, scan_run_id="manual-999")) await session.commit() # Pipeline expects "pipeline-A"; the manual scan's id won the last write. - await _mark_scan(session, started=now - timedelta(minutes=1), - completed=now, run_id="manual-999") + await _mark_scan(session, completed=now, run_id="manual-999") summary = await shadow_book_service.open_shadow_positions( session, activation_config=_CONFIG, expected_run_id="pipeline-A" @@ -334,10 +359,9 @@ class TestPipelineScanBinding: """If the pipeline's own scan failed, the stored id is a prior run's.""" ids = await _seed(session, ["AAA"]) now = datetime.now(timezone.utc) - session.add(_setup(ids["AAA"], rank=0.9, detected=now)) + session.add(_setup(ids["AAA"], rank=0.9, detected=now, scan_run_id="yesterday")) await session.commit() - await _mark_scan(session, started=now - timedelta(minutes=1), - completed=now, run_id="yesterday") + await _mark_scan(session, completed=now, run_id="yesterday") summary = await shadow_book_service.open_shadow_positions( session, activation_config=_CONFIG, expected_run_id="pipeline-today"