| import random
|
| import json
|
|
|
| class EloRank:
|
| def __init__(self, initial_rating=1000, k_factor=32):
|
| """
|
| Initialize the EloRank class.
|
| :param initial_rating: Initial ELO rating for each model.
|
| :param k_factor: The K-factor that determines the sensitivity of rating changes.
|
| """
|
| self.ratings = {}
|
| self.initial_rating = initial_rating
|
| self.k_factor = k_factor
|
| self.wins = {}
|
|
|
| def add_model(self, model_id):
|
| """
|
| Add a new model with the initial rating.
|
| :param model_id: Unique identifier for the model.
|
| """
|
| self.ratings[model_id] = self.initial_rating
|
| self.wins[model_id] = 0
|
|
|
| def record_match(self, winner, loser):
|
| """
|
| Update the ratings based on a match result.
|
| :param winner: Model ID of the winner.
|
| :param loser: Model ID of the loser.
|
| """
|
| rating_winner = self.ratings[winner]
|
| rating_loser = self.ratings[loser]
|
|
|
| expected_winner = self.expected_score(rating_winner, rating_loser)
|
| expected_loser = self.expected_score(rating_loser, rating_winner)
|
|
|
| self.ratings[winner] += self.k_factor * (1 - expected_winner)
|
| self.ratings[loser] += self.k_factor * (0 - expected_loser)
|
|
|
|
|
| self.wins[winner] += 1
|
|
|
| def expected_score(self, rating_a, rating_b):
|
| """
|
| Calculate the expected score for a model.
|
| :param rating_a: Rating of model A.
|
| :param rating_b: Rating of model B.
|
| :return: Expected score.
|
| """
|
| return 1 / (1 + 10 ** ((rating_b - rating_a) / 400))
|
|
|
| def get_rating(self, model_id):
|
| """
|
| Get the current rating of a model.
|
| :param model_id: Unique identifier for the model.
|
| :return: Current rating of the model.
|
| """
|
| return self.ratings.get(model_id, None)
|
|
|
| def get_wins(self, model_id):
|
| """
|
| Get the number of wins of a model.
|
| :param model_id: Unique identifier for the model.
|
| :return: Number of wins of the model.
|
| """
|
| return self.wins.get(model_id, 0)
|
|
|
| def get_top_models(self, n=2):
|
| """
|
| Get the top N models by rating.
|
| :param n: Number of top models to retrieve.
|
| :return: List of model IDs of the top models.
|
| """
|
| return sorted(self.ratings, key=self.ratings.get, reverse=True)[:n]
|
|
|
| def sample_next_match(self):
|
| """
|
| Sample the next match based on the probability proportional to the current rating.
|
| This approach helps accelerate the convergence of ranking.
|
| :return: Tuple of two model IDs for the next match.
|
| """
|
| model_ids = list(self.ratings.keys())
|
| probabilities = [self.ratings[model_id] for model_id in model_ids]
|
| total_rating = sum(probabilities)
|
| probabilities = [rating / total_rating for rating in probabilities]
|
|
|
|
|
| next_match = random.choices(model_ids, probabilities, k=2)
|
| while next_match[0] == next_match[1]:
|
| next_match = random.choices(model_ids, probabilities, k=2)
|
|
|
| return tuple(next_match)
|
|
|
| def process_match_records(self, file_path):
|
| """
|
| Process match records from a JSON file and update ratings and win counts accordingly.
|
| :param file_path: Path to the JSON file containing match records.
|
| """
|
| with open(file_path, 'r') as file:
|
| match_records = json.load(file)
|
|
|
| for record in match_records:
|
| winner = record['winner']
|
| model_1 = record['model_1']
|
| model_2 = record['model_2']
|
|
|
|
|
| if model_1 not in self.ratings:
|
| self.add_model(model_1)
|
| if model_2 not in self.ratings:
|
| self.add_model(model_2)
|
|
|
|
|
| if winner == model_1:
|
| self.record_match(model_1, model_2)
|
| elif winner == model_2:
|
| self.record_match(model_2, model_1)
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
