Add brains/sovereign-v3/league.py

brains/sovereign-v3/league.py

@@ -0,0 +1,157 @@
+"""
+league.py — league-lite self-play manager (fictitious self-play).
+
+The last-survivor objective punishes myopic bot-exploit policies, so the
+pipeline keeps a pool of FROZEN PAST SELVES and lets the current agent
+train against them (league-lite: a cheap league a 2xT4 budget can afford).
+
+Usage (pipeline side):
+
+    from league import League
+    lg = League("weights/league")
+    lg.add("main")                       # the current agent
+    # after each eval round:
+    lg.update_elo([("main", "main-20250101-0000", 0.75), ...])
+    # if the current agent beats the pool:
+    if lg.maybe_promote("main", win_rate=0.62):
+        torch.save(model.state_dict(), lg.snapshot_path("main"))
+
+    # rollouts: fill the sim's bot slots with frozen selves
+    opponents = lg.pick_opponents(2)     # 2 past selves + 7 bots
+
+Elo is a rough ranking of checkpoint strength over time — not a science;
+the promotion threshold is the real gate. Pure python + json, no deps.
+"""
+
+from __future__ import annotations
+
+import json
+import time
+from pathlib import Path
+from typing import List, Optional
+
+
+class LeagueMember:
+    def __init__(self, name: str, path: Optional[str] = None,
+                 elo: float = 1000.0, kind: str = "past"):
+        self.name = name
+        self.path = path
+        self.elo = float(elo)
+        self.kind = kind
+
+
+class League:
+    def __init__(self, directory: str, main_name: str = "main"):
+        self.dir = Path(directory)
+        self.dir.mkdir(parents=True, exist_ok=True)
+        self.main_name = main_name
+        self.members: List[LeagueMember] = []
+        self.history: List[tuple] = []
+        self._load()
+
+    # -- persistence --------------------------------------------------------
+
+    def _load(self) -> None:
+        p = self.dir / "league.json"
+        if p.exists():
+            try:
+                data = json.loads(p.read_text())
+                self.members = [LeagueMember(**m)
+                                for m in data.get("members", [])]
+                self.history = [tuple(h) for h in data.get("history", [])]
+            except (json.JSONDecodeError, TypeError):
+                pass  # start fresh on a corrupt file
+
+    def save(self) -> None:
+        data = {"members": [vars(m) for m in self.members],
+                "history": [list(h) for h in self.history]}
+        (self.dir / "league.json").write_text(json.dumps(data, indent=2))
+
+    # -- membership ---------------------------------------------------------
+
+    def add(self, name: str, path: Optional[str] = None,
+            elo: float = 1000.0, kind: str = "past") -> LeagueMember:
+        for m in self.members:
+            if m.name == name:
+                return m
+        m = LeagueMember(name, str(path) if path else None, elo, kind)
+        self.members.append(m)
+        self.save()
+        return m
+
+    def snapshot_path(self, name: str) -> Path:
+        return self.dir / f"{name}_snapshot.pt"
+
+    # -- opponent selection -------------------------------------------------
+
+    def pick_opponents(self, k: int, exclude: Optional[str] = None
+                       ) -> List[LeagueMember]:
+        """k frozen opponents: alternating weakest/strongest for curriculum
+        pressure (always train against SOMEONE beatable and SOMEONE tough)."""
+        exclude = exclude or self.main_name
+        pool = [m for m in self.members if m.name != exclude]
+        if not pool:
+            return []
+        pool.sort(key=lambda m: m.elo)
+        picks: List[LeagueMember] = []
+        for i in range(min(k, len(pool))):
+            if i % 2 == 0:
+                picks.append(pool[min(i // 2, len(pool) - 1)])
+            else:
+                picks.append(pool[max(len(pool) - 1 - i // 2, 0)])
+        seen, out = set(), []
+        for p in picks:
+            if p.name not in seen:
+                out.append(p)
+                seen.add(p.name)
+        return out
+
+    # -- ratings ------------------------------------------------------------
+
+    def update_elo(self, results) -> None:
+        """results: iterable of (name_a, name_b, score_a) with score_a in
+        [0,1] = fraction of the match won by a (0.5 = draw)."""
+        K = 32.0
+        for a_name, b_name, score_a in results:
+            a = next((m for m in self.members if m.name == a_name), None)
+            b = next((m for m in self.members if m.name == b_name), None)
+            if a is None or b is None:
+                continue
+            ea = 1.0 / (1.0 + 10 ** ((b.elo - a.elo) / 400.0))
+            delta = K * (float(score_a) - ea)
+            a.elo += delta
+            b.elo -= delta
+        self.history.extend(results)
+        self.save()
+
+    def maybe_promote(self, name: str, win_rate: float,
+                      threshold: float = 0.55) -> bool:
+        """Freeze a snapshot of `name` into the pool when it beats it."""
+        if win_rate >= threshold:
+            stamp = time.strftime("%Y%m%d-%H%M%S")
+            self.add(f"{name}-{stamp}",
+                     path=str(self.snapshot_path(name)),
+                     elo=1200.0, kind="past")
+            return True
+        return False
+
+    def summary(self) -> str:
+        lines = [f"{m.name:32s} elo={m.elo:7.1f}  {m.kind}"
+                 for m in sorted(self.members, key=lambda m: -m.elo)]
+        return "\n".join(lines) if lines else "(empty league)"
+
+
+if __name__ == "__main__":
+    # smoke: create a temp league, add members, rate matches, promote
+    import tempfile
+    with tempfile.TemporaryDirectory() as td:
+        lg = League(td)
+        lg.add("main", elo=1000)
+        lg.add("past-1", elo=980)
+        lg.add("past-2", elo=1050)
+        lg.update_elo([("main", "past-1", 1.0), ("main", "past-2", 0.0)])
+        ops = lg.pick_opponents(2)
+        print("opponents:", [o.name for o in ops])
+        print("promoted:", lg.maybe_promote("main", 0.60))
+        print(lg.summary())
+    print("league.py smoke OK")