Spaces:
Running on Zero
Running on Zero
| """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()) | |