Adriano Dal Pastro
b6539802e0
- mv src/multi_swarm → src/multi_swarm_core/multi_swarm_core (member layout) - sed-replace globale degli import: from/import multi_swarm.* → multi_swarm_core.* - 115 occorrenze aggiornate in src/ scripts/ tests/ - multi_swarm_coevolutive (nome repo) preservato dal word boundary Pre-condizione per il setup uv workspace della Fase 3. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
100 lines
3.8 KiB
Python
100 lines
3.8 KiB
Python
"""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_core.agents.adversarial import AdversarialAgent
|
|
from multi_swarm_core.agents.falsification import FalsificationAgent
|
|
from multi_swarm_core.cerbero.client import CerberoClient
|
|
from multi_swarm_core.config import load_settings
|
|
from multi_swarm_core.data.cerbero_ohlcv import CerberoOHLCVLoader, OHLCVRequest
|
|
from multi_swarm_core.protocol.parser import parse_strategy
|
|
from multi_swarm_core.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()
|