rework, add wip kda rating

3 files changed, 261 insertions, 102 deletions

diff --git a/kdarate.py b/kdarate.py
new file mode 100644
index 0000000..c66ab30
--- /dev/null
+++ b/kdarate.py
@@ -0,0 +1,253 @@
+#!/usr/bin/python
+
+import os
+import itertools
+import json
+import asyncio
+import asyncpg
+import numpy as np
+import tensorflow as tf
+tf.logging.set_verbosity(tf.logging.INFO)
+
+
+class KDAClassifier(object):
+    """A DNN that classifies loss/win based on KDA."""
+    def __init__(self):
+        self.model = None
+        self.modeldir = "/tmp/vgstats-tf-actor-model"
+        self._pool = None
+
+        # DNN configuration
+        self._categories = []
+        self._continuous = ["kills", "deaths", "assists"]
+        self._classes = ["loss", "win"]
+        self._label = "win"
+
+        # learn configuration
+        self._steps = 200  # per batch
+        self._max_batches = 20  # number of batches to train
+
+        # database mappings
+        self._paths = {
+            "id": "id",
+            "kills": "(data->'attributes'->'stats'->>'kills')::int",
+            "deaths": "(data->'attributes'->'stats'->>'deaths')::int",
+            "assists": "(data->'attributes'->'stats'->>'assists')::int",
+#            "actor": "data->'attributes'->>'actor'",
+            "win": "(data->'attributes'->'stats'->>'winner')::bool::int"
+        }
+#        for n in range(0, 6):
+#            self._paths["item"+str(n)] = \
+#            "COALESCE(data->'attributes'->'stats'->'items'->>" + str(n) +", '')"
+
+        self._step = 0  # saves batch learning state
+
+    async def _async_connect(self, dbstring):
+        """Connect to the database."""
+        # TODO with current synchronous implementation, pooling isn't needed
+        self._pool = await asyncpg.create_pool(dbstring)
+
+        # create layout
+        # maps participant -> role jsonb
+        async with self._pool.acquire() as conn:
+            async with conn.transaction():
+                await conn.execute(
+                    # TODO do this globally etc
+                    "CREATE TABLE IF NOT EXISTS enhanced_participant_wip" +
+                    "(participant_id TEXT PRIMARY KEY, actor JSONB)")
+
+    def connect(self, dbstring):
+        """See _async_connect."""
+        asyncio.get_event_loop().run_until_complete(
+            self._async_connect(dbstring))
+
+    # TODO maybe don't use asyncpg
+    async def _async_get_sample(self, size="ALL", offset=0):
+        """Return a data set from the database.
+        :param size: (optional) The number of items to get. Defaults to `All`.
+        :type size: int or str
+        :param offset: (optional) The SQL OFFSET parameter.
+        :type offset: int or str
+        :type identified: bool
+        """
+        query = "SELECT "
+        elements = []
+        for name, path in self._paths.items():
+            elements.append(
+                """
+                ARRAY(SELECT
+                  {3}
+                  FROM participant
+                  ORDER BY id
+                  LIMIT {0} OFFSET {1}
+                ) AS {2}
+                """.format(size, offset, name, path))
+
+        query += ", ".join(elements)
+
+        async with self._pool.acquire() as conn:
+            async with conn.transaction():
+                return (await conn.fetch(query))[0]
+
+    def _get_sample(self, size="ALL", offset=0):
+        """See _async_get_sample."""
+        return asyncio.get_event_loop().run_until_complete(
+            self._async_get_sample(size, offset))
+
+    def _model_setup(self):
+        """Sets up a model that takes the `num_features` features as input."""
+        feature_columns = [
+            tf.contrib.layers.sparse_column_with_hash_bucket(feat, 100)
+            for feat in self._categories
+        ]
+        emb_columns = [
+            tf.contrib.layers.embedding_column(
+                sparse_id_column=col,
+                dimension=5  # log_2(number of unique features)  TODO
+            )
+            for col in feature_columns
+        ] + [
+            tf.contrib.layers.real_valued_column(feat)
+            for feat in self._continuous
+        ]
+
+        self.model = tf.contrib.learn.DNNClassifier(
+            feature_columns=emb_columns,
+            hidden_units=[2],  # http://stats.stackexchange.com/a/1097  TODO inp+outp / 2
+            n_classes=len(self._classes),
+            model_dir=self.modeldir,
+            config=tf.contrib.learn.RunConfig(
+                save_checkpoints_secs=2
+            )
+        )
+
+    def _toinput(self, sample, train=True):
+        """Convert sample to Tensors.
+        :param sample: Data dictionary.
+        :type sample: dict
+        :param train: (optional) Whether to return labels too.
+        :type train: bool
+        :return: features, label
+        :rtype: tuple
+        """
+        continuous = {k: tf.constant(sample[k])
+                      for k in self._continuous}
+        categories = {k: tf.SparseTensor(
+            indices=[[i, 0] for i in range(len(sample[k]))],
+            values=sample[k],
+            shape=[len(sample[k]), 1])
+                    for k in self._categories}
+
+        features = {**continuous, **categories}
+
+        if train:
+            label = tf.constant(sample[self._label])
+            return features, label
+        else:
+            return features
+
+    # TODO use asyncpg cursor https://magicstack.github.io/asyncpg/current/api/index.html#cursors
+    def _more(self, batchsize=500, limit=None):
+        """Fetches up to `limit` number of training items
+           from the database in batches."""
+        # TODO maybe you can use an iterator?
+        if limit:
+            if self._step > limit:
+#               self._step = 0
+                # TODO ????
+                raise tf.errors.OutOfRangeError
+        sample = self._get_sample(
+            size=batchsize, offset=self._step)
+        self._step += batchsize
+        return self._toinput(sample, train=True)
+
+    def train(self):
+        """Train a DNN on a static, algorithmic guess."""
+        self._model_setup()
+
+        if os.path.isdir(self.modeldir):
+            print("already trained, not training again")
+            return
+
+        # get one batch of testing data
+        # TODO either terminate with `eval_steps` or with OutOfRangeError
+        validation_monitor = tf.contrib.learn.monitors.ValidationMonitor(
+            input_fn=lambda: self._toinput(self._get_sample(size=1000, offset=101000), train=True),
+            eval_steps=1,
+            every_n_steps=20,
+            early_stopping_metric="accuracy",
+            early_stopping_metric_minimize=False,
+            early_stopping_rounds=100
+        )
+
+        # validation monitor stops learning if the accuracy does not increase of 100 steps
+        for _ in range(self._max_batches):
+            self.model.fit(
+                input_fn=lambda: self._more(),
+                steps=self._steps,
+                monitors=[validation_monitor]
+            )
+
+    def classify(self, sample, only_best=False):
+        """Classify a data set.
+
+        :param only_best: (optional) Return the predicted result
+                          instead of a dict of propabilities.
+        :type only_best: bool
+        :return: Prediction results.
+        :rtype: list of dict or list
+        """
+        if only_best:
+            return self.model.predict(input_fn=lambda: self._toinput(sample, train=False))
+        else:
+            return self.model.predict_proba(input_fn=lambda: self._toinput(sample, train=False))
+
+    async def classify_db(self):
+        # TODO rewrite
+        """Classify all data in the data base and insert."""
+        sample = self._get_sample(identified=True)
+
+        # split sample (with participant ids) into data and ids
+        data = np.array(
+            [[val for key, val in s.items() if key != "id"] for s in sample]
+        )
+        ids = [s["id"] for s in sample]
+        # let the DNN do the work
+        predicts = self.classify(data)
+
+        # convert numpy->dict->json for db
+        predicts = [
+            json.dumps({
+                k: float(v) for k, v in p.items()
+            }) for p in predicts
+        ]
+        dbdata = list(zip(ids, predicts))
+        async with self._pool.acquire() as conn:
+            async with conn.transaction():
+                await conn.executemany(
+                    """INSERT INTO enhanced_participant_wip(participant_id, actor)
+                       VALUES ($1, $2)
+                       ON CONFLICT (participant_id) DO
+                       UPDATE SET data=$2
+                    """, dbdata)
+
+
+def main():
+    classifier = KDAClassifier()
+    classifier.connect("postgres://vgstats@localhost/vgstats")
+    # TODO train only for the latest patch.
+    print("training")
+    classifier.train()
+    print("done training")
+    sample = classifier._get_sample(size=10, offset=102000) # DEBUG
+    print(sample)
+    # TODO train batches/fixed number
+    print("classifying debug data")
+    d = classifier.classify(sample)
+    for l in itertools.islice(d, 10):
+        print(l)
+    print(sample["win"])
+    print("classified")
+    #classifier.classify_db()
+
+main()
diff --git a/requirements.txt b/requirements.txt
new file mode 100644
index 0000000..dacd504
--- /dev/null
+++ b/requirements.txt
@@ -0,0 +1,8 @@
+appdirs==1.4.0
+asyncpg==0.8.4
+numpy==1.12.0
+packaging==16.8
+protobuf==3.2.0
+pyparsing==2.1.10
+six==1.10.0
+tensorflow-gpu==0.12.1
diff --git a/tf.py b/tf.py
deleted file mode 100644
index ab72ad1..0000000
--- a/tf.py
+++ /dev/null
@@ -1,102 +0,0 @@
-#!/usr/bin/python
-"""
-Work in progress
- - Tensorflow Linear Regression to predict win/loss based on KDA
-"""
-
-import asyncio
-import pandas
-import tensorflow as tf
-
-import api.database
-
-
-db = api.database.Database()
-
-
-async def get_winrate_sample():
-    return await db.select(
-        """
-        SELECT 
-            (data->'attributes'->'stats'->>'winner')::bool AS win,
-            data->'attributes'->>'actor' AS hero,
-            (data->'attributes'->'stats'->>'farm')::float AS cs,
-            (data->'attributes'->'stats'->>'kills')::int AS k,
-            (data->'attributes'->'stats'->>'deaths')::int AS d,
-            (data->'attributes'->'stats'->>'assists')::int AS a
-        FROM participant
-        """
-    )
-
-def to_input_winrate(df):
-    LABEL = "win"
-    CATEGORIES = ["hero"]
-    CONTINUOUS = ["cs", "k", "d", "a"]
-
-    # convert constants to tensors
-    conts = {k: tf.constant(df[k])
-             for k in CONTINUOUS}
-    # convert categories to sparse tensors
-    cats = {k: tf.SparseTensor(
-        indices=[[i, 0] for i in range(len(df[k]))],
-        values=df[k],
-        shape=[len(df[k]), 1])
-            for k in CATEGORIES}
-
-    df[LABEL] = [int(d) for d in df[LABEL]]  # bool to 0 / 1
-
-    features = {**conts, **cats}
-    label = tf.constant(df[LABEL])
-    return features, label
-
-async def train_winrate():
-    data = await get_winrate_sample()  # list of dict
-    trainlimit = int(len(data) * 0.8)
-
-    # convert to dict of list
-    train_sample = {}
-    for it in data[:trainlimit]:
-        for k, v in it.items():
-            try:
-                train_sample[k].append(v)
-            except KeyError:
-                train_sample[k] = [v]
-    test_sample = {}
-    for it in data[trainlimit:]:
-        for k, v in it.items():
-            try:
-                test_sample[k].append(v)
-            except KeyError:
-                test_sample[k] = [v]
-
-    # create input layers
-    cs = tf.contrib.layers.real_valued_column("cs")
-    k = tf.contrib.layers.real_valued_column("k")
-    d = tf.contrib.layers.real_valued_column("d")
-    a = tf.contrib.layers.real_valued_column("a")
-    hero = tf.contrib.layers.sparse_column_with_hash_bucket("hero", hash_bucket_size=50)
-
-    wide_columns = [
-        hero, cs, k, d, a
-    ]
-
-    model = tf.contrib.learn.LinearClassifier(
-        feature_columns=wide_columns
-    )
-
-    def train():
-        return to_input_winrate(train_sample)
-    def test():
-        return to_input_winrate(test_sample)
-
-    model.fit(input_fn=train, steps=200)
-    results = model.evaluate(input_fn=test, steps=1)
-    for key in sorted(results):
-        print("%s: %s" % (key, results[key]))
-
-
-tf.logging.set_verbosity(tf.logging.INFO)
-
-loop = asyncio.get_event_loop()
-loop.run_until_complete(db.connect("postgres://vgstats@localhost/vgstats"))
-loop.run_until_complete(train_winrate())