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/024_paper_trade_book.py

@@ -0,0 +1,59 @@
+"""paper trade book tag (manual vs shadow) + weekday cron repair
+
+Revision ID: 024
+Revises: 023
+Create Date: 2026-07-20 00:00:00.000000
+
+Two things ship together because both are corrections to 023's stored state.
+
+1. ``paper_trades.book`` separates the discretionary book from the automatic
+   shadow book. Everything that exists today was opened by hand, so the
+   backfill value is "manual".
+
+2. 023 wrote weekday crons with a numeric day-of-week. APScheduler's
+   from_crontab() feeds field 5 to its own day_of_week where 0=Monday, so
+   "1-5" resolved to Tue-Sat: every Monday was skipped and the scanner ran on
+   Saturdays against stale data. Rewrite only the rows that still hold the
+   broken numeric form, so a hand-corrected setting is never clobbered.
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+revision: str = "024"
+down_revision: Union[str, None] = "023"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+# key -> (broken numeric form written by 023, corrected named form)
+_CRON_REPAIR: dict[str, tuple[str, str]] = {
+    "schedule_near_close_pipeline_cron": ("30 15 * * 1-5", "30 15 * * mon-fri"),
+    "schedule_after_close_pipeline_cron": ("45 16 * * 1-5", "45 16 * * mon-fri"),
+    "schedule_intraday_pipeline_cron": ("0 10-15 * * 1-5", "0 10-15 * * mon-fri"),
+    "schedule_fundamentals_cron": ("0 1 * * 1", "0 1 * * mon"),
+}
+
+
+def upgrade() -> None:
+    # server_default backfills existing rows, so no separate UPDATE is needed.
+    op.add_column(
+        "paper_trades",
+        sa.Column("book", sa.String(length=10), nullable=False, server_default="manual"),
+    )
+
+    # Literals are inlined rather than bound because bound parameters render as
+    # NULL under `alembic upgrade --sql`, which would silently produce a script
+    # that matches nothing. Every value here is a constant defined above.
+    for key, (broken, fixed) in _CRON_REPAIR.items():
+        op.execute(
+            f"UPDATE system_settings SET value = '{fixed}' "  # noqa: S608
+            f"WHERE key = '{key}' AND value = '{broken}'"
+        )
+
+
+def downgrade() -> None:
+    op.drop_column("paper_trades", "book")
+    # Crons are deliberately left corrected — restoring the numeric form would
+    # reintroduce the skipped-Monday bug.

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/paper_trade.py

@@ -49,3 +49,12 @@ class PaperTrade(Base):
     # Execution era for forward vs backtest comparison:
     # null/legacy = pre-cutover morning-scan, "near_close" = post near-close cutover.
     fill_mode: Mapped[str | None] = mapped_column(String(20), nullable=True)
+    # Which book this trade belongs to:
+    #   "manual" — discretionary, opened by the user from a qualified setup
+    #   "shadow" — opened automatically by the validated strategy (top-ranked
+    #              qualified up to capacity, 1% risk). The shadow book is the
+    #              faithful live twin of the backtest; the two books share the
+    #              same exit policy so the only difference is *selection*.
+    # Gate-reset re-entry state is tracked per book — the books diverge as soon
+    # as their entries differ, and each must see its own trade history.
+    book: Mapped[str] = mapped_column(String(10), nullable=False, default="manual")

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/routers/admin.py

@@ -16,8 +16,10 @@ from app.schemas.admin import (
     JobTriggerRequest,
     JobToggle,
     RecommendationConfigUpdate,
+    PerformanceConfigUpdate,
     ScheduleConfigUpdate,
     SentimentConfigUpdate,
+    ShadowBookConfigUpdate,
     SentimentTestRequest,
     PasswordReset,
     RegistrationToggle,
@@ -201,6 +203,50 @@ async def update_schedule_settings(
     return APIEnvelope(status="success", data=updated)
 
 
+@router.get("/admin/settings/performance", response_model=APIEnvelope)
+async def get_performance_settings(
+    _admin: User = Depends(require_admin),
+    db: AsyncSession = Depends(get_db),
+):
+    return APIEnvelope(
+        status="success", data=await admin_service.get_performance_config(db)
+    )
+
+
+@router.put("/admin/settings/performance", response_model=APIEnvelope)
+async def update_performance_settings(
+    body: PerformanceConfigUpdate,
+    _admin: User = Depends(require_admin),
+    db: AsyncSession = Depends(get_db),
+):
+    updated = await admin_service.update_performance_config(
+        db, body.model_dump(exclude_unset=True)
+    )
+    return APIEnvelope(status="success", data=updated)
+
+
+@router.get("/admin/settings/shadow-book", response_model=APIEnvelope)
+async def get_shadow_book_settings(
+    _admin: User = Depends(require_admin),
+    db: AsyncSession = Depends(get_db),
+):
+    return APIEnvelope(
+        status="success", data=await admin_service.get_shadow_book_config(db)
+    )
+
+
+@router.put("/admin/settings/shadow-book", response_model=APIEnvelope)
+async def update_shadow_book_settings(
+    body: ShadowBookConfigUpdate,
+    _admin: User = Depends(require_admin),
+    db: AsyncSession = Depends(get_db),
+):
+    updated = await admin_service.update_shadow_book_config(
+        db, body.model_dump(exclude_unset=True, exclude_none=True)
+    )
+    return APIEnvelope(status="success", data=updated)
+
+
 @router.get("/admin/settings/sentiment", response_model=APIEnvelope)
 async def get_sentiment_settings(
     _admin: User = Depends(require_admin),

app/routers/paper_trades.py

@@ -65,6 +65,18 @@ async def paper_trade_equity_curve(
     )
 
 
+@router.get("/paper-trades/performance", response_model=APIEnvelope)
+async def paper_trade_performance(
+    user: User = Depends(require_access),
+    db: AsyncSession = Depends(get_db),
+) -> APIEnvelope:
+    """Shadow book vs discretionary book vs SPY since the configured start date."""
+    return APIEnvelope(
+        status="success",
+        data=await paper_trade_service.performance_summary(db, user.id),
+    )
+
+
 @router.put("/paper-trades/exit-policy", response_model=APIEnvelope)
 async def write_exit_policy(
     body: ExitPolicyUpdate,

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

app/scheduler.py

@@ -33,7 +33,14 @@ from app.exceptions import ProviderError
 from app.providers.alpaca import AlpacaOHLCVProvider
 from app.providers.fundamentals_chain import build_fundamental_provider_chain
 from app.providers.protocol import SentimentData
-from app.services import fundamental_service, ingestion_service, sentiment_service, settings_store
+from app.services import (
+    fundamental_service,
+    ingestion_service,
+    pipeline_run,
+    sentiment_service,
+    settings_store,
+    shadow_book_service,
+)
 from app.services.alert_service import dispatch_alerts
 from app.services.backtest_service import (
     BACKTEST_TARGET_MODELS,
@@ -610,6 +617,64 @@ async def backfill_ohlcv() -> None:
     await collect_ohlcv(full_backfill=True, job_name="data_backfill")
 
 
+async def run_shadow_book() -> None:
+    """Open the strategy's own positions from the latest qualifying scan.
+
+    The shadow book is the faithful live twin of the backtest: top-ranked
+    qualified setups, up to capacity, 1% risk, no human input. It runs straight
+    after the near-close scan so its entries are marked at the same near-close
+    prices the discretionary book sees, leaving *selection* as the only
+    difference between the two books.
+
+    When run as a pipeline step it acts only on the scan that stamped *this
+    pipeline's* run id (``expected_run_id``): if the pipeline's own scan was
+    disabled or failed, the stored run id is some other scan's — including a
+    manual scan that overlapped and finished last — and shadow refuses.
+    Triggered directly from Admin (no pipeline context) it falls back to the
+    scan-freshness window — an explicit operator action.
+
+    Opt-in (``shadow_book_enabled``) because it writes live trades.
+    """
+    job_name = "shadow_book"
+    expected_run_id = pipeline_run.current()
+    _log_event(logging.INFO, "job_start", job=job_name)
+    _runtime_start(job_name, total=1)
+
+    try:
+        async with async_session_factory() as db:
+            if not await _is_job_enabled(db, job_name):
+                _log_event(logging.INFO, "job_skipped", job=job_name, reason="disabled")
+                _runtime_finish(job_name, "skipped", processed=0, total=1, message="Disabled")
+                return
+            if not await shadow_book_service.is_enabled(db):
+                _log_event(logging.INFO, "job_skipped", job=job_name, reason="not enabled in settings")
+                _runtime_finish(job_name, "skipped", processed=0, total=1, message="Not enabled")
+                return
+
+            from app.services.admin_service import get_activation_config
+
+            activation_config = await get_activation_config(db)
+            summary = await shadow_book_service.open_shadow_positions(
+                db,
+                activation_config=activation_config,
+                expected_run_id=expected_run_id,
+            )
+            symbols = await shadow_book_service.symbols_for(db, summary["symbols"])
+
+        _runtime_progress(job_name, processed=1, total=1)
+        _runtime_finish(
+            job_name, "completed", processed=1, total=1,
+            message=(
+                f"Opened {summary['opened']} ({', '.join(symbols) if symbols else 'none'}); "
+                f"held {summary['skipped_held']}, gate-locked {summary['skipped_locked']}"
+            ),
+        )
+        _log_event(logging.INFO, "job_complete", job=job_name, opened=summary["opened"], symbols=symbols)
+    except Exception as exc:
+        _runtime_finish(job_name, "error", processed=0, total=1, message=str(exc))
+        _log_event(logging.ERROR, "job_error", job=job_name, error_type=type(exc).__name__, message=str(exc))
+
+
 async def collect_ohlcv_final() -> None:
     """After-close OHLCV refresh that replaces the day's partial bar.
 
@@ -1238,6 +1303,9 @@ _NEAR_CLOSE_PIPELINE_STEPS = [
     # back to the previous close and execution degrades to the stale_close floor.
     ("data_collector", "collect_ohlcv"),
     ("rr_scanner", "scan_rr"),
+    # Straight after the scan so shadow entries mark at the same near-close
+    # prices the discretionary book is looking at.
+    ("shadow_book", "run_shadow_book"),
     ("alerts", "dispatch_alerts_job"),
 ]
 
@@ -1267,6 +1335,11 @@ async def _run_pipeline(job_name: str, steps: list[tuple[str, str]]) -> None:
 
     Each step respects its own enable flag and manages its own runtime status; a
     failing step is logged and the pipeline continues with the next one.
+
+    A unique run id is bound for the invocation and visible to every step via the
+    shared task context: the scan step stamps it into its completion markers and
+    the shadow step requires an exact match, so only a scan that ran inside this
+    pipeline can drive the shadow book.
     """
     _log_event(logging.INFO, "job_start", job=job_name)
     async with async_session_factory() as db:
@@ -1280,6 +1353,7 @@ async def _run_pipeline(job_name: str, steps: list[tuple[str, str]]) -> None:
 
     funcs = globals()
     done = 0
+    token = pipeline_run.bind(pipeline_run.new_run_id())
     try:
         for step_name, func_name in steps:
             _runtime_progress(job_name, processed=done, total=total, current_ticker=step_name)
@@ -1293,6 +1367,8 @@ async def _run_pipeline(job_name: str, steps: list[tuple[str, str]]) -> None:
     except Exception as exc:
         _runtime_finish(job_name, "error", processed=done, total=total, message=str(exc))
         _log_event(logging.ERROR, "job_error", job=job_name, error_type=type(exc).__name__, message=str(exc))
+    finally:
+        pipeline_run.release(token)
 
 
 async def run_daily_pipeline() -> None:
@@ -1456,6 +1532,7 @@ def configure_scheduler(schedule_config: dict[str, str] | None = None) -> None:
         (collect_benchmark, "benchmark_collector", "Benchmark Collector"),
         (collect_sentiment, "sentiment_collector", "Sentiment Collector"),
         (scan_rr, "rr_scanner", "R:R Scanner"),
+        (run_shadow_book, "shadow_book", "Shadow Book (auto-traded strategy)"),
         (evaluate_outcomes, "outcome_evaluator", "Outcome Evaluator"),
         (compute_market_regime, "market_regime", "Market Regime"),
         (compute_regime_monitor, "regime_monitor", "Regime Monitor"),

app/schemas/admin.py

@@ -84,6 +84,25 @@ class ScheduleConfigUpdate(BaseModel):
     schedule_fundamentals_cron: str | None = Field(default=None, max_length=120)
 
 
+class PerformanceConfigUpdate(BaseModel):
+    """Window for the Performance comparison.
+
+    ``start_date`` is an ISO date, or empty string to show all history. The
+    strategy has been revised repeatedly; pinning a start keeps the shadow-vs-
+    manual comparison inside one configuration instead of averaging across
+    rules that no longer exist.
+    """
+    start_date: str | None = Field(default=None, max_length=10)
+
+
+class ShadowBookConfigUpdate(BaseModel):
+    """Auto-traded shadow book: the validated strategy with no human input."""
+    enabled: bool | None = None
+    capacity: int | None = Field(default=None, ge=1, le=100)
+    risk_pct: float | None = Field(default=None, gt=0, le=10)
+    start_equity: float | None = Field(default=None, ge=1000)
+
+
 class SentimentConfigUpdate(BaseModel):
     """Runtime sentiment LLM config. api_key is write-only; omit/empty to keep
     the stored key."""

app/services/admin_service.py

@@ -204,6 +204,61 @@ async def update_activation_config(
     return await get_activation_config(db)
 
 
+# ---------------------------------------------------------------------------
+# Performance window + shadow book
+# ---------------------------------------------------------------------------
+
+async def get_performance_config(db: AsyncSession) -> dict:
+    """Start date for the Performance comparison ('' = all history)."""
+    from app.services.paper_trade_service import KEY_PERFORMANCE_START
+
+    return {"start_date": await settings_store.get_value(db, KEY_PERFORMANCE_START, "") or ""}
+
+
+async def update_performance_config(db: AsyncSession, updates: dict) -> dict:
+    """Set (or clear) the performance start date. Empty string means all history."""
+    from datetime import date as _date
+
+    from app.services.paper_trade_service import KEY_PERFORMANCE_START
+
+    if "start_date" in updates:
+        raw = (updates.get("start_date") or "").strip()
+        if raw:
+            try:
+                _date.fromisoformat(raw)
+            except ValueError as exc:
+                raise ValidationError("start_date must be an ISO date (YYYY-MM-DD)") from exc
+        await update_setting(db, KEY_PERFORMANCE_START, raw)
+    return await get_performance_config(db)
+
+
+async def get_shadow_book_config(db: AsyncSession) -> dict:
+    """Shadow book switch + sizing, with the validated defaults filled in."""
+    from app.services import shadow_book_service
+
+    config = await shadow_book_service.get_config(db)
+    config["enabled"] = await shadow_book_service.is_enabled(db)
+    return config
+
+
+async def update_shadow_book_config(db: AsyncSession, updates: dict) -> dict:
+    """Update the shadow book. Enabling it starts automatic live entries."""
+    from app.services import shadow_book_service
+
+    if "enabled" in updates:
+        await update_setting(
+            db, shadow_book_service.KEY_ENABLED, "true" if updates["enabled"] else "false"
+        )
+    for key, storage_key in (
+        ("capacity", shadow_book_service.KEY_CAPACITY),
+        ("risk_pct", shadow_book_service.KEY_RISK_PCT),
+        ("start_equity", shadow_book_service.KEY_START_EQUITY),
+    ):
+        if key in updates:
+            await update_setting(db, storage_key, str(updates[key]))
+    return await get_shadow_book_config(db)
+
+
 # ---------------------------------------------------------------------------
 # Pipeline schedule (cron)
 # ---------------------------------------------------------------------------
@@ -569,6 +624,7 @@ VALID_JOB_NAMES = {
     "near_close_pipeline",
     "after_close_pipeline",
     "intraday_pipeline",
+    "shadow_book",
 }
 
 JOB_LABELS = {
@@ -589,6 +645,7 @@ JOB_LABELS = {
     "near_close_pipeline": "Near-Close Pipeline (scan+alert)",
     "after_close_pipeline": "After-Close Pipeline (outcome)",
     "intraday_pipeline": "Intraday Pipeline",
+    "shadow_book": "Shadow Book (auto-traded strategy)",
 }
 
 # Jobs driven by a pipeline (in order) rather than their own auto timer.
@@ -601,6 +658,7 @@ PIPELINE_MEMBERS = {
     "alerts",
     "market_regime",
     "regime_monitor",
+    "shadow_book",
 }
 
 

app/services/alert_service.py

@@ -29,6 +29,7 @@ from app.config import settings
 from app.models.alert import AlertLog
 from app.models.ohlcv import OHLCVRecord
 from app.models.paper_trade import PaperTrade
+from app.services.trade_policy import MANUAL_BOOK
 from app.models.score import CompositeScore
 from app.models.sr_level import SRLevel
 from app.models.ticker import Ticker
@@ -632,6 +633,10 @@ async def _collect_closed_trades(db: AsyncSession) -> list[ClosedTradeItem]:
             PaperTrade.closed_at.is_not(None),
             PaperTrade.closed_at > cutoff,
             PaperTrade.close_reason.in_(("trailing", "stop", "target", "time")),
+            # Your own positions only — shadow trades are a research record, not
+            # something you hold, and mixing them in unlabelled reads as if you
+            # were stopped out of a name you never took.
+            PaperTrade.book == MANUAL_BOOK,
         )
         .order_by(PaperTrade.closed_at.desc())
     )
@@ -642,8 +647,14 @@ async def _collect_closed_trades(db: AsyncSession) -> list[ClosedTradeItem]:
 
 
 async def _paper_book_value(db: AsyncSession) -> float:
-    """Paper-trade equity: fixed capital plus realized/unrealized P&L."""
+    """Paper-trade equity: fixed capital plus realized/unrealized P&L.
-    result = await db.execute(select(PaperTrade))
+
+    Discretionary book only — the shadow book runs on its own notional equity
+    and folding it in would report a number matching neither book.
+    """
+    result = await db.execute(
+        select(PaperTrade).where(PaperTrade.book == MANUAL_BOOK)
+    )
     trades = list(result.scalars().all())
     latest: dict[int, float | None] = {}
     for trade in trades:

app/services/paper_trade_service.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import bisect
+import logging
 from datetime import date, datetime, timezone
 
 from sqlalchemy import and_, func, select
@@ -20,7 +21,9 @@ from app.services.outcome_service import (
     Bar,
     evaluate_setup_against_bars,
 )
-from app.services.trade_policy import get_reentry_gate_locks
+from app.services.trade_policy import MANUAL_BOOK, SHADOW_BOOK, get_reentry_gate_locks
+
+logger = logging.getLogger(__name__)
 
 # Exit policy for OPEN paper trades (auto-close). Production defaults to the
 # July 2026 promoted strategy: initial stop + 3x ATR trailing stop, with a max
@@ -399,7 +402,15 @@ async def list_trades(
     db: AsyncSession,
     user_id: int | None = None,
     status: str | None = None,
+    book: str | None = MANUAL_BOOK,
 ) -> list[dict]:
+    """Trades for the UI. Defaults to the discretionary book.
+
+    Shadow trades are attached to a user row for FK reasons only — they are not
+    that person's decisions. Listing them alongside manual trades would mix two
+    different books in one P&L and let the autonomous record be edited by hand.
+    Pass ``book=None`` to deliberately span both.
+    """
     stmt = (
         select(PaperTrade, Ticker.symbol)
         .join(Ticker, PaperTrade.ticker_id == Ticker.id)
@@ -408,6 +419,8 @@ async def list_trades(
         stmt = stmt.where(PaperTrade.user_id == user_id)
     if status is not None:
         stmt = stmt.where(PaperTrade.status == status)
+    if book is not None:
+        stmt = stmt.where(PaperTrade.book == book)
     stmt = stmt.order_by(PaperTrade.opened_at.desc())
 
     rows = (await db.execute(stmt)).all()
@@ -490,6 +503,13 @@ async def close_trade(
     trade = result.scalar_one_or_none()
     if trade is None:
         raise NotFoundError(f"Paper trade not found: {trade_id}")
+    if trade.book == SHADOW_BOOK:
+        # The shadow book's value is that no human touched it. A hand-closed
+        # position would make its record something other than what the strategy
+        # would have done; it exits only via the automatic exit policy.
+        raise ValidationError(
+            "Shadow book trades are closed by the exit policy, not by hand"
+        )
     if trade.status == "closed":
         raise ValidationError("Trade is already closed")
 
@@ -690,6 +710,66 @@ def build_equity_curve(
     return out
 
 
+KEY_PERFORMANCE_START = "performance_start_date"
+
+
+async def get_performance_start(db: AsyncSession) -> date | None:
+    """Date the performance view starts from, or None for 'all history'.
+
+    The strategy has been revised repeatedly, so early trades were taken under
+    rules that no longer exist. Pinning a start date keeps the comparison inside
+    one regime instead of averaging across configurations that were replaced.
+    """
+    raw = await settings_store.get_value(db, KEY_PERFORMANCE_START, "")
+    if not raw or not str(raw).strip():
+        return None
+    try:
+        return date.fromisoformat(str(raw).strip())
+    except ValueError:
+        logger.warning("invalid %s: %r", KEY_PERFORMANCE_START, raw)
+        return None
+
+
+def trade_r_multiple(trade, mark: float | None) -> float | None:
+    """Result in R — profit measured in units of the trade's own initial risk.
+
+    R is the only sizing-independent yardstick available here: the shadow book
+    sizes at a fixed 1% of equity while manual trades were sized by hand, so
+    currency P&L cannot compare them. Open trades are marked to ``mark``.
+    """
+    risk_per_share = abs(trade.entry_price - trade.stop_loss)
+    if risk_per_share <= 0:
+        return None
+    exit_price = trade.close_price if trade.status == "closed" else mark
+    if exit_price is None:
+        return None
+    per_share = (
+        exit_price - trade.entry_price
+        if trade.direction == "long"
+        else trade.entry_price - exit_price
+    )
+    return per_share / risk_per_share
+
+
+def book_stats(trades: list, marks: dict[int, float]) -> dict:
+    """Sizing-independent summary of one book: counts, win rate, R-multiples."""
+    rs = [
+        r
+        for r in (trade_r_multiple(t, marks.get(t.ticker_id)) for t in trades)
+        if r is not None
+    ]
+    closed = [t for t in trades if t.status == "closed"]
+    wins = [r for r in rs if r > 0]
+    return {
+        "trades": len(trades),
+        "closed": len(closed),
+        "open": len(trades) - len(closed),
+        "win_rate": round(100.0 * len(wins) / len(rs), 1) if rs else None,
+        "total_r": round(sum(rs), 2) if rs else 0.0,
+        "avg_r": round(sum(rs) / len(rs), 3) if rs else None,
+    }
+
+
 async def equity_curve(db: AsyncSession, user_id: int) -> list[dict]:
     """Equity-curve series for a user's paper book (empty without benchmark data)."""
     trades = (
@@ -714,3 +794,118 @@ async def equity_curve(db: AsyncSession, user_id: int) -> list[dict]:
     for tid, day, close in rows.all():
         ticker_closes.setdefault(tid, {})[day] = float(close)
     return build_equity_curve(list(trades), ticker_closes, benchmark_closes)
+
+
+def _cumulative_pnl(trades: list, ticker_closes: dict, days: list[date]) -> list[float]:
+    """Cumulative realized + mark-to-market P&L of one book on each day."""
+    sorted_dates = {tid: sorted(c) for tid, c in ticker_closes.items()}
+    out: list[float] = []
+    for d in days:
+        total = 0.0
+        for t in trades:
+            if t.opened_at.date() > d:
+                continue
+            closed_on = (
+                t.closed_at.date()
+                if (t.status == "closed" and t.closed_at is not None)
+                else None
+            )
+            if closed_on is not None and closed_on <= d and t.close_price is not None:
+                ref = float(t.close_price)
+            else:
+                ref = _value_on_or_before(
+                    sorted_dates.get(t.ticker_id) or [],
+                    ticker_closes.get(t.ticker_id) or {},
+                    d,
+                )
+                if ref is None:
+                    continue
+            per_share = (
+                ref - t.entry_price if t.direction == "long" else t.entry_price - ref
+            )
+            total += per_share * t.shares
+        out.append(round(total, 2))
+    return out
+
+
+async def performance_summary(db: AsyncSession, user_id: int | None = None) -> dict:
+    """Shadow book vs discretionary book vs SPY, from the configured start date.
+
+    Currency P&L is reported per book but is *not* the comparison — the books
+    size differently, so the honest read is the R-multiple stats. SPY is a plain
+    buy-and-hold reference over the same window rather than a per-trade
+    counterfactual, so one line serves both books.
+    """
+    start = await get_performance_start(db)
+    stmt = select(PaperTrade)
+    if start is not None:
+        stmt = stmt.where(func.date(PaperTrade.opened_at) >= start)
+    if user_id is not None:
+        # "Your picks" must be *yours*. The shadow book is a single autonomous
+        # book with no owner, so it is never scoped to a user.
+        stmt = stmt.where(
+            (PaperTrade.book == SHADOW_BOOK) | (PaperTrade.user_id == user_id)
+        )
+    trades = list((await db.execute(stmt)).scalars().all())
+
+    benchmark_closes = await benchmark_service.load_benchmark_closes(db)
+    empty = {
+        "start_date": start.isoformat() if start else None,
+        "series": [],
+        "stats": {},
+    }
+    if not trades or not benchmark_closes:
+        return empty
+
+    first = min(t.opened_at.date() for t in trades)
+    if start is not None:
+        first = max(first, start)
+    days = [d for d in sorted(benchmark_closes) if d >= first]
+    if not days:
+        return empty
+
+    ticker_ids = {t.ticker_id for t in trades}
+    rows = await db.execute(
+        select(OHLCVRecord.ticker_id, OHLCVRecord.date, OHLCVRecord.close).where(
+            OHLCVRecord.ticker_id.in_(ticker_ids), OHLCVRecord.date >= first
+        )
+    )
+    ticker_closes: dict[int, dict[date, float]] = {}
+    for tid, day, close in rows.all():
+        ticker_closes.setdefault(tid, {})[day] = float(close)
+
+    books = {
+        MANUAL_BOOK: [t for t in trades if (t.book or MANUAL_BOOK) == MANUAL_BOOK],
+        SHADOW_BOOK: [t for t in trades if t.book == SHADOW_BOOK],
+    }
+    pnl = {
+        name: _cumulative_pnl(book_trades, ticker_closes, days)
+        for name, book_trades in books.items()
+    }
+
+    bench_dates = sorted(benchmark_closes)
+    spy0 = _value_on_or_before(bench_dates, benchmark_closes, days[0])
+    spy_pct = [
+        round(100.0 * (benchmark_closes[d] / spy0 - 1.0), 2) if spy0 else 0.0
+        for d in days
+    ]
+
+    # Latest close per ticker, for marking open positions in the R stats.
+    marks = {
+        tid: closes[max(closes)] for tid, closes in ticker_closes.items() if closes
+    }
+    stats = {name: book_stats(bt, marks) for name, bt in books.items()}
+    for name in books:
+        stats[name]["pnl"] = pnl[name][-1] if pnl[name] else 0.0
+    stats["spy"] = {"pct": spy_pct[-1] if spy_pct else 0.0}
+
+    series = [
+        {
+            "date": d.isoformat(),
+            "manual_pnl": pnl[MANUAL_BOOK][i],
+            "shadow_pnl": pnl[SHADOW_BOOK][i],
+            "spy_pct": spy_pct[i],
+        }
+        for i, d in enumerate(days)
+    ]
+    return {"start_date": start.isoformat() if start else None, "series": series, "stats": stats}

app/services/pipeline_run.py

@@ -0,0 +1,47 @@
+"""Per-invocation identity for pipeline runs.
+
+A pipeline invocation stamps a unique run id into the task context. The scan it
+runs records that id alongside its completion markers, and the shadow book
+requires an *exact* match before acting on the scan's batch.
+
+This is what timestamp comparison cannot provide. A manually triggered scan and
+the scheduled near-close pipeline are separate APScheduler jobs, and
+``max_instances=1`` only serialises a job against itself — not two different
+jobs. So a manual scan can start just before the pipeline and finish just after
+it began, leaving a completion timestamp later than the pipeline's start even
+though its batch is unrelated. Matching on a run id generated by the pipeline,
+and stamped only by the scan running inside that pipeline, removes the ambiguity.
+
+Lives in its own module so the scheduler (which sets the id), the scanner (which
+stamps it), and the shadow book (which checks it) can all import it without an
+import cycle.
+"""
+
+from __future__ import annotations
+
+import contextvars
+import uuid
+
+_run_id: contextvars.ContextVar[str | None] = contextvars.ContextVar(
+    "pipeline_run_id", default=None
+)
+
+
+def new_run_id() -> str:
+    """A fresh, collision-free run id."""
+    return uuid.uuid4().hex
+
+
+def current() -> str | None:
+    """Run id of the pipeline invocation on the current task, if any."""
+    return _run_id.get()
+
+
+def bind(run_id: str) -> contextvars.Token:
+    """Set the current run id; pass the returned token to ``release``."""
+    return _run_id.set(run_id)
+
+
+def release(token: contextvars.Token) -> None:
+    """Restore the previous run id (call in a finally)."""
+    _run_id.reset(token)

app/services/rr_scanner_service.py

@@ -30,7 +30,10 @@ 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,
     get_reentry_gate_locks,
     observe_reentry_gate_transitions,
 )
@@ -44,6 +47,14 @@ from app.services.recommendation_service import (
 
 logger = logging.getLogger(__name__)
 
+# 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"
+
 STRATEGY_VERSION = "residual_highvol_80_20_atr_trail3_v1"
 
 # A setup counts as live only while the daily scan keeps re-emitting it. The
@@ -514,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.
 
@@ -680,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()
@@ -740,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)
@@ -772,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:
@@ -788,12 +811,18 @@ async def scan_all_tickers(
             logger.exception("Error scanning ticker %s", symbol)
 
     if activation is not None:
-        transitioned_ticker_ids = await observe_reentry_gate_transitions(
+        # Both books, from the same observation: gate-reset state is per book,
-            db,
+        # so observing only the manual book would leave shadow stop-outs stuck
-            evaluated_ticker_ids=evaluated_ticker_ids,
+        # with a fail timestamp that never requalifies — permanently ineligible.
-            qualified_ticker_ids=qualified_ticker_ids,
+        transitioned_ticker_ids: set[int] = set()
-            observed_at=gate_observation_started_at,
+        for book in (MANUAL_BOOK, SHADOW_BOOK):
-        )
+            transitioned_ticker_ids |= await observe_reentry_gate_transitions(
+                db,
+                evaluated_ticker_ids=evaluated_ticker_ids,
+                qualified_ticker_ids=qualified_ticker_ids,
+                observed_at=gate_observation_started_at,
+                book=book,
+            )
         await db.commit()
         if transitioned_ticker_ids:
             logger.info(
@@ -804,6 +833,16 @@ async def scan_all_tickers(
     if progress_callback is not None and total:
         progress_callback(total, total, "")
 
+    # 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, scan_run_id)
+    await db.commit()
+
     return all_setups
 
 
@@ -815,6 +854,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]:
@@ -843,11 +883,23 @@ async def get_trade_setups(
     excluded_ticker_ids: set[int] = set()
     reentry_gate_locks: dict[int, datetime] = {}
     if exclude_open_trade_tickers:
-        open_trade_result = await db.execute(
+        # Manual book only. The shadow book holds the *top-ranked* names by
+        # 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_stmt = (
             select(PaperTrade.ticker_id)
-            .where(PaperTrade.status == "open")
+            .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()
         )

app/services/shadow_book_service.py

@@ -0,0 +1,363 @@
+"""Shadow book — the validated strategy, traded automatically.
+
+The discretionary paper book only ever contains trades the user chose to take,
+inside a ~20 minute window, on days they were available. The backtest that
+validated this strategy does none of that: it takes the top-ranked qualified
+setups up to capacity, every session, with no human involved. That difference
+makes the manual book unusable as out-of-sample evidence — it measures the
+strategy *plus* discretion and availability.
+
+The shadow book closes that gap. It mirrors ``_simulate_portfolio``'s selection
+rule exactly and shares the manual book's exit policy, so the only difference
+between the two books is *which* qualified setups get taken.
+
+Parity is the load-bearing property here. Selection ordering comes from the
+stored ``strategy_rank`` the scanner already wrote (the same 80/20
+momentum/vol blend the backtest ranks on) rather than being recomputed, so the
+two cannot drift apart.
+"""
+
+from __future__ import annotations
+
+import logging
+from datetime import datetime, timedelta, timezone
+
+from sqlalchemy import select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.models.paper_trade import PaperTrade
+from app.models.ticker import Ticker
+from app.models.trade_setup import TradeSetup
+from app.models.user import User
+from app.services import settings_store
+from app.services.qualification import setup_qualifies
+from app.services.trade_policy import SHADOW_BOOK, get_reentry_gate_locks
+
+logger = logging.getLogger(__name__)
+
+KEY_ENABLED = "shadow_book_enabled"
+KEY_CAPACITY = "shadow_book_capacity"
+KEY_RISK_PCT = "shadow_book_risk_pct"
+KEY_START_EQUITY = "shadow_book_start_equity"
+
+# Matches the validated configuration: 10-position book, 1% fixed-fractional
+# risk. Start equity is only a sizing base — comparisons are drawn in percent
+# and R-multiples, never in raw currency.
+DEFAULT_CAPACITY = 10
+DEFAULT_RISK_PCT = 1.0
+DEFAULT_START_EQUITY = 100_000.0
+
+# Mirrors ``_simulate_portfolio``'s SIM_NOTIONAL_CAP: no single position may
+# exceed this fraction of equity, and the book never uses margin. Without the
+# cap, a setup with a tight stop turns 1% risk into a position several times
+# equity — a leveraged trade the validated strategy would never have taken.
+NOTIONAL_CAP = 0.20
+
+# 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:
+    """Shadow book sizing/capacity config, falling back to validated defaults."""
+    raw = await settings_store.get_map(
+        db, [KEY_CAPACITY, KEY_RISK_PCT, KEY_START_EQUITY]
+    )
+
+    def _num(key: str, default: float, *, minimum: float, maximum: float) -> float:
+        try:
+            value = float(raw.get(key) or default)
+        except (TypeError, ValueError):
+            return default
+        return max(minimum, min(maximum, value))
+
+    return {
+        "capacity": int(_num(KEY_CAPACITY, DEFAULT_CAPACITY, minimum=1, maximum=100)),
+        "risk_pct": _num(KEY_RISK_PCT, DEFAULT_RISK_PCT, minimum=0.05, maximum=10.0),
+        "start_equity": _num(
+            KEY_START_EQUITY, DEFAULT_START_EQUITY, minimum=1000.0, maximum=1e9
+        ),
+    }
+
+
+async def is_enabled(db: AsyncSession) -> bool:
+    """Shadow book writes trades to the live book, so it is opt-in."""
+    value = await settings_store.get_value(db, KEY_ENABLED, "false")
+    return str(value).strip().lower() in {"1", "true", "yes", "on"}
+
+
+async def equity_and_cash(
+    db: AsyncSession, start_equity: float, positions: list[PaperTrade]
+) -> tuple[float, float]:
+    """Marked equity and free cash, matching ``_simulate_portfolio``.
+
+    The simulator sizes from *marked* equity — cash plus open positions at their
+    latest close — and spends from cash, so a book that is fully invested cannot
+    keep buying. Sizing from realized P&L alone would drift away from the
+    backtest as soon as positions were held across a scan.
+    """
+    from app.services.paper_trade_service import _latest_closes
+
+    result = await db.execute(
+        select(PaperTrade).where(
+            PaperTrade.book == SHADOW_BOOK,
+            PaperTrade.status == "closed",
+            PaperTrade.close_price.is_not(None),
+        )
+    )
+    realized = 0.0
+    for trade in result.scalars():
+        per_share = (
+            trade.close_price - trade.entry_price
+            if trade.direction == "long"
+            else trade.entry_price - trade.close_price
+        )
+        realized += per_share * trade.shares
+
+    open_cost = sum(p.entry_price * p.shares for p in positions)
+    marks = await _latest_closes(db, {p.ticker_id for p in positions})
+    open_value = sum(
+        (marks.get(p.ticker_id) or p.entry_price) * p.shares for p in positions
+    )
+
+    cash = start_equity + realized - open_cost
+    return cash + open_value, cash
+
+
+def position_shares(
+    equity: float,
+    risk_pct: float,
+    entry: float,
+    stop: float,
+    *,
+    cash_available: float | None = None,
+) -> float:
+    """Shares to buy, sized exactly as ``_simulate_portfolio`` sizes them.
+
+    Fixed-fractional risk first, then the two caps the simulator applies: no
+    position may exceed ``NOTIONAL_CAP`` of equity, and the book cannot spend
+    cash it does not have. Dropping either cap lets a tight stop produce a
+    leveraged position and breaks compounding parity with the backtest.
+    """
+    risk_per_share = abs(entry - stop)
+    if risk_per_share <= 0 or equity <= 0 or entry <= 0:
+        return 0.0
+
+    shares = (equity * risk_pct / 100.0) / risk_per_share
+    shares = min(shares, (equity * NOTIONAL_CAP) / entry)
+    if cash_available is not None:
+        shares = min(shares, max(0.0, cash_available) / entry)
+    # Dust guard, as in the simulator: sub-$1 positions are noise, not trades.
+    return shares if shares * entry >= 1.0 else 0.0
+
+
+async def _open_positions(db: AsyncSession) -> list[PaperTrade]:
+    result = await db.execute(
+        select(PaperTrade).where(
+            PaperTrade.book == SHADOW_BOOK, PaperTrade.status == "open"
+        )
+    )
+    return list(result.scalars().all())
+
+
+async def _shadow_user_id(db: AsyncSession) -> int | None:
+    """Shadow trades are not owned by a person; attach them to the first user."""
+    result = await db.execute(select(User.id).order_by(User.id.asc()).limit(1))
+    row = result.first()
+    return int(row[0]) if row else None
+
+
+async def _scan_run_to_trade(
+    db: AsyncSession,
+    *,
+    now: datetime,
+    expected_run_id: str | None = 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
+      failed in *this* pipeline never stamped this id, and a concurrent manual
+      scan (a separate APScheduler job, not serialised against the pipeline)
+      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 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.
+
+    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
+
+    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 completed is None or not run_id:
+        return None
+    if expected_run_id is not None:
+        return run_id if run_id == expected_run_id else None
+    if now - completed > MAX_SCAN_AGE:
+        return None
+    return run_id
+
+
+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,
+    expected_run_id: str | None = None,
+) -> list[TradeSetup]:
+    """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 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
+       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).
+    """
+    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.scan_run_id == run_id)
+    )
+    rows = [s for s in result.scalars() if (s.direction or "long") == "long"]
+
+    latest: dict[int, TradeSetup] = {}
+    for setup in rows:
+        held = latest.get(setup.ticker_id)
+        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(
+        qualified,
+        key=lambda s: (
+            s.strategy_rank if s.strategy_rank is not None else float("-inf")
+        ),
+        reverse=True,
+    )
+
+
+async def open_shadow_positions(
+    db: AsyncSession,
+    *,
+    activation_config: dict,
+    opened_at: datetime | None = None,
+    expected_run_id: str | None = None,
+) -> dict:
+    """Fill free capacity with the top-ranked qualified setups.
+
+    Mirrors the backtest: rank the qualified cross-section, walk it top-down,
+    skip anything already held or locked out by post-stop gate-reset, and stop
+    at capacity. Returns a summary for the job log.
+
+    ``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
+    ``_scan_run_to_trade``.
+    """
+    summary = {
+        "opened": 0,
+        "skipped_held": 0,
+        "skipped_locked": 0,
+        "skipped_no_cash": 0,
+        "symbols": [],
+    }
+    config = await get_config(db)
+
+    positions = await _open_positions(db)
+    held = {p.ticker_id for p in positions}
+    free_slots = config["capacity"] - len(positions)
+    if free_slots <= 0:
+        return summary
+
+    user_id = await _shadow_user_id(db)
+    if user_id is None:
+        logger.warning("shadow book skipped: no user to attach trades to")
+        return summary
+
+    locks = await get_reentry_gate_locks(db, book=SHADOW_BOOK)
+    equity, cash = await equity_and_cash(db, config["start_equity"], positions)
+    timestamp = opened_at or datetime.now(timezone.utc)
+
+    candidates = await _todays_qualified_setups(
+        db, activation_config, now=timestamp, expected_run_id=expected_run_id
+    )
+    for setup in candidates:
+        if free_slots <= 0:
+            break
+        if setup.ticker_id in held:
+            summary["skipped_held"] += 1
+            continue
+        if setup.ticker_id in locks:
+            summary["skipped_locked"] += 1
+            continue
+
+        entry = float(setup.entry_price or 0.0)
+        stop = float(setup.stop_loss or 0.0)
+        shares = position_shares(
+            equity, config["risk_pct"], entry, stop, cash_available=cash
+        )
+        if shares <= 0:
+            summary["skipped_no_cash"] += 1
+            continue
+        cash -= shares * entry
+
+        db.add(
+            PaperTrade(
+                user_id=user_id,
+                ticker_id=setup.ticker_id,
+                direction=setup.direction,
+                entry_price=entry,
+                shares=shares,
+                stop_loss=stop,
+                target=float(setup.target or 0.0),
+                status="open",
+                opened_at=timestamp,
+                fill_mode="near_close",
+                book=SHADOW_BOOK,
+            )
+        )
+        held.add(setup.ticker_id)
+        free_slots -= 1
+        summary["opened"] += 1
+        summary["symbols"].append(setup.ticker_id)
+
+    if summary["opened"]:
+        await db.commit()
+    return summary
+
+
+async def symbols_for(db: AsyncSession, ticker_ids: list[int]) -> list[str]:
+    """Resolve ticker ids to symbols for logging."""
+    if not ticker_ids:
+        return []
+    result = await db.execute(select(Ticker.symbol).where(Ticker.id.in_(ticker_ids)))
+    return [row[0] for row in result.all()]

app/services/trade_policy.py

@@ -22,12 +22,22 @@ def _ny_trading_date(moment: datetime) -> date:
     return moment.astimezone(_REENTRY_DAY_TZ).date()
 
 
+MANUAL_BOOK = "manual"
+SHADOW_BOOK = "shadow"
+
+
 async def _latest_initial_stop_trades(
     db: AsyncSession,
     *,
     closed_before: datetime | None = None,
+    book: str = MANUAL_BOOK,
 ) -> dict[int, PaperTrade]:
-    """Return a ticker's latest closed trade only when it was an initial stop."""
+    """Return a ticker's latest closed trade only when it was an initial stop.
+
+    Scoped to one ``book``: the discretionary and shadow books diverge as soon
+    as their entries differ, so each must see only its own stop history when
+    deciding whether a ticker is locked out of re-entry.
+    """
     ranked_stmt = (
         select(
             PaperTrade.id.label("trade_id"),
@@ -41,6 +51,7 @@ async def _latest_initial_stop_trades(
         .where(
             PaperTrade.status == "closed",
             PaperTrade.closed_at.is_not(None),
+            PaperTrade.book == book,
         )
     )
     if closed_before is not None:
@@ -58,7 +69,9 @@ async def _latest_initial_stop_trades(
     return {trade.ticker_id: trade for trade in result.scalars()}
 
 
-async def get_reentry_gate_locks(db: AsyncSession) -> dict[int, datetime]:
+async def get_reentry_gate_locks(
+    db: AsyncSession, *, book: str = MANUAL_BOOK
+) -> dict[int, datetime]:
     """Return tickers still waiting for a post-stop gate failure.
 
     A later qualified setup is actionable only after the daily scanner has
@@ -66,7 +79,7 @@ async def get_reentry_gate_locks(db: AsyncSession) -> dict[int, datetime]:
     then a fresh qualification. The returned timestamp is the stop time and is
     useful for diagnostics; callers normally only need the keys.
     """
-    latest = await _latest_initial_stop_trades(db)
+    latest = await _latest_initial_stop_trades(db, book=book)
     return {
         ticker_id: trade.closed_at
         for ticker_id, trade in latest.items()
@@ -80,6 +93,7 @@ async def observe_reentry_gate_transitions(
     evaluated_ticker_ids: Iterable[int],
     qualified_ticker_ids: Iterable[int],
     observed_at: datetime | None = None,
+    book: str = MANUAL_BOOK,
 ) -> set[int]:
     """Persist gate-failure and later requalification observations.
 
@@ -93,7 +107,7 @@ async def observe_reentry_gate_transitions(
         return set()
     qualified = {int(ticker_id) for ticker_id in qualified_ticker_ids}
     timestamp = observed_at or datetime.now(timezone.utc)
-    latest = await _latest_initial_stop_trades(db, closed_before=timestamp)
+    latest = await _latest_initial_stop_trades(db, closed_before=timestamp, book=book)
     updated: set[int] = set()
     for ticker_id in evaluated:
         trade = latest.get(ticker_id)

frontend/src/api/admin.ts

@@ -92,6 +92,41 @@ export function updateScheduleSettings(payload: Partial<ScheduleConfig>) {
     .then((r) => r.data);
 }
 
+export interface PerformanceConfig {
+  start_date: string;
+}
+
+export function getPerformanceSettings() {
+  return apiClient
+    .get<PerformanceConfig>('admin/settings/performance')
+    .then((r) => r.data);
+}
+
+export function updatePerformanceSettings(payload: Partial<PerformanceConfig>) {
+  return apiClient
+    .put<PerformanceConfig>('admin/settings/performance', payload)
+    .then((r) => r.data);
+}
+
+export interface ShadowBookConfig {
+  enabled: boolean;
+  capacity: number;
+  risk_pct: number;
+  start_equity: number;
+}
+
+export function getShadowBookSettings() {
+  return apiClient
+    .get<ShadowBookConfig>('admin/settings/shadow-book')
+    .then((r) => r.data);
+}
+
+export function updateShadowBookSettings(payload: Partial<ShadowBookConfig>) {
+  return apiClient
+    .put<ShadowBookConfig>('admin/settings/shadow-book', payload)
+    .then((r) => r.data);
+}
+
 export function getSentimentSettings() {
   return apiClient
     .get<SentimentProviderConfig>('admin/settings/sentiment')

frontend/src/api/paperTrades.ts

@@ -38,6 +38,39 @@ export function getEquityCurve() {
   return apiClient.get<EquityPoint[]>('paper-trades/equity-curve').then((r) => r.data);
 }
 
+export interface PerfPoint {
+  date: string;
+  manual_pnl: number;
+  shadow_pnl: number;
+  spy_pct: number;
+}
+
+export interface BookStats {
+  trades: number;
+  closed: number;
+  open: number;
+  win_rate: number | null;
+  total_r: number;
+  avg_r: number | null;
+  pnl: number;
+}
+
+export interface PerformanceSummary {
+  start_date: string | null;
+  series: PerfPoint[];
+  stats: {
+    manual?: BookStats;
+    shadow?: BookStats;
+    spy?: { pct: number };
+  };
+}
+
+export function getPerformance() {
+  return apiClient
+    .get<PerformanceSummary>('paper-trades/performance')
+    .then((r) => r.data);
+}
+
 export function closePaperTrade(id: number, closePrice?: number) {
   return apiClient
     .post<{ id: number; status: string }>(`paper-trades/${id}/close`, {

frontend/src/components/admin/PerformanceSettings.tsx

@@ -0,0 +1,167 @@
+import { useEffect, useState } from 'react';
+import { useMutation, useQuery, useQueryClient } from '@tanstack/react-query';
+import {
+  getPerformanceSettings,
+  getShadowBookSettings,
+  updatePerformanceSettings,
+  updateShadowBookSettings,
+  type ShadowBookConfig,
+} from '../../api/admin';
+import { SkeletonCard } from '../ui/Skeleton';
+
+/** Performance window + the auto-traded shadow book.
+ *
+ *  These belong together: the shadow book is what the comparison measures, and
+ *  the start date is what keeps the comparison inside a single strategy
+ *  configuration.
+ */
+export function PerformanceSettings() {
+  const qc = useQueryClient();
+  const window = useQuery({ queryKey: ['admin', 'performance'], queryFn: getPerformanceSettings });
+  const shadow = useQuery({ queryKey: ['admin', 'shadow-book'], queryFn: getShadowBookSettings });
+
+  const [startDate, setStartDate] = useState('');
+  const [book, setBook] = useState<ShadowBookConfig | null>(null);
+
+  useEffect(() => {
+    if (window.data) setStartDate(window.data.start_date ?? '');
+  }, [window.data]);
+  useEffect(() => {
+    if (shadow.data) setBook(shadow.data);
+  }, [shadow.data]);
+
+  const saveWindow = useMutation({
+    mutationFn: () => updatePerformanceSettings({ start_date: startDate }),
+    onSuccess: () => {
+      qc.invalidateQueries({ queryKey: ['admin', 'performance'] });
+      qc.invalidateQueries({ queryKey: ['paper-trades', 'performance'] });
+    },
+  });
+
+  const saveBook = useMutation({
+    mutationFn: (payload: Partial<ShadowBookConfig>) => updateShadowBookSettings(payload),
+    onSuccess: (data) => {
+      setBook(data);
+      qc.invalidateQueries({ queryKey: ['admin', 'shadow-book'] });
+      qc.invalidateQueries({ queryKey: ['paper-trades', 'performance'] });
+    },
+  });
+
+  if (window.isLoading || shadow.isLoading || !book) return <SkeletonCard />;
+
+  return (
+    <div className="glass space-y-5 p-5">
+      <div>
+        <h3 className="text-sm font-semibold text-gray-200">Performance &amp; Shadow Book</h3>
+        <p className="mt-1 text-xs leading-relaxed text-gray-500">
+          The <span className="text-gray-300">shadow book</span> trades the validated strategy with no
+          human input: top-ranked qualified setups up to capacity, sized to a fixed risk, entered right
+          after the near-close scan. It shares the paper exit policy with your own trades, so the only
+          difference between the two books is <span className="text-gray-300">which setups get taken</span>.
+        </p>
+      </div>
+
+      <label className="block space-y-1">
+        <span className="text-xs text-gray-400">Performance since</span>
+        <div className="flex gap-2">
+          <input
+            type="date"
+            value={startDate}
+            onChange={(e) => setStartDate(e.target.value)}
+            className="input-glass w-48 px-3 py-2 text-sm"
+          />
+          <button
+            type="button"
+            onClick={() => saveWindow.mutate()}
+            disabled={saveWindow.isPending}
+            className="btn-glass px-3 py-2 text-sm"
+          >
+            {saveWindow.isPending ? 'Saving…' : 'Save'}
+          </button>
+          {startDate && (
+            <button
+              type="button"
+              onClick={() => {
+                setStartDate('');
+                updatePerformanceSettings({ start_date: '' }).then(() => {
+                  qc.invalidateQueries({ queryKey: ['admin', 'performance'] });
+                  qc.invalidateQueries({ queryKey: ['paper-trades', 'performance'] });
+                });
+              }}
+              className="btn-glass px-3 py-2 text-sm text-gray-400"
+            >
+              Clear
+            </button>
+          )}
+        </div>
+        <span className="block text-[11px] leading-relaxed text-gray-500">
+          Trades opened before this date are excluded from the Performance card. The strategy has been
+          revised repeatedly — pinning a start keeps the comparison inside one configuration instead of
+          averaging across rules that no longer exist. Empty shows all history.
+        </span>
+      </label>
+
+      <div className="border-t border-white/5 pt-4">
+        <label className="flex items-start gap-3">
+          <input
+            type="checkbox"
+            checked={book.enabled}
+            onChange={(e) => saveBook.mutate({ enabled: e.target.checked })}
+            className="mt-0.5"
+          />
+          <span>
+            <span className="text-sm text-gray-200">Shadow book enabled</span>
+            <span className="block text-[11px] leading-relaxed text-gray-500">
+              Starts opening real paper positions automatically on the next near-close scan. Verify its
+              first selections match a backtest of that day's cross-section before trusting the curve.
+            </span>
+          </span>
+        </label>
+
+        <div className="mt-4 grid gap-4 md:grid-cols-3">
+          <label className="block space-y-1">
+            <span className="text-xs text-gray-400">Capacity (positions)</span>
+            <input
+              type="number"
+              min={1}
+              max={100}
+              value={book.capacity}
+              onChange={(e) => setBook({ ...book, capacity: Number(e.target.value) })}
+              onBlur={() => saveBook.mutate({ capacity: book.capacity })}
+              className="input-glass w-full px-3 py-2 text-sm"
+            />
+          </label>
+          <label className="block space-y-1">
+            <span className="text-xs text-gray-400">Risk per trade (%)</span>
+            <input
+              type="number"
+              step="0.05"
+              min={0.05}
+              max={10}
+              value={book.risk_pct}
+              onChange={(e) => setBook({ ...book, risk_pct: Number(e.target.value) })}
+              onBlur={() => saveBook.mutate({ risk_pct: book.risk_pct })}
+              className="input-glass w-full px-3 py-2 text-sm"
+            />
+          </label>
+          <label className="block space-y-1">
+            <span className="text-xs text-gray-400">Start equity ($)</span>
+            <input
+              type="number"
+              min={1000}
+              step={1000}
+              value={book.start_equity}
+              onChange={(e) => setBook({ ...book, start_equity: Number(e.target.value) })}
+              onBlur={() => saveBook.mutate({ start_equity: book.start_equity })}
+              className="input-glass w-full px-3 py-2 text-sm"
+            />
+          </label>
+        </div>
+        <p className="mt-2 text-[11px] leading-relaxed text-gray-500">
+          Defaults match the validated configuration: 10 positions, 1% fixed-fractional risk. Start
+          equity is only a sizing base — the books are compared in R-multiples, not currency.
+        </p>
+      </div>
+    </div>
+  );
+}

frontend/src/components/dashboard/PerfChart.tsx

@@ -1,11 +1,17 @@
 import { useMemo, useRef, useState } from 'react';
 import { useQuery } from '@tanstack/react-query';
-import { getEquityCurve } from '../../api/paperTrades';
+import { getPerformance, type BookStats } from '../../api/paperTrades';
 import { Section } from '../ui/Section';
 
-const W = 760;
+const W = 1040;
-const H = 220;
+const H = 260;
-const PAD = { top: 14, right: 84, bottom: 26, left: 56 };
+const PAD = { top: 16, right: 92, bottom: 28, left: 60 };
+
+const COLORS = {
+  shadow: 'var(--up)',
+  manual: 'var(--accent, #7aa2f7)',
+  spy: 'var(--ink-3)',
+} as const;
 
 function money(v: number): string {
   const sign = v > 0 ? '+' : v < 0 ? '−' : '';
@@ -25,17 +31,54 @@ function niceTicks(lo: number, hi: number, count = 4): number[] {
   return out;
 }
 
-/** Paper book vs the same dollars riding SPY — cumulative P&L since first trade. */
+/** R-multiple stats read straight across; currency P&L does not, because the
+ *  books size differently. Kept adjacent so the comparison is hard to misread. */
+function StatCell({ label, stats, color }: { label: string; stats?: BookStats; color: string }) {
+  if (!stats || stats.trades === 0) {
+    return (
+      <div className="min-w-[7rem]">
+        <div className="flex items-center gap-1.5 text-xs text-gray-400">
+          <i className="inline-block h-2 w-2 rounded-full" style={{ background: color }} />
+          {label}
+        </div>
+        <div className="num mt-1 text-sm text-gray-500">no trades yet</div>
+      </div>
+    );
+  }
+  return (
+    <div className="min-w-[7rem]">
+      <div className="flex items-center gap-1.5 text-xs text-gray-400">
+        <i className="inline-block h-2 w-2 rounded-full" style={{ background: color }} />
+        {label}
+      </div>
+      <div className="num mt-1 text-lg font-semibold text-gray-100">
+        {stats.total_r > 0 ? '+' : ''}
+        {stats.total_r.toFixed(2)}R
+      </div>
+      <div className="num text-[11px] leading-relaxed text-gray-500">
+        {stats.trades} trades · {stats.win_rate ?? '—'}% win
+        <br />
+        avg {stats.avg_r === null ? '—' : `${stats.avg_r > 0 ? '+' : ''}${stats.avg_r.toFixed(2)}R`} · {money(stats.pnl)}
+      </div>
+    </div>
+  );
+}
+
+/** Shadow book (the strategy, traded automatically) vs the discretionary book
+ *  vs SPY. Both books share an exit policy, so the only difference is which
+ *  qualified setups get taken. */
 export function PerfChart() {
-  const curve = useQuery({ queryKey: ['paper-trades', 'equity-curve'], queryFn: getEquityCurve });
+  const perf = useQuery({ queryKey: ['paper-trades', 'performance'], queryFn: getPerformance });
   const [hover, setHover] = useState<number | null>(null);
   const svgRef = useRef<SVGSVGElement>(null);
 
-  const data = curve.data ?? [];
+  const data = perf.data?.series ?? [];
+  const stats = perf.data?.stats ?? {};
+  const startDate = perf.data?.start_date ?? null;
 
   const geom = useMemo(() => {
     if (data.length < 2) return null;
-    const values = data.flatMap((p) => [p.book_pnl, p.benchmark_pnl, 0]);
+    const values = data.flatMap((p) => [p.manual_pnl, p.shadow_pnl, 0]);
     const lo = Math.min(...values);
     const hi = Math.max(...values);
     const pad = (hi - lo) * 0.08 || 1;
@@ -45,9 +88,20 @@ export function PerfChart() {
     const plotH = H - PAD.top - PAD.bottom;
     const px = (i: number) => PAD.left + (i / (data.length - 1)) * plotW;
     const py = (v: number) => PAD.top + plotH - ((v - yLo) / (yHi - yLo)) * plotH;
-    const line = (key: 'book_pnl' | 'benchmark_pnl') =>
+    const line = (key: 'manual_pnl' | 'shadow_pnl') =>
       data.map((p, i) => `${i === 0 ? 'M' : 'L'}${px(i).toFixed(1)},${py(p[key]).toFixed(1)}`).join(' ');
-    // Month boundaries for the x axis.
+
+    // SPY is a percentage reference, so it rides its own scale pinned to the
+    // same zero line — otherwise a flat book would squash it out of view.
+    const spyLo = Math.min(...data.map((p) => p.spy_pct), 0);
+    const spyHi = Math.max(...data.map((p) => p.spy_pct), 0);
+    const spySpan = Math.max(Math.abs(spyLo), Math.abs(spyHi)) || 1;
+    const bookSpan = Math.max(Math.abs(yLo), Math.abs(yHi)) || 1;
+    const spyY = (v: number) => py((v / spySpan) * bookSpan);
+    const spyLine = data
+      .map((p, i) => `${i === 0 ? 'M' : 'L'}${px(i).toFixed(1)},${spyY(p.spy_pct).toFixed(1)}`)
+      .join(' ');
+
     const xTicks: { i: number; label: string }[] = [];
     let lastMonth = '';
     data.forEach((p, i) => {
@@ -57,15 +111,24 @@ export function PerfChart() {
         xTicks.push({ i, label: new Date(`${p.date}T00:00:00`).toLocaleDateString('en-US', { month: 'short' }) });
       }
     });
-    if (xTicks.length > 8) {
+    if (xTicks.length > 10) {
-      const keep = Math.ceil(xTicks.length / 8);
+      const keep = Math.ceil(xTicks.length / 10);
       for (let k = xTicks.length - 1; k >= 0; k--) if (k % keep !== 0) xTicks.splice(k, 1);
     }
-    return { yLo, yHi, plotH, px, py, line, yTicks: niceTicks(yLo, yHi), xTicks };
+    return { yLo, yHi, plotH, px, py, line, spyLine, spyY, yTicks: niceTicks(yLo, yHi), xTicks };
   }, [data]);
 
-  if (!geom) return null;
+  if (!geom) {
-  const { px, py, line, yTicks, xTicks, plotH } = geom;
+    return (
+      <Section title="Performance" hint="shadow book vs your picks vs SPY">
+        <div className="glass p-5 text-sm text-gray-500">
+          No trades in the selected window yet
+          {startDate ? ` (since ${startDate})` : ''}. The shadow book starts recording once enabled.
+        </div>
+      </Section>
+    );
+  }
+  const { px, py, line, spyLine, spyY, yTicks, xTicks, plotH } = geom;
 
   const onMove = (e: React.MouseEvent<SVGSVGElement>) => {
     const rect = svgRef.current?.getBoundingClientRect();
@@ -81,11 +144,32 @@ export function PerfChart() {
     new Date(`${iso}T00:00:00`).toLocaleDateString('en-US', { month: 'short', day: 'numeric' });
 
   return (
-    <Section title="Performance" hint="paper book vs the same dollars in SPY · cumulative P&L">
+    <Section
+      title="Performance"
+      hint={`shadow book vs your picks vs SPY${startDate ? ` · since ${startDate}` : ''}`}
+    >
       <div className="glass p-5 pb-2">
-        <div className="flex justify-end gap-4 text-xs text-gray-400">
+        <div className="mb-3 flex flex-wrap items-start justify-between gap-x-8 gap-y-3">
-          <span><i className="mr-1.5 inline-block h-2 w-2 rounded-full align-middle" style={{ background: 'var(--up)' }} /> Book</span>
+          <div className="flex flex-wrap gap-x-8 gap-y-3">
-          <span><i className="mr-1.5 inline-block h-2 w-2 rounded-full align-middle" style={{ background: 'var(--ink-3)' }} /> Same $ in SPY</span>
+            <StatCell label="Shadow (strategy)" stats={stats.shadow} color={COLORS.shadow} />
+            <StatCell label="Your picks" stats={stats.manual} color={COLORS.manual} />
+            <div className="min-w-[7rem]">
+              <div className="flex items-center gap-1.5 text-xs text-gray-400">
+                <i className="inline-block h-2 w-2 rounded-full" style={{ background: COLORS.spy }} />
+                SPY
+              </div>
+              <div className="num mt-1 text-lg font-semibold text-gray-300">
+                {(stats.spy?.pct ?? 0) > 0 ? '+' : ''}
+                {(stats.spy?.pct ?? 0).toFixed(1)}%
+              </div>
+              <div className="num text-[11px] text-gray-500">buy &amp; hold</div>
+            </div>
+          </div>
+          <p className="max-w-[22rem] text-[11px] leading-relaxed text-gray-500">
+            Books share the same exit, so the difference is <b className="text-gray-400">selection</b>.
+            Compare on <b className="text-gray-400">R</b>, not $ — sizing differs. Expect months of
+            noise before a gap means anything.
+          </p>
         </div>
         <svg
           ref={svgRef}
@@ -94,7 +178,9 @@ export function PerfChart() {
           onMouseMove={onMove}
           onMouseLeave={() => setHover(null)}
           role="img"
-          aria-label={`Paper book cumulative P&L ${money(data[last].book_pnl)} versus ${money(data[last].benchmark_pnl)} for the same dollars in SPY`}
+          aria-label={`Shadow book ${money(data[last].shadow_pnl)}, your picks ${money(
+            data[last].manual_pnl,
+          )}, SPY ${(data[last].spy_pct ?? 0).toFixed(1)} percent`}
         >
           {yTicks.map((t) => (
             <g key={t}>
@@ -104,35 +190,40 @@ export function PerfChart() {
               </text>
             </g>
           ))}
-          {/* zero baseline slightly stronger when it's inside the plot */}
           <line x1={PAD.left} x2={W - PAD.right} y1={py(0)} y2={py(0)} stroke="var(--ink-3)" strokeWidth="1" opacity="0.5" />
           {xTicks.map(({ i, label }) => (
             <text key={`${label}-${i}`} x={px(i)} y={H - 6} textAnchor="middle" className="num" fill="var(--ink-3)" fontSize="10">
               {label}
             </text>
           ))}
-          <path d={line('benchmark_pnl')} fill="none" stroke="var(--ink-3)" strokeWidth="2" strokeLinejoin="round" />
+          <path d={spyLine} fill="none" stroke={COLORS.spy} strokeWidth="1.5" strokeDasharray="4 3" strokeLinejoin="round" />
-          <path d={line('book_pnl')} fill="none" stroke="var(--up)" strokeWidth="2" strokeLinejoin="round" />
+          <path d={line('manual_pnl')} fill="none" stroke={COLORS.manual} strokeWidth="2" strokeLinejoin="round" />
+          <path d={line('shadow_pnl')} fill="none" stroke={COLORS.shadow} strokeWidth="2" strokeLinejoin="round" />
           {hover !== null && (
             <g>
               <line x1={px(hover)} x2={px(hover)} y1={PAD.top} y2={PAD.top + plotH} stroke="var(--ink-3)" strokeWidth="1" />
-              <circle cx={px(hover)} cy={py(data[hover].book_pnl)} r="4.5" fill="var(--up)" stroke="var(--surface)" strokeWidth="2" />
+              <circle cx={px(hover)} cy={py(data[hover].shadow_pnl)} r="4.5" fill={COLORS.shadow} stroke="var(--surface)" strokeWidth="2" />
-              <circle cx={px(hover)} cy={py(data[hover].benchmark_pnl)} r="4.5" fill="var(--ink-3)" stroke="var(--surface)" strokeWidth="2" />
+              <circle cx={px(hover)} cy={py(data[hover].manual_pnl)} r="4.5" fill={COLORS.manual} stroke="var(--surface)" strokeWidth="2" />
+              <circle cx={px(hover)} cy={spyY(data[hover].spy_pct)} r="4" fill={COLORS.spy} stroke="var(--surface)" strokeWidth="2" />
             </g>
           )}
-          <circle cx={px(last)} cy={py(data[last].book_pnl)} r="4" fill="var(--up)" stroke="var(--surface)" strokeWidth="2" />
+          <text x={px(last) + 10} y={py(data[last].shadow_pnl) + 4} className="num" fill="var(--ink)" fontSize="11" fontWeight="600">
-          <circle cx={px(last)} cy={py(data[last].benchmark_pnl)} r="4" fill="var(--ink-3)" stroke="var(--surface)" strokeWidth="2" />
+            {money(data[last].shadow_pnl)}
-          <text x={px(last) + 10} y={py(data[last].book_pnl) + 4} className="num" fill="var(--ink)" fontSize="11" fontWeight="600">
-            {money(data[last].book_pnl)}
           </text>
-          <text x={px(last) + 10} y={py(data[last].benchmark_pnl) + 4} className="num" fill="var(--ink-2)" fontSize="11">
+          <text x={px(last) + 10} y={py(data[last].manual_pnl) + 4} className="num" fill="var(--ink-2)" fontSize="11">
-            {money(data[last].benchmark_pnl)}
+            {money(data[last].manual_pnl)}
           </text>
         </svg>
         <p className="num px-1 pb-1 pt-1.5 text-[11px] text-gray-500" aria-live="polite">
-          {hb
+          {hb ? (
-            ? <>{fmtDate(hb.date)} — book <b className="text-gray-200">{money(hb.book_pnl)}</b> · SPY <b className="text-gray-200">{money(hb.benchmark_pnl)}</b></>
+            <>
-            : <>hover for daily values · realized + mark-to-market, since first paper trade</>}
+              {fmtDate(hb.date)} — shadow <b className="text-gray-200">{money(hb.shadow_pnl)}</b> · yours{' '}
+              <b className="text-gray-200">{money(hb.manual_pnl)}</b> · SPY{' '}
+              <b className="text-gray-200">{hb.spy_pct.toFixed(1)}%</b>
+            </>
+          ) : (
+            <>hover for daily values · realized + mark-to-market{startDate ? ` · window starts ${startDate}` : ''}</>
+          )}
         </p>
       </div>
     </Section>

frontend/src/pages/AdminPage.tsx

@@ -5,6 +5,7 @@ import { AlertSettings } from '../components/admin/AlertSettings';
 import { SentimentProviderSettings } from '../components/admin/SentimentProviderSettings';
 import { DataCleanup } from '../components/admin/DataCleanup';
 import { JobControls } from '../components/admin/JobControls';
+import { PerformanceSettings } from '../components/admin/PerformanceSettings';
 import { PipelineReadinessPanel } from '../components/admin/PipelineReadinessPanel';
 import { SystemEventsPanel } from '../components/admin/SystemEventsPanel';
 import { RecommendationSettings } from '../components/admin/RecommendationSettings';
@@ -36,6 +37,7 @@ export default function AdminPage() {
           <div className="space-y-4">
             <ActivationSettings />
             <ExitPolicySettings />
+            <PerformanceSettings />
             <AlertSettings />
             <SentimentProviderSettings />
             <TickerUniverseBootstrap />

tests/unit/test_alpaca_provider_window.py

@@ -43,11 +43,19 @@ def _utcnow() -> datetime:
     return datetime.now(timezone.utc).replace(tzinfo=None)
 
 
+def _utc_today() -> date:
+    """The provider clamps against UTC, so the fixture must use the UTC date.
+
+    ``date.today()`` is local and runs a day ahead in a UTC+hh timezone just
+    after midnight, which would assert against a day the clamp cannot reach."""
+    return datetime.now(timezone.utc).date()
+
+
 @pytest.mark.asyncio
 async def test_todays_in_progress_bar_is_inside_the_window():
     """The whole near-close design depends on today's bar being fetchable."""
     provider, client = _provider()
-    today = date.today()
+    today = _utc_today()
 
     await provider.fetch_ohlcv("AAPL", today - timedelta(days=5), today)
 
@@ -60,7 +68,7 @@ async def test_window_stays_out_of_the_delayed_data_period():
     """A window reaching the last ~15 minutes fails the entire request."""
     provider, client = _provider()
 
-    await provider.fetch_ohlcv("AAPL", date.today() - timedelta(days=5), date.today())
+    await provider.fetch_ohlcv("AAPL", _utc_today() - timedelta(days=5), _utc_today())
 
     assert client.request.end <= _utcnow() - timedelta(minutes=15)
 
@@ -69,7 +77,7 @@ async def test_window_stays_out_of_the_delayed_data_period():
 async def test_completed_past_day_is_fully_covered():
     """Clamping must not swallow the last day of a historical window."""
     provider, client = _provider()
-    end_date = date.today() - timedelta(days=3)
+    end_date = _utc_today() - timedelta(days=3)
 
     await provider.fetch_ohlcv("AAPL", end_date - timedelta(days=5), end_date)
 
@@ -80,7 +88,7 @@ async def test_completed_past_day_is_fully_covered():
 async def test_window_collapsing_to_nothing_skips_the_call():
     """A start inside the delayed period yields no request at all, not an error."""
     provider, client = _provider()
-    tomorrow = date.today() + timedelta(days=1)
+    tomorrow = _utc_today() + timedelta(days=2)
 
     records = await provider.fetch_ohlcv("AAPL", tomorrow, tomorrow)
 

tests/unit/test_book_separation.py

@@ -0,0 +1,181 @@
+"""The two books must not leak into each other.
+
+Regression cover for the review of ba2df8b. Each test here pins a way the
+shadow book could quietly corrupt the discretionary record — or be corrupted
+by it — which would invalidate the comparison the shadow book exists to make.
+"""
+
+from __future__ import annotations
+
+from datetime import datetime, timedelta, timezone
+
+import pytest
+
+from app.exceptions import ValidationError
+from app.models.paper_trade import PaperTrade
+from app.models.ticker import Ticker
+from app.models.user import User
+from app.services import paper_trade_service as pts
+from app.services.trade_policy import (
+    MANUAL_BOOK,
+    SHADOW_BOOK,
+    get_reentry_gate_locks,
+    observe_reentry_gate_transitions,
+)
+
+
+@pytest.fixture
+async def session():
+    from tests.conftest import _test_session_factory
+
+    async with _test_session_factory() as session:
+        yield session
+
+
+async def _seed(session) -> int:
+    session.add(User(id=1, username="owner", password_hash="x"))
+    session.add(Ticker(id=1, symbol="AAA", name="AAA"))
+    await session.commit()
+    return 1
+
+
+def _trade(*, book, status="open", close_reason=None, closed_at=None, user_id=1):
+    now = datetime.now(timezone.utc)
+    return PaperTrade(
+        user_id=user_id,
+        ticker_id=1,
+        direction="long",
+        entry_price=100.0,
+        shares=10.0,
+        stop_loss=95.0,
+        target=115.0,
+        status=status,
+        opened_at=now - timedelta(days=3),
+        closed_at=closed_at,
+        close_price=95.0 if status == "closed" else None,
+        close_reason=close_reason,
+        book=book,
+    )
+
+
+class TestManualEndpoints:
+    @pytest.mark.asyncio
+    async def test_list_excludes_shadow_by_default(self, session):
+        """Open Positions must not silently mix the autonomous book in."""
+        await _seed(session)
+        session.add_all([_trade(book=MANUAL_BOOK), _trade(book=SHADOW_BOOK)])
+        await session.commit()
+
+        rows = await pts.list_trades(session, user_id=1)
+
+        assert len(rows) == 1
+
+    @pytest.mark.asyncio
+    async def test_list_can_span_both_books_deliberately(self, session):
+        await _seed(session)
+        session.add_all([_trade(book=MANUAL_BOOK), _trade(book=SHADOW_BOOK)])
+        await session.commit()
+
+        rows = await pts.list_trades(session, user_id=1, book=None)
+
+        assert len(rows) == 2
+
+    @pytest.mark.asyncio
+    async def test_shadow_trades_cannot_be_closed_by_hand(self, session):
+        """A hand-closed shadow trade is no longer what the strategy would do."""
+        await _seed(session)
+        trade = _trade(book=SHADOW_BOOK)
+        session.add(trade)
+        await session.commit()
+
+        with pytest.raises(ValidationError, match="exit policy"):
+            await pts.close_trade(session, 1, trade.id, 110.0)
+
+
+class TestGateResetCycle:
+    @pytest.mark.asyncio
+    async def test_shadow_stop_completes_fail_then_requalify(self, session):
+        """The full cycle must advance for the shadow book, or a stopped ticker
+        stays locked out forever."""
+        await _seed(session)
+        closed = datetime.now(timezone.utc) - timedelta(days=2)
+        session.add(
+            _trade(book=SHADOW_BOOK, status="closed", close_reason="stop", closed_at=closed)
+        )
+        await session.commit()
+
+        assert 1 in await get_reentry_gate_locks(session, book=SHADOW_BOOK)
+
+        # Day 1: ticker no longer qualifies → failure observed.
+        await observe_reentry_gate_transitions(
+            session,
+            evaluated_ticker_ids=[1],
+            qualified_ticker_ids=[],
+            observed_at=closed + timedelta(days=1),
+            book=SHADOW_BOOK,
+        )
+        assert 1 in await get_reentry_gate_locks(session, book=SHADOW_BOOK)
+
+        # Day 2: qualifies again → lock releases.
+        await observe_reentry_gate_transitions(
+            session,
+            evaluated_ticker_ids=[1],
+            qualified_ticker_ids=[1],
+            observed_at=closed + timedelta(days=2),
+            book=SHADOW_BOOK,
+        )
+        assert 1 not in await get_reentry_gate_locks(session, book=SHADOW_BOOK)
+
+    @pytest.mark.asyncio
+    async def test_observing_one_book_does_not_move_the_other(self, session):
+        await _seed(session)
+        closed = datetime.now(timezone.utc) - timedelta(days=2)
+        session.add_all(
+            [
+                _trade(book=SHADOW_BOOK, status="closed", close_reason="stop", closed_at=closed),
+                _trade(book=MANUAL_BOOK, status="closed", close_reason="stop", closed_at=closed),
+            ]
+        )
+        await session.commit()
+
+        await observe_reentry_gate_transitions(
+            session,
+            evaluated_ticker_ids=[1],
+            qualified_ticker_ids=[],
+            observed_at=closed + timedelta(days=1),
+            book=SHADOW_BOOK,
+        )
+        await observe_reentry_gate_transitions(
+            session,
+            evaluated_ticker_ids=[1],
+            qualified_ticker_ids=[1],
+            observed_at=closed + timedelta(days=2),
+            book=SHADOW_BOOK,
+        )
+
+        # Shadow released; manual never observed, so it stays locked.
+        assert 1 not in await get_reentry_gate_locks(session, book=SHADOW_BOOK)
+        assert 1 in await get_reentry_gate_locks(session, book=MANUAL_BOOK)
+
+
+class TestPerformanceScoping:
+    @pytest.mark.asyncio
+    async def test_manual_side_is_scoped_to_the_caller(self, session):
+        """'Your picks' must not aggregate another user's discretionary book."""
+        await _seed(session)
+        session.add(User(id=2, username="other", password_hash="x"))
+        await session.commit()
+        session.add_all(
+            [
+                _trade(book=MANUAL_BOOK, user_id=1),
+                _trade(book=MANUAL_BOOK, user_id=2),
+                _trade(book=SHADOW_BOOK, user_id=1),
+            ]
+        )
+        await session.commit()
+
+        mine = await pts.list_trades(session, user_id=1)
+        theirs = await pts.list_trades(session, user_id=2)
+
+        assert len(mine) == 1
+        assert len(theirs) == 1

tests/unit/test_paper_trade_service.py

@@ -16,6 +16,16 @@ from app.services import paper_trade_service as svc
 from tests.conftest import _test_session_factory  # type: ignore
 
 
+def _today() -> date:
+    """UTC date — trades are stamped in UTC, so fixtures must use it too.
+
+    ``date.today()`` is local; in a UTC+hh timezone it runs a day ahead between
+    midnight and the offset, which silently desynchronises bar and benchmark
+    fixtures from the UTC ``opened_at`` the service reads.
+    """
+    return datetime.now(timezone.utc).date()
+
+
 @pytest.fixture
 async def session():
     async with _test_session_factory() as s:
@@ -30,7 +40,7 @@ async def _seed(session, symbol: str, close: float) -> int:
     t = Ticker(symbol=symbol)
     session.add(t)
     await session.flush()
-    session.add(OHLCVRecord(ticker_id=t.id, date=date.today(),
+    session.add(OHLCVRecord(ticker_id=t.id, date=_today(),
                             open=close, high=close, low=close, close=close, volume=1))
     await session.commit()
     return t.id
@@ -51,7 +61,7 @@ async def test_create_and_list_open(session):
 async def test_create_trade_enforces_post_stop_gate_reset_at_service_boundary(session):
     blocked_id = await _seed(session, "LOCKQ", close=100.0)
     released_id = await _seed(session, "FREEQ", close=100.0)
-    today = date.today()
+    today = _today()
 
     def stopped_trade(ticker_id: int, *, gate_reset_complete: bool) -> PaperTrade:
         closed_on = today - timedelta(days=10)
@@ -167,7 +177,7 @@ async def test_resolve_closes_on_target(session):
     trade = await svc.create_trade(session, 1, symbol="AAA", direction="long",
                                    entry_price=100.0, shares=10, stop_loss=95.0, target=110.0)
     # later bars: a day that trades up through 110
-    await _add_bars(session, tid, [(103, 101), (111, 108)], start=date.today())
+    await _add_bars(session, tid, [(103, 101), (111, 108)], start=_today())
     closed = await svc.resolve_open_trades(session)
     assert closed == 1
     await session.refresh(trade)
@@ -180,7 +190,7 @@ async def test_resolve_closes_on_stop(session):
     tid = await _seed(session, "AAA", close=100.0)
     trade = await svc.create_trade(session, 1, symbol="AAA", direction="long",
                                    entry_price=100.0, shares=10, stop_loss=95.0, target=110.0)
-    await _add_bars(session, tid, [(101, 94)], start=date.today())  # low pierces stop
+    await _add_bars(session, tid, [(101, 94)], start=_today())  # low pierces stop
     closed = await svc.resolve_open_trades(session)
     assert closed == 1
     await session.refresh(trade)
@@ -192,7 +202,7 @@ async def test_resolve_leaves_open_when_neither_hit(session):
     tid = await _seed(session, "AAA", close=100.0)
     await svc.create_trade(session, 1, symbol="AAA", direction="long",
                            entry_price=100.0, shares=10, stop_loss=95.0, target=110.0)
-    await _add_bars(session, tid, [(103, 98), (104, 99)], start=date.today())  # range-bound
+    await _add_bars(session, tid, [(103, 98), (104, 99)], start=_today())  # range-bound
     closed = await svc.resolve_open_trades(session)
     assert closed == 0
     rows = await svc.list_trades(session, 1, status="open")
@@ -217,7 +227,7 @@ async def _add_open_trade(session, ticker_id: int, direction: str, *, entry: flo
 
 async def test_alpha_long_open(session):
     tid = await _seed(session, "AAA", close=110.0)  # current price 110 → +10% on a 100 entry
-    today = date.today()
+    today = _today()
     await _seed_benchmark(session, {today - timedelta(days=10): 400.0, today: 420.0})  # SPY +5%
     await _add_open_trade(session, tid, "long", entry=100.0, shares=10, days_ago=10)
 
@@ -229,7 +239,7 @@ async def test_alpha_long_open(session):
 
 async def test_alpha_short_and_missing_benchmark(session):
     tid = await _seed(session, "BBB", close=90.0)  # price fell to 90 → short +10%
-    today = date.today()
+    today = _today()
     await _add_open_trade(session, tid, "short", entry=100.0, shares=4, days_ago=10)
 
     # No benchmark data yet → alpha unset, not an error.
@@ -389,7 +399,7 @@ async def test_resolve_time_mode_closes_at_horizon(session):
     tid = await _seed(session, "AAA", close=100.0)
     trade = await svc.create_trade(session, 1, symbol="AAA", direction="long",
                                    entry_price=100.0, shares=10, stop_loss=95.0, target=200.0)
-    await _add_bars(session, tid, [(103, 101), (105, 102)], start=date.today())
+    await _add_bars(session, tid, [(103, 101), (105, 102)], start=_today())
     assert await svc.resolve_open_trades(session) == 1
     await session.refresh(trade)
     assert trade.status == "closed"
@@ -402,7 +412,7 @@ async def test_resolve_time_mode_stop_still_governs(session):
     tid = await _seed(session, "AAA", close=100.0)
     trade = await svc.create_trade(session, 1, symbol="AAA", direction="long",
                                    entry_price=100.0, shares=10, stop_loss=95.0, target=200.0)
-    await _add_bars(session, tid, [(101, 94)], start=date.today())  # low pierces the stop
+    await _add_bars(session, tid, [(101, 94)], start=_today())  # low pierces the stop
     assert await svc.resolve_open_trades(session) == 1
     await session.refresh(trade)
     assert trade.close_reason == "stop"
@@ -413,7 +423,7 @@ async def test_resolve_trailing_closes_with_reason(session):
     await svc.set_exit_policy(session, mode="trailing", trailing_pct=12.0)
     tid = await _seed(session, "AAA", close=100.0)
     await _add_open_trade(session, tid, "long", entry=100.0, shares=10, days_ago=10)
-    await _add_bars(session, tid, [(120, 110), (130, 100)], start=date.today())  # run up, pull back
+    await _add_bars(session, tid, [(120, 110), (130, 100)], start=_today())  # run up, pull back
     assert await svc.resolve_open_trades(session) == 1
     closed = await svc.list_trades(session, 1, status="closed")
     assert closed[0]["close_reason"] == "trailing"
@@ -424,7 +434,7 @@ async def test_resolve_atr_trailing_closes_with_reason(session, monkeypatch):
     await svc.set_exit_policy(session, mode="atr_trailing", atr_multiplier=3.0)
     tid = await _seed(session, "AAA", close=100.0)
     await _add_open_trade(session, tid, "long", entry=100.0, shares=10, days_ago=10)
-    await _add_bars(session, tid, [(121, 114), (107, 101)], start=date.today())
+    await _add_bars(session, tid, [(121, 114), (107, 101)], start=_today())
     assert await svc.resolve_open_trades(session) == 1
     closed = await svc.list_trades(session, 1, status="closed")
     assert closed[0]["close_reason"] == "trailing"
@@ -443,7 +453,7 @@ async def test_list_open_exposes_trailing_stop(session):
     await svc.set_exit_policy(session, mode="trailing", trailing_pct=12.0)
     tid = await _seed(session, "AAA", close=120.0)
     await _add_open_trade(session, tid, "long", entry=100.0, shares=10, days_ago=10)
-    await _add_bars(session, tid, [(125, 118)], start=date.today())  # peak 125
+    await _add_bars(session, tid, [(125, 118)], start=_today())  # peak 125
     row = (await svc.list_trades(session, 1, status="open"))[0]
     assert row["trailing_stop"] == pytest.approx(110.0)  # 125 * (1 - 0.12)
     assert row["trailing_distance_pct"] is not None
@@ -454,7 +464,7 @@ async def test_list_open_exposes_atr_trailing_stop(session, monkeypatch):
     await svc.set_exit_policy(session, mode="atr_trailing", atr_multiplier=3.0)
     tid = await _seed(session, "AAA", close=120.0)
     await _add_open_trade(session, tid, "long", entry=100.0, shares=10, days_ago=10)
-    await _add_bars(session, tid, [(125, 118)], start=date.today())
+    await _add_bars(session, tid, [(125, 118)], start=_today())
     row = (await svc.list_trades(session, 1, status="open"))[0]
     assert row["trailing_stop"] == pytest.approx(106.5)  # latest close 121.5 - 3 * 5
     assert row["trailing_distance_pct"] is not None

tests/unit/test_performance_summary.py

@@ -0,0 +1,101 @@
+"""Performance comparison: per-book series, R-multiples, and the start-date window."""
+
+from __future__ import annotations
+
+from datetime import date, datetime, timedelta, timezone
+from types import SimpleNamespace
+
+import pytest
+
+from app.services import paper_trade_service as pts
+from app.services.trade_policy import MANUAL_BOOK, SHADOW_BOOK
+
+
+def _trade(*, book, entry=100.0, stop=95.0, close=None, shares=10.0, opened_days_ago=5):
+    now = datetime.now(timezone.utc)
+    return SimpleNamespace(
+        ticker_id=1,
+        direction="long",
+        entry_price=entry,
+        stop_loss=stop,
+        shares=shares,
+        book=book,
+        status="closed" if close is not None else "open",
+        close_price=close,
+        opened_at=now - timedelta(days=opened_days_ago),
+        closed_at=now if close is not None else None,
+    )
+
+
+class TestRMultiple:
+    def test_winner_measured_in_units_of_initial_risk(self):
+        # Entry 100, stop 95 → 5 of risk. Exit 115 → +15 → +3R.
+        trade = _trade(book=SHADOW_BOOK, close=115.0)
+        assert pts.trade_r_multiple(trade, None) == pytest.approx(3.0)
+
+    def test_full_stop_is_minus_one_r(self):
+        trade = _trade(book=SHADOW_BOOK, close=95.0)
+        assert pts.trade_r_multiple(trade, None) == pytest.approx(-1.0)
+
+    def test_open_trade_marks_to_the_latest_close(self):
+        trade = _trade(book=SHADOW_BOOK)
+        assert pts.trade_r_multiple(trade, 110.0) == pytest.approx(2.0)
+
+    def test_no_risk_distance_has_no_r(self):
+        trade = _trade(book=SHADOW_BOOK, entry=100.0, stop=100.0, close=120.0)
+        assert pts.trade_r_multiple(trade, None) is None
+
+
+class TestBookStats:
+    def test_r_is_independent_of_position_size(self):
+        """The whole point: a 10-share and a 1000-share book compare equally."""
+        small = pts.book_stats([_trade(book=SHADOW_BOOK, close=115.0, shares=10)], {})
+        large = pts.book_stats([_trade(book=MANUAL_BOOK, close=115.0, shares=1000)], {})
+        assert small["total_r"] == large["total_r"] == pytest.approx(3.0)
+
+    def test_counts_and_win_rate(self):
+        trades = [
+            _trade(book=SHADOW_BOOK, close=115.0),
+            _trade(book=SHADOW_BOOK, close=95.0),
+            _trade(book=SHADOW_BOOK),
+        ]
+        stats = pts.book_stats(trades, {1: 110.0})
+        assert stats["trades"] == 3
+        assert stats["closed"] == 2
+        assert stats["open"] == 1
+        # +3R, -1R, +2R marked → 2 of 3 positive.
+        assert stats["win_rate"] == pytest.approx(66.7)
+        assert stats["total_r"] == pytest.approx(4.0)
+
+
+class TestPerformanceStartDate:
+    @pytest.fixture
+    async def session(self):
+        from tests.conftest import _test_session_factory
+
+        async with _test_session_factory() as session:
+            yield session
+
+    @pytest.mark.asyncio
+    async def test_unset_means_all_history(self, session):
+        assert await pts.get_performance_start(session) is None
+
+    @pytest.mark.asyncio
+    async def test_reads_an_iso_date(self, session):
+        await pts.settings_store.upsert_setting(
+            session, pts.KEY_PERFORMANCE_START, "2026-07-20"
+        )
+        assert await pts.get_performance_start(session) == date(2026, 7, 20)
+
+    @pytest.mark.asyncio
+    async def test_garbage_falls_back_to_all_history(self, session):
+        """A bad setting must not blank the whole performance card."""
+        await pts.settings_store.upsert_setting(
+            session, pts.KEY_PERFORMANCE_START, "not-a-date"
+        )
+        assert await pts.get_performance_start(session) is None
+
+    @pytest.mark.asyncio
+    async def test_empty_string_means_all_history(self, session):
+        await pts.settings_store.upsert_setting(session, pts.KEY_PERFORMANCE_START, "")
+        assert await pts.get_performance_start(session) is None

tests/unit/test_pipeline_run.py

@@ -0,0 +1,72 @@
+"""Pipeline run-id context and the scanner stamping it into scan markers."""
+
+from __future__ import annotations
+
+import asyncio
+
+import pytest
+
+from app.services import pipeline_run
+
+
+def test_no_run_id_by_default():
+    assert pipeline_run.current() is None
+
+
+def test_bind_and_release_restore_previous():
+    assert pipeline_run.current() is None
+    token = pipeline_run.bind("run-1")
+    try:
+        assert pipeline_run.current() == "run-1"
+    finally:
+        pipeline_run.release(token)
+    assert pipeline_run.current() is None
+
+
+def test_new_run_ids_are_unique():
+    ids = {pipeline_run.new_run_id() for _ in range(100)}
+    assert len(ids) == 100
+
+
+@pytest.mark.asyncio
+async def test_run_id_propagates_to_awaited_coroutines():
+    """The scan and shadow steps are awaited inside the pipeline's task, so they
+    must observe the id the pipeline bound."""
+
+    async def step() -> str | None:
+        return pipeline_run.current()
+
+    token = pipeline_run.bind("run-42")
+    try:
+        assert await step() == "run-42"
+    finally:
+        pipeline_run.release(token)
+
+
+@pytest.mark.asyncio
+async def test_run_id_does_not_leak_into_an_independent_task():
+    """A manual scan is a separate APScheduler job, started independently of the
+    pipeline. Modelled here as a task created before the bind: it captures its
+    own context and never observes the id the pipeline binds afterwards."""
+    seen: dict[str, str | None] = {}
+    manual_started = asyncio.Event()
+    let_manual_finish = asyncio.Event()
+
+    async def manual_job() -> None:
+        manual_started.set()
+        await let_manual_finish.wait()
+        seen["manual"] = pipeline_run.current()
+
+    # Created with no id in context — the manual job predates the pipeline bind.
+    task = asyncio.create_task(manual_job())
+    await manual_started.wait()
+
+    token = pipeline_run.bind("pipeline")
+    try:
+        assert pipeline_run.current() == "pipeline"
+        let_manual_finish.set()
+        await task
+    finally:
+        pipeline_run.release(token)
+
+    assert seen["manual"] is None

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

tests/unit/test_scheduler.py

@@ -107,6 +107,7 @@ class TestConfigureScheduler:
             "sentiment_collector",
             "fundamental_collector",
             "rr_scanner",
+            "shadow_book",
             "ticker_universe_sync",
             "outcome_evaluator",
             "alerts",
@@ -143,5 +144,6 @@ class TestConfigureScheduler:
             "outcome_evaluator",
             "rr_scanner",
             "sentiment_collector",
+            "shadow_book",
             "ticker_universe_sync",
         ])

tests/unit/test_shadow_book_service.py

@@ -0,0 +1,431 @@
+"""Shadow book selection, sizing and book isolation.
+
+The shadow book only has evidentiary value if it selects what the backtest
+would select: top-ranked qualified setups, up to capacity, skipping held names
+and post-stop gate-reset lockouts. These tests pin that contract.
+"""
+
+from __future__ import annotations
+
+import json
+from datetime import datetime, timedelta, timezone
+
+import pytest
+
+from app.models.paper_trade import PaperTrade
+from app.models.ticker import Ticker
+from app.models.trade_setup import TradeSetup
+from app.models.user import User
+from app.services import shadow_book_service
+from app.services.trade_policy import MANUAL_BOOK, SHADOW_BOOK, get_reentry_gate_locks
+
+
+@pytest.fixture
+async def session():
+    from tests.conftest import _test_session_factory
+
+    async with _test_session_factory() as session:
+        yield session
+
+
+# Floors the gate applies; every setup below clears them so tests exercise
+# ranking rather than qualification.
+_CONFIG = {
+    "min_rr": 2.0,
+    "min_confidence": 0.0,
+    "min_momentum_percentile": 80.0,
+    "exclude_neutral": False,
+}
+
+
+async def _seed(session, symbols: list[str]) -> dict[str, int]:
+    session.add(User(id=1, username="owner", password_hash="x"))
+    ids: dict[str, int] = {}
+    for i, symbol in enumerate(symbols, start=1):
+        ticker = Ticker(id=i, symbol=symbol, name=symbol)
+        session.add(ticker)
+        ids[symbol] = i
+    await session.commit()
+    return ids
+
+
+def _setup(
+    ticker_id: int,
+    *,
+    rank: float,
+    detected: datetime,
+    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
+    return TradeSetup(
+        ticker_id=ticker_id,
+        direction=direction,
+        entry_price=entry,
+        stop_loss=stop,
+        target=target,
+        rr_ratio=3.0,
+        composite_score=70.0,
+        confidence_score=70.0,
+        detected_at=detected,
+        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}]
+        ),
+    )
+
+
+async def _mark_scan(
+    session,
+    *,
+    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.
+
+    ``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 or datetime.now(timezone.utc)
+    await shadow_book_service.settings_store.upsert_setting(
+        session, rr.KEY_LAST_SCAN_COMPLETED, completed.isoformat()
+    )
+    await shadow_book_service.settings_store.upsert_setting(
+        session, rr.KEY_LAST_SCAN_RUN_ID, run_id
+    )
+    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)
+        assert shares == pytest.approx(200.0)  # $1,000 risk / $5 per share
+
+    def test_zero_risk_distance_takes_no_position(self):
+        assert shadow_book_service.position_shares(100_000, 1.0, 100.0, 100.0) == 0.0
+
+    def test_tight_stop_is_capped_at_the_notional_limit(self):
+        """Without the cap, 1% risk on a $0.50 stop is a 2x-equity position."""
+        shares = shadow_book_service.position_shares(100_000, 1.0, 100.0, 99.5)
+        # Risk sizing alone wants 2,000 shares ($200k); the 20% cap allows 200.
+        assert shares == pytest.approx(200.0)
+        assert shares * 100.0 <= 100_000 * shadow_book_service.NOTIONAL_CAP
+
+    def test_cannot_spend_cash_it_does_not_have(self):
+        shares = shadow_book_service.position_shares(
+            100_000, 1.0, 100.0, 95.0, cash_available=5_000
+        )
+        assert shares == pytest.approx(50.0)
+
+    def test_no_cash_means_no_position(self):
+        assert (
+            shadow_book_service.position_shares(
+                100_000, 1.0, 100.0, 95.0, cash_available=0
+            )
+            == 0.0
+        )
+
+
+class TestSelection:
+    @pytest.mark.asyncio
+    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),
+                _setup(ids["BBB"], rank=0.90, detected=now),
+                _setup(ids["CCC"], rank=0.50, detected=now),
+            ]
+        )
+        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
+        )
+
+        assert summary["opened"] == 2
+        # Highest strategy_rank first — the backtest's ordering key.
+        assert summary["symbols"] == [ids["BBB"], ids["CCC"]]
+
+    @pytest.mark.asyncio
+    async def test_skips_names_already_held(self, session):
+        ids = await _seed(session, ["AAA", "BBB"])
+        now = datetime.now(timezone.utc)
+        session.add_all(
+            [_setup(ids["AAA"], rank=0.9, detected=now), _setup(ids["BBB"], rank=0.5, detected=now)]
+        )
+        session.add(
+            PaperTrade(
+                user_id=1, ticker_id=ids["AAA"], direction="long", entry_price=100.0,
+                shares=10.0, stop_loss=95.0, target=115.0, status="open",
+                opened_at=now, book=SHADOW_BOOK,
+            )
+        )
+        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
+        )
+
+        assert summary["skipped_held"] == 1
+        assert summary["symbols"] == [ids["BBB"]]
+
+    @pytest.mark.asyncio
+    async def test_respects_post_stop_gate_lock(self, session):
+        ids = await _seed(session, ["AAA"])
+        now = datetime.now(timezone.utc)
+        session.add(_setup(ids["AAA"], rank=0.9, detected=now))
+        # Stopped out and never requalified — locked out of re-entry.
+        session.add(
+            PaperTrade(
+                user_id=1, ticker_id=ids["AAA"], direction="long", entry_price=100.0,
+                shares=10.0, stop_loss=95.0, target=115.0, status="closed",
+                opened_at=now - timedelta(days=5), closed_at=now - timedelta(days=1),
+                close_price=95.0, close_reason="stop", book=SHADOW_BOOK,
+            )
+        )
+        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
+        )
+
+        assert summary["opened"] == 0
+        assert summary["skipped_locked"] == 1
+
+
+class TestScanFreshness:
+    @pytest.mark.asyncio
+    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"])
+        session.add(_setup(ids["AAA"], rank=0.9, detected=datetime.now(timezone.utc)))
+        await session.commit()
+
+        summary = await shadow_book_service.open_shadow_positions(
+            session, activation_config=_CONFIG
+        )
+
+        assert summary["opened"] == 0
+
+    @pytest.mark.asyncio
+    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(_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"] == 0
+
+    @pytest.mark.asyncio
+    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)
+        session.add_all(
+            [
+                _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, completed=now, run_id="scan-run")
+
+        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)
+        # 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, completed=now, run_id="scan-run")
+
+        summary = await shadow_book_service.open_shadow_positions(
+            session, activation_config=_CONFIG
+        )
+
+        assert summary["opened"] == 0
+
+
+class TestPipelineScanBinding:
+    @pytest.mark.asyncio
+    async def test_pipeline_scan_run_id_match_is_accepted(self, session):
+        """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, scan_run_id="pipeline-A"))
+        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["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):
+        """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, 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, completed=now, run_id="manual-999")
+
+        summary = await shadow_book_service.open_shadow_positions(
+            session, activation_config=_CONFIG, expected_run_id="pipeline-A"
+        )
+
+        assert summary["opened"] == 0
+
+    @pytest.mark.asyncio
+    async def test_pipeline_scan_failed_leaves_prior_run_id(self, session):
+        """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, scan_run_id="yesterday"))
+        await session.commit()
+        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"
+        )
+
+        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:
+    @pytest.mark.asyncio
+    async def test_gate_locks_do_not_leak_between_books(self, session):
+        """A manual stop must not lock the shadow book out of the same name."""
+        ids = await _seed(session, ["AAA"])
+        now = datetime.now(timezone.utc)
+        session.add(
+            PaperTrade(
+                user_id=1, ticker_id=ids["AAA"], direction="long", entry_price=100.0,
+                shares=10.0, stop_loss=95.0, target=115.0, status="closed",
+                opened_at=now - timedelta(days=5), closed_at=now - timedelta(days=1),
+                close_price=95.0, close_reason="stop", book=MANUAL_BOOK,
+            )
+        )
+        await session.commit()
+
+        assert ids["AAA"] in await get_reentry_gate_locks(session, book=MANUAL_BOOK)
+        assert ids["AAA"] not in await get_reentry_gate_locks(session, book=SHADOW_BOOK)
+
+    @pytest.mark.asyncio
+    async def test_shadow_equity_ignores_manual_pnl(self, session):
+        ids = await _seed(session, ["AAA"])
+        now = datetime.now(timezone.utc)
+        session.add(
+            PaperTrade(
+                user_id=1, ticker_id=ids["AAA"], direction="long", entry_price=100.0,
+                shares=100.0, stop_loss=95.0, target=115.0, status="closed",
+                opened_at=now - timedelta(days=5), closed_at=now,
+                close_price=150.0, close_reason="trailing", book=MANUAL_BOOK,
+            )
+        )
+        await session.commit()
+
+        equity, cash = await shadow_book_service.equity_and_cash(session, 100_000.0, [])
+
+        assert equity == 100_000.0
+        assert cash == 100_000.0