""" Arena del gioco-SESSION: 100 agenti ciechi cercano pattern ORARI intraday (fascia di controllo -> finestra successiva) su due serie anonime (A=BTC, B=ETH). Torneo standard (3 finestre, 90 epoche, cull 10%/10) col motore session_engine. uv run python -m scripts.games.session_arena # 100 random (test) GAME_SPECS_DIR=... GAME_OUT=... uv run python -m scripts.games.session_arena --from-specs """ from __future__ import annotations import json import os import random import sys from pathlib import Path from scripts.games.session_engine import load_session, splits3, evaluate OUT = Path("data/games"); OUT.mkdir(parents=True, exist_ok=True) PSPACE = dict(ctrl_hour=(0, 23, "i"), ctrl_len=(1, 6, "i"), entry_thr=(0.0, 1.5, "f"), hold=(1, 12, "i")) SERIES = ["A", "B"] DIRECTIONS = ["trend", "reversion"] def _rand(rng, lo, hi, typ): return int(rng.randint(int(lo), int(hi))) if typ == "i" else round(rng.uniform(lo, hi), 3) def random_spec(rng): p = {k: _rand(rng, *v) for k, v in PSPACE.items()} return {"series": rng.choice(SERIES), "direction": rng.choice(DIRECTIONS), "params": p} def _normalize(spec): out = {"series": spec.get("series") if spec.get("series") in SERIES else "A", "direction": spec.get("direction") if spec.get("direction") in DIRECTIONS else "trend", "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, 3) return out def _flat(spec): return {"direction": spec["direction"], **spec["params"]} def mutate(spec, rng, strength=0.25): 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, 3) if rng.random() < 0.12: s["direction"] = rng.choice(DIRECTIONS) 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 = self.valid_fit = -1e9; self.metrics = self.vmetrics = {}; self.alive = True @property def series(self): return self.spec["series"] def score(self, datasets, sm): d = datasets[self.series]; tr, va, _ = sm[self.series] self.metrics = evaluate(d, _flat(self.spec), tr); self.vmetrics = evaluate(d, _flat(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="session_result.json", log=print): rng = random.Random(seed) datasets = {"A": load_session("BTC"), "B": load_session("ETH")} sm = {k: splits3(datasets[k]) for k in datasets} 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())} | best VALID {max(a.valid_fit for a in agents):.1f}") for ep in range(1, epochs + 1): for a in alive(): cand = _normalize(mutate(a.spec, rng)) d = datasets[cand["series"]]; tr, va, _ = sm[cand["series"]] m = evaluate(d, _flat(cand), tr) if m["fitness"] > a.train_fit: a.spec, a.metrics, a.train_fit = cand, m, m["fitness"] a.vmetrics = evaluate(d, _flat(cand), 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 log(f"[epoch {ep:2d}] cull {k:2d} -> {len(alive()):3d} | best VALID " f"{max(a.valid_fit for a in alive()):.1f} | worst {min(a.valid_fit for a in alive()):.1f}") survivors = sorted(alive(), key=lambda a: a.valid_fit, reverse=True) results = [] for rank, a in enumerate(survivors, 1): d = datasets[a.series]; _, _, te = sm[a.series] results.append({"rank": rank, "agent": a.id, "spec": a.spec, "brief": a.brief, "series": a.series, "train": a.metrics, "valid": a.vmetrics, "test": evaluate(d, _flat(a.spec), te), "full": evaluate(d, _flat(a.spec), None)}) payload = {"n_agents": len(specs), "survivors": len(survivors), "results": results, "reveal": {"A": "BTC", "B": "ETH"}, "game": "session"} (OUT / out_name).write_text(json.dumps(payload, indent=2)) return payload def leaderboard(payload, top=10, log=print): log("\n===== CLASSIFICA SESSION (top %d) — fascia controllo -> finestra dopo =====" % top) log(f"{'#':>2} {'ag':>4} {'ser':>3} {'h':>3} {'len':>3} {'thr%':>5} {'hold':>4} {'dir':>9} " f"{'TEpnl%':>8} {'TEwin':>5} {'TEtpm':>6} {'TEsh':>6}") for r in payload["results"][:top]: sp = r["spec"]; te = r["test"]; p = sp["params"] log(f"{r['rank']:>2} {r['agent']:>4} {sp['series']:>3} {p['ctrl_hour']:>3} {p['ctrl_len']:>3} " f"{p['entry_thr']:>5.2f} {p['hold']:>4} {sp['direction']:>9} {te['pnl_pct']:>8.0f} " f"{te['win_rate']*100:>4.0f}% {te['tpm']:>6.0f} {te['sharpe']:>6.1f}") def load_specs(specs_dir, n=100): 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()) params = {k: raw.get(k, raw.get("params", {}).get(k)) for k in PSPACE} spec = _normalize({"series": {"X": "A", "Y": "B"}.get(raw.get("series"), raw.get("series")), "direction": raw.get("direction"), "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 -> random)") specs.append(spec) return specs, briefs def main(): if "--from-specs" in sys.argv: sd = os.environ.get("GAME_SPECS_DIR", "data/games/specs_sess") on = os.environ.get("GAME_OUT", "session_result.json") specs, briefs = load_specs(sd) print(f"caricati {sum(1 for b in briefs if 'mancante' not in b)}/100 spec da agenti reali") payload = run_tournament(specs, briefs=briefs, out_name=on) else: rng = random.Random(42) payload = run_tournament([random_spec(rng) for _ in range(100)], seed=42) leaderboard(payload) rev = payload["reveal"]; w = payload["results"][0]; p = w["spec"]["params"] print(f"\n>>> RIVELAZIONE: A={rev['A']}, B={rev['B']}. <<<") print(f"VINCITORE: #{w['agent']} {w['series']} fascia h{p['ctrl_hour']} len{p['ctrl_len']} " f"-> {w['spec']['direction']} hold{p['hold']}h thr{p['entry_thr']}%") print(f" ipotesi: {w['brief']}") print(f" TEST: PnL {w['test']['pnl_pct']:.0f}% | win {w['test']['win_rate']*100:.0f}% | " f"{w['test']['tpm']:.0f} tr/mese | Sharpe {w['test']['sharpe']:.1f}") if __name__ == "__main__": main()