Adriano
56a631f38a
Stringe le soglie esistenti e aggiunge due check HIGH per killare le strategie degeneri scoperte nel run v5 (top-1 +2.66% vs BTC B&H +106%, flat 99.8% del tempo, fees 69% del lordo). - overtrading: soglia da n_bars/5 a n_bars/20 (MEDIUM) - undertrading: HIGH se n_trades < 10 (era MEDIUM <5) — sample troppo piccolo per distinguere edge da rumore (lucky shot) - flat_too_long (NEW, HIGH): signal attivo per <5% delle bar — la strategia ha mancato il regime, e' una non-strategia - fees_eat_alpha (NEW, HIGH): gross_pnl > 0 ma fees > 50% del lordo — margine sottile non sostenibile in produzione Test count: 141 -> 145 (+4 nuovi test deterministici via monkeypatch). Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
341 lines
11 KiB
Python
341 lines
11 KiB
Python
import json
|
|
|
|
import numpy as np
|
|
import pandas as pd
|
|
import pytest
|
|
|
|
from multi_swarm.agents.adversarial import (
|
|
AdversarialAgent,
|
|
AdversarialReport,
|
|
Severity,
|
|
)
|
|
from multi_swarm.backtest.engine import BacktestResult
|
|
from multi_swarm.backtest.orders import Side, Trade
|
|
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 = json.dumps(
|
|
{
|
|
"rules": [
|
|
{
|
|
"condition": {
|
|
"op": "gt",
|
|
"args": [
|
|
{"kind": "feature", "name": "close"},
|
|
{"kind": "literal", "value": -1e9},
|
|
],
|
|
},
|
|
"action": "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 = json.dumps(
|
|
{
|
|
"rules": [
|
|
{
|
|
"condition": {
|
|
"op": "gt",
|
|
"args": [
|
|
{"kind": "indicator", "name": "rsi", "params": [14]},
|
|
{"kind": "literal", "value": 70.0},
|
|
],
|
|
},
|
|
"action": "entry-short",
|
|
},
|
|
{
|
|
"condition": {
|
|
"op": "lt",
|
|
"args": [
|
|
{"kind": "indicator", "name": "rsi", "params": [14]},
|
|
{"kind": "literal", "value": 30.0},
|
|
],
|
|
},
|
|
"action": "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 = json.dumps(
|
|
{
|
|
"rules": [
|
|
{
|
|
"condition": {
|
|
"op": "gt",
|
|
"args": [
|
|
{"kind": "feature", "name": "close"},
|
|
{"kind": "literal", "value": 1e9},
|
|
],
|
|
},
|
|
"action": "entry-long",
|
|
}
|
|
]
|
|
}
|
|
)
|
|
ast = parse_strategy(src)
|
|
agent = AdversarialAgent()
|
|
report = agent.review(ast, ohlcv)
|
|
assert any(f.name == "no_trades" for f in report.findings)
|
|
|
|
|
|
# AST minimale valido (parser-acceptable). Usato nei test che monkeypatchano
|
|
# compile_strategy/BacktestEngine.run: il contenuto della strategia e'
|
|
# irrilevante perche' il signal/result viene iniettato.
|
|
_MINIMAL_STRATEGY_SRC = json.dumps(
|
|
{
|
|
"rules": [
|
|
{
|
|
"condition": {
|
|
"op": "gt",
|
|
"args": [
|
|
{"kind": "feature", "name": "close"},
|
|
{"kind": "literal", "value": 0.0},
|
|
],
|
|
},
|
|
"action": "entry-long",
|
|
}
|
|
]
|
|
}
|
|
)
|
|
|
|
|
|
def _make_trade(
|
|
entry_ts: pd.Timestamp,
|
|
exit_ts: pd.Timestamp,
|
|
entry_price: float,
|
|
exit_price: float,
|
|
side: Side = Side.LONG,
|
|
fees_bp: float = 5.0,
|
|
) -> Trade:
|
|
return Trade(
|
|
entry_ts=entry_ts.to_pydatetime() if hasattr(entry_ts, "to_pydatetime") else entry_ts,
|
|
exit_ts=exit_ts.to_pydatetime() if hasattr(exit_ts, "to_pydatetime") else exit_ts,
|
|
side=side,
|
|
size=1.0,
|
|
entry_price=entry_price,
|
|
exit_price=exit_price,
|
|
fees_bp=fees_bp,
|
|
)
|
|
|
|
|
|
def test_undertrading_under_10_is_high(monkeypatch: pytest.MonkeyPatch,
|
|
ohlcv: pd.DataFrame) -> None:
|
|
"""5 trade su 500 bar -> HIGH undertrading (Phase 1.5: era MEDIUM <5)."""
|
|
fake_trades = [
|
|
_make_trade(
|
|
ohlcv.index[i * 50],
|
|
ohlcv.index[i * 50 + 10],
|
|
entry_price=100.0,
|
|
exit_price=101.0,
|
|
)
|
|
for i in range(5)
|
|
]
|
|
fake_signals = pd.Series(
|
|
[Side.LONG] * 250 + [Side.FLAT] * 250, index=ohlcv.index, dtype=object
|
|
)
|
|
|
|
def fake_run(self, ohlcv: pd.DataFrame, signals: pd.Series) -> BacktestResult: # type: ignore[no-untyped-def]
|
|
return BacktestResult(
|
|
equity_curve=pd.Series([0.0] * len(ohlcv), index=ohlcv.index, name="equity"),
|
|
returns=pd.Series([0.0] * len(ohlcv), index=ohlcv.index, name="returns"),
|
|
trades=fake_trades,
|
|
)
|
|
|
|
def fake_compile(strategy): # type: ignore[no-untyped-def]
|
|
return lambda df: fake_signals
|
|
|
|
monkeypatch.setattr(
|
|
"multi_swarm.agents.adversarial.BacktestEngine.run", fake_run
|
|
)
|
|
monkeypatch.setattr(
|
|
"multi_swarm.agents.adversarial.compile_strategy", fake_compile
|
|
)
|
|
|
|
src = _MINIMAL_STRATEGY_SRC
|
|
ast = parse_strategy(src)
|
|
agent = AdversarialAgent()
|
|
report = agent.review(ast, ohlcv)
|
|
assert any(
|
|
f.name == "undertrading" and f.severity == Severity.HIGH
|
|
for f in report.findings
|
|
)
|
|
|
|
|
|
def test_overtrading_with_tighter_threshold(monkeypatch: pytest.MonkeyPatch,
|
|
ohlcv: pd.DataFrame) -> None:
|
|
"""n_trades > n_bars/20 -> MEDIUM overtrading (Phase 1.5: era /5)."""
|
|
# 500 bar / 20 = 25. Forziamo 30 trade.
|
|
n = 30
|
|
fake_trades = [
|
|
_make_trade(
|
|
ohlcv.index[i * 10],
|
|
ohlcv.index[i * 10 + 5],
|
|
entry_price=100.0,
|
|
exit_price=100.5,
|
|
)
|
|
for i in range(n)
|
|
]
|
|
# Signal alternato per evitare flat_too_long: 50% LONG, 50% FLAT.
|
|
fake_signals = pd.Series(
|
|
[Side.LONG if i % 2 == 0 else Side.FLAT for i in range(len(ohlcv))],
|
|
index=ohlcv.index,
|
|
dtype=object,
|
|
)
|
|
|
|
def fake_run(self, ohlcv: pd.DataFrame, signals: pd.Series) -> BacktestResult: # type: ignore[no-untyped-def]
|
|
return BacktestResult(
|
|
equity_curve=pd.Series([0.0] * len(ohlcv), index=ohlcv.index, name="equity"),
|
|
returns=pd.Series([0.0] * len(ohlcv), index=ohlcv.index, name="returns"),
|
|
trades=fake_trades,
|
|
)
|
|
|
|
def fake_compile(strategy): # type: ignore[no-untyped-def]
|
|
return lambda df: fake_signals
|
|
|
|
monkeypatch.setattr(
|
|
"multi_swarm.agents.adversarial.BacktestEngine.run", fake_run
|
|
)
|
|
monkeypatch.setattr(
|
|
"multi_swarm.agents.adversarial.compile_strategy", fake_compile
|
|
)
|
|
|
|
src = _MINIMAL_STRATEGY_SRC
|
|
ast = parse_strategy(src)
|
|
agent = AdversarialAgent()
|
|
report = agent.review(ast, ohlcv)
|
|
assert any(
|
|
f.name == "overtrading" and f.severity == Severity.MEDIUM
|
|
for f in report.findings
|
|
)
|
|
|
|
|
|
def test_flat_too_long_flagged(monkeypatch: pytest.MonkeyPatch,
|
|
ohlcv: pd.DataFrame) -> None:
|
|
"""Signal flat per >95% delle bar -> HIGH flat_too_long."""
|
|
n_bars = len(ohlcv)
|
|
# 96% flat: 480 FLAT + 20 LONG = 96% flat ratio
|
|
n_active = 20
|
|
sig_values = [Side.LONG] * n_active + [Side.FLAT] * (n_bars - n_active)
|
|
fake_signals = pd.Series(sig_values, index=ohlcv.index, dtype=object)
|
|
# 15 trade per evitare undertrading HIGH.
|
|
fake_trades = [
|
|
_make_trade(
|
|
ohlcv.index[i * 30],
|
|
ohlcv.index[i * 30 + 1],
|
|
entry_price=100.0,
|
|
exit_price=101.0,
|
|
)
|
|
for i in range(15)
|
|
]
|
|
|
|
def fake_run(self, ohlcv: pd.DataFrame, signals: pd.Series) -> BacktestResult: # type: ignore[no-untyped-def]
|
|
return BacktestResult(
|
|
equity_curve=pd.Series([0.0] * len(ohlcv), index=ohlcv.index, name="equity"),
|
|
returns=pd.Series([0.0] * len(ohlcv), index=ohlcv.index, name="returns"),
|
|
trades=fake_trades,
|
|
)
|
|
|
|
def fake_compile(strategy): # type: ignore[no-untyped-def]
|
|
return lambda df: fake_signals
|
|
|
|
monkeypatch.setattr(
|
|
"multi_swarm.agents.adversarial.BacktestEngine.run", fake_run
|
|
)
|
|
monkeypatch.setattr(
|
|
"multi_swarm.agents.adversarial.compile_strategy", fake_compile
|
|
)
|
|
|
|
src = _MINIMAL_STRATEGY_SRC
|
|
ast = parse_strategy(src)
|
|
agent = AdversarialAgent()
|
|
report = agent.review(ast, ohlcv)
|
|
assert any(
|
|
f.name == "flat_too_long" and f.severity == Severity.HIGH
|
|
for f in report.findings
|
|
)
|
|
|
|
|
|
def test_fees_eat_alpha_flagged(monkeypatch: pytest.MonkeyPatch,
|
|
ohlcv: pd.DataFrame) -> None:
|
|
"""gross_pnl > 0 ma fees > 50% del lordo -> HIGH fees_eat_alpha."""
|
|
# Costruisco trade con gross piccolo e fees alti via fees_bp esagerato.
|
|
# entry=100, exit=100.05, size=1 -> gross=0.05
|
|
# fees_bp=200 (2%) su (100+100.05)*1*200/10000 = 4.001 fees per trade
|
|
# In aggregato: gross=15*0.05=0.75, fees=15*4.001=60 -> ratio enorme.
|
|
n = 15
|
|
fake_trades = [
|
|
_make_trade(
|
|
ohlcv.index[i * 30],
|
|
ohlcv.index[i * 30 + 1],
|
|
entry_price=100.0,
|
|
exit_price=100.05,
|
|
fees_bp=200.0,
|
|
)
|
|
for i in range(n)
|
|
]
|
|
# Signal misto per evitare flat_too_long. 50% attivo.
|
|
fake_signals = pd.Series(
|
|
[Side.LONG if i % 2 == 0 else Side.FLAT for i in range(len(ohlcv))],
|
|
index=ohlcv.index,
|
|
dtype=object,
|
|
)
|
|
|
|
def fake_run(self, ohlcv: pd.DataFrame, signals: pd.Series) -> BacktestResult: # type: ignore[no-untyped-def]
|
|
return BacktestResult(
|
|
equity_curve=pd.Series([0.0] * len(ohlcv), index=ohlcv.index, name="equity"),
|
|
returns=pd.Series([0.0] * len(ohlcv), index=ohlcv.index, name="returns"),
|
|
trades=fake_trades,
|
|
)
|
|
|
|
def fake_compile(strategy): # type: ignore[no-untyped-def]
|
|
return lambda df: fake_signals
|
|
|
|
monkeypatch.setattr(
|
|
"multi_swarm.agents.adversarial.BacktestEngine.run", fake_run
|
|
)
|
|
monkeypatch.setattr(
|
|
"multi_swarm.agents.adversarial.compile_strategy", fake_compile
|
|
)
|
|
|
|
src = _MINIMAL_STRATEGY_SRC
|
|
ast = parse_strategy(src)
|
|
agent = AdversarialAgent()
|
|
report = agent.review(ast, ohlcv)
|
|
assert any(
|
|
f.name == "fees_eat_alpha" and f.severity == Severity.HIGH
|
|
for f in report.findings
|
|
)
|