worker: true skill rating

2 files changed, 114 insertions, 123 deletions

diff --git a/requirements.txt b/requirements.txt
index 05a189a..93125e0 100644
--- a/requirements.txt
+++ b/requirements.txt
@@ -1,5 +1,6 @@
 appdirs==1.4.3
 cymysql==0.8.9
+mpmath==0.19
 numpy==1.12.1
 packaging==16.8
 pika==0.10.0
@@ -8,4 +9,5 @@ pyparsing==2.2.0
 six==1.10.0
 SQLAlchemy==1.1.9
 tensorflow==1.1.0rc1
+trueskill==0.4.4
 Werkzeug==0.12.1
diff --git a/worker.py b/worker.py
index 0e0b70d..009cbc1 100644
--- a/worker.py
+++ b/worker.py
@@ -1,34 +1,37 @@
 #!/usr/bin/python3
 import os
 import time
-import random
 import logging
-import itertools
 
-from sqlalchemy.orm import Session, relationship
+from sqlalchemy.orm import Session, relationship, subqueryload, load_only
 from sqlalchemy.ext.automap import automap_base
 from sqlalchemy.exc import OperationalError
 from sqlalchemy import create_engine
 
-import tensorflow as tf
-import numpy as np
+import pika
+import trueskill
 
 
+RABBITMQ_URI = os.environ.get("RABBITMQ_URI") or "amqp://localhost"
 DATABASE_URI = os.environ["DATABASE_URI"]
-MODEL_ROOT = os.path.join(os.getcwd(), os.path.dirname(__file__))\
-        + "/models/"
+BATCHSIZE = os.environ.get("BATCHSIZE") or 500  # matches
+IDLE_TIMEOUT = os.environ.get("IDLE_TIMEOUT") or 1  # s
 
 # ORM definitions
 Match = Roster = Participant = ParticipantStats = Player = None
-db = None
+db = rabbit = channel = None
+
+# batch storage
+queue = []
+timer = None
 
 # models
 mvpmodel = None
 
 
 def connect():
-    global Match, Roster, Participant, Hero, ParticipantStats, Player
-    global db
+    global Match, Roster, Participant, ParticipantStats, Player
+    global db, rabbit, channel
 
     # generate schema from db
     Base = automap_base()
@@ -44,21 +47,29 @@ def connect():
     # definitions
     # TODO check whether the primaryjoin clause is the best method to do this
     Match = Base.classes.match
+    Match.rosters = relationship(
+        "roster", foreign_keys="roster.match_api_id",
+        primaryjoin="and_(match.api_id == roster.match_api_id)")
+    Match.participants = relationship(
+        "participant", foreign_keys="participant.match_api_id",
+        primaryjoin="and_(match.api_id == participant.match_api_id)")
     Roster = Base.classes.roster
     Roster.match = relationship(
         "match", foreign_keys="match.api_id",
         primaryjoin="and_(match.api_id == roster.match_api_id)")
+    Roster.participants = relationship(
+        "participant", foreign_keys="participant.roster_api_id",
+        primaryjoin="and_(roster.api_id == participant.roster_api_id)")
     Participant = Base.classes.participant
     Participant.roster = relationship(
         "roster", foreign_keys="roster.api_id",
         primaryjoin="and_(roster.api_id == participant.roster_api_id)")
+    Participant.match = relationship(
+        "match", foreign_keys="match.api_id",
+        primaryjoin="and_(match.api_id == participant.match_api_id)")
     Participant.player = relationship(
         "player", foreign_keys="player.api_id",
         primaryjoin="and_(player.api_id == participant.player_api_id)")
-    Participant.hero = relationship(
-        "hero", foreign_keys="hero.id",
-        primaryjoin="and_(hero.id == participant.hero_id)")
-    Hero = Base.classes.hero
     ParticipantStats = Base.classes.participant_stats
     Participant.participant_stats = relationship(
         "participant_stats", foreign_keys="participant_stats.participant_api_id",
@@ -67,126 +78,104 @@ def connect():
 
     db = Session(engine)
 
+    while True:
+        try:
+            rabbit = pika.BlockingConnection(pika.URLParameters(RABBITMQ_URI))
+            break
+        except pika.exceptions.ConnectionClosed as err:
+            logging.error(err)
+            time.sleep(5)
+    channel = rabbit.channel()
+    channel.queue_declare(queue="analyze", durable=True)
+    channel.basic_qos(prefetch_count=BATCHSIZE)
+    channel.basic_consume(newjob, queue="analyze")
 
-class Model(object):
-    # override this configuration
-    features = []
-    label = ""
-    type = ""
-    batches = 1
-    batchsize = 1
-    steps = 0
-    id = "unlabeled"
-
-    def __init__(self, db):
-        self._db = db
-        self._feature_cols = [tf.contrib.layers.real_valued_column(
-            feat, dimension=1) for feat in self.features]
-        if self.type == "linear":
-            self._model = tf.contrib.learn.LinearClassifier(
-                feature_columns=self._feature_cols,
-                model_dir=MODEL_ROOT + self.id,
-                config=tf.contrib.learn.RunConfig(
-                    save_checkpoints_secs=1))
-
-    def _from_record(self, path, record):
-        table, column = path.split(".")
-        if table == "participant":
-            return vars(record)[column]
-        if table == "participant_stats":
-            return vars(record.participant_stats[0])[column]
-        if table == "player":
-            return vars(record.player[0])[column]
-        if table == "roster":
-            return vars(record.roster[0])[column]
-        if table == "match":
-            return vars(record.roster[0].match[0])[column]
-        raise KeyError("Invalid path " + path)
-
-    def _batch(self, ids=None, size=None):
-        if ids is None:  # training, get random sample
-            size = size or self.batchsize
-            # train a model one batch
-            offset = random.random() * self._db.query(Participant)\
-                .count()
-            # have a bit of randomness in the sample
-            records = self._db.query(Participant)\
-                .offset(offset).limit(size)\
-                .all()
-        else:
-            records = self._db.query(Participant)\
-                .filter(Participant.api_id.in_(ids))\
-                .order_by(Participant.api_id.desc())\
-                .all()
-
-        data = {}
-        labels = []
-        # populate from db records
-        for record in records:
-            labels.append(self.estimate(record))
-            for path in self.features:
-                if path not in data:
-                    data[path] = []
-                data[path].append(self._from_record(path, record))
-
-        # convert to numpy arrs
-        for key in data:
-            data[key] = np.array(data[key])
-        labels = np.array(labels)
-
-        logging.info("sample size: %s", len(labels))
-        if ids is not None:
-            assert len(ids) == len(labels), "got nonexisting participant"
 
-        return tf.contrib.learn.io.numpy_input_fn(
-            data, labels, batch_size=self.batchsize,
-            num_epochs=self.steps)
+def newjob(_, method, properties, body):
+    global timer, queue
+    queue.append((method, properties, body))
+    if timer is None:
+        timer = rabbit.add_timeout(IDLE_TIMEOUT, process)
+    if len(queue) == BATCHSIZE:
+        process()
 
-    # TODO at the moment, it's tied to Participant
-    def train(self, force=False):
-        if force or not os.path.isdir(MODEL_ROOT + self.id):
-            monitor = tf.contrib.learn.monitors.ValidationMonitor(
-                input_fn=self._batch(),
-                eval_steps=1, every_n_steps=20)
-            for _ in range(self.batches):
-                self._model.fit(input_fn=self._batch(),
-                                steps=self.steps,
-                                monitors=[monitor])
 
-    def predict(self, ids):
-        return itertools.islice(
-            self._model.predict_proba(input_fn=self._batch(ids)),
-            len(ids))
+def process():
+    global timer, queue, db, mvpmodel
+    if timer is not None:
+        rabbit.remove_timeout(timer)
+        timer = None
+    jobs = queue[:]
+    queue = []
 
-    def estimate(self, record):
-        # override: calculate or return the label's value
-        pass
+    logging.info("analyzing batch %s", str(len(jobs)))
+    ids = list(set([str(id, "utf-8") for _, _, id in jobs]))
 
+    with db.no_autoflush:
+        env = trueskill.TrueSkill(
+            backend="mpmath",
+            mu=11.0/30*3000,
+            sigma=3000/3,
+            beta=11.0/30*3000 /2,
+            tau=3000/3 /100
+        )
+        for match in db.query(Match).options(\
+            load_only("api_id")\
+            .subqueryload(Match.rosters)\
+                .load_only("api_id", "match_api_id", "winner")\
+            .subqueryload(Roster.participants)\
+                .load_only("api_id", "match_api_id", "roster_api_id",
+                           "player_api_id", "skill_tier",
+                           "trueskill_sigma", "trueskill_mu")\
+                .subqueryload(Participant.player)\
+                    .load_only("api_id", "trueskill_sigma", "trueskill_mu")\
+         ).filter(Match.api_id.in_(ids)):
+            matchup = []
+            for roster in match.rosters:
+                team = []
+                for participant in roster.participants:
+                    player = participant.player[0]
+                    mu = participant.trueskill_mu or player.trueskill_mu
+                    sigma = participant.trueskill_sigma or player.trueskill_sigma
+                    if mu is None:
+                        # no data -> approximate ts by VST
+                        mu = participant.skill_tier * 100
+                        if mu < 1:
+                            mu = 1  # unranked
+                        sigma = mu / 3
+                        player.trueskill_mu = mu
+                        player.trueskill_sigma = sigma
+                    # store pre match values
+                    participant.trueskill_mu = mu
+                    participant.trueskill_sigma = sigma
 
-class MVPScoreModel(Model):
-    def __init__(self, db):
-        self.features = ["participant_stats.non_jungle_minion_kills"]
-        self.label = "participant_stats.impact_score"
-        self.type = "linear"
-        self.batches = 1
-        self.batchsize = 500
-        self.steps = 500
-        self.id = "kda-win"
-        super().__init__(db)
+                    team.append(env.create_rating(float(mu), float(sigma)))
+                matchup.append(team)
 
-    def estimate(self, record):
-        # for training, rating = participant.winner
-        return record.winner
+            # store the fairness of the match
+            match.trueskill_quality = env.quality(matchup)
+            for team, roster in zip(env.rate(matchup, ranks=[int(not r.winner) for r in match.rosters]),
+                                    match.rosters):
+                # lower rank is better = winner!
+                for rating, participant in zip(team, roster.participants):
+                    player = participant.player[0]
+                    if player.trueskill_mu == participant.trueskill_mu \
+                       and player.trueskill_sigma == participant.trueskill_sigma:
+                        # match hasn't been rated before
+                        player.trueskill_mu = rating.mu
+                        player.trueskill_sigma = rating.sigma
 
-    def rate(self, ids):
-        return [w for l, w in self.predict(ids)]
+    db.commit()
+    # ack all until this one
+    logging.info("acking batch")
+    channel.basic_ack(jobs[-1][0].delivery_tag, multiple=True)
+    # notify web
+    for api_id in ids:
+        channel.basic_publish("amq.topic", "participant." + api_id,
+                              "stats_update")
 
 
 logging.basicConfig(level=logging.INFO)
 if __name__ == "__main__":
     connect()
-    mvpmodel = MVPScoreModel(db)
-    mvpmodel.train(force=True)
-    for name in mvpmodel._model.get_variable_names():
-        logging.info("%s: %s", name,
-                     mvpmodel._model.get_variable_value(name))
+    channel.start_consuming()