ml/predictor.py

import os.path
from operator import itemgetter
from typing import Text, Tuple

import numpy as np
import pandas as pd
import requests

from configs.config import cfg
from configs.constants import DATA_ROOT
from ml.model import MLModel
from ml.utils import load_pickle

from datetime import tzinfo, timedelta, datetime

ZERO = timedelta(0)


class UTC(tzinfo):
    def utcoffset(self, dt):
        return ZERO

    def tzname(self, dt):
        return "UTC"

    def dst(self, dt):
        return ZERO


class Predictor:
    """
    A match predictor using ML
    """

    def __init__(self, base_df: pd.DataFrame, model: MLModel):
        self.model = model
        self.base_df = base_df

    def find_stats(self, team):
        """

        :param team: Name of the team, eg: Qatar, etc.
        :return:
        """

        last_game = self.base_df[
            (self.base_df["home_team"] == team) | (self.base_df["away_team"] == team)
        ].tail(1)

        if last_game["home_team"].values[0] == team:
            team_rank = last_game["rank_home"].values[0]
            team_goals = last_game["home_goals_mean"].values[0]
            team_goals_l5 = last_game["home_goals_mean_l5"].values[0]
            team_goals_suf = last_game["home_goals_suf_mean"].values[0]
            team_goals_suf_l5 = last_game["home_goals_suf_mean_l5"].values[0]
            team_rank_suf = last_game["home_rank_mean"].values[0]
            team_rank_suf_l5 = last_game["home_rank_mean_l5"].values[0]
            team_gp_rank = last_game["home_game_points_rank_mean"].values[0]
            team_gp_rank_l5 = last_game["home_game_points_rank_mean_l5"].values[0]
        else:
            team_rank = last_game["rank_away"].values[0]
            team_goals = last_game["away_goals_mean"].values[0]
            team_goals_l5 = last_game["away_goals_mean_l5"].values[0]
            team_goals_suf = last_game["away_goals_suf_mean"].values[0]
            team_goals_suf_l5 = last_game["away_goals_suf_mean_l5"].values[0]
            team_rank_suf = last_game["away_rank_mean"].values[0]
            team_rank_suf_l5 = last_game["away_rank_mean_l5"].values[0]
            team_gp_rank = last_game["away_game_points_rank_mean"].values[0]
            team_gp_rank_l5 = last_game["away_game_points_rank_mean_l5"].values[0]

        return [
            team_rank,
            team_goals,
            team_goals_l5,
            team_goals_suf,
            team_goals_suf_l5,
            team_rank_suf,
            team_rank_suf_l5,
            team_gp_rank,
            team_gp_rank_l5,
        ]

    @staticmethod
    def find_features(team_1, team_2):
        """

        :param team_1:
        :param team_2:
        :return:
        """
        rank_dif = team_1[0] - team_2[0]
        goals_dif = team_1[1] - team_2[1]
        goals_dif_l5 = team_1[2] - team_2[2]
        goals_suf_dif = team_1[3] - team_2[3]
        goals_suf_dif_l5 = team_1[4] - team_2[4]
        goals_per_ranking_dif = (team_1[1] / team_1[5]) - (team_2[1] / team_2[5])
        dif_rank_agst = team_1[5] - team_2[5]
        dif_rank_agst_l5 = team_1[6] - team_2[6]
        dif_gp_rank = team_1[7] - team_2[7]
        dif_gp_rank_l5 = team_1[8] - team_2[8]

        return [
            rank_dif,
            goals_dif,
            goals_dif_l5,
            goals_suf_dif,
            goals_suf_dif_l5,
            goals_per_ranking_dif,
            dif_rank_agst,
            dif_rank_agst_l5,
            dif_gp_rank,
            dif_gp_rank_l5,
            1,
            0,
        ]

    def __predict(self, team_1: Text, team_2: Text):

        team_1_stat = self.find_stats(team_1)
        team_2_stat = self.find_stats(team_2)

        features_g1 = self.find_features(team_1_stat, team_2_stat)
        features_g2 = self.find_features(team_2_stat, team_1_stat)

        probs_g1 = self.model.predict_proba([features_g1])
        probs_g2 = self.model.predict_proba([features_g2])
        team_1_prob_g1 = probs_g1[0][0]
        team_1_prob_g2 = probs_g2[0][1]
        team_2_prob_g1 = probs_g1[0][1]
        team_2_prob_g2 = probs_g2[0][0]

        team_1_prob = (probs_g1[0][0] + probs_g2[0][1]) / 2
        team_2_prob = (probs_g2[0][0] + probs_g1[0][1]) / 2

        return (
            team_1_prob_g1,
            team_1_prob_g2,
            team_1_prob,
            team_2_prob,
            team_2_prob_g1,
            team_2_prob_g2,
        )

    def predict(self, team_1: Text, team_2: Text) -> Tuple[bool, Text, float]:
        """

        :param team_1:
        :param team_2:
        :return:
        """
        draw = False
        (
            team_1_prob_g1,
            team_1_prob_g2,
            team_1_prob,
            team_2_prob,
            team_2_prob_g1,
            team_2_prob_g2,
        ) = self.__predict(team_1, team_2)
        winner, winner_proba = "", 0.0
        if ((team_1_prob_g1 > team_2_prob_g1) & (team_2_prob_g2 > team_1_prob_g2)) | (
            (team_1_prob_g1 < team_2_prob_g1) & (team_2_prob_g2 < team_1_prob_g2)
        ):
            draw = True

        elif team_1_prob > team_2_prob:
            winner = team_1
            winner_proba = team_1_prob

        elif team_2_prob > team_1_prob:
            winner = team_2
            winner_proba = team_2_prob
        return draw, winner, winner_proba

    def predict_all_matches(self) -> Text:
        """
        Predict all the matches in the tournament
        :return:
        """
        result = ""
        data = load_pickle(os.path.join(DATA_ROOT, cfg.data.table_matches))
        table = data["table"]
        matches = data["matches"]
        advanced_group, last_group = [], ""

        for teams in matches:
            draw = False
            (
                team_1_prob_g1,
                team_1_prob_g2,
                team_1_prob,
                team_2_prob,
                team_2_prob_g1,
                team_2_prob_g2,
            ) = self.__predict(teams[1], teams[2])
            winner, winner_proba = "", 0.0
            if (
                (team_1_prob_g1 > team_2_prob_g1) & (team_2_prob_g2 > team_1_prob_g2)
            ) | ((team_1_prob_g1 < team_2_prob_g1) & (team_2_prob_g2 < team_1_prob_g2)):
                draw = True
                for i in table[teams[0]]:
                    if i[0] == teams[1] or i[0] == teams[2]:
                        i[1] += 1

            elif team_1_prob > team_2_prob:
                winner = teams[1]
                winner_proba = team_1_prob
                for i in table[teams[0]]:
                    if i[0] == teams[1]:
                        i[1] += 3

            elif team_2_prob > team_1_prob:
                winner = teams[2]
                winner_proba = team_2_prob
                for i in table[teams[0]]:
                    if i[0] == teams[2]:
                        i[1] += 3

            for i in table[teams[0]]:  # adding tiebreaker (probs per game)
                if i[0] == teams[1]:
                    i[2].append(team_1_prob)
                if i[0] == teams[2]:
                    i[2].append(team_2_prob)

            if last_group != teams[0]:
                if last_group != "":
                    result += "\n"
                    result += "Group %s advanced: \n" % last_group
                    for i in table[last_group]:  # adding tiebreaker
                        i[2] = np.mean(i[2])

                    final_points = table[last_group]
                    final_table = sorted(
                        final_points, key=itemgetter(1, 2), reverse=True
                    )
                    advanced_group.append([final_table[0][0], final_table[1][0]])
                    for i in final_table:
                        result += "%s -------- %d\n" % (i[0], i[1])
                result += "\n"
                result += (
                    "-" * 10
                    + " Starting Analysis for Group %s " % (teams[0])
                    + "-" * 10
                    + "\n"
                )

            if draw is False:
                result += "Group %s - %s vs. %s: Winner %s with %.2f probability\n" % (
                    teams[0],
                    teams[1],
                    teams[2],
                    winner,
                    winner_proba,
                )
            else:
                result += "Group %s - %s vs. %s: Draw\n" % (
                    teams[0],
                    teams[1],
                    teams[2],
                )
            last_group = teams[0]
        result += "\n"
        result += "Group %s advanced: \n" % last_group

        for i in table[last_group]:  # adding tiebreaker
            i[2] = np.mean(i[2])

        final_points = table[last_group]
        final_table = sorted(final_points, key=itemgetter(1, 2), reverse=True)
        advanced_group.append([final_table[0][0], final_table[1][0]])
        for i in final_table:
            result += "%s -------- %d\n" % (i[0], i[1])

        advanced = advanced_group
        playoffs = {
            "Round of 16": [],
            "Quarter-Final": [],
            "Semi-Final": [],
            "Final": [],
        }

        for p in playoffs.keys():
            playoffs[p] = []

        actual_round = ""
        next_rounds = []

        for p in playoffs.keys():
            if p == "Round of 16":
                control = []
                for a in range(0, len(advanced * 2), 1):
                    if a < len(advanced):
                        if a % 2 == 0:
                            control.append((advanced * 2)[a][0])
                        else:
                            control.append((advanced * 2)[a][1])
                    else:
                        if a % 2 == 0:
                            control.append((advanced * 2)[a][1])
                        else:
                            control.append((advanced * 2)[a][0])
                playoffs[p] = [
                    [control[c], control[c + 1]]
                    for c in range(0, len(control) - 1, 1)
                    if c % 2 == 0
                ]

                for i in range(0, len(playoffs[p]), 1):
                    game = playoffs[p][i]

                    home = game[0]
                    away = game[1]

                    (
                        team_1_prob_g1,
                        team_1_prob_g2,
                        team_1_prob,
                        team_2_prob,
                        team_2_prob_g1,
                        team_2_prob_g2,
                    ) = self.__predict(home, away)
                    if actual_round != p:
                        result += "-" * 10 + "\n"
                        result += "Starting simulation of %s\n" % p
                        result += "-" * 10 + "\n"

                    if team_1_prob < team_2_prob:
                        result += "%s vs. %s: %s advances with prob %.2f\n" % (
                            home,
                            away,
                            away,
                            team_2_prob,
                        )
                        next_rounds.append(away)
                    else:
                        result += "%s vs. %s: %s advances with prob %.2f\n" % (
                            home,
                            away,
                            home,
                            team_1_prob,
                        )
                        next_rounds.append(home)

                    game.append([team_1_prob, team_2_prob])
                    playoffs[p][i] = game
                    actual_round = p

            else:
                playoffs[p] = [
                    [next_rounds[c], next_rounds[c + 1]]
                    for c in range(0, len(next_rounds) - 1, 1)
                    if c % 2 == 0
                ]
                next_rounds = []
                for i in range(0, len(playoffs[p])):
                    game = playoffs[p][i]
                    home = game[0]
                    away = game[1]

                    (
                        team_1_prob_g1,
                        team_1_prob_g2,
                        team_1_prob,
                        team_2_prob,
                        team_2_prob_g1,
                        team_2_prob_g2,
                    ) = self.__predict(home, away)
                    if actual_round != p:
                        result += "-" * 10 + "\n"
                        result += "Starting simulation of %s\n" % p
                        result += "-" * 10 + "\n"

                    if team_1_prob < team_2_prob:
                        result += "%s vs. %s: %s advances with prob %.2f \n" % (
                            home,
                            away,
                            away,
                            team_2_prob,
                        )
                        next_rounds.append(away)
                    else:
                        result += "%s vs. %s: %s advances with prob %.2f \n" % (
                            home,
                            away,
                            home,
                            team_1_prob,
                        )
                        next_rounds.append(home)
                    game.append([team_1_prob, team_2_prob])
                    playoffs[p][i] = game
                    actual_round = p

        print(result)
        return result