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) <noreply@anthropic.com>
This commit is contained in:
2026-05-09 20:07:56 +02:00
parent 15cb6b37aa
commit 3fbd5eba5e
2 changed files with 163 additions and 0 deletions
+111
View File
@@ -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
+52
View File
@@ -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)