Adriano
242724ba05
Due fondamenta scientifiche per filtrare overfit e lucky-shot:
1) undertrading_threshold parametrico (era hardcoded 10):
- AdversarialAgent.__init__(undertrading_threshold=10)
- CLI flag --undertrading-threshold
- Aggiunto a hard_kill_findings v2 default
{"no_trades", "degenerate", "undertrading"}: ora un genome con 1 trade
fortunato (es. genome 80be6bcc-1trade-fit-0.21 di fitness-v2-combo) viene
killato anche sotto fitness v2 soft-kill.
- Test parametric: undertrading_threshold=25 → 15 trade triggerano HIGH.
2) Walk-Forward Validation (WFA):
- RunConfig.wfa_train_split (None=off, 0<x<1=on) + wfa_top_k=5
- run_phase1: split ohlcv in train/test; GA usa solo train; a fine GA
i top_k genomi (by fitness in-sample, fitness>0) vengono rivalutati
sul test_ohlcv via falsification+adversarial+compute_fitness.
- Schema migration: evaluations + fitness_oos, sharpe_oos, return_oos,
max_dd_oos, n_trades_oos (ALTER TABLE con try/except per DB pre-2.6).
- Repository.update_evaluation_oos helper per popolare colonne OOS.
- CLI flags --wfa-train-split, --wfa-top-k.
- Test integration: train_split=0.7 → fitness_oos popolato per top_k.
Motivazione: la fase 2.5 ha generato 17 run con fitness fino a 0.36 + DSR
positivo, ma OOS test su 7 anni mostra che flat-ablation top crolla -37%
mentre fitness-v2 top regge (+143%). WFA in-run permette ora di vedere
direttamente il degradation train→test senza eseguire backtest separati,
rendendo possibile filtrare overfit early durante l'ottimizzazione.
Tests (+2 → 193 totale):
- test_undertrading_threshold_parametric
- test_e2e_wfa_populates_fitness_oos
Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
478 lines
16 KiB
Python
478 lines
16 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_undertrading_threshold_parametric(monkeypatch: pytest.MonkeyPatch,
|
|
ohlcv: pd.DataFrame) -> None:
|
|
"""undertrading_threshold=25 → 15 trade vengono killati come HIGH."""
|
|
fake_trades = [
|
|
_make_trade(
|
|
ohlcv.index[i * 30],
|
|
ohlcv.index[i * 30 + 10],
|
|
entry_price=100.0,
|
|
exit_price=101.0,
|
|
)
|
|
for i in range(15)
|
|
]
|
|
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)
|
|
|
|
ast = parse_strategy(_MINIMAL_STRATEGY_SRC)
|
|
# Default threshold 10: 15 trade NON killato
|
|
agent_default = AdversarialAgent()
|
|
rep_default = agent_default.review(ast, ohlcv)
|
|
assert not any(f.name == "undertrading" for f in rep_default.findings)
|
|
# Threshold 25: 15 trade killato
|
|
agent_strict = AdversarialAgent(undertrading_threshold=25)
|
|
rep_strict = agent_strict.review(ast, ohlcv)
|
|
assert any(
|
|
f.name == "undertrading" and f.severity == Severity.HIGH
|
|
for f in rep_strict.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
|
|
)
|
|
|
|
|
|
def test_time_in_market_too_high_flagged(monkeypatch: pytest.MonkeyPatch,
|
|
ohlcv: pd.DataFrame) -> None:
|
|
"""Signal LONG per >80% delle bar -> HIGH time_in_market_too_high."""
|
|
n_bars = len(ohlcv)
|
|
# 90% LONG, 10% FLAT iniziali (warmup-like) per evitare degenerate.
|
|
n_flat = int(n_bars * 0.10)
|
|
sig_values = [Side.FLAT] * n_flat + [Side.LONG] * (n_bars - n_flat)
|
|
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 == "time_in_market_too_high" and f.severity == Severity.HIGH
|
|
for f in report.findings
|
|
)
|
|
|
|
|
|
def test_reasonable_balanced_strategy_not_flagged(monkeypatch: pytest.MonkeyPatch,
|
|
ohlcv: pd.DataFrame) -> None:
|
|
"""Mix ~50% flat, ~25% long, ~25% short: no HIGH sui gate temporali."""
|
|
n_bars = len(ohlcv)
|
|
# Pattern ciclico: 2 flat, 1 long, 1 short per ogni gruppo da 4 bar.
|
|
# Risultato: ~50% FLAT, ~25% LONG, ~25% SHORT. flat_ratio=0.5 < 0.95,
|
|
# active_ratio=0.5 < 0.80.
|
|
pattern = [Side.FLAT, Side.FLAT, Side.LONG, Side.SHORT]
|
|
sig_values = [pattern[i % 4] for i in range(n_bars)]
|
|
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)
|
|
# I due gate temporali non devono triggerare.
|
|
names = [f.name for f in report.findings]
|
|
assert "flat_too_long" not in names
|
|
assert "time_in_market_too_high" not in names
|