PythagorasGoal/scripts/games/session_arena.py

"""
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()