""" Arena — tournament orchestrator per il gioco "Blind Traders". 100 agenti partono da una spec di strategia (creata alla cieca: vedi agent_brief.py / workflow). L'orchestratore valuta ogni spec con il backtest deterministico (engine.evaluate) su TRAIN, da' epoche di elaborazione (ogni agente affina la propria strategia via hill-climb sui parametri) e OGNI 10 EPOCHE blocca il 10% meno profittevole. Restano i 10 piu' profittevoli. Punteggio = fitness su PNL + %win, con vincolo >=10 trade/mese (engine). """ from __future__ import annotations import json import random from pathlib import Path import numpy as np from scripts.games.engine import load_anon, splits3, evaluate OUT = Path("data/games") OUT.mkdir(parents=True, exist_ok=True) # Spazio parametri per famiglia (min, max, tipo) SPACE = { "zscore": dict(lookback=(10, 100, "i"), entry_thr=(1.0, 3.5, "f"), tp_atr=(0.5, 4.0, "f"), sl_atr=(1.0, 5.0, "f"), max_bars=(6, 72, "i")), "breakout": dict(lookback=(12, 120, "i"), entry_thr=(0.0, 0.0, "f"), tp_atr=(0.5, 4.0, "f"), sl_atr=(1.0, 5.0, "f"), max_bars=(6, 72, "i")), "ma_cross": dict(lookback=(5, 50, "i"), slow_mult=(2.0, 6.0, "f"), entry_thr=(0.0, 0.0, "f"), tp_atr=(0.5, 4.0, "f"), sl_atr=(1.0, 5.0, "f"), max_bars=(6, 72, "i")), "rsi": dict(lookback=(7, 30, "i"), entry_thr=(1.0, 4.0, "f"), tp_atr=(0.5, 4.0, "f"), sl_atr=(1.0, 5.0, "f"), max_bars=(6, 72, "i")), "momentum": dict(lookback=(6, 72, "i"), entry_thr=(1.0, 6.0, "f"), tp_atr=(0.5, 4.0, "f"), sl_atr=(1.0, 5.0, "f"), max_bars=(6, 72, "i")), "pairs": dict(lookback=(20, 120, "i"), entry_thr=(1.5, 3.0, "f"), exit_thr=(0.2, 1.0, "f"), max_bars=(24, 120, "i")), } SINGLE_FAMILIES = ["zscore", "breakout", "ma_cross", "rsi", "momentum"] DIRECTIONS = ["reversion", "trend"] TIMEFRAMES = ["1h", "15m", "5m"] # timing diversi su cui competono gli agenti def _rand_param(rng, lo, hi, typ): if typ == "i": return int(rng.randint(int(lo), int(hi))) return round(rng.uniform(lo, hi), 3) def random_spec(rng): if rng.random() < 0.25: fam = "pairs" else: fam = rng.choice(SINGLE_FAMILIES) params = {} for k, (lo, hi, typ) in SPACE[fam].items(): params[k] = _rand_param(rng, lo, hi, typ) spec = {"family": fam, "params": params, "tf": rng.choice(TIMEFRAMES)} if fam == "pairs": spec["series"] = "AB" else: spec["series"] = rng.choice(["A", "B"]) spec["params"]["direction"] = rng.choice(DIRECTIONS) return spec def mutate(spec, rng, strength=0.25): """Perturba la spec (hill-climb). Per lo piu' numerica; raramente cambia direzione/serie. La famiglia resta fissa (identita' dell'agente).""" s = json.loads(json.dumps(spec)) fam = s["family"] # perturba 1-2 parametri numerici keys = [k for k in SPACE[fam] if SPACE[fam][k][0] != SPACE[fam][k][1]] for k in rng.sample(keys, k=min(len(keys), rng.randint(1, 2))): lo, hi, typ = SPACE[fam][k] cur = s["params"][k] span = (hi - lo) * strength nv = cur + rng.uniform(-span, span) nv = max(lo, min(hi, nv)) s["params"][k] = int(round(nv)) if typ == "i" else round(nv, 3) if fam != "pairs": if rng.random() < 0.10: s["params"]["direction"] = rng.choice(DIRECTIONS) if rng.random() < 0.05: s["series"] = rng.choice(["A", "B"]) # il timeframe resta l'identita' dell'agente (timing fisso) -> non muta return s def _normalize(spec): """Completa/ripulisce una spec proposta da un agente (robustezza).""" fam = spec.get("family") if fam not in SPACE: fam = "zscore" out = {"family": fam, "params": {}} for k, (lo, hi, typ) in SPACE[fam].items(): v = spec.get("params", {}).get(k, (lo + hi) / 2) try: v = float(v) except Exception: v = (lo + hi) / 2 v = max(lo, min(hi, v)) out["params"][k] = int(round(v)) if typ == "i" else round(v, 3) out["tf"] = spec.get("tf") if spec.get("tf") in TIMEFRAMES else "1h" if fam == "pairs": out["series"] = "AB" else: out["series"] = spec.get("series", "A") if spec.get("series") in ("A", "B") else "A" d = spec.get("params", {}).get("direction") or spec.get("direction") out["params"]["direction"] = d if d in DIRECTIONS else "reversion" return out class Agent: def __init__(self, aid, spec, brief=""): self.id = aid self.spec = _normalize(spec) self.brief = brief # cosa "dice" l'agente (ipotesi NL) self.train_fit = -1e9 # criterio di hill-climb (l'agente ottimizza qui) self.valid_fit = -1e9 # criterio dell'orchestratore (cull + rank) self.metrics = {} # metriche TRAIN self.vmetrics = {} # metriche VALID self.alive = True self.culled_epoch = None @property def tf(self): return self.spec.get("tf", "1h") def score(self, datasets, splits_map): data = datasets[self.tf] tr, va, _ = splits_map[self.tf] self.metrics = evaluate(data, self.spec, tr) self.vmetrics = evaluate(data, self.spec, va) self.train_fit = self.metrics["fitness"] self.valid_fit = self.vmetrics["fitness"] def run_tournament(specs, briefs=None, seed=7, epochs=90, cull_every=10, cull_n=10, log=print): rng = random.Random(seed) # carica solo i timeframe effettivamente usati dagli agenti used_tfs = sorted({_normalize(s).get("tf", "1h") for s in specs}) datasets = {tf: load_anon(tf) for tf in used_tfs} splits_map = {tf: splits3(datasets[tf], 0.60, 0.20) for tf in used_tfs} briefs = briefs or [""] * len(specs) agents = [Agent(i, s, briefs[i] if i < len(briefs) else "") for i, s in enumerate(specs)] for a in agents: a.score(datasets, splits_map) alive = lambda: [a for a in agents if a.alive] log(f"[epoch 0] {len(alive())} agenti | best VALID fit " f"{max(a.valid_fit for a in agents):.1f}") history = [] for ep in range(1, epochs + 1): # elaborazione: l'agente affina sul TRAIN (cio' che vede); ricalcola VALID for a in alive(): cand = mutate(a.spec, rng) data = datasets[a.tf] tr, va, _ = splits_map[a.tf] m = evaluate(data, cand, tr) if m["fitness"] > a.train_fit: a.spec = _normalize(cand) a.metrics, a.train_fit = m, m["fitness"] a.vmetrics = evaluate(data, a.spec, va) a.valid_fit = a.vmetrics["fitness"] # cull ogni N epoche: l'ORCHESTRATORE blocca il 10% meno profittevole # in VALIDATION (generalizzazione, non overfit sul train) if ep % cull_every == 0: av = sorted(alive(), key=lambda a: a.valid_fit) k = cull_n if len(av) - cull_n >= 10 else max(0, len(av) - 10) for a in av[:k]: a.alive = False a.culled_epoch = ep log(f"[epoch {ep:2d}] cull {k:2d} -> {len(alive()):3d} vivi | " f"best VALID {max(a.valid_fit for a in alive()):.1f} | " f"worst-alive {min(a.valid_fit for a in alive()):.1f}") history.append({"epoch": ep, "alive": len(alive()), "best_valid": max(a.valid_fit for a in alive())}) survivors = sorted(alive(), key=lambda a: a.valid_fit, reverse=True) # report finale: TEST = OOS puro mai toccato dall'ottimizzazione results = [] for rank, a in enumerate(survivors, 1): data = datasets[a.tf] _, _, te = splits_map[a.tf] test = evaluate(data, a.spec, te) full = evaluate(data, a.spec, None) results.append({ "rank": rank, "agent": a.id, "spec": a.spec, "brief": a.brief, "tf": a.tf, "train": a.metrics, "valid": a.vmetrics, "test": test, "full": full, }) payload = {"n_agents": len(specs), "epochs": epochs, "survivors": len(survivors), "results": results, "history": history, "reveal": {"A": "BTC", "B": "ETH", "tf": "1h"}} (OUT / "tournament_result.json").write_text(json.dumps(payload, indent=2)) return payload def leaderboard(payload, top=10, log=print): log("\n================ CLASSIFICA FINALE (top %d) ================" % top) log("VALID = finestra su cui l'orchestratore giudica | TEST = OOS puro (mai ottimizzato)") log(f"{'#':>2} {'ag':>4} {'tf':>3} {'famiglia':>9} {'ser':>3} {'dir':>9} " f"{'TEpnl%':>8} {'TEwin':>5} {'TEtpm':>6} {'TEsh':>5} {'VApnl%':>8} {'VAwin':>5}") for r in payload["results"][:top]: sp = r["spec"]; te = r["test"]; va = r["valid"] d = sp["params"].get("direction", "-") log(f"{r['rank']:>2} {r['agent']:>4} {sp.get('tf','1h'):>3} {sp['family']:>9} " f"{sp['series']:>3} {d:>9} {te['pnl_pct']:>8.0f} {te['win_rate']*100:>4.0f}% " f"{te['tpm']:>6.1f} {te['sharpe']:>5.1f} {va['pnl_pct']:>8.0f} " f"{va['win_rate']*100:>4.0f}%") if __name__ == "__main__": import sys # modalita' test: 100 agenti random rng = random.Random(42) specs = [random_spec(rng) for _ in range(100)] payload = run_tournament(specs, seed=42) leaderboard(payload)