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 <noreply@anthropic.com>

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")

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")
 

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
+# Markers of the most recent *successful* scan, written together only when
-# scan_all_tickers completes: STARTED bounds which setups belong to the run
+# scan_all_tickers completes. COMPLETED gives its freshness; RUN_ID identifies
-# (detected_at >= STARTED), COMPLETED gives its freshness, and RUN_ID identifies
+# the run — the same id stamped on every setup row it produced. The shadow book
-# the pipeline invocation that produced it (or a fresh id for a manual scan).
+# matches RUN_ID exactly and then selects setups by that id, so neither a
-# The shadow book matches RUN_ID exactly rather than trusting timestamps, so a
+# concurrent manual scan nor a stale prior run can be mistaken for it.
-# concurrent manual scan cannot be mistaken for the pipeline's own.
-KEY_LAST_SCAN_STARTED = "last_scan_run_started_at"
 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
+    # Publish the run markers only now that the scan has completed: COMPLETED for
-    # the run id of the pipeline this scan ran inside (or a fresh id when run
+    # freshness and RUN_ID (the same id stamped on this run's setup rows) for
-    # standalone — a manual scan then can never match a pipeline's expected id).
+    # identity, in one commit. A hard failure above leaves the previous,
-    # All three markers are written in one commit so a reader sees a consistent
+    # now-superseded, markers in place — so the shadow book will not match.
-    # (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()
-    )
     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

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:
+) -> str | None:
-    """Start of the scan we may act on, or None if there is 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
+      window on the last scan's own id. There is no pipeline scan to bind to, so
-      the operator's explicit choice.
+      acting on a recent scan is the operator's explicit choice.
 
-    The three markers are written in one commit, so the returned STARTED belongs
+    Setups are then selected by ``scan_run_id`` equal to the returned id, so a
-    to the same run as the matched RUN_ID and correctly bounds its setups.
+    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 run_id if run_id == expected_run_id else None
-            return None
+    if now - completed > MAX_SCAN_AGE:
-    elif 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
+    1. Take only rows the matched scan produced (``scan_run_id == run id``). A
-       previous run's setups sit ~24h earlier and are excluded, so a stale row
+       previous run, or a manual scan overlapping in time, carries a different
-       can never be traded even if it once qualified.
+       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(
+    run_id = await _scan_run_to_trade(db, now=now, expected_run_id=expected_run_id)
-        db, now=now, expected_run_id=expected_run_id
+    if run_id is None:
-    )
-    if run_start 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,

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
+    completed = completed or started or datetime.now(timezone.utc)
-    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()
     )
@@ -242,20 +245,20 @@ class TestScanFreshness:
         assert summary["opened"] == 0
 
     @pytest.mark.asyncio
-    async def test_setups_before_this_run_are_excluded(self, session):
+    async def test_setups_from_another_run_are_excluded_by_identity(self, session):
-        """A qualified row from a previous run (before the current scan start)
+        """A row from a different scan run must not be traded even if its
-        must not be traded even though the current scan completed."""
+        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["AAA"], rank=0.9, detected=now, scan_run_id="scan-run"),
-                _setup(ids["BBB"], rank=0.8, detected=now - timedelta(hours=20)),  # prior 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),
+        await _mark_scan(session, completed=now, run_id="pipeline-A")
-                         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
+        """A manual rr_scanner overlaps the pipeline and writes the markers last.
-        pipeline and writes the markers last. Its completion timestamp is later
+        Its completion timestamp is later than the pipeline start, but its run id
-        than the pipeline start, but its run id is not the pipeline's — so the
+        is not the pipeline's — refuse, though a timestamp check would accept."""
-        shadow book must refuse, even 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),
+        await _mark_scan(session, completed=now, run_id="manual-999")
-                         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),
+        await _mark_scan(session, completed=now, run_id="yesterday")
-                         completed=now, run_id="yesterday")
 
         summary = await shadow_book_service.open_shadow_positions(
             session, activation_config=_CONFIG, expected_run_id="pipeline-today"