signal-platform/app/services/shadow_book_service.py

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