#!/usr/bin/env python3 """Fit Bradley-Terry ratings to completed Pokemon Bench legs.""" from __future__ import annotations import argparse import json import math from collections import Counter from pathlib import Path from pokemon_bench.models import BestOfThreeResult, TournamentSummary from pokemon_bench.providers import GENERATIVE_DECISION_PROTOCOL def solve_linear_system(matrix: list[list[float]], vector: list[float]) -> list[float]: """Solve a small dense linear system with partial-pivot elimination.""" size = len(vector) augmented = [row[:] + [value] for row, value in zip(matrix, vector, strict=True)] for column in range(size): pivot = max(range(column, size), key=lambda row: abs(augmented[row][column])) if abs(augmented[pivot][column]) < 1e-12: raise ValueError("Bradley-Terry information matrix is singular") augmented[column], augmented[pivot] = augmented[pivot], augmented[column] divisor = augmented[column][column] augmented[column] = [value / divisor for value in augmented[column]] for row in range(size): if row == column: continue factor = augmented[row][column] augmented[row] = [ value - factor * pivot_value for value, pivot_value in zip( augmented[row], augmented[column], strict=True ) ] return [row[-1] for row in augmented] def invert_matrix(matrix: list[list[float]]) -> list[list[float]]: """Invert a small dense matrix using the same deterministic solver.""" size = len(matrix) columns = [ solve_linear_system(matrix, [float(row == column) for row in range(size)]) for column in range(size) ] return [[columns[column][row] for column in range(size)] for row in range(size)] def fit_results( results: list[BestOfThreeResult], decision_protocols: dict[str, set[str]] | None = None, *, prior_std: float | None = None, ) -> dict[str, object]: """Fit centered ratings: MLE/Wald by default, MAP/Laplace with a prior.""" models = sorted({model for result in results for model in result.models}) model_index = {model: index for index, model in enumerate(models)} matches: list[tuple[int, int]] = [] records = {model: Counter(wins=0, losses=0) for model in models} for result in results: winner = result.winner_model loser = next(model for model in result.models if model != winner) matches.append((model_index[winner], model_index[loser])) records[winner]["wins"] += 1 records[loser]["losses"] += 1 count = len(models) if count < 2: raise ValueError('At least two models are required') if prior_std is not None and (not math.isfinite(prior_std) or prior_std <= 0): raise ValueError('prior_std must be positive and finite') parameters = [0.0] * (count - 1) for _ in range(100): abilities = parameters + [0.0] gradient = [0.0] * count information = [[0.0] * count for _ in range(count)] if prior_std is not None: # Isotropic Gaussian prior on centered log-strengths, sum(beta)=0. # Unlike fixing a prior on the reference model, this is label invariant. precision = 1.0 / prior_std**2 mean = sum(abilities) / count for row in range(count): gradient[row] -= precision * (abilities[row] - mean) for column in range(count): information[row][column] += precision * (float(row == column) - 1.0 / count) for winner, loser in matches: difference = abilities[winner] - abilities[loser] probability = 1.0 / (1.0 + math.exp(-difference)) residual = 1.0 - probability weight = probability * (1.0 - probability) gradient[winner] += residual gradient[loser] -= residual information[winner][winner] += weight information[loser][loser] += weight information[winner][loser] -= weight information[loser][winner] -= weight reduced_information = [row[:-1] for row in information[:-1]] step = solve_linear_system(reduced_information, gradient[:-1]) parameters = [ value + delta for value, delta in zip(parameters, step, strict=True) ] if max(abs(delta) for delta in step) < 1e-10: break uncentered = parameters + [0.0] mean_ability = sum(uncentered) / count centered = [ability - mean_ability for ability in uncentered] covariance_reduced = invert_matrix(reduced_information) centering = [ [float(row == column) - 1.0 / count for column in range(count - 1)] for row in range(count) ] covariance = [ [ sum( centering[row][left] * covariance_reduced[left][right] * centering[column][right] for left in range(count - 1) for right in range(count - 1) ) for column in range(count) ] for row in range(count) ] elo_scale = 400.0 / math.log(10.0) ratings = [1500.0 + elo_scale * ability for ability in centered] standard_errors = [ elo_scale * math.sqrt(max(covariance[index][index], 0.0)) for index in range(count) ] ability_weights = [math.exp(ability) for ability in centered] total_weight = sum(ability_weights) ability_shares = [weight / total_weight for weight in ability_weights] rows = [] for index, model in enumerate(models): wins = records[model]["wins"] losses = records[model]["losses"] rows.append( { "model": model, "rating": round(float(ratings[index]), 1), "ci_low": round( float(ratings[index] - 1.96 * standard_errors[index]), 1 ), "ci_high": round( float(ratings[index] + 1.96 * standard_errors[index]), 1 ), "ability_share": round(float(ability_shares[index]), 5), "wins": wins, "losses": losses, "win_rate": round(wins / (wins + losses), 4), } ) rows.sort(key=lambda row: row["rating"], reverse=True) previous_rating: float | None = None previous_rank = 0 for position, row in enumerate(rows, start=1): if row["rating"] != previous_rating: previous_rank = position previous_rating = row["rating"] row["rank"] = previous_rank output: dict[str, object] = { "method": ("Bradley-Terry MAP with centered Gaussian prior; Elo scale centered at 1500" if prior_std is not None else "Bradley-Terry MLE; Elo scale centered at 1500"), "legs": len(matches), "ratings": rows, } if prior_std is not None: output['prior_std_log_strength'] = prior_std output['interval_method'] = '95% Laplace posterior intervals, conditional on observed series' if decision_protocols is not None: normalized = { model: sorted(decision_protocols.get(model, {"unknown"})) for model in models } distinct = {protocol for values in normalized.values() for protocol in values} output["decision_protocols"] = normalized output["mixed_decision_protocols"] = len(distinct) > 1 if len(distinct) > 1: output["comparability_note"] = ( "Ratings measure observed game performance across a connected " "match graph, but models used different recorded decision interfaces." ) return output def fit(summary: TournamentSummary) -> dict[str, object]: """Fit one tournament summary (kept for callers using the original API).""" return fit_results(summary.results) def main() -> None: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("summaries", type=Path, nargs="+") parser.add_argument("--prior-std", type=float, default=None, help="Centered log-strength Gaussian prior; use 2 to reproduce the public leaderboard") arguments = parser.parse_args() summaries = [ TournamentSummary.model_validate_json(path.read_text(encoding="utf-8")) for path in arguments.summaries ] results = [result for summary in summaries for result in summary.results] protocols: dict[str, set[str]] = {} for summary_path in arguments.summaries: manifest_path = summary_path.parent / "manifest.json" raw = json.loads(manifest_path.read_text(encoding="utf-8")) for model, config in raw["model_configs"].items(): protocol = config.get( "decision_protocol", GENERATIVE_DECISION_PROTOCOL ) protocols.setdefault(model, set()).add(protocol) print(json.dumps(fit_results(results, protocols, prior_std=arguments.prior_std), indent=2)) if __name__ == "__main__": main()