feat: trailing-stop auto-exit for paper trades + close/digest alerts

Applies the backtest-validated trailing stop to live paper trading, and surfaces
it transparently.

Exit (A):
- New paper-trade exit policy (paper_exit_mode=trailing, paper_trailing_pct=12),
  tunable in Admin → Paper-Trade Exit. resolve_open_trades runs a trailing stop
  (initial stop as floor, ratchets up from the peak; target ignored — the
  validated rule) and records close_reason (trailing|stop|target|manual; +migration
  013).
- list_trades enriches open trades with the live trailing-stop level + distance %.
  Open Trades panel shows the active tactic and a Trail Stop column.

Alerts (B):
- Daily digest now lists open trades with unrealized gain, trailing stop, and how
  far away it is.
- New "trade closed" alert: one summary per auto-close (trailing/target/stop, not
  manual) — direction, reason, days held, P&L abs+%/R — covering wins AND
  stop-loss losses. Deduped by trade id; toggle in Admin alerts.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

app/services/alert_service.py

@@ -26,6 +26,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
 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.models.score import CompositeScore
 from app.models.sr_level import SRLevel
 from app.models.ticker import Ticker
@@ -47,6 +48,7 @@ KEY_SR = "alerts_sr_proximity_enabled"
 KEY_SCORE_DROP = "alerts_score_drop_enabled"
 KEY_DIGEST = "alerts_digest_enabled"
 KEY_REGIME_QUADRANT = "alerts_regime_quadrant_enabled"
+KEY_TRADE_CLOSED = "alerts_trade_closed_enabled"
 
 _BOOL_DEFAULTS = {
     KEY_ENABLED: False,
@@ -55,8 +57,15 @@ _BOOL_DEFAULTS = {
     KEY_SCORE_DROP: True,
     KEY_DIGEST: True,
     KEY_REGIME_QUADRANT: True,
+    KEY_TRADE_CLOSED: True,
 }
 
+# Paper-trade auto-close alert: catch every close at least once (the job runs
+# hourly), then never re-send the same trade (a huge cooldown ≈ once-per-trade).
+CLOSED_LOOKBACK_HOURS = 26
+CLOSED_ALERT_COOLDOWN_HOURS = 24 * 365 * 5
+TRADE_CLOSED_TYPE = "trade_closed"
+
 # Tunables (kept as constants for now; promote to settings if needed)
 SR_PROXIMITY_PCT = 2.0      # within this % of a strong zone → alert
 SR_MIN_STRENGTH = 60        # only strong zones are alert-worthy
@@ -90,7 +99,7 @@ def _as_bool(value: str | None, default: bool) -> bool:
 
 
 async def _resolve(db: AsyncSession) -> dict:
-    keys = [KEY_ENABLED, KEY_TOKEN, KEY_CHAT_ID, KEY_QUALIFIED, KEY_SR, KEY_SCORE_DROP, KEY_DIGEST, KEY_REGIME_QUADRANT]
+    keys = [KEY_ENABLED, KEY_TOKEN, KEY_CHAT_ID, KEY_QUALIFIED, KEY_SR, KEY_SCORE_DROP, KEY_DIGEST, KEY_REGIME_QUADRANT, KEY_TRADE_CLOSED]
     stored = await settings_store.get_map(db, keys)
 
     db_token = (stored.get(KEY_TOKEN) or "").strip()
@@ -113,6 +122,7 @@ async def _resolve(db: AsyncSession) -> dict:
         "score_drop": _as_bool(stored.get(KEY_SCORE_DROP), _BOOL_DEFAULTS[KEY_SCORE_DROP]),
         "digest": _as_bool(stored.get(KEY_DIGEST), _BOOL_DEFAULTS[KEY_DIGEST]),
         "regime_quadrant": _as_bool(stored.get(KEY_REGIME_QUADRANT), _BOOL_DEFAULTS[KEY_REGIME_QUADRANT]),
+        "trade_closed": _as_bool(stored.get(KEY_TRADE_CLOSED), _BOOL_DEFAULTS[KEY_TRADE_CLOSED]),
     }
 
 
@@ -129,6 +139,7 @@ async def get_alert_config(db: AsyncSession) -> dict:
         "score_drop_enabled": r["score_drop"],
         "digest_enabled": r["digest"],
         "regime_quadrant_enabled": r["regime_quadrant"],
+        "trade_closed_enabled": r["trade_closed"],
     }
 
 
@@ -143,6 +154,7 @@ async def update_alert_config(
     score_drop_enabled: bool | None = None,
     digest_enabled: bool | None = None,
     regime_quadrant_enabled: bool | None = None,
+    trade_closed_enabled: bool | None = None,
 ) -> dict:
     """Persist config. An empty/omitted bot_token leaves the stored token intact."""
     bool_updates = {
@@ -152,6 +164,7 @@ async def update_alert_config(
         KEY_SCORE_DROP: score_drop_enabled,
         KEY_DIGEST: digest_enabled,
         KEY_REGIME_QUADRANT: regime_quadrant_enabled,
+        KEY_TRADE_CLOSED: trade_closed_enabled,
     }
     for key, val in bool_updates.items():
         if val is not None:
@@ -376,9 +389,81 @@ async def _collect_digest(db: AsyncSession) -> tuple[str, str] | None:
             )
     else:
         lines.append("No qualified setups today.")
+
+    # Open paper trades: unrealized gain + the live trailing stop and how far away.
+    from app.services import paper_trade_service
+
+    open_trades = await paper_trade_service.list_trades(db, status="open")
+    if open_trades:
+        lines.append("")
+        lines.append(f"💼 <b>{len(open_trades)} open trade(s):</b>")
+        for t in open_trades:
+            entry = t["entry_price"]
+            cur = t.get("current_price")
+            sign = 1.0 if t["direction"] == "long" else -1.0
+            if cur and entry:
+                gain_pct = (cur - entry) / entry * 100.0 * sign
+                gain_usd = (cur - entry) * t["shares"] * sign
+                gain = f"{gain_pct:+.1f}% ({'+' if gain_usd >= 0 else '−'}${abs(gain_usd):.0f})"
+            else:
+                gain = "n/a"
+            ts = t.get("trailing_stop")
+            if ts is not None:
+                dist = t.get("trailing_distance_pct")
+                stop_txt = f"trail {ts:.2f}" + (f" ({dist:.1f}% away)" if dist is not None else "")
+            else:
+                stop_txt = f"stop {t['stop_loss']:.2f}"
+            lines.append(f"• {t['symbol']} {t['direction'].upper()} {gain} · {stop_txt}")
+
     return key, "\n".join(lines)
 
 
+# ---------------------------------------------------------------------------
+# Paper-trade close trigger (one summary per auto-closed trade)
+# ---------------------------------------------------------------------------
+
+def _format_closed_trade(trade: PaperTrade, symbol: str) -> str:
+    sign = 1.0 if trade.direction == "long" else -1.0
+    entry = trade.entry_price
+    exit_price = trade.close_price if trade.close_price is not None else entry
+    per_share = (exit_price - entry) * sign
+    pnl_pct = (per_share / entry * 100.0) if entry else 0.0
+    pnl_usd = per_share * trade.shares
+    risk = abs(entry - trade.stop_loss)
+    r_mult = (per_share / risk) if risk > 0 else None
+    win = per_share > 0
+    money = f"{'+' if pnl_usd >= 0 else '−'}${abs(pnl_usd):.2f}"
+    r_txt = f" · {r_mult:+.2f}R" if r_mult is not None else ""
+    days = (trade.closed_at - trade.opened_at).days if (trade.closed_at and trade.opened_at) else None
+    held = f" · held {days}d" if days is not None else ""
+    reason = {"trailing": "trailing stop", "stop": "stop-loss", "target": "target"}.get(
+        trade.close_reason or "", trade.close_reason or "closed"
+    )
+    return (
+        f"{'✅' if win else '🔴'} <b>{symbol} {trade.direction.upper()} closed</b> ({reason})\n"
+        f"{pnl_pct:+.1f}% · {money}{r_txt}{held}\n"
+        f"{entry:.2f} → {exit_price:.2f}"
+    )
+
+
+async def _collect_closed_trades(db: AsyncSession) -> list[tuple[str, str]]:
+    """One alert per auto-closed paper trade (trailing / stop / target). Manual
+    closes are skipped — you already know about those. Dedup is by trade id."""
+    cutoff = datetime.now(timezone.utc) - timedelta(hours=CLOSED_LOOKBACK_HOURS)
+    result = await db.execute(
+        select(PaperTrade, Ticker.symbol)
+        .join(Ticker, PaperTrade.ticker_id == Ticker.id)
+        .where(
+            PaperTrade.status == "closed",
+            PaperTrade.closed_at.is_not(None),
+            PaperTrade.closed_at > cutoff,
+            PaperTrade.close_reason.in_(("trailing", "stop", "target")),
+        )
+        .order_by(PaperTrade.closed_at.desc())
+    )
+    return [(str(trade.id), _format_closed_trade(trade, symbol)) for trade, symbol in result.all()]
+
+
 # ---------------------------------------------------------------------------
 # Regime quadrant-change trigger (hysteresis + cooldown)
 # ---------------------------------------------------------------------------
@@ -500,6 +585,11 @@ async def dispatch_alerts(db: AsyncSession) -> dict:
         for key, text in await _collect_regime_quadrant(db):
             outgoing.append((QUAD_TYPE, key, text))
 
+    if cfg["trade_closed"]:
+        for key, text in await _collect_closed_trades(db):
+            if not await _recently_alerted(db, TRADE_CLOSED_TYPE, key, cooldown_hours=CLOSED_ALERT_COOLDOWN_HOURS):
+                outgoing.append((TRADE_CLOSED_TYPE, key, text))
+
     sent = 0
     if outgoing:
         async with httpx.AsyncClient(timeout=15) as client:

app/services/paper_trade_service.py

@@ -11,7 +11,7 @@ from app.exceptions import NotFoundError, ValidationError
 from app.models.ohlcv import OHLCVRecord
 from app.models.paper_trade import PaperTrade
 from app.models.ticker import Ticker
-from app.services import benchmark_service
+from app.services import benchmark_service, settings_store
 from app.services.outcome_service import (
     OUTCOME_AMBIGUOUS,
     OUTCOME_STOP_HIT,
@@ -20,6 +20,45 @@ from app.services.outcome_service import (
     evaluate_setup_against_bars,
 )
 
+# Exit policy for OPEN paper trades (auto-close). "trailing" rides a trailing stop
+# (validated as the best exit in the backtest); "target" closes at the setup's
+# stop/target. Stored in SystemSetting so it's tunable + transparent in the UI.
+KEY_EXIT_MODE = "paper_exit_mode"
+KEY_TRAILING_PCT = "paper_trailing_pct"
+DEFAULT_EXIT_MODE = "trailing"
+DEFAULT_TRAILING_PCT = 12.0
+
+
+async def get_exit_policy(db: AsyncSession) -> dict:
+    """Active auto-exit policy: {'mode': 'trailing'|'target', 'trailing_pct': float}."""
+    mode = (await settings_store.get_value(db, KEY_EXIT_MODE, DEFAULT_EXIT_MODE)).strip().lower()
+    if mode not in ("trailing", "target"):
+        mode = DEFAULT_EXIT_MODE
+    raw = await settings_store.get_value(db, KEY_TRAILING_PCT, str(DEFAULT_TRAILING_PCT))
+    try:
+        pct = float(raw)
+    except (TypeError, ValueError):
+        pct = DEFAULT_TRAILING_PCT
+    pct = max(0.5, min(90.0, pct))
+    return {"mode": mode, "trailing_pct": pct}
+
+
+async def set_exit_policy(
+    db: AsyncSession, *, mode: str | None = None, trailing_pct: float | None = None
+) -> dict:
+    """Persist the auto-exit policy (admin). Validates inputs."""
+    if mode is not None:
+        mode = mode.strip().lower()
+        if mode not in ("trailing", "target"):
+            raise ValidationError("mode must be 'trailing' or 'target'")
+        await settings_store.upsert_setting(db, KEY_EXIT_MODE, mode)
+    if trailing_pct is not None:
+        if not 0.5 <= float(trailing_pct) <= 90.0:
+            raise ValidationError("trailing_pct must be between 0.5 and 90")
+        await settings_store.upsert_setting(db, KEY_TRAILING_PCT, str(float(trailing_pct)))
+    await db.commit()
+    return await get_exit_policy(db)
+
 
 async def _get_ticker(db: AsyncSession, symbol: str) -> Ticker:
     normalised = symbol.strip().upper()
@@ -51,6 +90,41 @@ async def _latest_closes(db: AsyncSession, ticker_ids: set[int]) -> dict[int, fl
     return {tid: float(close) for tid, close in result.all()}
 
 
+async def _max_high_after(db: AsyncSession, ticker_id: int, since: date) -> float | None:
+    """Highest high strictly after ``since`` — the running peak for a trailing stop."""
+    result = await db.execute(
+        select(func.max(OHLCVRecord.high)).where(
+            OHLCVRecord.ticker_id == ticker_id, OHLCVRecord.date > since
+        )
+    )
+    v = result.scalar()
+    return float(v) if v is not None else None
+
+
+def _trailing_close(
+    direction: str, entry: float, init_stop: float, trail_frac: float, bars: list[Bar]
+) -> tuple[float, date, str] | None:
+    """Walk post-entry bars; return (price, date, reason) when the trailing or initial
+    stop is hit, else None. The stop only ratchets up: max(init_stop, peak*(1-trail))
+    for a long. reason = 'trailing' once it's above the initial stop, else 'stop'."""
+    long = direction == "long"
+    peak = entry
+    for b in bars:
+        if long:
+            level = max(init_stop, peak * (1 - trail_frac))
+            if b.low <= level:
+                return level, b.date, ("trailing" if level > init_stop else "stop")
+            if b.high > peak:
+                peak = b.high
+        else:
+            level = min(init_stop, peak * (1 + trail_frac))
+            if b.high >= level:
+                return level, b.date, ("trailing" if level < init_stop else "stop")
+            if b.low < peak:
+                peak = b.low
+    return None
+
+
 async def create_trade(
     db: AsyncSession,
     user_id: int,
@@ -91,6 +165,7 @@ def _to_dict(
     symbol: str,
     current_price: float | None,
     benchmark_closes: dict[date, float] | None = None,
+    trailing: tuple[float, float | None] | None = None,
 ) -> dict:
     # For open trades, mark to market; for closed, the realized exit price.
     ref = current_price if trade.status == "open" else trade.close_price
@@ -130,19 +205,23 @@ def _to_dict(
         "benchmark_return_pct": benchmark_return,
         "alpha_pct": alpha_pct,
         "alpha_usd": alpha_usd,
+        "close_reason": trade.close_reason,
+        "trailing_stop": trailing[0] if trailing else None,
+        "trailing_distance_pct": trailing[1] if trailing else None,
     }
 
 
 async def list_trades(
     db: AsyncSession,
-    user_id: int,
+    user_id: int | None = None,
     status: str | None = None,
 ) -> list[dict]:
     stmt = (
         select(PaperTrade, Ticker.symbol)
         .join(Ticker, PaperTrade.ticker_id == Ticker.id)
-        .where(PaperTrade.user_id == user_id)
     )
+    if user_id is not None:  # None → all users (single-user app; used by the digest)
+        stmt = stmt.where(PaperTrade.user_id == user_id)
     if status is not None:
         stmt = stmt.where(PaperTrade.status == status)
     stmt = stmt.order_by(PaperTrade.opened_at.desc())
@@ -156,7 +235,32 @@ async def list_trades(
     # makes a provider call).
     benchmark_closes = await benchmark_service.load_benchmark_closes(db)
 
-    return [_to_dict(t, sym, prices.get(t.ticker_id), benchmark_closes) for t, sym in rows]
+    # Current trailing-stop level + distance for open trades (when trailing is active).
+    policy = await get_exit_policy(db)
+    trailing_info: dict[int, tuple[float, float | None]] = {}
+    if policy["mode"] == "trailing":
+        trail_frac = policy["trailing_pct"] / 100.0
+        for t, _ in rows:
+            if t.status != "open":
+                continue
+            max_high = await _max_high_after(db, t.ticker_id, t.opened_at.date())
+            peak = max(t.entry_price, max_high) if max_high is not None else t.entry_price
+            long = t.direction == "long"
+            level = (
+                max(t.stop_loss, peak * (1 - trail_frac))
+                if long
+                else min(t.stop_loss, peak * (1 + trail_frac))
+            )
+            cur = prices.get(t.ticker_id)
+            dist = None
+            if cur:
+                dist = ((cur - level) / cur * 100.0) if long else ((level - cur) / cur * 100.0)
+            trailing_info[t.id] = (level, dist)
+
+    return [
+        _to_dict(t, sym, prices.get(t.ticker_id), benchmark_closes, trailing_info.get(t.id))
+        for t, sym in rows
+    ]
 
 
 async def close_trade(
@@ -185,6 +289,7 @@ async def close_trade(
 
     trade.status = "closed"
     trade.close_price = float(close_price)
+    trade.close_reason = "manual"
     trade.closed_at = datetime.now(timezone.utc)
     await db.commit()
     await db.refresh(trade)
@@ -204,6 +309,10 @@ async def resolve_open_trades(db: AsyncSession) -> int:
     if not open_trades:
         return 0
 
+    policy = await get_exit_policy(db)
+    mode = policy["mode"]
+    trail_frac = policy["trailing_pct"] / 100.0
+
     closed = 0
     for trade in open_trades:
         bars_result = await db.execute(
@@ -218,21 +327,27 @@ async def resolve_open_trades(db: AsyncSession) -> int:
         if not bars:
             continue
 
-        # max_bars beyond the data so a still-open trade returns undecided (not "expired").
-        outcome, outcome_date = evaluate_setup_against_bars(
-            trade.direction, trade.stop_loss, trade.target, bars, max_bars=len(bars) + 1
-        )
-        if outcome == OUTCOME_TARGET_HIT:
-            trade.close_price = trade.target
-        elif outcome in (OUTCOME_STOP_HIT, OUTCOME_AMBIGUOUS):
-            trade.close_price = trade.stop_loss
+        if mode == "trailing":
+            hit = _trailing_close(trade.direction, trade.entry_price, trade.stop_loss, trail_frac, bars)
+            if hit is None:
+                continue  # neither the trailing nor the initial stop reached yet
+            close_price, close_date, reason = hit
         else:
-            continue
+            # max_bars beyond the data so a still-open trade returns undecided (not "expired").
+            outcome, outcome_date = evaluate_setup_against_bars(
+                trade.direction, trade.stop_loss, trade.target, bars, max_bars=len(bars) + 1
+            )
+            if outcome == OUTCOME_TARGET_HIT:
+                close_price, close_date, reason = trade.target, outcome_date, "target"
+            elif outcome in (OUTCOME_STOP_HIT, OUTCOME_AMBIGUOUS):
+                close_price, close_date, reason = trade.stop_loss, outcome_date, "stop"
+            else:
+                continue
 
         trade.status = "closed"
-        trade.closed_at = datetime.combine(
-            outcome_date, datetime.min.time(), tzinfo=timezone.utc
-        )
+        trade.close_price = float(close_price)
+        trade.close_reason = reason
+        trade.closed_at = datetime.combine(close_date, datetime.min.time(), tzinfo=timezone.utc)
         closed += 1
 
     if closed: