Cerbero-Bite/src/cerbero_bite/runtime/auto_pause.py

"""Auto-pause circuit breaker (§7-bis F).

Pure-function evaluation that consults `system_state.auto_pause_until`
and the rolling P/L of the last N closed positions to decide whether
the engine should skip an entry cycle.

Two responsibilities, both deterministic at call time:

* :func:`is_paused` — returns ``True`` when the persisted
  ``auto_pause_until`` is in the future. Independent from the kill
  switch, which targets technical errors.
* :func:`evaluate_drawdown_breach` — given the last N closed P/Ls and
  the current capital, returns whether the rolling drawdown breached
  the configured ``max_drawdown_pct`` threshold. The orchestrator
  layer is the one that flips the persisted state on breach (this
  module stays I/O-free for testability).

The two are separated on purpose: ``is_paused`` is the cheap,
read-only gate consulted at the start of every entry cycle; the
breach evaluation runs once per cycle right after the entry
filtering, before the entry is actually placed.
"""

from __future__ import annotations

from dataclasses import dataclass
from datetime import datetime, timedelta
from decimal import Decimal

from cerbero_bite.config.schema import AutoPauseConfig
from cerbero_bite.state.models import SystemStateRecord

__all__ = [
    "AutoPauseDecision",
    "PauseStatus",
    "evaluate_drawdown_breach",
    "is_paused",
    "pause_until",
]


@dataclass(frozen=True)
class PauseStatus:
    """Snapshot del flag di auto-pausa al momento della valutazione."""

    paused: bool
    until: datetime | None
    reason: str | None


@dataclass(frozen=True)
class AutoPauseDecision:
    """Esito di :func:`evaluate_drawdown_breach`."""

    should_pause: bool
    cumulative_pnl_usd: Decimal
    drawdown_pct: Decimal
    threshold_pct: Decimal
    reason: str | None


def is_paused(
    state: SystemStateRecord | None, *, now: datetime
) -> PauseStatus:
    """Restituisce lo stato della pausa rispetto a ``now``.

    ``state == None`` o ``auto_pause_until == None`` o
    ``auto_pause_until <= now`` ⇒ engine attivo.
    """
    if state is None or state.auto_pause_until is None:
        return PauseStatus(paused=False, until=None, reason=None)
    until = state.auto_pause_until
    if until.tzinfo is not None and now.tzinfo is None:
        # Coerenza: se il valore persistito è tz-aware, normalizziamo.
        return PauseStatus(
            paused=until > now.replace(tzinfo=until.tzinfo),
            until=until,
            reason=state.auto_pause_reason,
        )
    return PauseStatus(
        paused=until > now,
        until=until,
        reason=state.auto_pause_reason,
    )


def pause_until(now: datetime, weeks: int) -> datetime:
    """Calcola la scadenza della pausa (``now + weeks``).

    Estratto in funzione separata per facilitare i test e per ricordare
    che la pausa è espressa in **settimane** (la strategia ha cron
    settimanale; pause più corte non avrebbero modo di evitare una
    settimana di entry).
    """
    return now + timedelta(weeks=max(1, weeks))


def evaluate_drawdown_breach(
    *,
    cfg: AutoPauseConfig,
    recent_pnl_usd: list[Decimal],
    capital_usd: Decimal,
) -> AutoPauseDecision:
    """Decide se la pausa va armata ora dato il rolling P/L.

    Regola: se la somma dei P/L delle ultime ``cfg.lookback_trades``
    posizioni chiuse è negativa e in valore assoluto eccede
    ``cfg.max_drawdown_pct × capital_usd``, ritorna
    ``should_pause=True``. Tutte le altre condizioni → False.

    ``cfg.enabled=False`` → ritorna sempre False (filtro disabilitato).
    Lookback insufficiente → ritorna False (non scattiamo finché non
    abbiamo abbastanza storia per giudicare).
    """
    threshold_pct = cfg.max_drawdown_pct
    cumulative = sum((p for p in recent_pnl_usd), start=Decimal("0"))

    if not cfg.enabled:
        return AutoPauseDecision(
            should_pause=False,
            cumulative_pnl_usd=cumulative,
            drawdown_pct=Decimal("0"),
            threshold_pct=threshold_pct,
            reason=None,
        )

    if len(recent_pnl_usd) < cfg.lookback_trades:
        return AutoPauseDecision(
            should_pause=False,
            cumulative_pnl_usd=cumulative,
            drawdown_pct=Decimal("0"),
            threshold_pct=threshold_pct,
            reason=None,
        )

    if capital_usd <= 0:
        return AutoPauseDecision(
            should_pause=False,
            cumulative_pnl_usd=cumulative,
            drawdown_pct=Decimal("0"),
            threshold_pct=threshold_pct,
            reason=None,
        )

    # Solo perdite ci interessano: vincite cumulate non scattano la pausa.
    if cumulative >= 0:
        return AutoPauseDecision(
            should_pause=False,
            cumulative_pnl_usd=cumulative,
            drawdown_pct=cumulative / capital_usd,
            threshold_pct=threshold_pct,
            reason=None,
        )

    drawdown_pct = (-cumulative) / capital_usd
    if drawdown_pct >= threshold_pct:
        return AutoPauseDecision(
            should_pause=True,
            cumulative_pnl_usd=cumulative,
            drawdown_pct=drawdown_pct,
            threshold_pct=threshold_pct,
            reason=(
                f"rolling DD {drawdown_pct:.2%} ≥ {threshold_pct:.2%} "
                f"(last {cfg.lookback_trades} trades, "
                f"cumulative {cumulative} USD)"
            ),
        )

    return AutoPauseDecision(
        should_pause=False,
        cumulative_pnl_usd=cumulative,
        drawdown_pct=drawdown_pct,
        threshold_pct=threshold_pct,
        reason=None,
    )