"""Ratings for the arena: sequential Elo replayed in vote order. Each model starts from its benchmark prior. A vote moves the two models by K * (score - expected), so the change depends on the rating gap, and K is per model (large for new models, shrinking with battles), so models don't move in identical fixed steps. "Both bad" scores as a draw but is tallied separately. """ import math import threading from typing import Dict, Iterable, List, Tuple import registry ANCHOR = 1000.0 # rating of an unplayed model SCALE = 400.0 # Elo points per factor of 10 in odds PRIOR_WEIGHT = 4.0 # virtual draws against the model's own prior rating PRIOR_PER_INDEX = 8.0 # Elo points per TTS Index point away from INDEX_CENTER INDEX_CENTER = 82.0 def prior_elo(repo_id: str) -> float: """Starting rating from the automatic benchmark (TTS Index), so unplayed or evenly matched models don't tie.""" return ANCHOR + PRIOR_PER_INDEX * (registry.spec(repo_id).tts_index - INDEX_CENTER) FIT_ITERATIONS = 500 FIT_TOLERANCE = 1e-10 VOTES = ("a", "b", "tie", "both_bad") SCORE_FOR = {"a": 1.0, "b": 0.0, "tie": 0.5, "both_bad": 0.5} K_MAX = 48.0 # K for a model's first battle K_MIN = 12.0 # K once a model is well established K_HALF = 15.0 # battles after which K has fallen halfway def k_factor(battles: int) -> float: """Per-model step size: new models move fast, established ones slowly.""" return K_MIN + (K_MAX - K_MIN) * K_HALF / (K_HALF + battles) def expected(ra: float, rb: float) -> float: return 1.0 / (1.0 + 10 ** ((rb - ra) / SCALE)) def _blank(repo_id: str) -> Dict: return { "repo_id": repo_id, "elo": prior_elo(repo_id), "battles": 0, "wins": 0, "losses": 0, "ties": 0, "both_bad": 0, } def _pair_key(repo_a: str, repo_b: str) -> Tuple[str, bool]: """Order-independent key for a pair, plus whether the args were swapped.""" if repo_a <= repo_b: return f"{repo_a}|{repo_b}", False return f"{repo_b}|{repo_a}", True def fit_ratings(pairs: Dict[str, List[float]], repos: Iterable[str]) -> Dict[str, float]: """Bradley-Terry MLE via the MM algorithm (Hunter 2004), anchored. `pairs` maps "lo|hi" to [games, score_of_lo]. Returns Elo-scale ratings. Strengths are held relative to the anchor, whose strength is fixed at 1.0, so the scale is pinned and no normalisation step is needed. """ repos = list(repos) phi = {r: 10 ** ((prior_elo(r) - ANCHOR) / SCALE) for r in repos} theta = dict(phi) # Per model: total score, and the opponents it actually faced. score: Dict[str, float] = {r: 0.0 for r in repos} opponents: Dict[str, List[Tuple[str, float]]] = {r: [] for r in repos} for key, (games, score_lo) in pairs.items(): lo, hi = key.split("|", 1) if lo not in theta or hi not in theta: continue # a model that has since left the roster score[lo] += score_lo score[hi] += games - score_lo opponents[lo].append((hi, games)) opponents[hi].append((lo, games)) for _ in range(FIT_ITERATIONS): shift = 0.0 for repo in repos: wins = score[repo] + 0.5 * PRIOR_WEIGHT denom = PRIOR_WEIGHT / (theta[repo] + phi[repo]) for other, games in opponents[repo]: denom += games / (theta[repo] + theta[other]) updated = wins / denom shift = max(shift, abs(updated - theta[repo])) theta[repo] = updated if shift < FIT_TOLERANCE: break return {r: ANCHOR + SCALE * math.log10(t) for r, t in theta.items()} class EloStore: """Battle record and fitted ratings for every arena.""" def __init__(self, state: Dict | None = None): self._lock = threading.RLock() self.arenas: Dict[str, Dict] = {} for arena, models in registry.ARENAS.items(): saved = (state or {}).get("arenas", {}).get(arena, {}) or {} saved_models = saved.get("models") or {} entries = {} for spec in models: entry = _blank(spec.repo_id) entry.update({k: v for k, v in saved_models.get(spec.repo_id, {}).items() if k in entry}) entry["repo_id"] = spec.repo_id entries[spec.repo_id] = entry self.arenas[arena] = { "models": entries, "pairs": {k: list(v) for k, v in (saved.get("pairs") or {}).items()}, } self._refit_all() # -- state ------------------------------------------------------------ def to_dict(self) -> Dict: with self._lock: return { "version": 2, "rating": { "method": "sequential Elo, gap-dependent, per-model K", "anchor": ANCHOR, "scale": SCALE, "prior_weight": PRIOR_WEIGHT, }, "arenas": { arena: { "models": {rid: dict(e) for rid, e in data["models"].items()}, "pairs": {k: list(v) for k, v in data["pairs"].items()}, } for arena, data in self.arenas.items() }, } # -- fitting ---------------------------------------------------------- def _step(self, arena: str, repo_a: str, repo_b: str, vote: str) -> None: """Sequential Elo: the change depends on the rating gap and on each model's own experience.""" models = self.arenas[arena]["models"] a, b = models[repo_a], models[repo_b] ea = expected(a["elo"], b["elo"]) sa = SCORE_FOR[vote] ka, kb = k_factor(a["battles"]), k_factor(b["battles"]) a["elo"] += ka * (sa - ea) b["elo"] += kb * ((1.0 - sa) - (1.0 - ea)) def _refit_all(self) -> None: pass # ratings come from the saved state or from replay() # -- updates ---------------------------------------------------------- def _tally(self, arena: str, repo_a: str, repo_b: str, vote: str) -> None: models = self.arenas[arena]["models"] a = models.setdefault(repo_a, _blank(repo_a)) b = models.setdefault(repo_b, _blank(repo_b)) key, swapped = _pair_key(repo_a, repo_b) score_a = SCORE_FOR[vote] entry = self.arenas[arena]["pairs"].setdefault(key, [0.0, 0.0]) entry[0] += 1 entry[1] += (1.0 - score_a) if swapped else score_a a["battles"] += 1 b["battles"] += 1 if vote == "a": a["wins"] += 1 b["losses"] += 1 elif vote == "b": b["wins"] += 1 a["losses"] += 1 elif vote == "tie": a["ties"] += 1 b["ties"] += 1 else: a["both_bad"] += 1 b["both_bad"] += 1 def record_vote(self, arena: str, repo_a: str, repo_b: str, vote: str) -> Dict: """Apply one vote and re-fit. Returns rating deltas for the two models. Only the two models in the battle move; how far depends on the gap between them and on how many battles each has already played. """ if vote not in VOTES: raise ValueError(f"unknown vote {vote!r}") with self._lock: before = {r: e["elo"] for r, e in self.arenas[arena]["models"].items()} self._step(arena, repo_a, repo_b, vote) self._tally(arena, repo_a, repo_b, vote) after = {r: e["elo"] for r, e in self.arenas[arena]["models"].items()} return { repo: {"before": before.get(repo, prior_elo(repo)), "after": after[repo]} for repo in (repo_a, repo_b) } def replay(self, records) -> int: """Rebuild every rating from a vote log, in log order.""" applied = 0 with self._lock: for arena, models in registry.ARENAS.items(): self.arenas[arena] = { "models": {s.repo_id: _blank(s.repo_id) for s in models}, "pairs": {}, } for record in records: try: arena = record["arena"] repo_a = record["model_a"]["repo_id"] repo_b = record["model_b"]["repo_id"] vote = record["vote"] except (KeyError, TypeError): continue if arena not in self.arenas or vote not in VOTES: continue if repo_a not in registry.BY_REPO or repo_b not in registry.BY_REPO: continue self._step(arena, repo_a, repo_b, vote) self._tally(arena, repo_a, repo_b, vote) applied += 1 return applied # -- reads ------------------------------------------------------------ def rows(self, arena: str): """Leaderboard rows, best first.""" with self._lock: entries = [dict(e) for e in self.arenas[arena]["models"].values()] out = [] # Tie-break on evidence: a rating backed by more battles ranks higher. for e in sorted(entries, key=lambda x: (-x["elo"], -x["battles"], x["repo_id"])): s = registry.spec(e["repo_id"]) decided = e["wins"] + e["losses"] out.append( { "model": s.name, "repo_id": s.repo_id, "params": s.params, "params_m": s.params_m, "elo": e["elo"], "elo_per_m": e["elo"] / s.params_m if s.params_m else 0.0, "battles": e["battles"], "wins": e["wins"], "losses": e["losses"], "ties": e["ties"], "both_bad": e["both_bad"], "win_rate": (e["wins"] / decided) if decided else 0.0, } ) return out def total_battles(self, arena: str) -> int: with self._lock: return sum(int(n) for n, _ in self.arenas[arena]["pairs"].values())