riix.models.elo

The Elo rating system

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

Implements the original Elo rating system.

Elo( competitors: list, initial_rating: float = 1500.0, k: float = 32.0, alpha: float = 0.005756462732485115, update_method: str = 'iterative', dtype=<class 'numpy.float64'>)
16    def __init__(
17        self,
18        competitors: list,
19        initial_rating: float = 1500.0,
20        k: float = 32.0,
21        alpha: float = math.log(10.0) / 400.0,
22        update_method: str = 'iterative',
23        dtype=np.float64,
24    ):
25        """
26        Initializes the Elo rating system with the given parameters.
27
28        Parameters:
29            competitors (list): A list of competitors to be rated within the system.
30            initial_rating (float, optional): The initial Elo rating for new competitors. Defaults to 1500.0.
31            k (float, optional): The K-factor, which controls the rate at which ratings change. Defaults to 32.0.
32            alpha (float, optional): Scaling factor used in the calculation of expected scores. Defaults to log(10) / 400.
33            update_method (str, optional): Method used to update ratings ('iterative' or other methods if implemented). Defaults to 'iterative'.
34            dtype: The data type for internal numpy computations. Defaults to np.float64.
35
36        Initializes an Elo rating system with customizable settings for initial ratings, K-factor, and update method.
37        """
38        super().__init__(competitors)
39        self.k = k
40        self.alpha = alpha
41        self.ratings = np.zeros(shape=self.num_competitors, dtype=dtype) + initial_rating
42        self.cache = {'probs': None}
43        if update_method == 'batched':
44            self.update = self.batched_update
45        elif update_method == 'iterative':
46            self.update = self.iterative_update

Initializes the Elo rating system with the given parameters.

Arguments:
  • competitors (list): A list of competitors to be rated within the system.
  • initial_rating (float, optional): The initial Elo rating for new competitors. Defaults to 1500.0.
  • k (float, optional): The K-factor, which controls the rate at which ratings change. Defaults to 32.0.
  • alpha (float, optional): Scaling factor used in the calculation of expected scores. Defaults to log(10) / 400.
  • update_method (str, optional): Method used to update ratings ('iterative' or other methods if implemented). Defaults to 'iterative'.
  • dtype: The data type for internal numpy computations. Defaults to np.float64.

Initializes an Elo rating system with customizable settings for initial ratings, K-factor, and update method.

rating_dim = 1
k
alpha
ratings
cache
def predict( self, matchups: numpy.ndarray, time_step: int = None, set_cache: bool = False):
48    def predict(self, matchups: np.ndarray, time_step: int = None, set_cache: bool = False):
49        """
50        Generates predictions for a series of matchups between competitors.
51
52        This method calculates the probability of the first competitor in each matchup winning
53        based on their Elo ratings. The probabilities are computed using the sigmoid function
54        applied to the rating differences, scaled by the alpha parameter.
55
56        Parameters:
57            matchups (np.ndarray): A NumPy array of matchups, where each row represents a matchup
58                                    and contains two integers indicating the indices of the competitors
59                                    in the 'ratings' array.
60            time_step (int, optional): A time step at which the predictions are made. This parameter
61                                    is not used in the current implementation but can be utilized
62                                    for time-dependent predictions. Defaults to None.
63            set_cache (bool, optional): If True, caches the computed probabilities in the 'cache'
64                                        attribute under the key 'probs'. Defaults to False.
65
66        Returns:
67            np.ndarray: A NumPy array containing the predicted probabilities for the first competitor
68                        in each matchup winning against the second.
69        """
70        ratings_1 = self.ratings[matchups[:, 0]]
71        ratings_2 = self.ratings[matchups[:, 1]]
72        probs = sigmoid(self.alpha * (ratings_1 - ratings_2))
73        if set_cache:
74            self.cache['probs'] = probs
75        return probs

Generates predictions for a series of matchups between competitors.

This method calculates the probability of the first competitor in each matchup winning based on their Elo ratings. The probabilities are computed using the sigmoid function applied to the rating differences, scaled by the alpha parameter.

Arguments:
  • matchups (np.ndarray): A NumPy array of matchups, where each row represents a matchup and contains two integers indicating the indices of the competitors in the 'ratings' array.
  • time_step (int, optional): A time step at which the predictions are made. This parameter is not used in the current implementation but can be utilized for time-dependent predictions. Defaults to None.
  • set_cache (bool, optional): If True, caches the computed probabilities in the 'cache' attribute under the key 'probs'. Defaults to False.
Returns:

np.ndarray: A NumPy array containing the predicted probabilities for the first competitor in each matchup winning against the second.

def get_pre_match_ratings(self, matchups: numpy.ndarray, **kwargs):
77    def get_pre_match_ratings(self, matchups: np.ndarray, **kwargs):
78        return self.ratings[matchups]

Returns the ratings for competitors at the timestep of the matchups Useful when using pre-match ratings as features in downstream ML pipelines

Arguments:
  • matchups (np.ndarray of shape (n,2)): competitor indices
  • time_step (optional int)
Returns:

np.ndarray of shape (n,2): ratings for specified competitors

def batched_update(self, matchups, outcomes, use_cache, **kwargs):
80    def batched_update(self, matchups, outcomes, use_cache, **kwargs):
81        """
82        Apply a single update based on all results of the rating period.
83
84        Parameters:
85            matchups: Matchup information for the rating period.
86            outcomes: Results of the matchups.
87            use_cache: Flag to use cached probabilities or calculate anew.
88        """
89        active_in_period = np.unique(matchups)
90        masks = np.equal(matchups[:, :, None], active_in_period[None, :])  # N x 2 x active
91        if use_cache:
92            probs = self.cache['probs']
93        else:
94            probs = self.predict(time_step=None, matchups=matchups, set_cache=False)
95        per_match_diff = (outcomes - probs)[:, None]
96        per_match_diff = np.hstack([per_match_diff, -per_match_diff])
97        per_competitor_diff = (per_match_diff[:, :, None] * masks).sum(axis=(0, 1))
98        self.ratings[active_in_period] += self.k * per_competitor_diff

Apply a single update based on all results of the rating period.

Arguments:
  • matchups: Matchup information for the rating period.
  • outcomes: Results of the matchups.
  • use_cache: Flag to use cached probabilities or calculate anew.
def iterative_update(self, matchups, outcomes, **kwargs):
100    def iterative_update(self, matchups, outcomes, **kwargs):
101        """
102        Treats the matchups in the rating period as sequential events.
103
104        Parameters:
105            matchups: Sequential matchups in the rating period.
106            outcomes: Results of each matchup.
107            **kwargs: Additional parameters (not used).
108        """
109        for idx in range(matchups.shape[0]):
110            comp_1, comp_2 = matchups[idx]
111            diff = self.ratings[comp_1] - self.ratings[comp_2]
112            prob = sigmoid_scalar(self.alpha * diff)
113            update = self.k * (outcomes[idx] - prob)
114            self.ratings[comp_1] += update
115            self.ratings[comp_2] -= update

Treats the matchups in the rating period as sequential events.

Arguments:
  • matchups: Sequential matchups in the rating period.
  • outcomes: Results of each matchup.
  • **kwargs: Additional parameters (not used).
def print_leaderboard(self, num_places):
117    def print_leaderboard(self, num_places):
118        sorted_idxs = np.argsort(-self.ratings)[:num_places]
119        max_len = min(np.max([len(comp) for comp in self.competitors] + [10]), 25)
120        print(f'{"competitor": <{max_len}}\t{"rating"}')
121        for p_idx in range(num_places):
122            comp_idx = sorted_idxs[p_idx]
123            print(f'{self.competitors[comp_idx]: <{max_len}}\t{self.ratings[comp_idx]:.6f}')

Prints the leaderboard of the rating system.

Arguments:
  • num_places int: The number of top places to display on the leaderboard.