From 0766892430698c4ace6bf99c768ed7ab1c961003 Mon Sep 17 00:00:00 2001 From: schneefux Date: Wed, 15 Mar 2017 20:28:12 +0100 Subject: refactor, write role classification model (wip) --- api.py | 254 +++++------------------------------------------------------------ 1 file changed, 18 insertions(+), 236 deletions(-) (limited to 'api.py') diff --git a/api.py b/api.py index 7f294eb..49a2b0d 100644 --- a/api.py +++ b/api.py @@ -1,16 +1,12 @@ #!/usr/bin/python import os -import itertools -import json import asyncio -import asyncpg import logging -import numpy as np -import tensorflow as tf -import psycopg2 import joblib.worker +import classifier + #tf.logging.set_verbosity(tf.logging.WARNING) @@ -31,254 +27,40 @@ db_config = { } -# TODO create abstract class, move to own file -class KDAClassifier(object): - """A DNN that classifies loss/win based on KDA.""" - def __init__(self): - self.model = None - self.modeldir = os.path.realpath( - os.path.join(os.getcwd(), os.path.dirname(__file__))) + "/models/kda-win" - self._pool = None - - # DNN configuration - self._categories = [] - self._continuous = ["kills", "deaths", "assists", "cs", "teamkills"] - self._classes = ["loss", "win"] - self._label = "win" - - # learn configuration - self._steps = 200 # per batch - self._batchsize = 500 - self._max_batches = 5 # number of batches to train - - # database mappings - # TODO cache, rm duplicated code - self._trainquery = """ - SELECT - participant.kills, - participant.deaths, - participant.assists, - participant.farm, - roster.hero_kills, - participant.winner, - participant.api_id - FROM participant - TABLESAMPLE BERNOULLI(5) - JOIN roster on participant.roster_api_id=roster.api_id - JOIN match on roster.match_api_id=match.api_id - LIMIT %s - """ - self._testquery = """ - SELECT - participant.kills, - participant.deaths, - participant.assists, - participant.farm, - roster.hero_kills, - participant.winner, - participant.api_id - FROM participant - JOIN roster on participant.roster_api_id=roster.api_id - JOIN match on roster.match_api_id=match.api_id - WHERE participant.api_id=ANY(%s) - """ - - def connect(self, **args): - self._conn = psycopg2.connect(**args) - - def _get_sample(self, size=None, filter=None): - """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 - """ - cur = self._conn.cursor() - assert (filter is None) ^ (size is None) - if filter is None: - cur.execute(self._trainquery, (size,)) - else: - cur.execute(self._testquery, (filter,)) - ids = [] - data = {"kills": [], "deaths": [], "assists": [], "win": [], "cs": [], "teamkills": []} - for rec in cur: - data["kills"].append(rec[0]) - data["deaths"].append(rec[1]) - data["assists"].append(rec[2]) - data["cs"].append(rec[3]) - data["teamkills"].append(rec[4]) - data["win"].append(rec[5]) - ids.append(rec[6]) - cur.close() - logging.debug(data) - return data, ids - - 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 - ) - ) - - # TODO tf warning - dimensions are wrong - 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): - """Fetches up to `limit` number of training items - from the database in batches.""" - # TODO maybe you can use an iterator? - sample = self._get_sample(size=self._batchsize)[0] - 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): - logging.warning("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)[0], train=True), - eval_steps=1, - every_n_steps=10 - ) - - 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)) - - def windup(self): - pass - - def teardown(self, failed=False): - if failed: - pass - else: - self._conn.commit() - - def classify_db(self, objids): - """Classify all data in the data base and insert.""" - # split sample (with participant ids) into data and ids - # TODO use parameters - logging.error("sample size: %s", len(objids)) - sample, objids = self._get_sample(filter=objids) - d = self.classify(sample) - cnt = 0 - cur = self._conn.cursor() - for l in itertools.islice(d, len(objids)): - cur.execute(""" - INSERT INTO participant_stats - (patch_version, participant_api_id, score) - VALUES(2.2, %(objid)s, %(score)s) - ON CONFLICT(participant_api_id) DO - UPDATE SET score=%(score)s - """, {"objid": objids[cnt], "score": float(l[1])}) - cnt += 1 - cur.close() - - class Analyzer(joblib.worker.Worker): def __init__(self): - self._pool = None self._queries = {} super().__init__(jobtype="analyze") - self.classifier = None + self.classifiers = [] async def connect(self, dbconf, queuedb): """Connect to database.""" logging.warning("connecting to database") await super().connect(**queuedb) - self._pool = await asyncpg.create_pool(**dbconf) - self.classifier = KDAClassifier() - self.classifier.connect(**dbconf) - self.classifier._step = 1000 # TODO test batch size + cl = classifier.KDAClassifier() + cl.connect(**dbconf) + self.classifiers.append(cl) +# cl = classifier.RoleClassifier() +# cl.connect(**dbconf) +# self.classifiers.append(cl) async def setup(self): """Setup the model.""" - self.classifier.train() + for cl in self.classifiers: + cl.train() async def _windup(self): - self._con = await self._pool.acquire() - self._tr = self._con.transaction() - await self._tr.start() - self.classifier.windup() + for cl in self.classifiers: + cl.windup() self._participants = [] async def _teardown(self, failed): if len(self._participants) > 0: # TODO if this fails, job is still marked as finished - self.classifier.classify_db(self._participants) - - if failed: - await self._tr.rollback() - else: - await self._tr.commit() - await self._pool.release(self._con) - self.classifier.teardown() + for cl in self.classifiers: + cl.classify_db(self._participants) + for cl in self.classifiers: + cl.teardown() async def _execute_job(self, jobid, payload, priority): object_id = payload["id"] @@ -294,7 +76,7 @@ async def startup(): worker = Analyzer() await worker.connect(db_config, queue_db) await worker.setup() - await worker.start(batchlimit=1000) + await worker.start(batchlimit=10000) logging.basicConfig(level=logging.DEBUG) -- cgit v1.3.1