#!/usr/bin/python
# Copyright 2010 Doug Zongker
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
"""
Implements the player skill estimation algorithm from Herbrich et al.,
"TrueSkill(TM): A Bayesian Skill Rating System".
"""
from __future__ import print_function
__author__ = "Doug Zongker <dougz@isotropic.org>"
import sys
if sys.hexversion < 0x02060000:
print("requires Python 2.6 or higher")
sys.exit(1)
from scipy.stats.distributions import norm as scipy_norm
from math import sqrt
norm = scipy_norm()
pdf = norm.pdf
cdf = norm.cdf
icdf = norm.ppf # inverse CDF
# Update rules for approximate marginals for the win and draw cases,
# respectively.
def Vwin(t, e):
return pdf(t-e) / cdf(t-e)
def Wwin(t, e):
return Vwin(t, e) * (Vwin(t, e) + t - e)
def Vdraw(t, e):
return (pdf(-e-t) - pdf(e-t)) / (cdf(e-t) - cdf(-e-t))
def Wdraw(t, e):
return Vdraw(t, e) ** 2 + ((e-t) * pdf(e-t) + (e+t) * pdf(e+t)) / (cdf(e-t) - cdf(-e-t))
class Gaussian(object):
"""
Object representing a gaussian distribution. Create as:
Gaussian(mu=..., sigma=...)
or
Gaussian(pi=..., tau=...)
or
Gaussian() # gives 0 mean, infinite sigma
"""
def __init__(self, mu=None, sigma=None, pi=None, tau=None):
if pi is not None:
self.pi = pi
self.tau = tau
elif mu is not None:
self.pi = sigma ** -2
self.tau = self.pi * mu
else:
self.pi = 0
self.tau = 0
def __repr__(self):
return "N(pi={0.pi},tau={0.tau})".format(self)
def __str__(self):
if self.pi == 0.0:
return "N(mu=0,sigma=inf)"
else:
sigma = sqrt(1/self.pi)
mu = self.tau / self.pi
return "N(mu={0:.3f},sigma={1:.3f})".format(mu, sigma)
def MuSigma(self):
""" Return the value of this object as a (mu, sigma) tuple. """
if self.pi == 0.0:
return 0, float("inf")
else:
return self.tau / self.pi, sqrt(1/self.pi)
def __mul__(self, other):
return Gaussian(pi=self.pi+other.pi, tau=self.tau+other.tau)
def __div__(self, other):
return Gaussian(pi=self.pi-other.pi, tau=self.tau-other.tau)
__truediv__ = __div__
class Variable(object):
""" A variable node in the factor graph. """
def __init__(self):
self.value = Gaussian()
self.factors = {}
def AttachFactor(self, factor):
self.factors[factor] = Gaussian()
def UpdateMessage(self, factor, message):
old_message = self.factors[factor]
self.value = self.value / old_message * message
self.factors[factor] = message
def UpdateValue(self, factor, value):
old_message = self.factors[factor]
self.factors[factor] = value * old_message / self.value
self.value = value
def GetMessage(self, factor):
return self.factors[factor]
class Factor(object):
""" Base class for a factor node in the factor graph. """
def __init__(self, variables):
self.variables = variables
for v in variables:
v.AttachFactor(self)
# The following Factor classes implement the five update equations
# from Table 1 of the Herbrich et al. paper.
class PriorFactor(Factor):
""" Connects to a single variable, pushing a fixed (Gaussian) value
to that variable. """
def __init__(self, variable, param):
super(PriorFactor, self).__init__([variable])
self.param = param
def Start(self):
self.variables[0].UpdateValue(self, self.param)
class LikelihoodFactor(Factor):
""" Connects two variables, the value of one being the mean of the
message sent to the other. """
def __init__(self, mean_variable, value_variable, variance):
super(LikelihoodFactor, self).__init__([mean_variable, value_variable])
self.mean = mean_variable
self.value = value_variable
self.variance = variance
def UpdateValue(self):
""" Update the value after a change in the mean (going "down" in
the TrueSkill factor graph. """
y = self.mean.value
fy = self.mean.GetMessage(self)
a = 1.0 / (1.0 + self.variance * (y.pi - fy.pi))
self.value.UpdateMessage(self, Gaussian(pi=a*(y.pi - fy.pi),
tau=a*(y.tau - fy.tau)))
def UpdateMean(self):
""" Update the mean after a change in the value (going "up" in
the TrueSkill factor graph. """
# Note this is the same as UpdateValue, with self.mean and
# self.value interchanged.
x = self.value.value
fx = self.value.GetMessage(self)
a = 1.0 / (1.0 + self.variance * (x.pi - fx.pi))
self.mean.UpdateMessage(self, Gaussian(pi=a*(x.pi - fx.pi),
tau=a*(x.tau - fx.tau)))
class SumFactor(Factor):
""" A factor that connects a sum variable with 1 or more terms,
which are summed after being multiplied by fixed (real)
coefficients. """
def __init__(self, sum_variable, terms_variables, coeffs):
assert len(terms_variables) == len(coeffs)
self.sum = sum_variable
self.terms = terms_variables
self.coeffs = coeffs
super(SumFactor, self).__init__([sum_variable] + terms_variables)
def _InternalUpdate(self, var, y, fy, a):
new_pi = 1.0 / (sum(a[j]**2 / (y[j].pi - fy[j].pi) for j in range(len(a))))
new_tau = new_pi * sum(a[j] *
(y[j].tau - fy[j].tau) / (y[j].pi - fy[j].pi)
for j in range(len(a)))
var.UpdateMessage(self, Gaussian(pi=new_pi, tau=new_tau))
def UpdateSum(self):
""" Update the sum value ("down" in the factor graph). """
y = [t.value for t in self.terms]
fy = [t.GetMessage(self) for t in self.terms]
a = self.coeffs
self._InternalUpdate(self.sum, y, fy, a)
def UpdateTerm(self, index):
""" Update one of the term values ("up" in the factor graph). """
# Swap the coefficients around to make the term we want to update
# be the 'sum' of the other terms and the factor's sum, eg.,
# change:
#
# x = y_1 + y_2 + y_3
#
# to
#
# y_2 = x - y_1 - y_3
#
# then use the same update equation as for UpdateSum.
b = self.coeffs
a = [-b[i] / b[index] for i in range(len(b)) if i != index]
a.insert(index, 1.0 / b[index])
v = self.terms[:]
v[index] = self.sum
y = [i.value for i in v]
fy = [i.GetMessage(self) for i in v]
self._InternalUpdate(self.terms[index], y, fy, a)
class TruncateFactor(Factor):
""" A factor for (approximately) truncating the team difference
distribution based on a win or a draw (the choice of which is
determined by the functions you pass as V and W). """
def __init__(self, variable, V, W, epsilon):
super(TruncateFactor, self).__init__([variable])
self.var = variable
self.V = V
self.W = W
self.epsilon = epsilon
def Update(self):
x = self.var.value
fx = self.var.GetMessage(self)
c = x.pi - fx.pi
d = x.tau - fx.tau
sqrt_c = sqrt(c)
args = (d / sqrt_c, self.epsilon * sqrt_c)
V = self.V(*args)
W = self.W(*args)
new_val = Gaussian(pi=c / (1.0 - W), tau=(d + sqrt_c * V) / (1.0 - W))
self.var.UpdateValue(self, new_val)
def DrawProbability(epsilon, beta, total_players=2):
""" Compute the draw probability given the draw margin (epsilon). """
return 2 * cdf(epsilon / (sqrt(total_players) * beta)) - 1
def DrawMargin(p, beta, total_players=2):
""" Compute the draw margin (epsilon) given the draw probability. """
return icdf((p+1.0)/2) * sqrt(total_players) * beta
INITIAL_MU = 25.0
INITIAL_SIGMA = INITIAL_MU / 3.0
[docs]def SetParameters(beta=None, epsilon=None, draw_probability=None,
gamma=None):
"""
Sets three global parameters used in the TrueSkill algorithm.
beta is a measure of how random the game is. You can think of it as
the difference in skill (mean) needed for the better player to have
an ~80% chance of winning. A high value means the game is more
random (I need to be *much* better than you to consistently overcome
the randomness of the game and beat you 80% of the time); a low
value is less random (a slight edge in skill is enough to win
consistently). The default value of beta is half of INITIAL_SIGMA
(the value suggested by the Herbrich et al. paper).
epsilon is a measure of how common draws are. Instead of specifying
epsilon directly you can pass draw_probability instead (a number
from 0 to 1, saying what fraction of games end in draws), and
epsilon will be determined from that. The default epsilon
corresponds to a draw probability of 0.1 (10%). (You should pass a
value for either epsilon or draw_probability, not both.)
gamma is a small amount by which a player's uncertainty (sigma) is
increased prior to the start of each game. This allows us to
account for skills that vary over time; the effect of old games
on the estimate will slowly disappear unless reinforced by evidence
from new games.
"""
global BETA, EPSILON, GAMMA
if beta is None:
BETA = INITIAL_SIGMA / 2.0
else:
BETA = beta
if epsilon is None:
if draw_probability is None:
draw_probability = 0.10
EPSILON = DrawMargin(draw_probability, BETA)
else:
EPSILON = epsilon
if gamma is None:
GAMMA = INITIAL_SIGMA / 100.0
else:
GAMMA = gamma
SetParameters()
[docs]def AdjustPlayers(players):
"""
Adjust the skills of a list of players.
'players' is a list of player objects, for all the players who
participated in a single game. A 'player object' is any object with
a "skill" attribute (a (mu, sigma) tuple) and a "rank" attribute.
Lower ranks are better; the lowest rank is the overall winner of the
game. Equal ranks mean that the two players drew.
This function updates all the "skill" attributes of the player
objects to reflect the outcome of the game. The input list is not
altered.
"""
players = players[:]
# Sort players by rank, the factor graph will connect adjacent team
# performance variables.
players.sort(key=lambda p: p.rank)
# Create all the variable nodes in the graph. "Teams" are each a
# single player; there's a one-to-one correspondence between players
# and teams. (It would be straightforward to make multiplayer
# teams, but it's not needed for my current purposes.)
ss = [Variable() for p in players]
ps = [Variable() for p in players]
ts = [Variable() for p in players]
ds = [Variable() for p in players[:-1]]
# Create each layer of factor nodes. At the top we have priors
# initialized to the player's current skill estimate.
skill = [PriorFactor(s, Gaussian(mu=pl.skill[0],
sigma=sqrt(pl.skill[1] ** 2 + GAMMA ** 2)))
for (s, pl) in zip(ss, players)]
skill_to_perf = [LikelihoodFactor(s, p, BETA**2)
for (s, p) in zip(ss, ps)]
perf_to_team = [SumFactor(t, [p], [1])
for (p, t) in zip(ps, ts)]
team_diff = [SumFactor(d, [t1, t2], [+1, -1])
for (d, t1, t2) in zip(ds, ts[:-1], ts[1:])]
# At the bottom we connect adjacent teams with a 'win' or 'draw'
# factor, as determined by the rank values.
trunc = [TruncateFactor(d,
Vdraw if pl1.rank == pl2.rank else Vwin,
Wdraw if pl1.rank == pl2.rank else Wwin,
EPSILON)
for (d, pl1, pl2) in zip(ds, players[:-1], players[1:])]
# Start evaluating the graph by pushing messages 'down' from the
# priors.
for f in skill:
f.Start()
for f in skill_to_perf:
f.UpdateValue()
for f in perf_to_team:
f.UpdateSum()
# Because the truncation factors are approximate, we iterate,
# adjusting the team performance (t) and team difference (d)
# variables until they converge. In practice this seems to happen
# very quickly, so I just do a fixed number of iterations.
#
# This order of evaluation is given by the numbered arrows in Figure
# 1 of the Herbrich paper.
for i in range(5):
for f in team_diff:
f.UpdateSum() # arrows (1) and (4)
for f in trunc:
f.Update() # arrows (2) and (5)
for f in team_diff:
f.UpdateTerm(0) # arrows (3) and (6)
f.UpdateTerm(1)
# Now we push messages back up the graph, from the teams back to the
# player skills.
for f in perf_to_team:
f.UpdateTerm(0)
for f in skill_to_perf:
f.UpdateMean()
# Finally, the players' new skills are the new values of the s
# variables.
for s, pl in zip(ss, players):
pl.skill = s.value.MuSigma()
__all__ = ["AdjustPlayers", "SetParameters", "INITIAL_MU", "INITIAL_SIGMA"]