fix: grandfather pre-cutover stop episodes

alembic/versions/022_add_paper_trade_reentry_gate_reset.py

@@ -30,6 +30,22 @@ def upgrade() -> None:
             nullable=True,
         ),
     )
+    # The policy starts at this deployment. Historical NULL values mean the
+    # scanner never recorded reset observations, not that those old episodes
+    # are still active. Mark both transitions complete so only stops created
+    # after the migration can open a re-entry lock.
+    op.execute(
+        sa.text(
+            """
+            UPDATE paper_trades
+            SET reentry_gate_failed_at = closed_at,
+                reentry_gate_requalified_at = closed_at
+            WHERE status = 'closed'
+              AND close_reason = 'stop'
+              AND closed_at IS NOT NULL
+            """
+        )
+    )
 
 
 def downgrade() -> None:

app/services/trade_policy.py

@@ -5,7 +5,7 @@ from __future__ import annotations
 from collections.abc import Iterable
 from datetime import datetime, timezone
 
-from sqlalchemy import select
+from sqlalchemy import func, select
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.models.paper_trade import PaperTrade
@@ -16,27 +16,35 @@ async def _latest_initial_stop_trades(
     *,
     closed_before: datetime | None = None,
 ) -> dict[int, PaperTrade]:
-    """Return the most recent initial-stop trade for each ticker."""
-    stmt = (
-        select(PaperTrade)
+    """Return a ticker's latest closed trade only when it was an initial stop."""
+    ranked_stmt = (
+        select(
+            PaperTrade.id.label("trade_id"),
+            func.row_number()
+            .over(
+                partition_by=PaperTrade.ticker_id,
+                order_by=(PaperTrade.closed_at.desc(), PaperTrade.id.desc()),
+            )
+            .label("recency"),
+        )
         .where(
             PaperTrade.status == "closed",
-            PaperTrade.close_reason == "stop",
             PaperTrade.closed_at.is_not(None),
         )
-        .order_by(
-            PaperTrade.ticker_id.asc(),
-            PaperTrade.closed_at.desc(),
-            PaperTrade.id.desc(),
-        )
     )
     if closed_before is not None:
-        stmt = stmt.where(PaperTrade.closed_at <= closed_before)
+        ranked_stmt = ranked_stmt.where(PaperTrade.closed_at <= closed_before)
+    ranked = ranked_stmt.subquery()
+    stmt = (
+        select(PaperTrade)
+        .join(ranked, ranked.c.trade_id == PaperTrade.id)
+        .where(
+            ranked.c.recency == 1,
+            PaperTrade.close_reason == "stop",
+        )
+    )
     result = await db.execute(stmt)
-    latest: dict[int, PaperTrade] = {}
-    for trade in result.scalars():
-        latest.setdefault(trade.ticker_id, trade)
-    return latest
+    return {trade.ticker_id: trade for trade in result.scalars()}
 
 
 async def get_reentry_gate_locks(db: AsyncSession) -> dict[int, datetime]:

docs/research/post-stop-reentry.md

@@ -18,6 +18,14 @@ Scanner errors do not count as a gate failure. The two transitions are persisted
 on the latest initial-stop `PaperTrade`, so neither a service call nor a restart
 can bypass the rule.
 
+Migration 022 applies the policy prospectively. Existing initial-stop rows are
+grandfathered by marking both reset timestamps complete at their historical
+`closed_at`; otherwise their new NULL columns would be mistaken for active
+locks despite no scanner observations having existed. At runtime, only the
+actual latest closed trade per ticker can start a lock, and only when that exit
+was an initial stop. A newer trailing, time, target, or manual exit therefore
+cannot revive an older stop episode.
+
 This replaces the previously proposed fixed five-session lockdown. The normal
 reset counts an unqualified stop-day close when that close is observed after the
 stop. The stricter experiment, which required a failed close on a later session,

tests/unit/test_migration_022.py

@@ -0,0 +1,100 @@
+from __future__ import annotations
+
+import importlib.util
+from pathlib import Path
+
+import sqlalchemy as sa
+from alembic.migration import MigrationContext
+from alembic.operations import Operations
+
+
+def _load_migration_module():
+    path = (
+        Path(__file__).resolve().parents[2]
+        / "alembic"
+        / "versions"
+        / "022_add_paper_trade_reentry_gate_reset.py"
+    )
+    spec = importlib.util.spec_from_file_location("migration_022", path)
+    assert spec is not None and spec.loader is not None
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module
+
+
+def test_upgrade_grandfathers_only_preexisting_initial_stops():
+    migration = _load_migration_module()
+    engine = sa.create_engine("sqlite://")
+
+    with engine.begin() as connection:
+        connection.execute(
+            sa.text(
+                """
+                CREATE TABLE paper_trades (
+                    id INTEGER PRIMARY KEY,
+                    status VARCHAR NOT NULL,
+                    close_reason VARCHAR,
+                    closed_at DATETIME
+                )
+                """
+            )
+        )
+        connection.execute(
+            sa.text(
+                """
+                INSERT INTO paper_trades (id, status, close_reason, closed_at)
+                VALUES
+                    (1, 'closed', 'stop', '2026-07-01 12:00:00'),
+                    (2, 'closed', 'manual', '2026-07-02 12:00:00'),
+                    (3, 'open', NULL, NULL)
+                """
+            )
+        )
+
+        context = MigrationContext.configure(connection)
+        migration.op = Operations(context)
+        migration.upgrade()
+
+        historical = connection.execute(
+            sa.text(
+                """
+                SELECT closed_at, reentry_gate_failed_at,
+                       reentry_gate_requalified_at
+                FROM paper_trades
+                WHERE id = 1
+                """
+            )
+        ).one()
+        assert historical[1] == historical[0]
+        assert historical[2] == historical[0]
+
+        unaffected = connection.execute(
+            sa.text(
+                """
+                SELECT reentry_gate_failed_at, reentry_gate_requalified_at
+                FROM paper_trades
+                WHERE id IN (2, 3)
+                ORDER BY id
+                """
+            )
+        ).all()
+        assert unaffected == [(None, None), (None, None)]
+
+        connection.execute(
+            sa.text(
+                """
+                INSERT INTO paper_trades (id, status, close_reason, closed_at)
+                VALUES (4, 'closed', 'stop', '2026-07-18 12:00:00')
+                """
+            )
+        )
+        new_stop = connection.execute(
+            sa.text(
+                """
+                SELECT reentry_gate_failed_at, reentry_gate_requalified_at
+                FROM paper_trades
+                WHERE id = 4
+                """
+            )
+        ).one()
+        assert new_stop == (None, None)

tests/unit/test_trade_policy.py

@@ -24,6 +24,7 @@ def _stopped_trade(
     ticker_id: int,
     *,
     closed_at: datetime,
+    close_reason: str = "stop",
     gate_failed_at: datetime | None = None,
     gate_requalified_at: datetime | None = None,
 ) -> PaperTrade:
@@ -39,7 +40,7 @@ def _stopped_trade(
         opened_at=closed_at - timedelta(days=5),
         close_price=95.0,
         closed_at=closed_at,
-        close_reason="stop",
+        close_reason=close_reason,
         reentry_gate_failed_at=gate_failed_at,
         reentry_gate_requalified_at=gate_requalified_at,
     )
@@ -126,3 +127,33 @@ async def test_latest_stop_starts_a_new_gate_reset_episode(session):
     await session.commit()
 
     assert ticker.id in await get_reentry_gate_locks(session)
+
+
+async def test_newer_non_stop_exit_supersedes_historical_stop(session):
+    session.add(User(id=1, username="u", password_hash="x", role="user", has_access=True))
+    ticker = Ticker(symbol="LATEREXIT")
+    session.add(ticker)
+    await session.flush()
+
+    stopped_at = datetime.now(timezone.utc) - timedelta(days=20)
+    old_stop = _stopped_trade(ticker.id, closed_at=stopped_at)
+    later_manual_exit = _stopped_trade(
+        ticker.id,
+        closed_at=stopped_at + timedelta(days=10),
+        close_reason="manual",
+    )
+    session.add_all([old_stop, later_manual_exit])
+    await session.commit()
+
+    assert ticker.id not in await get_reentry_gate_locks(session)
+
+    observed_at = datetime.now(timezone.utc)
+    updated = await observe_reentry_gate_transitions(
+        session,
+        evaluated_ticker_ids={ticker.id},
+        qualified_ticker_ids=set(),
+        observed_at=observed_at,
+    )
+
+    assert updated == set()
+    assert old_stop.reentry_gate_failed_at is None