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}')
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.
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.
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.
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
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.
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).
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.