feat(scripts): backtest_strategy.py — esegue una strategia standalone su range esteso

Script utility per validare OOS strategie discovered durante run Phase 2.5.
Carica un JSON strategia (formato Hypothesis output), fetcha OHLCV via
Cerbero, esegue BacktestEngine + FalsificationReport + AdversarialReport,
stampa metriche annualizzate (CAGR, Sharpe, max DD, Calmar).

Esempio:
    uv run python scripts/backtest_strategy.py /tmp/strategy.json \
        --start 2018-09-01 --end 2026-01-01 --label my-strategy

Validato sui top 2 genomi Phase 2.5 (flat-ablation e fitness-v2-combo):
flat-ablation top overfit su 7y (-37%), fitness-v2 top regge (+143% in 7y,
CAGR +12.8%). Conferma che strategie con time gate temporal feature
generalizzano meglio di strategie con SMA crossover hard-tied al regime
del training period.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
2026-05-12 16:56:47 +02:00
parent cf42dd85f3
commit 4c184bb5f7
+99
View File
@@ -0,0 +1,99 @@
"""Backtest standalone di una strategia su range esteso.
Carica un JSON strategia (formato del Hypothesis Agent output), fetcha OHLCV
via Cerbero, esegue BacktestEngine + FalsificationReport + AdversarialReport,
stampa metriche annualizzate.
Esempio:
uv run python scripts/backtest_strategy.py /tmp/strategy_e52604ba.json \
--start 2019-01-01 --end 2026-01-01 --label flat-ablation-top
"""
from __future__ import annotations
import argparse
import json
import math
from datetime import datetime
from pathlib import Path
from multi_swarm.agents.adversarial import AdversarialAgent
from multi_swarm.agents.falsification import FalsificationAgent
from multi_swarm.cerbero.client import CerberoClient
from multi_swarm.config import load_settings
from multi_swarm.data.cerbero_ohlcv import CerberoOHLCVLoader, OHLCVRequest
from multi_swarm.protocol.parser import parse_strategy
from multi_swarm.protocol.validator import validate_strategy
def main() -> None:
p = argparse.ArgumentParser()
p.add_argument("strategy_file", type=Path)
p.add_argument("--start", default="2019-01-01T00:00:00+00:00")
p.add_argument("--end", default="2026-01-01T00:00:00+00:00")
p.add_argument("--exchange", default="deribit")
p.add_argument("--symbol", default="BTC-PERPETUAL")
p.add_argument("--timeframe", default="1h")
p.add_argument("--fees-bp", type=float, default=5.0)
p.add_argument("--n-trials-dsr", type=int, default=50)
p.add_argument("--label", default="strategy")
args = p.parse_args()
strategy_json = json.loads(args.strategy_file.read_text())
raw = json.dumps(strategy_json)
parsed = parse_strategy(raw)
validate_strategy(parsed)
print(f"Strategy '{args.label}' parsed OK: {len(parsed.rules)} rules")
settings = load_settings()
token = (
settings.cerbero_mainnet_token.get_secret_value()
if settings.cerbero_mainnet_token
else settings.cerbero_testnet_token.get_secret_value()
)
cerbero = CerberoClient(
base_url=settings.cerbero_base_url,
token=token,
bot_tag=settings.cerbero_bot_tag,
)
loader = CerberoOHLCVLoader(client=cerbero, cache_dir=settings.series_dir)
req = OHLCVRequest(
symbol=args.symbol,
timeframe=args.timeframe,
start=datetime.fromisoformat(args.start),
end=datetime.fromisoformat(args.end),
exchange=args.exchange,
)
ohlcv = loader.load(req)
n_bars = len(ohlcv)
years = n_bars / (24 * 365.25)
print(
f"OHLCV loaded: {n_bars} bars "
f"({ohlcv.index[0]}{ohlcv.index[-1]}, ~{years:.2f} anni)"
)
fals_agent = FalsificationAgent(fees_bp=args.fees_bp, n_trials_dsr=args.n_trials_dsr)
adv_agent = AdversarialAgent(fees_bp=args.fees_bp)
fals = fals_agent.evaluate(parsed, ohlcv)
adv = adv_agent.review(parsed, ohlcv)
cagr = (1.0 + float(fals.total_return)) ** (1.0 / years) - 1.0 if years > 0 else float("nan")
calmar = (cagr / float(fals.max_drawdown)) if fals.max_drawdown > 0 else float("inf")
print(f"\n=== {args.label} on {args.symbol} {args.timeframe} ({years:.2f} anni) ===")
print(f"n_trades: {fals.n_trades}")
print(f"total_return: {fals.total_return:+.4f} ({fals.total_return * 100:+.2f}%)")
print(f"CAGR: {cagr:+.4f} ({cagr * 100:+.2f}%)")
print(f"Sharpe (ann): {fals.sharpe:+.3f}")
print(f"DSR: {fals.dsr:.4f} (pvalue {fals.dsr_pvalue:.4f})")
print(f"max_drawdown: {fals.max_drawdown:.4f} ({fals.max_drawdown * 100:.2f}%)")
print(f"Calmar: {calmar:+.3f}")
print(f"\nAdversarial findings:")
if not adv.findings:
print(" (none)")
for f in adv.findings:
print(f" [{f.severity.value:6s}] {f.name:30s} {f.detail}")
if __name__ == "__main__":
main()