"""Keener Ratings implementation for the Elote library.
Keener's method is the score-based member of the classical global-fit family. Where
:class:`~elote.competitors.colley.ColleyMatrixCompetitor` and
:class:`~elote.competitors.massey.MasseyCompetitor` solve a linear system, Keener builds a
square *preference matrix* from the points competitors have scored on one another and reads
the ratings off that matrix's dominant eigenvector.
The construction has four steps. Let ``S_ij`` be the total number of points ``i`` has scored
against ``j`` across all their meetings.
1. **Smoothed preference.** Raw score shares are unstable for lopsided or barely-played
pairs, so Keener applies Laplace smoothing:
.. math::
a_{ij} = \\frac{S_{ij} + 1}{S_{ij} + S_{ji} + 2}
A pair that has never met gives ``a_ij = 1/2`` -- no evidence either way.
2. **Skew transform.** ``a_ij`` is pushed away from the middle by Keener's skew function
.. math::
h(x) = \\frac{1}{2} + \\frac{1}{2}\\,\\mathrm{sgn}\\!\\left(x - \\frac{1}{2}\\right)
\\sqrt{\\left|2x - 1\\right|}
which is monotone, fixes ``0``, ``1/2`` and ``1``, and satisfies ``h(x) + h(1 - x) = 1``,
so the matrix stays antisymmetric about ``1/2``. Its square root damps the influence of
very large margins, which is what stops the method rewarding running up the score without
limit.
3. **Games-played normalization.** Row ``i`` is divided by the number of games ``i`` played
plus a small prior, so a competitor cannot accumulate rating merely by playing more often,
and cannot accumulate it by playing barely at all either. The prior matters on a ragged
schedule: without it, a competitor's single game concentrates its entire row on one
opponent, which the eigenvector reads as strength.
4. **Stabilization.** A small positive constant is added to every entry. Keener's own
``A + eps * E`` perturbation makes the matrix strictly positive, so Perron-Frobenius
applies: the dominant eigenvalue is real and simple and its eigenvector is strictly
positive and unique up to scale. Without it a schedule whose graph is bipartite or
otherwise imprimitive has no single dominant eigenvector to read.
The ratings of a connected group are then that dominant eigenvector, scaled so they average
exactly ``1.0``.
References:
- Keener, J. P. (1993). The Perron-Frobenius Theorem and the Ranking of Football Teams.
SIAM Review, 35(1), 80-93.
- Langville, A. N., & Meyer, C. D. (2012). Who's #1? The Science of Rating and Ranking.
Princeton University Press, chapter 4.
"""
import math
from typing import Dict, Any, ClassVar, Optional, Sequence, Type, TypeVar, List, cast, Set
import numpy as np
from elote.competitors.base import BaseCompetitor, InvalidParameterException, InvalidRatingValueException
from elote.logging import logger
T = TypeVar("T", bound="KeenerCompetitor")
[docs]
class KeenerCompetitor(BaseCompetitor):
"""Keener Ratings competitor.
Keener rates a connected population from the dominant eigenvector of a pairwise
preference matrix built from points scored. Like Colley, Massey and Bradley-Terry -- and
unlike Elo -- ratings are not nudged after each result; the whole connected group is
re-fit from the complete match history, which makes the method order independent.
**Scores.** ``beat`` / ``lost_to`` / ``tied`` accept the common optional ``scores``
payload, the two competitors' scores in caller order. Keener is a score-based method, so
supplying real scores is what it is built for. When they are omitted the implementation
falls back to the same unit scores the rest of the library uses: the winner is credited
``1`` and the loser ``0``, and a draw credits each side ``0.5``. Unit-score Keener still
produces a sensible ranking, but it only sees who beat whom.
Key characteristics:
- Global eigenvector fit (does not depend on the order of results)
- Ratings are strictly positive and average exactly 1.0 within a connected group
- Large margins have a damped, not linear, influence thanks to the skew transform
Class Attributes:
_minimum_rating (float): The minimum allowed rating value. Default: 0.0. Keener
ratings are strictly positive, so no Elo-style floor applies.
_default_initial_rating (float): Default initial rating. Default: 1.0, which is the
mean the fitted ratings are normalized to, so an unplayed competitor sits exactly
at the population average.
_perturbation (float): Keener's ``eps``, added to every matrix entry so the matrix is
strictly positive and Perron-Frobenius applies. Default: 1e-4.
_games_prior (float): Pseudo-games added to the games-played denominator, shrinking a
competitor's average preference toward zero in proportion to how little evidence
backs it. Default: 2.0, matching the two pseudo-observations the Laplace-smoothed
score share already assumes. Without it, a competitor with a single game puts all
of its row weight on one opponent and the dominant eigenvector rewards that
concentration, so a one-game team can outrate an undefeated one.
_expected_score_scale (float): Logistic scale applied to the log rating ratio by
:meth:`expected_score`. Default: 1.0, which is the plain Keener share
``r_a / (r_a + r_b)``.
_round_decimals (int): Number of decimal places fitted ratings are canonicalized to,
so that solver noise cannot make mathematically identical records differ.
Default: 10, chosen because the eigen-solve's row-order noise is around 5e-15 --
five orders of magnitude below the rounding grid.
"""
_minimum_rating: ClassVar[float] = 0.0
_default_initial_rating: ClassVar[float] = 1.0
_perturbation: ClassVar[float] = 1e-4
_games_prior: ClassVar[float] = 2.0
_expected_score_scale: ClassVar[float] = 1.0
_round_decimals: ClassVar[int] = 10
# Ratings are strictly positive by construction; this only guards expected_score against
# a rating a caller has explicitly set to zero.
_probability_floor: ClassVar[float] = 1e-300
[docs]
def __init__(self, initial_rating: Optional[float] = None):
"""Initialize a new Keener competitor.
Args:
initial_rating (float, optional): The initial rating of this competitor.
Default: 1.0.
Raises:
InvalidRatingValueException: If the initial rating is not positive.
"""
super().__init__()
self._initial_rating = initial_rating if initial_rating is not None else self._default_initial_rating
if self._initial_rating <= 0:
raise InvalidRatingValueException("Keener ratings are strictly positive; initial_rating must be > 0")
self._rating = self._initial_rating
self._wins = 0
self._losses = 0
self._ties = 0
# Aggregate score totals across every game played. These are the serializable
# summary of the score history; the per-opponent breakdown below is not.
self._points_for = 0.0
self._points_against = 0.0
self._opponents: Dict["KeenerCompetitor", int] = {} # Opponent -> num games
# Opponent -> total points this competitor has scored against that opponent. This is
# the S_ij the preference matrix is built from.
self._scores_for: Dict["KeenerCompetitor", float] = {}
# True when recorded games are not yet reflected in _rating: the eigenvector fit is
# deferred to the first rating read that needs it (dirty-flag caching).
self._ratings_dirty = False
# Unique ID for hashing (instances are used as dict keys in the match graph).
self._id = id(self)
logger.debug("Initialized KeenerCompetitor %d with initial rating %.6f", self._id, self._initial_rating)
@property
def rating(self) -> float:
"""Get the current rating of this competitor.
Reading the rating is where deferred eigenvector fits happen: if games have been
recorded since the last read, the connected group is re-fit first.
Returns:
float: The current rating.
"""
self._ensure_current_ratings()
return self._rating
@rating.setter
def rating(self, value: float) -> None:
"""Set the current rating of this competitor.
Args:
value (float): The new rating value.
Raises:
InvalidRatingValueException: If the rating value is below the minimum rating.
"""
if value < self._minimum_rating:
logger.warning(
"Attempted to set rating %.6f below minimum %.6f for %d", value, self._minimum_rating, self._id
)
raise InvalidRatingValueException(f"Rating cannot be below the minimum rating of {self._minimum_rating}")
self._rating = value
@property
def num_games(self) -> int:
"""Get the total number of games played by this competitor.
Returns:
int: The total number of games played.
"""
return self._wins + self._losses + self._ties
def _ensure_current_ratings(self) -> None:
"""Re-fit the connected group's ratings if recorded games are not yet reflected.
Recording a game only marks the fit stale; the eigenvector computation runs once on
the first read that needs it, so repeated reads with no new games are O(1) amortized.
"""
if self._ratings_dirty:
self._recalculate_ratings()
def _mark_group_stale(self) -> None:
"""Mark every competitor in the connected group stale after a recorded game.
A new game changes the eigenvector fit of the whole connected component, not just
the two endpoints' cached ratings: endpoint-only flags would let a clean member
serve a rating that predates the game whenever it is read first, making output
depend on read order. One O(component) walk per recorded game keeps reads O(1).
"""
for comp in self._get_connected_competitors():
comp._ratings_dirty = True
[docs]
def expected_score(self, competitor: "BaseCompetitor") -> float:
"""Calculate the expected score against another competitor.
Keener ratings are positive strengths rather than probabilities, so the natural
mapping is the share ``r_self / (r_self + r_competitor)``. That share is computed
here as a logistic of the log rating ratio, which is the same quantity written in a
form that is exactly complementary in floating point: two equal ratings give exactly
0.5, and the two argument orders sum to exactly 1.0.
Args:
competitor (BaseCompetitor): The competitor to compare against.
Returns:
float: The expected score (probability of winning).
Raises:
MissMatchedCompetitorTypesException: If the competitor types don't match.
"""
self.verify_competitor_types(competitor)
mine = max(self.rating, self._probability_floor)
theirs = max(competitor.rating, self._probability_floor)
log_ratio = math.log(mine) - math.log(theirs)
return float(0.5 * (1.0 + np.tanh(0.5 * self._expected_score_scale * log_ratio)))
[docs]
def beat(self, competitor: BaseCompetitor, *, scores: Optional[Sequence[float]] = None) -> None:
"""Update ratings after this competitor has won against the given competitor.
The scores are recorded in the match graph and the whole connected group is re-fit
lazily, on the next rating read.
Args:
competitor (BaseCompetitor): The opponent competitor that lost.
scores (sequence of float, optional): The two scores in caller order,
``(self_score, competitor_score)``. When omitted the unit scores ``1`` and
``0`` are recorded.
Raises:
MissMatchedCompetitorTypesException: If the competitor types don't match.
ValueError: If ``scores`` does not describe a win for this competitor.
"""
logger.debug("Competitor %s beat %s", self, competitor)
self.verify_competitor_types(competitor)
validated = self._validate_scores(scores, 1.0)
opponent = cast(KeenerCompetitor, competitor)
mine, theirs = (1.0, 0.0) if validated is None else validated
self._wins += 1
opponent._losses += 1
self._record_game(opponent, mine, theirs)
logger.debug("Recorded win for %d (%.3f-%.3f), loss for %d", self._id, mine, theirs, opponent._id)
[docs]
def tied(self, competitor: BaseCompetitor, *, scores: Optional[Sequence[float]] = None) -> None:
"""Update ratings after this competitor has tied with the given competitor.
Args:
competitor (BaseCompetitor): The opponent competitor that tied.
scores (sequence of float, optional): The two scores in caller order,
``(self_score, competitor_score)``. They must be equal. When omitted each
side is credited the unit draw score ``0.5``.
Raises:
MissMatchedCompetitorTypesException: If the competitor types don't match.
ValueError: If ``scores`` does not describe a draw.
"""
logger.debug("Competitor %s tied with %s", self, competitor)
self.verify_competitor_types(competitor)
validated = self._validate_scores(scores, 0.5)
opponent = cast(KeenerCompetitor, competitor)
mine, theirs = (0.5, 0.5) if validated is None else validated
self._ties += 1
opponent._ties += 1
self._record_game(opponent, mine, theirs)
logger.debug("Recorded tie for %d and %d (%.3f-%.3f)", self._id, opponent._id, mine, theirs)
[docs]
def lost_to(self, competitor: BaseCompetitor, *, scores: Optional[Sequence[float]] = None) -> None:
"""Update ratings after this competitor has lost to the given competitor.
Args:
competitor (BaseCompetitor): The opponent competitor that won.
scores (sequence of float, optional): The two scores in caller order,
``(self_score, competitor_score)``. Reversed before being passed to the
winner's :meth:`beat`.
Raises:
MissMatchedCompetitorTypesException: If the competitor types don't match.
ValueError: If ``scores`` does not describe a loss for this competitor.
"""
logger.debug("Competitor %s lost to %s", self, competitor)
self.verify_competitor_types(competitor)
validated = self._validate_scores(scores, 0.0)
competitor.beat(self, scores=None if validated is None else (validated[1], validated[0]))
def _record_game(self, opponent: "KeenerCompetitor", mine: float, theirs: float) -> None:
"""Record one game's scores on both sides of the match graph.
Args:
opponent: The other competitor in the game.
mine: The points this competitor scored.
theirs: The points the opponent scored.
"""
self._points_for += mine
self._points_against += theirs
self._opponents[opponent] = self._opponents.get(opponent, 0) + 1
self._scores_for[opponent] = self._scores_for.get(opponent, 0.0) + mine
opponent._points_for += theirs
opponent._points_against += mine
opponent._opponents[self] = opponent._opponents.get(self, 0) + 1
opponent._scores_for[self] = opponent._scores_for.get(self, 0.0) + theirs
# The whole connected group's eigenvector fit is stale now; the re-fit is deferred
# to the first rating read that needs it. The opponent is in self's group.
self._mark_group_stale()
def _get_connected_competitors(self) -> List["KeenerCompetitor"]:
"""Get all competitors connected to this competitor in the match graph.
Returns:
List[KeenerCompetitor]: A list of all connected competitors.
"""
visited: Set["KeenerCompetitor"] = set()
to_visit: List["KeenerCompetitor"] = [self]
all_competitors = []
while to_visit:
current = to_visit.pop()
if current in visited:
continue
visited.add(current)
all_competitors.append(current)
for opponent in current._opponents:
if opponent not in visited:
to_visit.append(opponent)
logger.debug("Found %d connected competitors", len(all_competitors))
return all_competitors
@classmethod
def _preference_matrix(cls, scores: "np.ndarray", pair_games: "np.ndarray", games: "np.ndarray") -> "np.ndarray":
"""Build Keener's stabilized preference matrix from a raw score matrix.
Args:
scores: ``S``, where ``S[i, j]`` is the total number of points ``i`` scored
against ``j``.
pair_games: ``G``, where ``G[i, j]`` is the number of games ``i`` played
against ``j``.
games: The number of games each competitor played in total.
Returns:
numpy.ndarray: The strictly positive preference matrix.
"""
totals = scores + scores.T
# Laplace-smoothed score share.
proportions = (scores + 1.0) / (totals + 2.0)
# Keener's skew transform: monotone, fixes 1/2, and h(x) + h(1 - x) == 1.
centered = 2.0 * proportions - 1.0
preferences = 0.5 + 0.5 * np.sign(centered) * np.sqrt(np.abs(centered))
# A pair that never met expresses no preference, and neither does a competitor
# against itself. Leaving unplayed pairs at the "no evidence" value h(1/2) = 1/2
# instead would swamp the row: on a sparse schedule the n - games_i entries for
# opponents never faced dominate the games_i real ones, and the fit degenerates
# into a ranking by 1 / games_played.
preferences = np.where(pair_games > 0, preferences, 0.0)
np.fill_diagonal(preferences, 0.0)
# Divide by games played, making each row the competitor's average preference per
# game, so a longer schedule is not itself worth rating.
#
# The denominator carries a prior of `_games_prior` pseudo-games. Dividing by the raw
# count equalizes row sums but not row *concentration*: a competitor with one game
# places all of its weight on a single opponent, and the dominant eigenvector rewards
# that, so a one-game team could outrate an undefeated one with a full schedule. The
# prior shrinks a thin row toward zero by a factor of g / (g + prior) and leaves a
# full schedule essentially untouched. It is the same correction the Laplace-smoothed
# score share above already applies to a thin pairwise record.
preferences = preferences / (np.maximum(games, 0.0) + cls._games_prior)[:, None]
# Keener's eps * E perturbation: makes the matrix strictly positive, so its dominant
# eigenvalue is simple and its eigenvector strictly positive. This is also what
# connects competitors who never met, which the mask above disconnected.
return cast("np.ndarray", preferences + cls._perturbation)
def _recalculate_ratings(self) -> None:
"""Re-fit the Keener ratings for all connected competitors.
Builds the preference matrix for the connected group, takes the eigenvector of its
dominant eigenvalue, and scales the result so the group's ratings average 1.0.
"""
competitors = self._get_connected_competitors()
n = len(competitors)
if n <= 1:
logger.debug("Only one competitor in network, skipping recalculation.")
self._ratings_dirty = False
return
idx = {comp: i for i, comp in enumerate(competitors)}
scores = np.zeros((n, n), dtype=np.float64)
pair_games = np.zeros((n, n), dtype=np.float64)
games = np.zeros(n, dtype=np.float64)
for i, comp in enumerate(competitors):
games[i] = comp.num_games
for opponent, points in comp._scores_for.items():
j = idx.get(opponent)
if j is not None:
scores[i, j] = points
for opponent, count in comp._opponents.items():
j = idx.get(opponent)
if j is not None:
pair_games[i, j] = count
matrix = self._preference_matrix(scores, pair_games, games)
try:
eigenvalues, eigenvectors = np.linalg.eig(matrix)
except np.linalg.LinAlgError as e:
logger.warning(
"Keener eigenvector solve failed (%s). Falling back to mean-preference ratings for %d competitors.",
str(e),
n,
)
self._fallback_rating_calculation(competitors, matrix)
return
# The matrix is strictly positive, so by Perron-Frobenius the eigenvalue of largest
# magnitude is real, simple and positive, and its eigenvector can be taken positive.
dominant = int(np.argmax(eigenvalues.real))
vector = np.real(eigenvectors[:, dominant])
if vector.sum() < 0:
# LAPACK returns eigenvectors up to sign; orient the Perron vector positive.
vector = -vector
if not np.all(np.isfinite(vector)) or vector.sum() <= 0:
logger.warning("Keener eigenvector was degenerate for %d competitors; falling back.", n)
self._fallback_rating_calculation(competitors, matrix)
return
vector = np.maximum(vector, self._probability_floor)
ratings = vector * (n / vector.sum())
# LAPACK's pivoting follows row order, and row order is the order competitors were
# discovered in, so two runs over the same results can differ in the last few bits.
# Canonicalizing makes mathematically identical records compare exactly equal and
# makes the fit order independent.
ratings = np.round(ratings, decimals=self._round_decimals)
for i, comp in enumerate(competitors):
comp.rating = float(ratings[i])
# The eigensolve above is a completed fit: every connected competitor is current.
for comp in competitors:
comp._ratings_dirty = False
def _fallback_rating_calculation(self, competitors: List["KeenerCompetitor"], matrix: "np.ndarray") -> None:
"""Assign mean-preference ratings when the eigenvector cannot be computed.
Args:
competitors: The connected group of competitors to rate.
matrix: The preference matrix that was built for them.
"""
means = matrix.sum(axis=1)
total = float(means.sum())
n = len(competitors)
if total <= 0:
values = np.full(n, 1.0)
else:
values = means * (n / total)
for comp, value in zip(competitors, values, strict=True):
comp.rating = float(max(value, self._probability_floor))
# The fallback is itself a completed fit: every connected competitor is current.
for comp in competitors:
comp._ratings_dirty = False
def _export_parameters(self) -> Dict[str, Any]:
"""Export the parameters used to initialize this competitor.
Returns:
dict: A dictionary containing the initialization parameters.
"""
return {
"initial_rating": self._initial_rating,
}
def _export_current_state(self) -> Dict[str, Any]:
"""Export the current state variables of this competitor.
Returns:
dict: A dictionary containing the current state variables.
"""
# Never export a rating older than the recorded games.
self._ensure_current_ratings()
return {
"rating": self._rating,
"wins": self._wins,
"losses": self._losses,
"ties": self._ties,
"points_for": self._points_for,
"points_against": self._points_against,
}
def _import_parameters(self, parameters: Dict[str, Any]) -> None:
"""Import parameters from a state dictionary.
Args:
parameters (dict): A dictionary containing parameters.
"""
self._initial_rating = parameters.get("initial_rating", self._default_initial_rating)
self._rating = self._initial_rating
def _import_current_state(self, state: Dict[str, Any]) -> None:
"""Import current state variables from a state dictionary.
Note: opponent references cannot be restored from serialization, so the match graph is
reset; the rating and the aggregate score totals are preserved.
Args:
state (dict): A dictionary containing state variables.
"""
self._rating = state.get("rating", self._initial_rating)
self._wins = state.get("wins", 0)
self._losses = state.get("losses", 0)
self._ties = state.get("ties", 0)
self._points_for = state.get("points_for", 0.0)
self._points_against = state.get("points_against", 0.0)
self._opponents = {}
self._scores_for = {}
# The imported rating is current for the imported (edge-less) state; reset the
# deferred-fit cache rather than marking it stale.
self._ratings_dirty = False
@classmethod
def _create_from_parameters(cls: Type[T], parameters: Dict[str, Any]) -> T:
"""Create a new competitor instance from parameters.
Args:
parameters (dict): A dictionary containing parameters.
Returns:
KeenerCompetitor: A new competitor instance.
"""
return cls(initial_rating=parameters.get("initial_rating", cls._default_initial_rating))
[docs]
def reset(self) -> None:
"""Reset this competitor to its initial state."""
logger.info("Resetting KeenerCompetitor %d to initial state.", self._id)
self._rating = self._initial_rating
self._wins = 0
self._losses = 0
self._ties = 0
self._points_for = 0.0
self._points_against = 0.0
self._opponents = {}
self._scores_for = {}
self._ratings_dirty = False
def __repr__(self) -> str:
"""Return a string representation of this competitor."""
return f"<KeenerCompetitor: rating={self._rating:.4f}, W/L/T={self._wins}/{self._losses}/{self._ties}>"
def __str__(self) -> str:
"""Return a string representation of this competitor."""
return f"<KeenerCompetitor: rating={self._rating:.4f}>"
def __eq__(self, other: Any) -> bool:
"""Check if two competitors are the same object."""
return self is other
def __hash__(self) -> int:
"""Get a hash value for this competitor based on its object identity."""
return object.__hash__(self)