From 3d85cf5e4d82b7d9e369f907562ae80aa7ce1cc8 Mon Sep 17 00:00:00 2001 From: schneefux Date: Mon, 6 Feb 2017 10:55:06 +0100 Subject: rework, add wip kda rating --- kdarate.py | 253 +++++++++++++++++++++++++++++++++++++++++++++++++++++++ requirements.txt | 8 ++ tf.py | 102 ---------------------- 3 files changed, 261 insertions(+), 102 deletions(-) create mode 100644 kdarate.py create mode 100644 requirements.txt delete mode 100644 tf.py 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()) -- cgit v1.3.1