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

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