def make_ratings_table()

in fairdiplomacy/data/build_metadata.py [0:0]

• 72 lines of code
• 23 McCabe index (conditional complexity)

def make_ratings_table(db):
    member_rows = [
        MemberRow(*row)
        for row in db.execute(
            f"SELECT hashed_userID, hashed_gameID, countryID, status FROM {TABLE_MEMBERS}"
        ).fetchall()
    ]
    press_type = {
        game_id: press_type
        for game_id, press_type in db.execute(
            f"SELECT hashed_id, pressType from redacted_games"
        ).fetchall()
    }

    user_ids = list(set(r.user for r in member_rows))

    max_userid = max(user_ids)
    user_stats = [defaultdict(float) for _ in range(max_userid + 1)]

    member_dict = {(r.game, r.country): r for r in member_rows}
    good_games = list(find_good_games(db))

    print(f"Found {len(good_games)} good games.")

    WIN_STATI = ("Won", "Drawn")
    game_stats = {}
    for ii, game_id in enumerate(good_games):
        if ii & (ii - 1) == 0:
            print(f"Done {ii} / {len(good_games)} games")

        user_ids, winners = [], []
        for country_id in range(1, 7 + 1):
            k = (game_id, country_id)
            if k in member_dict:
                member_row = member_dict[k]
                this_user_stats = user_stats[member_row.user]
                this_user_stats["total"] += 1
                if member_row.status not in STATUS_TO_RANK:
                    continue
                this_user_stats[member_row.status] += 1
                if member_row.status in WIN_STATI:
                    winners.append(this_user_stats)
                user_ids.append(member_dict[k].user)

        # allot points to winners
        for winner in winners:
            winner["total_points"] += 1.0 / len(winners)

        this_game_stats = {"id": game_id, "press_type": press_type[game_id]}
        for country_id in range(1, 7 + 1):
            pwr = COUNTRY_ID_TO_POWER[country_id]
            k = (game_id, country_id)
            if k in member_dict:
                member_row = member_dict[k]
                u = member_row.user
                this_game_stats[pwr] = {
                    "id": u,
                    "points": 1.0 / len(winners) if member_row.status in WIN_STATI else 0,
                    "status": member_row.status,
                }
            else:
                this_game_stats[pwr] = None

        game_stats[game_id] = this_game_stats

    print("Computing logit scores")
    ratings = compute_logit_ratings(game_stats)
    for i in range(len(user_stats)):
        user_stats[i]["logit_rating"] = ratings[i]

    print("Adding final stats")
    for ii, game_id in enumerate(good_games):
        this_game_stats = game_stats[game_id]
        if this_game_stats is None:
            continue
        for country_id in range(1, 7 + 1):
            pwr = COUNTRY_ID_TO_POWER[country_id]
            k = (game_id, country_id)
            if k in member_dict:
                this_game_stats[pwr].update(**user_stats[member_dict[k].user])

    user_stats = [
        {**user_stats[u], "id": u} for u in range(max_userid) if user_stats[u]["total"] > 0
    ]
    return game_stats, user_stats