feat: shadow book + shadow-vs-manual performance comparison

The manual paper book only contains trades taken by hand, inside a 20
minute window, on days someone was available. The backtest that validated
this strategy auto-takes the top-ranked qualified setups up to capacity
every session. The forward record was therefore measuring strategy plus
discretion plus availability -- and degrading silently on busy days.

The shadow book closes that gap: it mirrors the backtest's selection rule
(top strategy_rank qualified, up to capacity, 1% fixed-fractional risk)
and shares the manual book's exit policy, so the only difference between
the two books is which setups get taken. Selection ordering reuses the
strategy_rank the scanner already stores rather than recomputing it, so
the two cannot drift apart. It runs as a near-close pipeline step right
after the scan, marking entries at the same prices a human would see.

Gate-reset re-entry state is now scoped per book -- the books diverge as
soon as their entries differ, and each must see only its own stops.

Performance view rewritten around the comparison:
  - three series (shadow, manual, SPY) from a new endpoint
  - SPY changes from a per-trade cost-basis counterfactual to plain
    buy-and-hold %, since one line has to serve two books
  - headline stats are R-multiples, not currency: the books size
    differently, so only R compares across them
  - configurable start date, because the strategy has been revised
    repeatedly and pre-cutover trades ran under rules that no longer
    exist

Migration 024 also repairs the numeric weekday crons written by 023,
rewriting only rows still holding the broken form so hand-corrected
settings survive. Its literals are inlined because bound parameters
render as NULL under 'alembic upgrade --sql'.

The shadow book is opt-in and writes nothing until enabled. Verify its
first selections match a backtest of that day's cross-section before
trusting any point on the curve.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

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/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,17 @@ 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)
+    )
+
+
 @router.put("/paper-trades/exit-policy", response_model=APIEnvelope)
 async def write_exit_policy(
     body: ExitPolicyUpdate,

app/scheduler.py

@@ -33,7 +33,13 @@ 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,
+    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 +616,54 @@ 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.
+
+    Opt-in (``shadow_book_enabled``) because it writes live trades.
+    """
+    job_name = "shadow_book"
+    _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
+            )
+            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 +1292,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"),
 ]
 

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/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
@@ -690,6 +693,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 +777,112 @@ 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) -> 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)
+    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/shadow_book_service.py

@@ -0,0 +1,231 @@
+"""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, timezone
+
+from sqlalchemy import func, 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
+
+
+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 current_equity(db: AsyncSession, start_equity: float) -> float:
+    """Start equity plus realized P&L of closed shadow trades.
+
+    Open positions are deliberately excluded: sizing off marked-to-market equity
+    would let an unrealized gain inflate the next position, which is not what the
+    backtest does.
+    """
+    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
+    return start_equity + realized
+
+
+def position_shares(equity: float, risk_pct: float, entry: float, stop: float) -> float:
+    """Fixed-fractional sizing: risk ``risk_pct`` of equity down to the stop."""
+    risk_per_share = abs(entry - stop)
+    if risk_per_share <= 0 or equity <= 0:
+        return 0.0
+    return (equity * risk_pct / 100.0) / risk_per_share
+
+
+async def _open_ticker_ids(db: AsyncSession) -> set[int]:
+    result = await db.execute(
+        select(PaperTrade.ticker_id).where(
+            PaperTrade.book == SHADOW_BOOK, PaperTrade.status == "open"
+        )
+    )
+    return {row[0] for row in result.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 _todays_qualified_setups(db: AsyncSession, config: dict) -> list[TradeSetup]:
+    """Latest setup per ticker from the most recent scan, gate-qualified.
+
+    Ordered by ``strategy_rank`` descending — the ordering the backtest selects
+    on. Setups without a rank sort last; they cannot be compared to ranked ones.
+    """
+    latest_scan = await db.execute(select(func.max(TradeSetup.detected_at)))
+    newest = latest_scan.scalar()
+    if newest is None:
+        return []
+
+    # Everything written by the same scan run (same calendar day, NY-agnostic:
+    # one qualifying scan per day is a hard invariant of the schedule).
+    result = await db.execute(
+        select(TradeSetup).where(
+            func.date(TradeSetup.detected_at) == func.date(newest),
+        )
+    )
+    setups = [s for s in result.scalars() if setup_qualifies(s, config)]
+    setups.sort(
+        key=lambda s: (
+            s.strategy_rank if s.strategy_rank is not None else float("-inf")
+        ),
+        reverse=True,
+    )
+    return setups
+
+
+async def open_shadow_positions(
+    db: AsyncSession,
+    *,
+    activation_config: dict,
+    opened_at: datetime | 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.
+    """
+    summary = {"opened": 0, "skipped_held": 0, "skipped_locked": 0, "symbols": []}
+    config = await get_config(db)
+
+    held = await _open_ticker_ids(db)
+    free_slots = config["capacity"] - len(held)
+    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 = await current_equity(db, config["start_equity"])
+    timestamp = opened_at or datetime.now(timezone.utc)
+
+    for setup in await _todays_qualified_setups(db, activation_config):
+        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)
+        if shares <= 0:
+            continue
+
+        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)