diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..3de1a8b --- /dev/null +++ b/.gitignore @@ -0,0 +1,7 @@ +*.csv +*.pkl +__py* +setup* +build/* +dist/* +whole_history_rating.egg-info/* \ No newline at end of file diff --git a/README.md b/README.md index 105611c..4bddaed 100644 --- a/README.md +++ b/README.md @@ -1,80 +1,161 @@ -# whole_history_rating -A python conversion from the ruby implementation of Rémi Coulom's Whole-History Rating (WHR) algorithm. -The original ruby code can be found [here](https://github.com/goshrine/whole_history_rating) +# Whole History Rating (WHR) Python Implementation +This Python library is a conversion from the original Ruby implementation of Rémi Coulom's Whole-History Rating (WHR) algorithm, designed to provide a dynamic rating system for games or matches where players' skills are continuously estimated over time. -Installation ------------- +The original Ruby code is available here at [goshrine](https://github.com/goshrine/whole_history_rating). + +## Installation + +To install the library, use the following command: ```shell pip install whole-history-rating ``` -Usage ------ +## Usage + +### Basic Setup + +Start by importing the library and initializing the base WHR object: -```py +```python from whr import whole_history_rating whr = whole_history_rating.Base() +``` + +### Creating Games + +Add games to the system using `create_game()` method. It takes the names of the black and white players, the winner ('B' for black, 'W' for white), the day number, and an optional handicap (generally less than 500 elo). -# Base.create_game() arguments: black player name, white player name, winner, day number, handicap -# Handicap should generally be less than 500 elo +```python whr.create_game("shusaku", "shusai", "B", 1, 0) whr.create_game("shusaku", "shusai", "W", 2, 0) whr.create_game("shusaku", "shusai", "W", 3, 0) +``` + + +### Refining Ratings Towards Stability + +To achieve accurate and stable ratings, the WHR algorithm allows for iterative refinement. This process can be controlled manually or handled automatically to adjust player ratings until they reach a stable state. + +#### Manual Iteration + +For manual control over the iteration process, specify the number of iterations you wish to perform. This approach gives you direct oversight over the refinement steps. -# Iterate the WHR algorithm towards convergence with more players/games, more iterations are needed. +```python whr.iterate(50) +``` + +This command will perform 50 iterations, incrementally adjusting player ratings towards stability with each step. + +#### Automatic Iteration + +For a more hands-off approach, the algorithm can automatically iterate until the Elo ratings stabilize within a specified precision. Automatic iteration is particularly useful when dealing with large datasets or when seeking to automate the rating process. + +```python +whr.auto_iterate(time_limit=10, precision=1e-3, batch_size=10) +``` + +- `time_limit` (optional): Sets a maximum duration (in seconds) for the iteration process. If `None` (the default), the algorithm will run indefinitely until the specified precision is achieved. +- `precision` (optional): Defines the desired level of accuracy for the ratings' stability. The default value is `0.001`, indicating that iteration will stop when changes between iterations are less than or equal to this threshold. +- `batch_size` (optional): Determines the number of iterations to perform before checking for convergence and, if a `time_limit` is set, before evaluating whether the time limit has been reached. The default value is `10`, balancing between frequent convergence checks and computational efficiency. + +This automated process allows the algorithm to efficiently converge to stable ratings, adjusting the number of iterations dynamically based on the complexity of the data and the specified precision and time constraints. + + +### Viewing Ratings + +Retrieve and view player ratings, which include the day number, elo rating, and uncertainty: + +```python +# Example output for whr.ratings_for_player("shusaku") +print(whr.ratings_for_player("shusaku")) +# Output: +# [[1, -43, 0.84], +# [2, -45, 0.84], +# [3, -45, 0.84]] + +# Example output for whr.ratings_for_player("shusai") +print(whr.ratings_for_player("shusai")) +# Output: +# [[1, 43, 0.84], +# [2, 45, 0.84], +# [3, 45, 0.84]] + +``` + +You can also view or retrieve all ratings in order: + +```python +whr.print_ordered_ratings(current=False) # Set `current=True` for the latest rankings only. +ratings = whr.get_ordered_ratings(current=False, compact=False) # Set `compact=True` for a condensed list. +``` + +### Predicting Match Outcomes + +Predict the outcome of future matches, including between non-existent players: + +```python +# Example of predicting a future match outcome +probability = whr.probability_future_match("shusaku", "shusai", 0) +print(f"Win probability: shusaku: {probability[0]*100}%; shusai: {probability[1]*100}%") +# Output: +# Win probability: shusaku: 37.24%; shusai: 62.76% <== this is printed +# (0.3724317501643667, 0.6275682498356332) +``` + + +### Enhanced Batch Loading of Games + +This feature facilitates the batch loading of multiple games simultaneously by accepting a list of strings, where each string encapsulates the details of a single game. To accommodate names with both first and last names and ensure flexibility in data formatting, you can specify a custom separator (e.g., a comma) to delineate the game attributes. + +#### Standard Loading + +Without specifying a separator, the default space (' ') is used to split the game details: + +```python +whr.load_games([ + "shusaku shusai B 1 0", # Game 1: Shusaku vs. Shusai, Black wins, Day 1, no handicap. + "shusaku shusai W 2 0", # Game 2: Shusaku vs. Shusai, White wins, Day 2, no handicap. + "shusaku shusai W 3 0" # Game 3: Shusaku vs. Shusai, White wins, Day 3, no handicap. +]) +``` + +#### Custom Separator for Complex Names + +When game details include names with spaces, such as first and last names, utilize the `separator` parameter to define an alternative delimiter, ensuring the integrity of each data point: + +```python +whr.load_games([ + "John Doe, Jane Smith, W, 1, 0", # Game 1: John Doe vs. Jane Smith, White wins, Day 1, no handicap. + "Emily Chen, Liam Brown, B, 2, 0" # Game 2: Emily Chen vs. Liam Brown, Black wins, Day 2, no handicap. +], separator=",") +``` + +This method allows for a clear and error-free way to load game data, especially when player names or game details include spaces, providing a robust solution for managing diverse datasets. + + +### Saving and Loading States + +Save the current state to a file and reload it later to avoid recalculating: -# Or let the module iterate until the elo is stable (precision by default 10E-3) with a time limit of 10 seconds by default -whr.auto_iterate(time_limit = 10, precision = 10E-3) - -# Results are stored in one triplet for each game: [day_number, elo_rating, uncertainty] -whr.ratings_for_player("shusaku") => - [[1, -43, 84], - [2, -45, 84], - [3, -45, 84]] -whr.ratings_for_player("shusai") => - [[1, 43, 84], - [2, 45, 84], - [3, 45, 84]] - -# You can print or get all ratings ordered -whr.print_ordered_ratings(current=False) # current to True to only get the last rank estimation -whr.get_ordered_ratings(current=False, compact=False) # compact to True to not have the name before each ranks - -# You can get a prediction for a future game between two players (even non existing players) -# Base.probability_future_match() arguments: black player name, white player name, handicap -whr.probability_future_match("shusaku", "shusai",0) => - win probability: shusaku:37.24%; shusai:62.76% - -# You can load several games all together using a file or a list of string representing the game -# all elements in list must be like: "black_name white_name winner time_step handicap extras" -# you can exclude handicap (default=0) and extras (default={}) -whr.load_games(["shusaku shusai B 1 0", "shusaku shusai W 2", "shusaku shusai W 3 0"]) -whr.load_games(["firstname1 name1, firstname2 name2, W, 1"], separator=",") - -# You can save and load a base (you don't have to redo all iterations) -whr.save_base(path) -whr2 = whole_history_rating.Base.load_base(path) +```python +whr.save_base('path_to_save.whr') +whr2 = whole_history_rating.Base.load_base('path_to_save.whr') ``` -Optional Configuration ----------------------- +## Optional Configuration -One of the meta parameters to WHR is the variance of rating change over one time step, `w2`, -which determines how much a player's rating is likely change in one day. Higher numbers allow for faster progress. -The default value is 300, which is fairly high. -Rémi Coulom in his paper, used w2=14 to get his [results](https://www.remi-coulom.fr/WHR/WHR.pdf) +Adjust the `w2` parameter, which influences the variance of rating change over time, allowing for faster or slower progression. The default is set to 300, but Rémi Coulom used a value of 14 in his paper to achieve his results. -```py -whr = whole_history_rating.Base({'w2':14}) +```python +whr = whole_history_rating.Base({'w2': 14}) ``` -You can also set the base not case sensitive. "shusaku" and "ShUsAkU" will be the same. +Enable case-insensitive player names to treat "shusaku" and "ShUsAkU" as the same entity: -```py -whr = whole_history_rating.Base({'uncased':True}) +```python +whr = whole_history_rating.Base({'uncased': True}) ``` diff --git a/tests/whr_test.py b/tests/whr_test.py index 2f45b97..a7f985a 100644 --- a/tests/whr_test.py +++ b/tests/whr_test.py @@ -1,3 +1,4 @@ +import copy import sys import os import pytest @@ -56,6 +57,24 @@ def test_winrates_should_be_inversely_proportional_with_handicap(): def test_output(): + whr = whole_history_rating.Base() + whr.create_game("shusaku", "shusai", "B", 1, 0) + whr.create_game("shusaku", "shusai", "W", 2, 0) + whr.create_game("shusaku", "shusai", "W", 3, 0) + whr.iterate(50) + assert [ + (1, -43, 0.84), + (2, -45, 0.84), + (3, -45, 0.84), + ] == whr.ratings_for_player("shusaku") + assert [ + (1, 43, 0.84), + (2, 45, 0.84), + (3, 45, 0.84), + ] == whr.ratings_for_player("shusai") + + +def test_output2(): whr = whole_history_rating.Base() whr.create_game("shusaku", "shusai", "B", 1, 0) whr.create_game("shusaku", "shusai", "W", 2, 0) @@ -64,16 +83,16 @@ def test_output(): whr.create_game("shusaku", "shusai", "W", 4, 0) whr.iterate(50) assert [ - [1, -92, 71], - [2, -94, 71], - [3, -95, 71], - [4, -96, 72], + (1, -92, 0.71), + (2, -94, 0.71), + (3, -95, 0.71), + (4, -96, 0.72), ] == whr.ratings_for_player("shusaku") assert [ - [1, 92, 71], - [2, 94, 71], - [3, 95, 71], - [4, 96, 72], + (1, 92, 0.71), + (2, 94, 0.71), + (3, 95, 0.71), + (4, 96, 0.72), ] == whr.ratings_for_player("shusai") @@ -136,22 +155,25 @@ def test_loading_several_games_at_once(capsys): whr.auto_iterate() # test getting ratings for player shusaku (day, elo, uncertainty) assert whr.ratings_for_player("shusaku") == [ - [1, 26.0, 70.0], - [2, 25.0, 70.0], - [3, 24.0, 70.0], + (1, 26.0, 0.70), + (2, 25.0, 0.70), + (3, 24.0, 0.70), ] # test getting ratings for player shusai, only current elo and uncertainty - assert whr.ratings_for_player("shusai", current=True) == (87.0, 84.0) + assert whr.ratings_for_player("shusai", current=True) == (87.0, 0.84) # test getting probability of future match between shusaku and nobody2 (which default to 1 win 1 loss) assert whr.probability_future_match("shusai", "nobody2", 0) == ( 0.6224906898220315, 0.3775093101779684, ) + display = "win probability: shusai:62.25%; nobody2:37.75%\n" + captured = capsys.readouterr() + assert display == captured.out # test getting log likelihood of base assert whr.log_likelihood() == 0.7431542354571272 # test printing ordered ratings whr.print_ordered_ratings() - display = "win probability: shusai:0.62%; nobody2:0.38%\nnobody => [-112.37545390067574]\nshusaku => [25.552142942931102, 24.669738398550702, 24.49953062693439]\nshusai => [84.74972643795506, 86.17200033461006, 86.88207745833284]\n" + display = "nobody => [-112.37545390067574]\nshusaku => [25.552142942931102, 24.669738398550702, 24.49953062693439]\nshusai => [84.74972643795506, 86.17200033461006, 86.88207745833284]\n" captured = capsys.readouterr() assert display == captured.out # test printing ordered ratings, only current elo @@ -176,6 +198,39 @@ def test_loading_several_games_at_once(capsys): # test loading base whr2 = whole_history_rating.Base.load_base("test_whr.pkl") # test inspecting the first game - whr_games = [x.inspect() for x in whr.games] - whr2_games = [x.inspect() for x in whr2.games] + whr_games = [str(x) for x in whr.games] + whr2_games = [str(x) for x in whr2.games] assert whr_games == whr2_games + + +def test_auto_iterate(capsys): + whr = whole_history_rating.Base() + # test loading several games at once + test_games = [ + "shusaku; shusai; B; 1", + "shusaku;shusai;W;2;0", + " shusaku ; shusai ;W ; 3; {'w2':300}", + "shusaku;nobody;B;3;0;{'w2':300}", + ] + whr.load_games(test_games, separator=";") + # test auto iterating to get convergence + whr1 = copy.deepcopy(whr) + whr2 = copy.deepcopy(whr) + whr3 = copy.deepcopy(whr) + whr4 = copy.deepcopy(whr) + whr5 = copy.deepcopy(whr) + iterations1, is_stable1 = whr1.auto_iterate(batch_size=1) + assert iterations1 == 9 + assert is_stable1 + iterations2, is_stable2 = whr2.auto_iterate() + assert iterations2 == 20 + assert is_stable2 + iterations3, is_stable3 = whr3.auto_iterate(precision=0.5, batch_size=1) + assert iterations3 == 6 + assert is_stable3 + iterations4, is_stable4 = whr4.auto_iterate(precision=0.9, batch_size=1) + assert iterations4 == 5 + assert is_stable4 + iterations5, is_stable5 = whr5.auto_iterate(time_limit=1, batch_size=1) + assert iterations5 == 9 + assert is_stable5 diff --git a/whr/game.py b/whr/game.py index 43781f3..be893e6 100644 --- a/whr/game.py +++ b/whr/game.py @@ -1,16 +1,30 @@ +from __future__ import annotations + import sys +from typing import Any + +from whr import player as P +from whr import playerday as PD class Game: - def __init__(self, black, white, winner, time_step, handicap=0, extras=None): + def __init__( + self, + black: P.Player, + white: P.Player, + winner: str, + time_step: int, + handicap: float = 0, + extras: dict[str, Any] | None = None, + ): self.day = time_step self.white_player = white self.black_player = black self.winner = winner.upper() self.handicap = handicap self.handicap_proc = handicap - self.bpd = None - self.wpd = None + self.bpd: PD.PlayerDay | None = None + self.wpd: PD.PlayerDay | None = None if extras is None: self.extras = dict() self.extras["komi"] = 6.5 @@ -19,7 +33,25 @@ def __init__(self, black, white, winner, time_step, handicap=0, extras=None): if self.extras.get("komi") is None: self.extras["komi"] = 6.5 - def opponents_adjusted_gamma(self, player): + def __str__(self) -> str: + return f"W:{self.white_player.name}(r={self.wpd.r if self.wpd is not None else '?'}) B:{self.black_player.name}(r={self.bpd.r if self.bpd is not None else '?'}) winner = {self.winner}, komi = {self.extras['komi']}, handicap = {self.handicap}" + + def opponents_adjusted_gamma(self, player: P.Player) -> float: + """ + Calculates the adjusted gamma value of a player's opponent. This is based on the opponent's + Elo rating adjusted for the game's handicap. + + Parameters: + player (P.Player): The player for whom to calculate the opponent's adjusted gamma. + + Returns: + float: The adjusted gamma value of the opponent. + + Raises: + AttributeError: If the player days are not set or the player is not part of the game. + """ + if self.bpd is None or self.wpd is None: + raise AttributeError("black player day and white player day must be set") if player == self.white_player: opponent_elo = self.bpd.elo + self.handicap elif player == self.black_player: @@ -35,12 +67,31 @@ def opponents_adjusted_gamma(self, player): raise AttributeError("bad adjusted gamma") return rval - def opponent(self, player): + def opponent(self, player: P.Player) -> P.Player: + """ + Returns the opponent of the specified player in this game. + + Parameters: + player (P.Player): The player whose opponent is to be found. + + Returns: + P.Player: The opponent player. + """ if player == self.white_player: return self.black_player return self.white_player - def prediction_score(self): + def prediction_score(self) -> float: + """ + Calculates the accuracy of the prediction for the game's outcome. + Returns a score based on the actual outcome compared to the predicted probabilities: + - Returns 1.0 if the prediction matches the actual outcome (white or black winning as predicted). + - Returns 0.5 if the win probability is exactly 0.5, indicating uncertainty. + - Returns 0.0 if the prediction does not match the actual outcome. + + Returns: + float: The prediction score of the game. + """ if self.white_win_probability() == 0.5: return 0.5 return ( @@ -52,15 +103,37 @@ def prediction_score(self): else 0.0 ) - def inspect(self): - return f"W:{self.white_player.name}(r={self.wpd.r if self.wpd is not None else '?'}) B:{self.black_player.name}(r={self.bpd.r if self.bpd is not None else '?'}) winner = {self.winner}, komi = {self.extras['komi']}, handicap = {self.handicap}" + def white_win_probability(self) -> float: + """ + Calculates the win probability for the white player based on their gamma value and + the adjusted gamma value of their opponent. + + Returns: + float: The win probability for the white player. + + Raises: + AttributeError: If the white player day is not set. + """ + if self.wpd is None: + raise AttributeError("white player day must be set") - def white_win_probability(self): return self.wpd.gamma() / ( self.wpd.gamma() + self.opponents_adjusted_gamma(self.white_player) ) - def black_win_probability(self): + def black_win_probability(self) -> float: + """ + Calculates the win probability for the black player based on their gamma value and + the adjusted gamma value of their opponent. + + Returns: + float: The win probability for the black player. + + Raises: + AttributeError: If the black player day is not set. + """ + if self.bpd is None: + raise AttributeError("black player day must be set") return self.bpd.gamma() / ( self.bpd.gamma() + self.opponents_adjusted_gamma(self.black_player) ) diff --git a/whr/player.py b/whr/player.py index 303d077..ba260a2 100644 --- a/whr/player.py +++ b/whr/player.py @@ -1,21 +1,33 @@ +from __future__ import annotations + import math import sys import bisect +from typing import Any import numpy as np +import numpy.typing as npt -from whr.playerday import PlayerDay from whr.utils import UnstableRatingException +from whr import playerday as PD +from whr import game as G class Player: - def __init__(self, name, config): + def __init__(self, name: str, config: dict[str, Any]): self.name = name self.debug = config["debug"] self.w2 = (math.sqrt(config["w2"]) * math.log(10) / 400) ** 2 - self.days = [] + self.days: list[PD.PlayerDay] = [] + + def log_likelihood(self) -> float: + """Computes the log likelihood of the player's ratings over all days. + + Incorporates both the likelihood of the observed game outcomes and the prior based on changes in rating over time. - def log_likelihood(self): + Returns: + float: The log likelihood value for the player's ratings. + """ result = 0.0 sigma2 = self.compute_sigma2() n = len(self.days) @@ -24,12 +36,12 @@ def log_likelihood(self): if i < (n - 1): rd = self.days[i].r - self.days[i + 1].r prior += (1 / (math.sqrt(2 * math.pi * sigma2[i]))) * math.exp( - -(rd ** 2) / 2 / sigma2[i] + -(rd**2) / 2 / sigma2[i] ) if i > 0: rd = self.days[i].r - self.days[i - 1].r prior += (1 / (math.sqrt(2 * math.pi * sigma2[i - 1]))) * math.exp( - -(rd ** 2) / (2 * sigma2[i - 1]) + -(rd**2) / (2 * sigma2[i - 1]) ) if prior == 0: result += self.days[i].log_likelihood() @@ -46,7 +58,18 @@ def log_likelihood(self): return result @staticmethod - def hessian(days, sigma2): + def hessian( + days: list[PD.PlayerDay], sigma2: list[float] + ) -> tuple[npt.NDArray[np.float64], npt.NDArray[np.float64]]: + """Computes the Hessian matrix for the log likelihood function. + + Args: + days (list[PD.PlayerDay]): A list of PD.PlayerDay instances for the player. + sigma2 (list[float]): A list of variance values between consecutive days. + + Returns: + tuple[npt.NDArray[np.float64], npt.NDArray[np.float64]]: A tuple containing the diagonal and sub-diagonal elements of the Hessian matrix. + """ n = len(days) diagonal = np.zeros((n,)) sub_diagonal = np.zeros((n - 1,)) @@ -59,9 +82,21 @@ def hessian(days, sigma2): diagonal[row] = days[row].log_likelihood_second_derivative() + prior - 0.001 for i in range(n - 1): sub_diagonal[i] = 1 / sigma2[i] - return [diagonal, sub_diagonal] + return (diagonal, sub_diagonal) + + def gradient( + self, r: list[float], days: list[PD.PlayerDay], sigma2: list[float] + ) -> list[float]: + """Computes the gradient of the log likelihood function. - def gradient(self, r, days, sigma2): + Args: + r (list[float]): A list of rating values for the player on different days. + days (list[PD.PlayerDay]): A list of PD.PlayerDay instances for the player. + sigma2 (list[float]): A list of variance values between consecutive days. + + Returns: + list[float]: A list containing the gradient of the log likelihood function. + """ g = [] n = len(days) for idx, day in enumerate(days): @@ -75,7 +110,8 @@ def gradient(self, r, days, sigma2): g.append(day.log_likelihood_derivative() + prior) return g - def run_one_newton_iteration(self): + def run_one_newton_iteration(self) -> None: + """Runs a single iteration of Newton's method to update player ratings.""" for day in self.days: day.clear_game_terms_cache() if len(self.days) == 1: @@ -83,13 +119,19 @@ def run_one_newton_iteration(self): elif len(self.days) > 1: self.update_by_ndim_newton() - def compute_sigma2(self): + def compute_sigma2(self) -> list[float]: + """Computes the variance values used as the prior for rating changes. + + Returns: + list[float]: A list of variance values between consecutive rating days. + """ sigma2 = [] for d1, d2 in zip(*(self.days[i:] for i in range(2))): sigma2.append(abs(d2.day - d1.day) * self.w2) return sigma2 - def update_by_ndim_newton(self): + def update_by_ndim_newton(self) -> None: + """Updates the player's ratings using a multidimensional Newton-Raphson method.""" # r r = [d.r for d in self.days] @@ -130,7 +172,12 @@ def update_by_ndim_newton(self): for idx, day in enumerate(self.days): day.r = day.r - x[idx] - def covariance(self): + def covariance(self) -> npt.NDArray[np.float64]: + """Computes the covariance matrix of the player's rating estimations. + + Returns: + The covariance matrix for the player's ratings. + """ r = [d.r for d in self.days] sigma2 = self.compute_sigma2() @@ -177,7 +224,14 @@ def covariance(self): return mat - def update_uncertainty(self): + def update_uncertainty(self) -> float | None: + """Updates the uncertainty measure for each day based on the covariance matrix. + + If the player has played on multiple days, this method calculates the variance for each day from the covariance matrix and updates each day's uncertainty value accordingly. If the player has not played on any day, a default uncertainty value is returned. + + Returns: + float | None: The default uncertainty value of 5 if the player has no recorded days, otherwise None after updating the uncertainty values for all recorded days. + """ if len(self.days) > 0: c = self.covariance() u = [c[i, i] for i in range(len(self.days))] # u = variance @@ -186,17 +240,22 @@ def update_uncertainty(self): return None return 5 - def add_game(self, game): + def add_game(self, game: G.Game) -> None: + """Adds a game to the player's record, updating or creating a new PD.PlayerDay instance as necessary. + + Args: + game (G.Game): The game to add to the player's record. + """ all_days = [x.day for x in self.days] if game.day not in all_days: day_index = bisect.bisect_right(all_days, game.day) - new_pday = PlayerDay(self, game.day) + new_pday = PD.PlayerDay(self, game.day) if len(self.days) == 0: new_pday.is_first_day = True new_pday.set_gamma(1) else: # still not perfect because gamma of day index can more farther if more games were not added in order - new_pday.set_gamma(self.days[day_index-1].gamma()) + new_pday.set_gamma(self.days[day_index - 1].gamma()) self.days.insert(day_index, new_pday) else: day_index = all_days.index(game.day) diff --git a/whr/playerday.py b/whr/playerday.py index 9ddd30f..2553156 100644 --- a/whr/playerday.py +++ b/whr/playerday.py @@ -1,9 +1,14 @@ +from __future__ import annotations + import math import sys +from whr import player as P +from whr import game as G + class PlayerDay: - def __init__(self, player, day): + def __init__(self, player: P.Player, day: int): self.day = day self.player = player self.is_first_day = False @@ -11,26 +16,53 @@ def __init__(self, player, day): self.lost_games = [] self._won_game_terms = None self._lost_game_terms = None + self.uncertainty: float = -1 + + def set_gamma(self, value: float) -> None: + """Sets the player's performance rating (gamma) for this day. - def set_gamma(self, value): + Args: + value (float): The new gamma value. + """ self.r = math.log(value) - def gamma(self): + def gamma(self) -> float: + """Calculates the player's performance rating (gamma) based on their rating. + + Returns: + float: The player's gamma value. + """ return math.exp(self.r) @property - def elo(self): + def elo(self) -> float: + """Calculates the ELO rating from the player's gamma value. + + Returns: + float: The ELO rating. + """ return (self.r * 400) / (math.log(10)) @elo.setter - def elo(self, value): + def elo(self, value: float) -> None: + """Sets the player's ELO rating, adjusting their internal rating accordingly. + + Args: + value (float): The new ELO rating. + """ self.r = value * (math.log(10) / 400) - def clear_game_terms_cache(self): + def clear_game_terms_cache(self) -> None: + """Clears the cached terms for games won and lost, forcing recalculation.""" self._won_game_terms = None self._lost_game_terms = None - def won_game_terms(self): + def won_game_terms(self) -> list[list[float]]: + """Calculates terms for games won by the player on this day. + + Returns: + list[list[float]]: A list of terms used for calculations, including the opponent's adjusted gamma. + """ if self._won_game_terms is None: self._won_game_terms = [] for g in self.won_games: @@ -45,7 +77,12 @@ def won_game_terms(self): self._won_game_terms.append([1.0, 0.0, 1.0, 1.0]) return self._won_game_terms - def lost_game_terms(self): + def lost_game_terms(self) -> list[list[float]]: + """Calculates terms for games lost by the player on this day. + + Returns: + list[list[float]]: A list of terms used for calculations, including the opponent's adjusted gamma. + """ if self._lost_game_terms is None: self._lost_game_terms = [] for g in self.lost_games: @@ -60,19 +97,34 @@ def lost_game_terms(self): self._lost_game_terms.append([0.0, 1.0, 1.0, 1.0]) return self._lost_game_terms - def log_likelihood_second_derivative(self): + def log_likelihood_second_derivative(self) -> float: + """Calculates the second derivative of the log likelihood of the player's rating. + + Returns: + float: The second derivative of the log likelihood. + """ result = 0.0 for _, _, c, d in self.won_game_terms() + self.lost_game_terms(): result += (c * d) / ((c * self.gamma() + d) ** 2.0) return -1 * self.gamma() * result - def log_likelihood_derivative(self): + def log_likelihood_derivative(self) -> float: + """Calculates the derivative of the log likelihood of the player's rating. + + Returns: + float: The derivative of the log likelihood. + """ tally = 0.0 for _, _, c, d in self.won_game_terms() + self.lost_game_terms(): tally += c / (c * self.gamma() + d) return len(self.won_game_terms()) - self.gamma() * tally - def log_likelihood(self): + def log_likelihood(self) -> float: + """Calculates the log likelihood of the player's rating based on games played. + + Returns: + float: The log likelihood. + """ tally = 0.0 for a, b, c, d in self.won_game_terms(): tally += math.log(a * self.gamma()) @@ -82,7 +134,12 @@ def log_likelihood(self): tally -= math.log(c * self.gamma() + d) return tally - def add_game(self, game): + def add_game(self, game: G.Game) -> None: + """Adds a game to this player's record, categorizing it as won or lost. + + Args: + game (G.Game): The game to add. + """ if (game.winner == "W" and game.white_player == self.player) or ( game.winner == "B" and game.black_player == self.player ): @@ -90,7 +147,8 @@ def add_game(self, game): else: self.lost_games.append(game) - def update_by_1d_newtons_method(self): + def update_by_1d_newtons_method(self) -> None: + """Updates the player's rating using one-dimensional Newton's method.""" dlogp = self.log_likelihood_derivative() d2logp = self.log_likelihood_second_derivative() dr = dlogp / d2logp diff --git a/whr/utils.py b/whr/utils.py index 0efca38..bb6f450 100644 --- a/whr/utils.py +++ b/whr/utils.py @@ -1,20 +1,28 @@ +from __future__ import annotations + + class UnstableRatingException(Exception): - pass + pass + + +def test_stability( + v1: list[list[float]], v2: list[list[float]], precision: float = 10e-3 +) -> bool: + """Tests whether two lists of lists of floats are approximately equal within a specified precision. + + This function flattens each list of lists into a single list and compares each corresponding element from the two lists. If the absolute difference between any pair of elements exceeds the given precision, the lists are considered not equal. -def test_stability(v1, v2, precision=10e-3): - """tests if two lists of lists of floats are equal but a certain precision - Args: - v1 (list[list[float]]): first list containing ints - v2 (list[list[float]]): second list containing ints - precision (float, optional): the precision after which v1 and v2 are not equal - + v1 (list[list[float]]): The first list of lists of floats. + v2 (list[list[float]]): The second list of lists of floats. + precision (float, optional): The precision threshold below which the values are considered equal. Defaults to 0.01. + Returns: - bool: True if the two lists are close enought, False otherwise + bool: True if the two lists are considered close enough, i.e., no pair of corresponding elements differs by more than the specified precision. False otherwise. """ - v1 = [x for y in v1 for x in y] - v2 = [x for y in v2 for x in y] - for x1, x2 in zip(v1, v2): + v1_flattened = [x for y in v1 for x in y] + v2_flattened = [x for y in v2 for x in y] + for x1, x2 in zip(v1_flattened, v2_flattened): if abs(x2 - x1) > precision: return False - return True \ No newline at end of file + return True diff --git a/whr/whole_history_rating.py b/whr/whole_history_rating.py index dbb8770..3d289bc 100644 --- a/whr/whole_history_rating.py +++ b/whr/whole_history_rating.py @@ -1,34 +1,29 @@ +from __future__ import annotations + import time -import math import ast import pickle -from collections import defaultdict +from typing import Any +from whr.utils import test_stability from whr.player import Player from whr.game import Game -from whr.utils import test_stability class Base: - def __init__(self, config=None): - if config is None: - self.config = defaultdict(lambda: None) - else: - self.config = config - if self.config.get("debug") is None: - self.config["debug"] = False - if self.config.get("w2") is None: - self.config["w2"] = 300.0 - if self.config.get("uncased") is None: - self.config["uncased"] = False + def __init__(self, config: dict[str, Any] | None = None): + self.config = config if config is not None else {} + self.config["debug"] = self.config.get("debug", False) + self.config["w2"] = self.config.get("w2", 300.0) + self.config["uncased"] = self.config.get("uncased", False) self.games = [] self.players = {} - def print_ordered_ratings(self, current=False): - """displays all ratings for each player (for each of his playing days) ordered + def print_ordered_ratings(self, current: bool = False) -> None: + """Displays all ratings for each player (for each of their playing days), ordered. Args: - current (bool, optional): True to let only the last estimation of the elo, False gets all estimation for each day played + current (bool, optional): If True, displays only the latest elo rating. If False, displays all elo ratings for each day played. """ players = [x for x in self.players.values() if len(x.days) > 0] players.sort(key=lambda x: x.days[-1].gamma()) @@ -39,15 +34,17 @@ def print_ordered_ratings(self, current=False): else: print(f"{p.name} => {[x.elo for x in p.days]}") - def get_ordered_ratings(self, current=False, compact=False): - """gets all ratings for each player (for each of his playing days) ordered - - Returns: - list[list[float]]: for each player and each of his playing day, the corresponding elo - + def get_ordered_ratings( + self, current: bool = False, compact: bool = False + ) -> list[list[float]]: + """Retrieves all ratings for each player (for each of their playing days), ordered. + Args: - current (bool, optional): True to let only the last estimation of the elo, False gets all estimation for each day played - compact (bool, optional): True to get only a list of elos, False to get the name before + current (bool, optional): If True, retrieves only the latest elo rating estimation. If False, retrieves all elo rating estimations for each day played. + compact (bool, optional): If True, returns only a list of elo ratings. If False, includes the player's name before their elo ratings. + + Returns: + list[list[float]]: A list containing the elo ratings for each player and each of their playing days. """ result = [] players = [x for x in self.players.values() if len(x.days) > 0] @@ -64,13 +61,13 @@ def get_ordered_ratings(self, current=False, compact=False): result.append((p.name, [x.elo for x in p.days])) return result - def log_likelihood(self): - """gets the likelihood of the current state - - the more iteration you do the higher the likelihood becomes - + def log_likelihood(self) -> float: + """Calculates the likelihood of the current state. + + The likelihood increases with more iterations. + Returns: - float: the likelihood + float: The likelihood. """ score = 0.0 for p in self.players.values(): @@ -78,14 +75,14 @@ def log_likelihood(self): score += p.log_likelihood() return score - def player_by_name(self, name): - """gets the player object corresponding to the name - + def player_by_name(self, name: str) -> Player: + """Retrieves the player object corresponding to the given name. + Args: - name (str): the name of the player - + name (str): The name of the player. + Returns: - Player: the corresponding player + Player: The corresponding player object. """ if self.config["uncased"]: name = name.lower() @@ -93,15 +90,17 @@ def player_by_name(self, name): self.players[name] = Player(name, self.config) return self.players[name] - def ratings_for_player(self, name, current=False): - """gets all rating for each day played for the player - + def ratings_for_player( + self, name, current: bool = False + ) -> list[tuple[int, float, float]] | tuple[float, float]: + """Retrieves all ratings for each day played by the specified player. + Args: - name (str): the player's name - current (bool, optional): True to let only the last estimation of the elo and uncertainty, False gets all estimation for each day played - + name (str): The name of the player. + current (bool, optional): If True, retrieves only the latest elo rating and uncertainty. If False, retrieves all elo ratings and uncertainties for each day played. + Returns: - list[list[int, float, float]]: for each day, the time_step the elo the uncertainty + list[tuple[int, float, float]] | tuple[float, float]: For each day, includes the time step, the elo rating, and the uncertainty if current is False, else just return the elo and uncertainty of the last day """ if self.config["uncased"]: name = name.lower() @@ -109,11 +108,19 @@ def ratings_for_player(self, name, current=False): if current: return ( round(player.days[-1].elo), - round(player.days[-1].uncertainty * 100), + round(player.days[-1].uncertainty, 2), ) - return [[d.day, round(d.elo), round(d.uncertainty * 100)] for d in player.days] + return [(d.day, round(d.elo), round(d.uncertainty, 2)) for d in player.days] - def _setup_game(self, black, white, winner, time_step, handicap, extras=None): + def _setup_game( + self, + black: str, + white: str, + winner: str, + time_step: int, + handicap: float, + extras: dict[str, Any] | None = None, + ) -> Game: if extras is None: extras = {} if black == white: @@ -123,19 +130,27 @@ def _setup_game(self, black, white, winner, time_step, handicap, extras=None): game = Game(black_player, white_player, winner, time_step, handicap, extras) return game - def create_game(self, black, white, winner, time_step, handicap, extras=None): - """creates a new game to be added to the base - + def create_game( + self, + black: str, + white: str, + winner: str, + time_step: int, + handicap: float, + extras: dict[str, Any] | None = None, + ) -> Game: + """Creates a new game to be added to the base. + Args: - black (str): the black name - white (str): the white name - winner (str): "B" if black won, "W" if white won - time_step (int): the day of the match from origin - handicap (float): the handicap (in elo) - extras (dict, optional): extra parameters - + black (str): The name of the black player. + white (str): The name of the white player. + winner (str): "B" if black won, "W" if white won. + time_step (int): The day of the match from the origin. + handicap (float): The handicap (in elo points). + extras (dict[str, Any] | None, optional): Extra parameters. + Returns: - Game: the added game + Game: The newly added game. """ if extras is None: extras = {} @@ -145,7 +160,7 @@ def create_game(self, black, white, winner, time_step, handicap, extras=None): game = self._setup_game(black, white, winner, time_step, handicap, extras) return self._add_game(game) - def _add_game(self, game): + def _add_game(self, game: Game) -> Game: game.white_player.add_game(game) game.black_player.add_game(game) if game.bpd is None: @@ -153,55 +168,61 @@ def _add_game(self, game): self.games.append(game) return game - def iterate(self, count): - """do a number of "count" iterations of the algorithm - + def iterate(self, count: int) -> None: + """Performs a specified number of iterations of the algorithm. + Args: - count (int): the number of iterations desired + count (int): The number of iterations to perform. """ for _ in range(count): self._run_one_iteration() for player in self.players.values(): player.update_uncertainty() - def auto_iterate(self, time_limit=10, precision=10e-3): - """iterates automatically until it converges or reaches the time limit - iterates iteratively ten by ten - + def auto_iterate( + self, + time_limit: int | None = None, + precision: float = 1e-3, + batch_size: int = 10, + ) -> tuple[int, bool]: + """Automatically iterates until the algorithm converges or reaches the time limit. + Args: - time_limit (int, optional): the maximal time after which no more iteration are launched - precision (float, optional): the precision of the stability desired - + time_limit (int | None, optional): The maximum time, in seconds, after which no more iterations will be launched. If None, no timeout is set + precision (float, optional): The desired precision of stability. + batch_size (int, optional): The number of iterations to perform at each step, with precision and timeout checks after each batch. + Returns: - tuple(int, bool): the number of iterations and True if it has reached stability, False otherwise + tuple[int, bool]: The number of iterations performed and a boolean indicating whether stability was reached. """ start = time.time() - self.iterate(10) - a = self.get_ordered_ratings(compact=True) - i = 10 + a = None + i = 0 while True: - self.iterate(10) - i += 10 + self.iterate(batch_size) + i += batch_size b = self.get_ordered_ratings(compact=True) - if test_stability(a, b, precision): + if a is not None and test_stability(a, b, precision): return i, True - if time.time() - start > time_limit: + if time_limit is not None and time.time() - start > time_limit: return i, False a = b - def probability_future_match(self, name1, name2, handicap=0): - """gets the probability of winning for an hypothetical match against name1 and name2 - - displays the probability of winning for name1 and name2 in percent rounded to the second decimal - + def probability_future_match( + self, name1: str, name2: str, handicap: float = 0 + ) -> tuple[float, float]: + """Calculates the winning probability for a hypothetical match between two players. + Args: - name1 (str): name1's name - name2 (str): name2's name - handicap (int, optional): the handicap (in elo) - extras (dict, optional): extra parameters - + name1 (str): The name of the first player. + name2 (str): The name of the second player. + handicap (float, optional): The handicap (in elo points). + Returns: - tuple(int, int): the probability between 0 and 1 for name1 first then name2 + tuple[float, float]: The winning probabilities for name1 and name2, respectively, as percentages rounded to the second decimal. + + Raises: + AttributeError: Raised if name1 and name2 are equal """ # Avoid self-played games (no info) if self.config["uncased"]: @@ -230,101 +251,80 @@ def probability_future_match(self, name1, name2, handicap=0): ) return player1_proba, player2_proba - def _run_one_iteration(self): - """runs one iteration of the whr algorithm - """ + def _run_one_iteration(self) -> None: + """Runs one iteration of the WHR algorithm.""" for player in self.players.values(): player.run_one_newton_iteration() - def load_games(self, games, separator=" "): - """loads all games at once - - given a string representing the path of a file or a list of string representing all games, - this function loads all games in the base - all match must comply to this format: - "black_name white_name winner time_step handicap extras" - black_name (required) - white_name (required) - winner is B or W (required) - time_step (required) - handicap (optional: default 0) - extras is a dict (optional) - + def load_games(self, games: list[str], separator: str = " ") -> None: + """Loads all games at once. + + Each game string must follow the format: "black_name white_name winner time_step handicap extras", + where handicap and extras are optional. Handicap defaults to 0 if not provided, and extras must be a valid dictionary. + Args: - games (str|list[str]): a path or a list of string representing games - separator (str, optional): the separator between all elements of a game, space by default (every element will be trim eventually) + games (list[str]): A list of strings representing games. + separator (str, optional): The separator used between elements of a game, defaulting to a space. + + Raises: + ValueError: If any game string does not comply with the expected format or if parsing fails. """ - data = None - if isinstance(games, str): - with open(games, "r") as f: - data = f.readlines() - else: - data = games - for line in data: + for line in games: + parts = [part.strip() for part in line.split(separator)] + if len(parts) < 4 or len(parts) > 6: + raise ValueError(f"Invalid game format: '{line}'") + + black, white, winner, time_step, *rest = parts handicap = 0 - extras = None - arguments = [x.strip() for x in line.split(separator)] - is_correct = False - if len(arguments) == 6: + extras = {} + + if len(rest) == 1: try: - black, white, winner, time_step, handicap, extras = arguments - extras = last = ast.literal_eval(extras) - if isinstance(extras, dict): - is_correct = True - except Exception as e: - raise ( - AttributeError( - f"the extras argument couldn't be evaluated as a dict: {extras}\n{e}" + handicap = int(rest[0]) + except ValueError: + try: + extras = ast.literal_eval(rest[0]) + if not isinstance(extras, dict): + raise ValueError() + except (ValueError, SyntaxError): + raise ValueError( + f"Invalid handicap or extra value in: '{line}'" ) - ) - if len(arguments) == 5: - black, white, winner, time_step, last = arguments + + if len(rest) == 2: try: - eval_last = ast.literal_eval(last) - if isinstance(eval_last, dict): - extras = eval_last - is_correct = True - elif isinstance(eval_last, int): - handicap = eval_last - is_correct = True - except Exception as e: - raise ( - AttributeError( - f"the last argument couldn't be evaluated as an int or a dict: {last}\n{e}" - ) - ) - if len(arguments) == 4: - black, white, winner, time_step = arguments - is_correct = True - if not is_correct: - raise ( - AttributeError( - f"loaded game must have this format: 'black_name white_name winner time_step handicap extras' with handicap (int or dict) and extras (dict) optional. the handicap|extras argument is: {last}" - ) - ) - time_step, handicap = int(time_step), int(handicap) + handicap = int(rest[0]) + except ValueError: + raise ValueError(f"Invalid handicap value in: '{line}'") + try: + extras = ast.literal_eval(rest[1]) + if not isinstance(extras, dict): + raise ValueError() + except (ValueError, SyntaxError): + raise ValueError(f"Invalid extras dictionary in: '{line}'") + if self.config["uncased"]: - black = black.lower() - white = white.lower() - self.create_game(black, white, winner, time_step, handicap, extras=extras) + black, white = black.lower(), white.lower() + + self.create_game(black, white, winner, int(time_step), handicap, extras) + + def save_base(self, path: str) -> None: + """Saves the current state of the base to a specified path. - def save_base(self, path): - """saves the current state of the base to a file at "path" - Args: - path (str): the path where to save the base + path (str): The path where the base will be saved. """ pickle.dump([self.players, self.games, self.config["w2"]], open(path, "wb")) @staticmethod - def load_base(path): - """loads a saved base - + def load_base(path: str) -> Base: + """Loads a saved base from a specified path. + Args: - path (str): the path to the saved base - + path (str): The path to the saved base. + Returns: - Base: the loaded base + Base: The loaded base. """ players, games, config = pickle.load(open(path, "rb")) result = Base()