riix.models.elo

  1"""The Elo rating system"""
  2import math
  3import numpy as np
  4from riix.core.base import OnlineRatingSystem
  5from riix.utils.math_utils import sigmoid, sigmoid_scalar
  6
  7
  8class Elo(OnlineRatingSystem):
  9    """
 10    Implements the original Elo rating system.
 11    """
 12
 13    rating_dim = 1
 14
 15    def __init__(
 16        self,
 17        competitors: list,
 18        initial_rating: float = 1500.0,
 19        k: float = 32.0,
 20        alpha: float = math.log(10.0) / 400.0,
 21        update_method: str = 'online',
 22        dtype=np.float64,
 23    ):
 24        """
 25        Initializes the Elo rating system with the given parameters.
 26
 27        Parameters:
 28            competitors (list): A list of competitors to be rated within the system.
 29            initial_rating (float, optional): The initial Elo rating for new competitors. Defaults to 1500.0.
 30            k (float, optional): The K-factor, which controls the rate at which ratings change. Defaults to 32.0.
 31            alpha (float, optional): Scaling factor used in the calculation of expected scores. Defaults to log(10) / 400.
 32            update_method (str, optional): Method used to update ratings ('online' or other methods if implemented). Defaults to 'online'.
 33            dtype: The data type for internal numpy computations. Defaults to np.float64.
 34
 35        Initializes an Elo rating system with customizable settings for initial ratings, K-factor, and update method.
 36        """
 37        super().__init__(competitors)
 38        self.k = k
 39        self.alpha = alpha
 40        self.ratings = np.zeros(shape=self.num_competitors, dtype=dtype) + initial_rating
 41        self.cache = {'probs': None}
 42        if update_method == 'batched':
 43            self.update = self.batched_update
 44        elif update_method == 'online':
 45            self.update = self.online_update
 46        else:
 47            raise ValueError(f'update_method must be one of online or batched, got {update_method}')
 48
 49    def predict(self, matchups: np.ndarray, time_step: int = None, set_cache: bool = False):
 50        """
 51        Generates predictions for a series of matchups between competitors.
 52
 53        This method calculates the probability of the first competitor in each matchup winning
 54        based on their Elo ratings. The probabilities are computed using the sigmoid function
 55        applied to the rating differences, scaled by the alpha parameter.
 56
 57        Parameters:
 58            matchups (np.ndarray): A NumPy array of matchups, where each row represents a matchup
 59                                    and contains two integers indicating the indices of the competitors
 60                                    in the 'ratings' array.
 61            time_step (int, optional): A time step at which the predictions are made. This parameter
 62                                    is not used in the current implementation but can be utilized
 63                                    for time-dependent predictions. Defaults to None.
 64            set_cache (bool, optional): If True, caches the computed probabilities in the 'cache'
 65                                        attribute under the key 'probs'. Defaults to False.
 66
 67        Returns:
 68            np.ndarray: A NumPy array containing the predicted probabilities for the first competitor
 69                        in each matchup winning against the second.
 70        """
 71        ratings_1 = self.ratings[matchups[:, 0]]
 72        ratings_2 = self.ratings[matchups[:, 1]]
 73        probs = sigmoid(self.alpha * (ratings_1 - ratings_2))
 74        if set_cache:
 75            self.cache['probs'] = probs
 76        return probs
 77
 78    def get_pre_match_ratings(self, matchups: np.ndarray, **kwargs):
 79        return self.ratings[matchups]
 80
 81    def batched_update(self, matchups, outcomes, use_cache, **kwargs):
 82        """
 83        Apply a single update based on all results of the rating period.
 84
 85        Parameters:
 86            matchups: Matchup information for the rating period.
 87            outcomes: Results of the matchups.
 88            use_cache: Flag to use cached probabilities or calculate anew.
 89        """
 90        active_in_period = np.unique(matchups)
 91        masks = np.equal(matchups[:, :, None], active_in_period[None, :])  # N x 2 x active
 92        if use_cache:
 93            probs = self.cache['probs']
 94        else:
 95            probs = self.predict(time_step=None, matchups=matchups, set_cache=False)
 96        per_match_diff = (outcomes - probs)[:, None]
 97        per_match_diff = np.hstack([per_match_diff, -per_match_diff])
 98        per_competitor_diff = (per_match_diff[:, :, None] * masks).sum(axis=(0, 1))
 99        self.ratings[active_in_period] += self.k * per_competitor_diff
100
101    def online_update(self, matchups, outcomes, **kwargs):
102        """
103        Treats the matchups in the rating period as sequential events.
104
105        Parameters:
106            matchups: Sequential matchups in the rating period.
107            outcomes: Results of each matchup.
108            **kwargs: Additional parameters (not used).
109        """
110        for idx in range(matchups.shape[0]):
111            comp_1, comp_2 = matchups[idx]
112            diff = self.ratings[comp_1] - self.ratings[comp_2]
113            prob = sigmoid_scalar(self.alpha * diff)
114            update = self.k * (outcomes[idx] - prob)
115            self.ratings[comp_1] += update
116            self.ratings[comp_2] -= update
117
118    def print_leaderboard(self, num_places):
119        sorted_idxs = np.argsort(-self.ratings)[:num_places]
120        max_len = min(np.max([len(comp) for comp in self.competitors] + [10]), 25)
121        print(f'{"competitor": <{max_len}}\t{"rating"}')
122        for p_idx in range(num_places):
123            comp_idx = sorted_idxs[p_idx]
124            print(f'{self.competitors[comp_idx]: <{max_len}}\t{self.ratings[comp_idx]:.6f}')