From 3fbd5eba5ebe27bbc79647beeb660e3db25b1127 Mon Sep 17 00:00:00 2001 From: AdrianoDev Date: Sat, 9 May 2026 20:07:56 +0200 Subject: [PATCH] feat(agents): hand-crafted adversarial with heuristic checks Implementa AdversarialAgent con check euristici hand-crafted: no_trades (HIGH), degenerate (HIGH), overtrading/undertrading (MEDIUM). Severity come StrEnum (UP042 clean), pipeline AST -> compile -> backtest -> findings allineata a FalsificationAgent. Co-Authored-By: Claude Opus 4.7 (1M context) --- src/multi_swarm/agents/adversarial.py | 111 ++++++++++++++++++++++++++ tests/unit/test_adversarial.py | 52 ++++++++++++ 2 files changed, 163 insertions(+) create mode 100644 src/multi_swarm/agents/adversarial.py create mode 100644 tests/unit/test_adversarial.py diff --git a/src/multi_swarm/agents/adversarial.py b/src/multi_swarm/agents/adversarial.py new file mode 100644 index 0000000..e545a8c --- /dev/null +++ b/src/multi_swarm/agents/adversarial.py @@ -0,0 +1,111 @@ +"""Adversarial agent: ispeziona una :class:`Strategy` con check euristici +hand-crafted per scovare patologie note (degenerate, no-trade, over/under +trading) 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. +""" + +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) -> None: + self._engine = BacktestEngine(fees_bp=fees_bp) + + 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 5 bar -> il segnale flippa cosi' spesso + # che le fees mangiano qualunque edge. + if n_trades > n_bars / 5: + report.findings.append( + Finding( + name="overtrading", + severity=Severity.MEDIUM, + detail=f"{n_trades} trades on {n_bars} bars (>1 per 5 bars)", + ) + ) + # Undertrading: < 5 trade -> sample size troppo piccolo per + # distinguere edge da rumore (lucky shot). + if n_trades < 5: + report.findings.append( + Finding( + name="undertrading", + severity=Severity.MEDIUM, + detail=f"only {n_trades} trades — likely lucky shot", + ) + ) + + return report diff --git a/tests/unit/test_adversarial.py b/tests/unit/test_adversarial.py new file mode 100644 index 0000000..feb94a0 --- /dev/null +++ b/tests/unit/test_adversarial.py @@ -0,0 +1,52 @@ +import numpy as np +import pandas as pd +import pytest + +from multi_swarm.agents.adversarial import AdversarialAgent, AdversarialReport, Severity +from multi_swarm.protocol.parser import parse_strategy + + +@pytest.fixture +def ohlcv() -> pd.DataFrame: + idx = pd.date_range("2024-01-01", periods=500, freq="1h", tz="UTC") + close = 100 + np.cumsum(np.random.RandomState(0).normal(0.0, 1.0, 500)) + return pd.DataFrame( + { + "open": close, + "high": close + 0.5, + "low": close - 0.5, + "close": close, + "volume": 1.0, + }, + index=idx, + ) + + +def test_degenerate_always_long_flagged(ohlcv: pd.DataFrame) -> None: + src = "(strategy (when (gt (feature close) -1e9) (entry-long)))" + ast = parse_strategy(src) + agent = AdversarialAgent() + report = agent.review(ast, ohlcv) + assert isinstance(report, AdversarialReport) + assert any(f.name == "degenerate" and f.severity == Severity.HIGH for f in report.findings) + + +def test_no_findings_on_reasonable_strategy(ohlcv: pd.DataFrame) -> None: + src = ( + "(strategy " + "(when (gt (indicator rsi 14) 70.0) (entry-short)) " + "(when (lt (indicator rsi 14) 30.0) (entry-long)))" + ) + ast = parse_strategy(src) + agent = AdversarialAgent() + report = agent.review(ast, ohlcv) + high_findings = [f for f in report.findings if f.severity == Severity.HIGH] + assert len(high_findings) == 0 + + +def test_zero_trade_strategy_flagged(ohlcv: pd.DataFrame) -> None: + src = "(strategy (when (gt (feature close) 1e9) (entry-long)))" + ast = parse_strategy(src) + agent = AdversarialAgent() + report = agent.review(ast, ohlcv) + assert any(f.name == "no_trades" for f in report.findings)