"""Adversarial agent: ispeziona una :class:`Strategy` con check euristici hand-crafted per scovare patologie note (degenerate, no-trade, over/under trading, flat-too-long, time-in-market-too-high, fees-eat-alpha) prima del training vero e proprio. Pipeline: AST -> compile_strategy -> signals -> BacktestEngine.run -> findings Le euristiche sono volutamente coarse: l'agente non rimpiazza la falsificazione, ma sega presto i casi degeneri (es. ``gt close -1e9`` → sempre long) che inquinerebbero il leaderboard del swarm. Phase 1.5 hardening: soglie strette per overtrading (n_trades > n_bars/20) e undertrading (HIGH se n_trades < 10), piu' tre nuovi check HIGH: ``flat_too_long`` (signal flat >95% delle bar), ``time_in_market_too_high`` (signal long/short >80% delle bar, di fatto leveraged buy-and-hold con funding/tail-risk cumulato) e ``fees_eat_alpha`` (fees > 50% del gross_pnl positivo). Killano le strategie "lucky shot", le sempre-in-market e quelle con margine sottile non sostenibile in produzione. """ from __future__ import annotations from dataclasses import dataclass, field from enum import StrEnum import pandas as pd # type: ignore[import-untyped] from ..backtest.engine import BacktestEngine from ..backtest.orders import Side from ..protocol.compiler import compile_strategy from ..protocol.parser import Strategy class Severity(StrEnum): LOW = "low" MEDIUM = "medium" HIGH = "high" @dataclass(frozen=True) class Finding: """Singolo problema identificato dall'agente avversariale.""" name: str severity: Severity detail: str @dataclass class AdversarialReport: """Esito della review: lista (eventualmente vuota) di :class:`Finding`.""" findings: list[Finding] = field(default_factory=list) class AdversarialAgent: """Agente hand-crafted che applica check euristici a una strategia.""" def __init__( self, fees_bp: float = 5.0, fees_eat_alpha_threshold: float = 0.5, flat_too_long_threshold: float = 0.95, undertrading_threshold: int = 10, ) -> None: self._engine = BacktestEngine(fees_bp=fees_bp) self._fees_eat_alpha_threshold = fees_eat_alpha_threshold self._flat_too_long_threshold = flat_too_long_threshold self._undertrading_threshold = undertrading_threshold def review(self, strategy: Strategy, ohlcv: pd.DataFrame) -> AdversarialReport: signal_fn = compile_strategy(strategy) signals = signal_fn(ohlcv) result = self._engine.run(ohlcv, signals) report = AdversarialReport() # No-trade: condizione mai vera o sempre flat -> niente da valutare. # Esce subito perche' i check successivi (degenerate, over/under) # presuppongono un signal stream non banale. if len(result.trades) == 0: report.findings.append( Finding( name="no_trades", severity=Severity.HIGH, detail="Strategy never opens a position on training data", ) ) return report # Degenerate: signals warmup (NaN) esclusi, l'unico valore non-NaN e' # LONG o SHORT. Non c'e' decisione, e' un buy-and-hold camuffato. non_na = signals.dropna() unique_signals = non_na.unique() if len(unique_signals) == 1 and unique_signals[0] in (Side.LONG, Side.SHORT): report.findings.append( Finding( name="degenerate", severity=Severity.HIGH, detail=f"Strategy is always {unique_signals[0].value}, no real decision", ) ) n_bars = len(ohlcv) n_trades = len(result.trades) # Overtrading: > 1 trade ogni 20 bar (Phase 1.5: era 1/5). # Soglia stretta per scovare strategie che flippano cosi' spesso # che le fees mangiano qualunque edge. if n_trades > n_bars / 20: report.findings.append( Finding( name="overtrading", severity=Severity.MEDIUM, detail=f"{n_trades} trades on {n_bars} bars (>1 per 20 bars)", ) ) # Undertrading: < 10 trade -> HIGH (Phase 1.5: era < 5 MEDIUM). # Sample size troppo piccolo per distinguere edge da rumore: e' # un "lucky shot" non riproducibile out-of-sample. if n_trades < self._undertrading_threshold: report.findings.append( Finding( name="undertrading", severity=Severity.HIGH, detail=( f"only {n_trades} trades — likely lucky shot " f"(<{self._undertrading_threshold} over training)" ), ) ) # Flat-too-long: signal attivo (LONG o SHORT) per <5% delle bar. # Anche se la strategia produce trade, una che e' inerte 19h su 20 # ha mancato il regime ed e' di fatto una non-strategia. # NaN (warmup) contano come "flat" perche' downstream l'engine # li riempie via ffill().fillna(Side.FLAT). n_active = int(((signals == Side.LONG) | (signals == Side.SHORT)).sum()) n_flat_or_nan = n_bars - n_active flat_ratio = n_flat_or_nan / n_bars if n_bars > 0 else 1.0 if flat_ratio > self._flat_too_long_threshold: report.findings.append( Finding( name="flat_too_long", severity=Severity.HIGH, detail=( f"Signal flat for {flat_ratio * 100:.1f}% of bars " f"(>{self._flat_too_long_threshold * 100:.0f}% threshold)" ), ) ) # Time-in-market-too-high: signal LONG o SHORT >80% delle bar. # Simmetrico opposto di flat_too_long: una strategia sempre-in-market # e' di fatto leveraged buy-and-hold, esposta a funding cumulato su # perp (paid ogni 8h), tail risk eventi notturni/weekend, nessuna # opportunity-cost flexibility. Sweet spot fitness positiva: 5-80% # time in market (combinato con flat_too_long). active_ratio = n_active / n_bars if n_bars > 0 else 0.0 if active_ratio > 0.80: report.findings.append( Finding( name="time_in_market_too_high", severity=Severity.HIGH, detail=( f"Signal long/short for {active_ratio * 100:.1f}% of bars " "(>80% threshold); esposizione cumulativa funding + tail risk, " "di fatto leveraged B&H" ), ) ) # Fees-eat-alpha: gross_pnl > 0 ma fees > 50% del lordo. # La strategia ha edge teorico ma il margine viene mangiato dai # costi di transazione: non sostenibile in produzione. # Se gross_pnl <= 0 il check non si applica (gia' perdente). gross_pnl = sum(t.gross_pnl for t in result.trades) total_fees = sum(t.fees for t in result.trades) if gross_pnl > 0 and total_fees / gross_pnl > self._fees_eat_alpha_threshold: report.findings.append( Finding( name="fees_eat_alpha", severity=Severity.HIGH, detail=( f"Fees ${total_fees:.2f} = " f"{total_fees / gross_pnl * 100:.1f}% of gross ${gross_pnl:.2f}" ), ) ) return report