""" Arena del gioco GRID TRADERS (sessione 3): 100 agenti ciechi configurano una griglia di trading secondo STRATEGIA_GRIGLIA.md su due serie anonime (A=BTC, B=ETH, mai rivelate; gli agenti le vedono come X/Y). Torneo standard: 3 finestre TRAIN/VALID/TEST, 90 epoche di hill-climb sul TRAIN, ogni 10 epoche l'orchestratore blocca il 10% meno profittevole in VALID, fino a 10 superstiti. TEST = OOS puro mai toccato dall'ottimizzazione. uv run python -m scripts.games.grid_arena # 100 random (smoke) GAME_SPECS_DIR=... GAME_OUT=... uv run python -m scripts.games.grid_arena --from-specs """ from __future__ import annotations import json import os import random import sys from pathlib import Path from scripts.games.engine import load_anon, splits3 from scripts.games import grid_engine from scripts.games.grid_engine import evaluate, max_levels OUT = Path("data/games") OUT.mkdir(parents=True, exist_ok=True) # Spazio parametri (min, max, tipo). range_*/sl_buf/tp_buf in FRAZIONE. PSPACE = dict( range_down=(0.02, 0.30, "f"), range_up=(0.02, 0.30, "f"), levels=(4, 30, "i"), sl_buf=(0.01, 0.15, "f"), tp_buf=(0.01, 0.15, "f"), max_bars=(48, 3000, "i"), ) SERIES = ["A", "B"] TIMEFRAMES = ["15m", "30m", "1h", "2h", "4h", "1d"] # no 5m (costo computazionale) def _rand(rng, lo, hi, typ): return int(rng.randint(int(lo), int(hi))) if typ == "i" else round(rng.uniform(lo, hi), 4) def random_spec(rng): p = {k: _rand(rng, *v) for k, v in PSPACE.items()} return {"series": rng.choice(SERIES), "tf": rng.choice(TIMEFRAMES), "params": p} def _normalize(spec): """Clampa la spec nello spazio valido e APPLICA il vincolo break-even (ยง4): se il passo e' troppo fitto riduce GRID_LEVELS (come da spec: 'vanno ridotti i GRID_LEVELS o allargato il range').""" out = {"series": spec.get("series") if spec.get("series") in SERIES else "A", "tf": spec.get("tf") if spec.get("tf") in TIMEFRAMES else "1h", "params": {}} src = spec.get("params", spec) for k, (lo, hi, typ) in PSPACE.items(): v = src.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, 4) p = out["params"] lmax = max_levels(p["range_down"], p["range_up"]) if lmax >= 2: p["levels"] = min(p["levels"], lmax) return out def mutate(spec, rng, strength=0.25): """Hill-climb: perturba 1-2 parametri; raramente cambia serie. Il timeframe e' l'identita' dell'agente -> non muta.""" s = json.loads(json.dumps(spec)) for k in rng.sample(list(PSPACE), k=rng.randint(1, 2)): lo, hi, typ = PSPACE[k] span = (hi - lo) * strength nv = max(lo, min(hi, s["params"][k] + rng.uniform(-span, span))) s["params"][k] = int(round(nv)) if typ == "i" else round(nv, 4) if rng.random() < 0.05: s["series"] = rng.choice(SERIES) return s class Agent: def __init__(self, aid, spec, brief=""): self.id = aid self.spec = _normalize(spec) self.brief = brief self.train_fit = -1e9 self.valid_fit = -1e9 self.metrics = {} self.vmetrics = {} self.alive = True self.culled_epoch = None @property def tf(self): return self.spec["tf"] def score(self, datasets, sm): data = datasets[self.tf] tr, va, _ = sm[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=2026, epochs=90, cull_every=10, cull_n=10, out_name="grid_result.json", log=print): rng = random.Random(seed) used_tfs = sorted({_normalize(s)["tf"] for s in specs}) datasets = {tf: load_anon(tf) for tf in used_tfs} sm = {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, sm) alive = lambda: [a for a in agents if a.alive] log(f"[epoch 0] {len(alive())} agenti | best VALID " f"{max(a.valid_fit for a in agents):.1f}") history = [] for ep in range(1, epochs + 1): for a in alive(): cand = _normalize(mutate(a.spec, rng)) data = datasets[a.tf] tr, va, _ = sm[a.tf] m = evaluate(data, cand, tr) if m["fitness"] > a.train_fit: a.spec, a.metrics, a.train_fit = cand, m, m["fitness"] a.vmetrics = evaluate(data, a.spec, va) a.valid_fit = a.vmetrics["fitness"] 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) results = [] for rank, a in enumerate(survivors, 1): data = datasets[a.tf] _, _, te = sm[a.tf] results.append({"rank": rank, "agent": a.id, "spec": a.spec, "brief": a.brief, "tf": a.tf, "train": a.metrics, "valid": a.vmetrics, "test": evaluate(data, a.spec, te), "full": evaluate(data, a.spec, None)}) payload = {"n_agents": len(specs), "epochs": epochs, "survivors": len(survivors), "results": results, "history": history, "game": "grid", "rule": "STRATEGIA_GRIGLIA.md", "reveal": {"A": "BTC", "B": "ETH"}} (OUT / out_name).write_text(json.dumps(payload, indent=2)) return payload def leaderboard(payload, top=10, log=print): log("\n========== CLASSIFICA GRID TRADERS (top %d) ==========" % top) log("VALID = finestra del giudice | TEST = OOS puro (mai ottimizzato)") log(f"{'#':>2} {'ag':>4} {'tf':>4} {'ser':>3} {'rng-/+':>11} {'lvl':>3} " f"{'sl/tp buf':>11} {'mbars':>5} {'TEpnl%':>7} {'TEwin':>5} " f"{'TEtpm':>6} {'TEsh':>5} {'VApnl%':>7}") for r in payload["results"][:top]: sp = r["spec"]; p = sp["params"]; te = r["test"]; va = r["valid"] log(f"{r['rank']:>2} {r['agent']:>4} {sp['tf']:>4} {sp['series']:>3} " f"{p['range_down']*100:>4.1f}/{p['range_up']*100:>4.1f}% {p['levels']:>3} " f"{p['sl_buf']*100:>4.1f}/{p['tp_buf']*100:>4.1f}% {p['max_bars']:>5} " f"{te['pnl_pct']:>7.0f} {te['win_rate']*100:>4.0f}% {te['tpm']:>6.1f} " f"{te['sharpe']:>5.1f} {va['pnl_pct']:>7.0f}") def load_specs(specs_dir, n=100): """Carica le spec proposte dagli agenti ciechi (X->A, Y->B, pct->frazione).""" rng = random.Random(7) specs, briefs = [], [] for i in range(n): f = Path(specs_dir) / f"agent_{i}.json" spec = None if f.exists(): try: raw = json.loads(f.read_text()) src = raw.get("params", raw) params = { "range_down": float(src.get("range_down_pct", src.get("range_down", 10))) , "range_up": float(src.get("range_up_pct", src.get("range_up", 10))), "levels": src.get("grid_levels", src.get("levels", 10)), "sl_buf": float(src.get("sl_buf_pct", src.get("sl_buf", 5))), "tp_buf": float(src.get("tp_buf_pct", src.get("tp_buf", 5))), "max_bars": src.get("max_bars", 500), } # gli agenti parlano in percentuale -> frazione for k in ("range_down", "range_up", "sl_buf", "tp_buf"): if params[k] > 1.0: params[k] = params[k] / 100.0 spec = _normalize({ "series": {"X": "A", "Y": "B"}.get(raw.get("series"), raw.get("series")), "tf": raw.get("tf", "1h"), "params": params}) briefs.append(str(raw.get("hypothesis", ""))[:300]) except Exception: spec = None if spec is None: spec = random_spec(rng) briefs.append("(spec mancante -> sostituto casuale)") specs.append(spec) n_real = sum(1 for b in briefs if "mancante" not in b) print(f"caricati {n_real}/{n} spec da agenti reali, {n - n_real} sostituiti casuali") return specs, briefs def main(): slip = float(os.environ.get("GAME_SLIP", "0.0")) grid_engine.set_slippage(slip) if slip > 0: print(f"SLIPPAGE attivo: {slip*100:.3f}%/lato") epochs = int(os.environ.get("GAME_EPOCHS", "90")) if "--from-specs" in sys.argv: sd = os.environ.get("GAME_SPECS_DIR", "data/games/specs_grid") on = os.environ.get("GAME_OUT", "grid_result.json") specs, briefs = load_specs(sd) payload = run_tournament(specs, briefs=briefs, epochs=epochs, out_name=on) else: rng = random.Random(42) payload = run_tournament([random_spec(rng) for _ in range(100)], seed=42, epochs=epochs) leaderboard(payload) rev = payload["reveal"] w = payload["results"][0] sp = w["spec"]; p = sp["params"] print(f"\n>>> RIVELAZIONE: Serie X = {rev['A']}, Serie Y = {rev['B']}. " f"Gli agenti non lo sapevano. <<<") print(f"\nVINCITORE: agente #{w['agent']} su {sp['tf']} serie {sp['series']} | " f"griglia -{p['range_down']*100:.1f}%/+{p['range_up']*100:.1f}% " f"x{p['levels']} livelli, SL buf {p['sl_buf']*100:.1f}%, " f"TP buf {p['tp_buf']*100:.1f}%, max {p['max_bars']} barre") print(f" ipotesi dell'agente: {w['brief']}") print(f" TEST(OOS): PnL {w['test']['pnl_pct']:.0f}% | win " f"{w['test']['win_rate']*100:.0f}% | {w['test']['tpm']:.1f} trade/mese | " f"Sharpe {w['test']['sharpe']:.1f}") if __name__ == "__main__": main()