trumpsignal-backend/app/services/circuit_breaker.py

"""
Circuit breaker — autonomous "stop trading" trigger per wallet.

Two trip conditions, evaluated AFTER every closed trade:

  1. Daily drawdown: cumulative PnL across all trades closed today (UTC)
     drops below -CB_DAILY_DD_PCT × (subscription.position_size_usd × 10).
     Default cap: -5% of "expected 10-trade notional" — i.e. if the user's
     base bet is $20, the daily DD limit is -$10 net realized.

  2. Consecutive losses: last CB_CONSEC_LOSSES closed trades for this wallet
     all have pnl_usd < 0. Default = 3.

When tripped:
  - manual_window_until is nulled (bot stops trading via manual override)
  - circuit_breaker_tripped_at = now
  - circuit_breaker_reason = "daily_dd" | "consecutive_losses"
  - WS broadcast a 'circuit_breaker_tripped' event so the UI lights up red

Gate: `is_tripped()` is checked by _execute_for_subscriber BEFORE opening any
new trade. A tripped wallet stays blocked for CB_LOCKOUT_HOURS regardless of
schedule, so a buggy scanner or a black-swan move can't drain the account
overnight.

User unblock path:
  POST /api/user/{wallet}/manual-window with any hours > 0 clears the trip.
  This is a deliberate human-in-the-loop — the user must consciously re-arm.
"""

from __future__ import annotations

import logging
from datetime import datetime, timedelta, timezone
from typing import Optional

from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession

from app.models import BotTrade, Subscription

logger = logging.getLogger(__name__)


# ─── Tunables ───────────────────────────────────────────────────────────────
CB_DAILY_DD_USD_PER_BASE = 0.5    # Daily DD limit as a multiple of base size.
                                  # base $20 → DD limit -$10. Tune to taste.
CB_CONSEC_LOSSES         = 3      # Consecutive losing trades that trip the CB.
CB_LOCKOUT_HOURS         = 24     # Once tripped, blocked for this many hours
                                  # unless the user explicitly clears it.


# ─── Trip detection ─────────────────────────────────────────────────────────


# Column indirection so one code path serves both independent breakers.
#   system "sys1" → circuit_breaker_tripped_at / circuit_breaker_reason
#   system "sys2" → sys2_cb_tripped_at        / sys2_cb_reason
_CB_COLS = {
    "sys1": ("circuit_breaker_tripped_at", "circuit_breaker_reason"),
    "sys2": ("sys2_cb_tripped_at",          "sys2_cb_reason"),
}


async def _trades_for_system(db: AsyncSession, wallet: str, since, system: str, limit: int | None = None):
    """Closed, priced trades for `wallet` since `since`, filtered to the
    given system by joining the trigger post's source.

    sys2 = trigger_post.source in System-2 set OR adopted trade (identified
    by hl_order_id starting with "adopted:" or sys2_mode being non-NULL).
    sys1 = everything else (currently only "truth").

    Important: adopted trades have trigger_post_id=NULL so their Post join
    returns NULL source — they must be identified by the trade row itself,
    not the linked post.
    """
    from app.models import Post
    from app.services.signal_categories import SYSTEM_2_SOURCES

    rows = await db.execute(
        select(BotTrade, Post.source)
        .outerjoin(Post, Post.id == BotTrade.trigger_post_id)
        .where(
            BotTrade.wallet_address == wallet,
            BotTrade.closed_at >= since,
            BotTrade.pnl_usd.is_not(None),
        )
        .order_by(BotTrade.closed_at.desc())
        .limit(limit * 4 if limit else None)  # fetch a small multiple to account for system filtering
    )
    out = []
    for trade, src in rows.all():
        # Adopted trades: trigger_post_id=NULL, hl_order_id starts with "adopted:"
        # or sys2_mode is set. Check trade row directly before falling back to source.
        is_s2 = (
            (trade.hl_order_id or "").startswith("adopted:")
            or trade.sys2_mode is not None
            or (src or "").lower() in SYSTEM_2_SOURCES
        )
        if (system == "sys2") == is_s2:
            out.append(trade)
    return out


async def check_and_trip(
    wallet: str, db: AsyncSession, system: str = "sys1",
) -> Optional[str]:
    """Run trip conditions for `wallet` within ONE system. Independent per
    system: a Trump losing streak can't lock out the reversal book.

    Called from close_and_finalize after a trade's pnl is written, with the
    system inferred from the closing trade's source.
    """
    col_at, col_reason = _CB_COLS[system]
    sub = (await db.execute(
        select(Subscription).where(Subscription.wallet_address == wallet)
    )).scalar_one_or_none()
    if sub is None:
        return None

    tripped_at = getattr(sub, col_at)
    if tripped_at is not None:
        age_hrs = (datetime.now(timezone.utc).replace(tzinfo=None) - tripped_at).total_seconds() / 3600
        if age_hrs < CB_LOCKOUT_HOURS:
            return None  # still in active lockout

    reason: Optional[str] = None
    now_naive = datetime.now(timezone.utc).replace(tzinfo=None)
    start_of_day = now_naive.replace(hour=0, minute=0, second=0, microsecond=0)

    today_trades = await _trades_for_system(db, wallet, start_of_day, system)
    today_pnl = sum(t.pnl_usd or 0 for t in today_trades)
    dd_limit = -CB_DAILY_DD_USD_PER_BASE * sub.position_size_usd
    if today_pnl < dd_limit:
        reason = "daily_dd"
        logger.warning("CB[%s] trip [daily_dd] %s: pnl=%.2f < %.2f",
                       system, wallet, today_pnl, dd_limit)

    if reason is None:
        # Consecutive losses within this system — only need the most recent
        # CB_CONSEC_LOSSES trades; no reason to scan full history.
        all_sys = await _trades_for_system(
            db, wallet, datetime(1970, 1, 1), system, limit=CB_CONSEC_LOSSES,
        )
        recent = all_sys[:CB_CONSEC_LOSSES]
        if len(recent) >= CB_CONSEC_LOSSES and all((t.pnl_usd or 0) < 0 for t in recent):
            reason = "consecutive_losses"
            logger.warning("CB[%s] trip [consecutive_losses] %s: last %d all negative",
                           system, wallet, CB_CONSEC_LOSSES)

    if reason is None:
        return None

    setattr(sub, col_at, now_naive)
    setattr(sub, col_reason, reason)
    # Only System 1 owns manual_window; nulling it on a sys2 trip would
    # wrongly disable the Trump book too. Sys2 trip just blocks sys2 entries.
    if system == "sys1":
        sub.manual_window_until = None
    await db.commit()

    try:
        from app.ws.manager import manager
        await manager.broadcast({
            "type":     "circuit_breaker_tripped",
            "wallet":   wallet,
            "system":   system,
            "reason":   reason,
            "tripped_at": now_naive.isoformat(),
            "unlock_at": (now_naive + timedelta(hours=CB_LOCKOUT_HOURS)).isoformat(),
        })
    except Exception as exc:
        logger.warning("CB broadcast failed: %s", exc)

    return reason


# ─── Gate (called by _execute_for_subscriber) ───────────────────────────────


def is_tripped(sub_dict: dict, system: str = "sys1") -> tuple[bool, str]:
    """Returns (tripped, reason) for the given system's breaker.
    `sub_dict` is the snapshot copy built in bot_engine."""
    col_at, col_reason = _CB_COLS[system]
    tripped_at = sub_dict.get(col_at)
    if tripped_at is None:
        return False, ""
    age_hrs = (datetime.now(timezone.utc).replace(tzinfo=None) - tripped_at).total_seconds() / 3600
    if age_hrs >= CB_LOCKOUT_HOURS:
        return False, ""  # lockout expired (auto-untrip on next refresh)
    reason = sub_dict.get(col_reason) or "unknown"
    remaining_hrs = CB_LOCKOUT_HOURS - age_hrs
    return True, f"cb[{system}]:{reason} (unlock in {remaining_hrs:.1f}h)"


# ─── Manual reset (called from /manual-window endpoint) ─────────────────────


async def clear_trip(
    wallet: str, db: AsyncSession, system: Optional[str] = None,
) -> bool:
    """Clear circuit-breaker trip state. Returns True if anything cleared.

    `system` selects which breaker to clear:
      None  → clear BOTH sys1 and sys2 (default for human-ack endpoints)
      "sys1"/"sys2" → clear that one only

    Critical: there are TWO independent breakers (Trump book / reversal book).
    The earlier implementation only cleared sys1; a sys2 trip therefore had
    no manual reset path and forced users to wait 24h. When the human
    re-arms via manual-window / auto-trade, they're acknowledging risk
    across the wallet — clear both.
    """
    sub = (await db.execute(
        select(Subscription).where(Subscription.wallet_address == wallet)
    )).scalar_one_or_none()
    if sub is None:
        return False

    targets = (system,) if system in ("sys1", "sys2") else ("sys1", "sys2")
    cleared_any = False
    for s in targets:
        col_at, col_reason = _CB_COLS[s]
        if getattr(sub, col_at) is not None:
            setattr(sub, col_at, None)
            setattr(sub, col_reason, None)
            cleared_any = True
            logger.info("CB[%s] cleared for %s", s, wallet)

    if cleared_any:
        await db.commit()
    return cleared_any