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

# Setups older than this mean the scan did not run in this pipeline pass.
MAX_SETUP_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 _todays_qualified_setups(
    db: AsyncSession, config: dict, *, now: datetime
) -> list[TradeSetup]:
    """Latest qualified setup per ticker from the scan that just ran.

    Freshness is a hard requirement, not a nicety: pipeline steps are allowed to
    fail independently, so if the scan is disabled or errors, the newest stored
    setups belong to a previous session. Trading those would enter yesterday's
    picks at yesterday's prices and quietly corrupt the record. Anything older
    than ``MAX_SETUP_AGE`` is treated as "no scan happened".

    Ordered by ``strategy_rank`` descending — the ordering the backtest selects
    on. Setups without a rank sort last; they cannot be compared to ranked ones.
    """
    cutoff = now - MAX_SETUP_AGE
    result = await db.execute(
        select(TradeSetup).where(TradeSetup.detected_at >= cutoff)
    )
    qualified = [s for s in result.scalars() if setup_qualifies(s, config)]

    # One setup per ticker — the most recent wins. A ticker can have several
    # rows in a scan (e.g. both directions); ranking over duplicates would let
    # one name occupy more than its share of the ordering.
    latest: dict[int, TradeSetup] = {}
    for setup in qualified:
        held = latest.get(setup.ticker_id)
        if held is None or setup.detected_at > held.detected_at:
            latest[setup.ticker_id] = setup

    return sorted(
        latest.values(),
        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,
) -> 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,
        "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)

    for setup in await _todays_qualified_setups(db, activation_config, now=timestamp):
        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()]