Stratz counterpicker

couchdb.py

@@ -52,30 +52,39 @@ def get_all_parsed_matches_more_recent_than(
 
 
 def get_all_matches_with_hero_after_start_time(
-    db: cloudant.database.CouchDatabase, start_time, hero_names=None
+    db: cloudant.database.CouchDatabase, start_time, hero_names=None, potential_hero_names=None
 ):
 
     if hero_names is None:
         hero_names = []
+    if potential_hero_names is None:
+        potential_hero_names = []
 
     hero_names = [name for name in hero_names if name]
+    potential_hero_names = [name for name in potential_hero_names if name]
     heroes = [opendota.find_hero(name) for name in hero_names]
+    potential_heroes = [opendota.find_hero(name) for name in potential_hero_names]
     query_dict = {
         "selector": {
             "start_time": {"$gt": start_time},
         },
         "sort": ["start_time"],
     }
-    if heroes:
-        if len(heroes) == 1:
-            query_dict["selector"]["players"] = {
-                "$elemMatch": {"hero_id": heroes[0]["id"]},
-            }
-        else:
-            selector = [
-                {"players": {"$elemMatch": {"hero_id": hero["id"]}}} for hero in heroes
-            ]
+    if len(heroes) + len(potential_heroes) == 1:
+        query_dict["selector"]["players"] = {
+            "$elemMatch": {"hero_id": heroes[0]["id"]},
+        }
+    elif heroes or potential_heroes:
+        selector = [
+            {"players": {"$elemMatch": {"hero_id": hero["id"]}}} for hero in heroes
+        ]
+        if selector:
             query_dict["selector"]["$and"] = selector
+        selector = [
+            {"players": {"$elemMatch": {"hero_id": hero["id"]}}} for hero in potential_heroes
+        ]
+        if selector:
+            query_dict["selector"]["$or"] = selector
     query = db.get_query_result(**query_dict)
     return query
 

counterpicker.py

@@ -0,0 +1,148 @@
+import argparse
+import math
+import itertools
+import json
+import collections
+import fuzzy_hero_names
+import tabulate
+
+import dateparser
+import tabulate
+
+import couchdb
+import matchlib
+import pprint
+import opendota
+import stratz
+
+def team_of_interest(game, hero_ids, potential_hero_ids):
+    heroes_on_radiant = [player['player_slot'] < 127 for player in game['players'] if player['hero_id'] in hero_ids]
+    potential_heroes_on_radiant = [player['player_slot'] < 127 for player in game['players'] if player['hero_id'] in potential_hero_ids]
+    heroes_all_on_radiant = all(heroes_on_radiant) and any(potential_heroes_on_radiant)
+    heroes_all_on_dire = bool(not any(heroes_on_radiant)) and bool(not all(potential_heroes_on_radiant))
+    return heroes_all_on_radiant, heroes_all_on_dire
+
+
+def games_with_heroes_on_same_team(dbquery, hero_ids, potential_hero_ids):
+    for game in dbquery:
+        heroes_all_on_radiant, heroes_all_on_dire = team_of_interest(game, hero_ids, potential_hero_ids)
+        heroes_on_same_team = heroes_all_on_radiant or heroes_all_on_dire
+        if heroes_on_same_team:
+            yield game
+
+def calculate_winrate_for_opposing_heroes(game, heroes, potential_heroes):
+    heroes_all_on_radiant, heroes_all_on_dire = team_of_interest(game, heroes, potential_heroes)
+    wins = collections.Counter()
+    games = collections.Counter()
+
+    if heroes_all_on_radiant:
+        dire_heroes = [player['hero_id'] for player in game['players'] if player['player_slot'] >= 127]
+        wins.update([hero_id for hero_id in dire_heroes if not game['radiant_win']])
+        games.update(dire_heroes)
+    if heroes_all_on_dire:
+        radiant_heroes = [player['hero_id'] for player in game['players'] if player['player_slot'] < 127]
+        wins.update([hero_id for hero_id in radiant_heroes if game['radiant_win']])
+        games.update(radiant_heroes)
+
+    return wins, games
+
+def calculate_winrates_from_localdb(dbquery, hero_ids, potential_hero_ids):
+    wins = collections.Counter()
+    games = collections.Counter()
+    for i, game in enumerate(games_with_heroes_on_same_team(dbquery, hero_ids, potential_hero_ids)):
+        hero_wins, hero_games = calculate_winrate_for_opposing_heroes(game, hero_ids, potential_hero_ids)
+        wins += hero_wins
+        games += hero_games
+    print(f'{i} games analyzed')
+
+    hero_winrate_table = []
+    for hero_id, game_count in games.items():
+        winrate = wins[hero_id] / game_count
+        hero = opendota.find_hero_by_id(hero_id)
+        standard_deviation = math.sqrt(
+            winrate * (1 - winrate) / game_count
+        )
+        hero_winrate_table.append([
+            hero['localized_name'],
+            wins['hero_id'],
+            game_count,
+            winrate,
+            standard_deviation,
+        ])
+
+    hero_winrate_table.sort(key=lambda row: row[3], reverse=True)
+    print(tabulate.tabulate(hero_winrate_table, headers=['Hero', 'Wins', 'Games', 'Winrate', 'Standard deviation']))
+
+def calculate_from_opendota(hero_ids):
+    hero_matchups = [opendota.get_matchups(hero_id) for hero_id in hero_ids]
+    wins = {}
+    games = {}
+    for matchup_payload in hero_matchups:
+        for hero_matchup in matchup_payload:
+            wins.setdefault(hero_matchup['hero_id'], 0)
+            wins[hero_matchup['hero_id']] += hero_matchup['wins']
+            games.setdefault(hero_matchup['hero_id'], 0)
+            games[hero_matchup['hero_id']] += hero_matchup['games_played']
+
+    hero_winrate_table = []
+    for hero_id, game_count in games.items():
+        hero_winrate_table.append([
+            opendota.find_hero_by_id(hero_id)['localized_name'],
+            wins[hero_id],
+            game_count,
+            wins[hero_id] / game_count,
+        ])
+    hero_winrate_table.sort(key=lambda row: row[3], reverse=True)
+    print(tabulate.tabulate(hero_winrate_table, headers=['Hero', 'Wins', 'Games', 'Winrate']))
+
+def suggest_counterpicks(heroes):
+    with_mtx, vs_mtx = stratz.load_fixed_matchups()
+    counterpick_score = {}
+    for hero in heroes:
+        counterpicks = vs_mtx[hero['id']]
+        for counterpick in counterpicks.values():
+            counterpick_score.setdefault(counterpick['heroId2'], {})
+            counterpick_score[counterpick['heroId2']][hero['id']] = counterpick['synergy']
+            counterpick_score[counterpick['heroId2']].setdefault('total', 0)
+            counterpick_score[counterpick['heroId2']]['total'] += counterpick['synergy']
+    counterpick_score = sorted(counterpick_score.items(), key=lambda x: x[1]['total'])
+    return counterpick_score[:20], counterpick_score[-20:]
+
+def counterpick_report(hero_names):
+    heroes = [fuzzy_hero_names.match(hero_name.strip()) for hero_name in hero_names]
+    counterpicks, bad_picks = suggest_counterpicks(heroes)
+    counterpick_table = []
+    for pick in counterpicks:
+        synergy_by_hero = [pick[1].get(hero['id']) for hero in heroes]
+        counterpick_table.append([
+            opendota.find_hero_by_id(pick[0])['localized_name'],
+            pick[1]['total'],
+            *synergy_by_hero
+        ])
+    print(tabulate.tabulate(counterpick_table, headers=['Hero', 'Total synergy'] + [hero['localized_name'] for hero in heroes]))
+
+    badpick_table = []
+    for pick in bad_picks:
+        synergy_by_hero = [pick[1].get(hero['id'], None) for hero in heroes]
+        badpick_table.append([
+            opendota.find_hero_by_id(pick[0])['localized_name'],
+            pick[1]['total'],
+            *synergy_by_hero
+        ])
+    print(tabulate.tabulate(badpick_table, headers=['Hero', 'Total synergy'] + [hero['localized_name'] for hero in heroes]))
+
+
+if __name__ == "__main__":
+    import sys
+    # parser = argparse.ArgumentParser()
+
+    # parser.add_argument("--process-queue", action="store_true")
+    # parser.add_argument("--populate-queue", action="store_true")
+    # parser.add_argument("--max-matches-to-queue", type=int, default=5)
+
+    # args = parser.parse_args()
+    # if args.process_queue:
+    #     pass
+    hero_names = sys.argv[1:]
+    counterpick_report(hero_names)
+

fuzzy_hero_names.py

@@ -0,0 +1,45 @@
+import opendota
+
+def _score_hero_name(hero_name, input):
+    normalized_hero_name = hero_name.lower()
+    normalized_input = input.strip().lower()
+    if normalized_input == normalized_hero_name:
+        return 100
+    if normalized_hero_name.startswith(normalized_input):
+        return 10 * len(normalized_input)
+
+    # Try to do acronym matching, ie. AM -> Anti-Mage
+    parts = normalized_hero_name.split()
+    if len(parts) == 2 and len(normalized_input) == 2:
+        return 25 * (normalized_input[0] == parts[0][0]) + 25 * (normalized_input[1] == parts[1][0])
+    if len(parts) == 2 and len(normalized_input) > 3 and parts[1].startswith(normalized_input):
+        return 5 * len(normalized_input)
+
+    parts = normalized_hero_name.split('-')
+    if len(parts) == 2 and len(normalized_input) == 2:
+        return 25 * (normalized_input[0] == parts[0][0]) + 25 * (normalized_input[1] == parts[1][0])
+    if len(parts) == 2 and len(normalized_input) > 3 and parts[1].startswith(normalized_input):
+        return 5 * len(normalized_input)
+
+    return 0
+
+def match(input):
+    heroes = opendota.load_hero_list().values()
+    best_score = 0
+    best_match = None
+    for hero in heroes:
+        score = _score_hero_name(hero['localized_name'], input)
+        if score > best_score:
+            best_score = score
+            best_match = hero
+    return best_match
+
+if __name__ == '__main__':
+    print(match('AM'))
+    print(match('grim'))
+    print(match('jugg'))
+    print(match('CM'))
+    print(match('maiden'))
+    print(match('primal'))
+    print(match('OD'))
+    print(match('Pango'))

hero.py

@@ -0,0 +1,13 @@
+#!python
+import opendota
+import fuzzy_hero_names
+import sys
+
+if __name__ == '__main__':
+    input = sys.argv[1]
+
+    try:
+        input = int(input)
+        print(opendota.find_hero_by_id(input))
+    except ValueError:
+        print(fuzzy_hero_names.match(input))

matchlib.py

@@ -163,6 +163,8 @@ def iterate_matches(date_string, limit=200, page_size=DEFAULT_QUERY_PAGE_SIZE):
 
 
 def is_fully_parsed(match):
+    if "players" not in match:
+        return False
     return bool(match["players"][0].get("purchase_log", None))
 
 

opendota.py

@@ -31,6 +31,16 @@ def request_parse(match_id):
     )
     return response.json()
 
+def check_job(job_id):
+    response = requests.get(
+        f"{API_ROOT}/request/{job_id}",
+        params=dict(api_key=secret.OPENDOTA_API_KEY),
+    )
+    return response
+
+def all_heroes():
+    return load_hero_list().values()
+
 
 def find_hero(heroname):
     for hero in load_hero_list().values():
@@ -44,6 +54,10 @@ def find_hero_by_id(hero_id):
     return load_hero_list().get(str(hero_id))
 
 
+def find_hero_name_by_id(hero_id):
+    return load_hero_list().get(str(hero_id))["localized_name"]
+
+
 def get_hero_id(heroname):
     return find_hero(heroname)["id"]
 
@@ -84,6 +98,36 @@ def get_heroes_table():
     )
     return response.json()
 
+@opendota_retry
+def get_abilities():
+    response = requests.get(
+        f"{API_ROOT}/constants/abilities",
+        params=dict(
+            api_key=secret.OPENDOTA_API_KEY,
+        ),
+    )
+    return response.json()
+
+@opendota_retry
+def get_ability_ids():
+    response = requests.get(
+        f"{API_ROOT}/constants/ability_ids",
+        params=dict(
+            api_key=secret.OPENDOTA_API_KEY,
+        ),
+    )
+    return response.json()
+
+@opendota_retry
+def get_matchups(hero_id):
+    response = requests.get(
+        f"{API_ROOT}/heroes/{hero_id}/matchups",
+        params=dict(
+            api_key=secret.OPENDOTA_API_KEY,
+        ),
+    )
+    return response.json()
+
 
 @opendota_retry
 def query_explorer(query):