#!/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()