ml/data_prepare.py

"""
The data process is base on https://www.kaggle.com/code/sslp23/predicting-fifa-2022-world-cup-with-ml
"""
import os.path

import pandas as pd
from sklearn.model_selection import train_test_split

from configs.config import cfg
from configs.constants import DATA_ROOT


def result_finder(home, away):
    """
    Encode the data
    :param home:
    :param away:
    :return:
    """
    if home > away:
        return pd.Series([0, 3, 0])
    if home < away:
        return pd.Series([1, 0, 3])
    else:
        return pd.Series([2, 1, 1])


def create_dataset(df: pd.DataFrame):
    """
    Create train, test dataset
    :param df:
    :return:
    """
    x_, y = df.iloc[:, 3:], df[["target"]]
    x_train, x_test, y_train, y_test = train_test_split(
        x_, y, test_size=0.22, random_state=100
    )
    return x_train, x_test, y_train, y_test


def data_preparing():
    """
    Data preparing
    :return:
    """
    try:
        df = pd.read_csv(cfg.data.result_url)
    except Exception as e:
        print(e)
        df = pd.read_csv(os.path.join(DATA_ROOT, cfg.data.result_file))
    df["date"] = pd.to_datetime(df["date"])
    df.dropna(inplace=True)
    df = df[(df["date"] >= cfg.day_get_result)].reset_index(drop=True)

    # RANK data prepare
    rank = pd.read_csv(os.path.join(DATA_ROOT, cfg.data.rank_file))
    rank["rank_date"] = pd.to_datetime(rank["rank_date"])
    rank = rank[(rank["rank_date"] >= cfg.day_get_rank)].reset_index(drop=True)
    rank["country_full"] = (
        rank["country_full"]
        .str.replace("IR Iran", "Iran")
        .str.replace("Korea Republic", "South Korea")
        .str.replace("USA", "United States")
    )

    # The merge is made in order to get a dataset FIFA games and its rankings.
    rank = (
        rank.set_index(["rank_date"])
        .groupby(["country_full"], group_keys=False)
        .resample("D")
        .first()
        .fillna(method="ffill")
        .reset_index()
    )
    df_wc_ranked = df.merge(
        rank[
            [
                "country_full",
                "total_points",
                "previous_points",
                "rank",
                "rank_change",
                "rank_date",
            ]
        ],
        left_on=["date", "home_team"],
        right_on=["rank_date", "country_full"],
    ).drop(["rank_date", "country_full"], axis=1)

    df_wc_ranked = df_wc_ranked.merge(
        rank[
            [
                "country_full",
                "total_points",
                "previous_points",
                "rank",
                "rank_change",
                "rank_date",
            ]
        ],
        left_on=["date", "away_team"],
        right_on=["rank_date", "country_full"],
        suffixes=("_home", "_away"),
    ).drop(["rank_date", "country_full"], axis=1)

    # Featuring
    df = df_wc_ranked

    df[["result", "home_team_points", "away_team_points"]] = df.apply(
        lambda x: result_finder(x["home_score"], x["away_score"]), axis=1
    )

    # we create columns that will help in the creation of the features: ranking difference,
    # points won at the game vs. team faced rank, and goals difference in the game.
    # All features that are not differences should be created for the two teams (away and home).
    df["rank_dif"] = df["rank_home"] - df["rank_away"]
    df["sg"] = df["home_score"] - df["away_score"]
    df["points_home_by_rank"] = df["home_team_points"] / df["rank_away"]
    df["points_away_by_rank"] = df["away_team_points"] / df["rank_home"]

    # In order to create the features, I'll separate the dataset in home team's and away team's dataset,
    # unify them and calculate the past game values.
    # After that, I'll separate again and merge them, retrieving the original dataset.
    # This process optimizes the creation of the features.
    home_team = df[
        [
            "date",
            "home_team",
            "home_score",
            "away_score",
            "rank_home",
            "rank_away",
            "rank_change_home",
            "total_points_home",
            "result",
            "rank_dif",
            "points_home_by_rank",
            "home_team_points",
        ]
    ]

    away_team = df[
        [
            "date",
            "away_team",
            "away_score",
            "home_score",
            "rank_away",
            "rank_home",
            "rank_change_away",
            "total_points_away",
            "result",
            "rank_dif",
            "points_away_by_rank",
            "away_team_points",
        ]
    ]
    home_team.columns = [
        h.replace("home_", "")
        .replace("_home", "")
        .replace("away_", "suf_")
        .replace("_away", "_suf")
        for h in home_team.columns
    ]

    away_team.columns = [
        a.replace("away_", "")
        .replace("_away", "")
        .replace("home_", "suf_")
        .replace("_home", "_suf")
        for a in away_team.columns
    ]
    team_stats = home_team.append(away_team)

    stats_val = []

    for index, row in team_stats.iterrows():
        team = row["team"]
        date = row["date"]
        past_games = team_stats.loc[
            (team_stats["team"] == team) & (team_stats["date"] < date)
        ].sort_values(by=["date"], ascending=False)
        last5 = past_games.head(5)

        goals = past_games["score"].mean()
        goals_l5 = last5["score"].mean()

        goals_suf = past_games["suf_score"].mean()
        goals_suf_l5 = last5["suf_score"].mean()

        rank = past_games["rank_suf"].mean()
        rank_l5 = last5["rank_suf"].mean()

        if len(last5) > 0:
            points = (
                past_games["total_points"].values[0]
                - past_games["total_points"].values[-1]
            )  # amount of points earned
            points_l5 = (
                last5["total_points"].values[0] - last5["total_points"].values[-1]
            )
        else:
            points = 0
            points_l5 = 0

        gp = past_games["team_points"].mean()
        gp_l5 = last5["team_points"].mean()

        gp_rank = past_games["points_by_rank"].mean()
        gp_rank_l5 = last5["points_by_rank"].mean()

        stats_val.append(
            [
                goals,
                goals_l5,
                goals_suf,
                goals_suf_l5,
                rank,
                rank_l5,
                points,
                points_l5,
                gp,
                gp_l5,
                gp_rank,
                gp_rank_l5,
            ]
        )

    stats_cols = [
        "goals_mean",
        "goals_mean_l5",
        "goals_suf_mean",
        "goals_suf_mean_l5",
        "rank_mean",
        "rank_mean_l5",
        "points_mean",
        "points_mean_l5",
        "game_points_mean",
        "game_points_mean_l5",
        "game_points_rank_mean",
        "game_points_rank_mean_l5",
    ]

    stats_df = pd.DataFrame(stats_val, columns=stats_cols)

    full_df = pd.concat(
        [team_stats.reset_index(drop=True), stats_df], axis=1, ignore_index=False
    )

    home_team_stats = full_df.iloc[: int(full_df.shape[0] / 2), :]
    away_team_stats = full_df.iloc[int(full_df.shape[0] / 2) :, :]

    home_team_stats = home_team_stats[home_team_stats.columns[-12:]]
    away_team_stats = away_team_stats[away_team_stats.columns[-12:]]

    home_team_stats.columns = ["home_" + str(col) for col in home_team_stats.columns]
    away_team_stats.columns = ["away_" + str(col) for col in away_team_stats.columns]

    # In order to unify the database, is needed to add home and away suffix for each column.
    # After that, the data is ready to be merged.
    match_stats = pd.concat(
        [home_team_stats, away_team_stats.reset_index(drop=True)],
        axis=1,
        ignore_index=False,
    )

    full_df = pd.concat(
        [df, match_stats.reset_index(drop=True)], axis=1, ignore_index=False
    )

    # Drop friendly game
    full_df["is_friendly"] = full_df["tournament"].apply(lambda x: find_friendly(x))
    full_df = pd.get_dummies(full_df, columns=["is_friendly"])

    base_df = full_df[
        [
            "date",
            "home_team",
            "away_team",
            "rank_home",
            "rank_away",
            "home_score",
            "away_score",
            "result",
            "rank_dif",
            "rank_change_home",
            "rank_change_away",
            "home_goals_mean",
            "home_goals_mean_l5",
            "home_goals_suf_mean",
            "home_goals_suf_mean_l5",
            "home_rank_mean",
            "home_rank_mean_l5",
            "home_points_mean",
            "home_points_mean_l5",
            "away_goals_mean",
            "away_goals_mean_l5",
            "away_goals_suf_mean",
            "away_goals_suf_mean_l5",
            "away_rank_mean",
            "away_rank_mean_l5",
            "away_points_mean",
            "away_points_mean_l5",
            "home_game_points_mean",
            "home_game_points_mean_l5",
            "home_game_points_rank_mean",
            "home_game_points_rank_mean_l5",
            "away_game_points_mean",
            "away_game_points_mean_l5",
            "away_game_points_rank_mean",
            "away_game_points_rank_mean_l5",
            "is_friendly_0",
            "is_friendly_1",
        ]
    ]

    df = base_df.dropna()

    df["target"] = df["result"].apply(lambda x: no_draw(x))

    model_db = create_db(df)

    return df, model_db


def find_friendly(x):
    """
    Return whether the match is friendly match or not.
    :param x:
    :return:
    """
    if x == "Friendly":
        return 1
    else:
        return 0


def create_db(df):
    """

    :param df:
    :return:
    """
    columns = [
        "home_team",
        "away_team",
        "target",
        "rank_dif",
        "home_goals_mean",
        "home_rank_mean",
        "away_goals_mean",
        "away_rank_mean",
        "home_rank_mean_l5",
        "away_rank_mean_l5",
        "home_goals_suf_mean",
        "away_goals_suf_mean",
        "home_goals_mean_l5",
        "away_goals_mean_l5",
        "home_goals_suf_mean_l5",
        "away_goals_suf_mean_l5",
        "home_game_points_rank_mean",
        "home_game_points_rank_mean_l5",
        "away_game_points_rank_mean",
        "away_game_points_rank_mean_l5",
        "is_friendly_0",
        "is_friendly_1",
    ]

    base = df.loc[:, columns]
    base.loc[:, "goals_dif"] = base["home_goals_mean"] - base["away_goals_mean"]
    base.loc[:, "goals_dif_l5"] = (
        base["home_goals_mean_l5"] - base["away_goals_mean_l5"]
    )
    base.loc[:, "goals_suf_dif"] = (
        base["home_goals_suf_mean"] - base["away_goals_suf_mean"]
    )
    base.loc[:, "goals_suf_dif_l5"] = (
        base["home_goals_suf_mean_l5"] - base["away_goals_suf_mean_l5"]
    )
    base.loc[:, "goals_per_ranking_dif"] = (
        base["home_goals_mean"] / base["home_rank_mean"]
    ) - (base["away_goals_mean"] / base["away_rank_mean"])
    base.loc[:, "dif_rank_agst"] = base["home_rank_mean"] - base["away_rank_mean"]
    base.loc[:, "dif_rank_agst_l5"] = (
        base["home_rank_mean_l5"] - base["away_rank_mean_l5"]
    )
    base.loc[:, "dif_points_rank"] = (
        base["home_game_points_rank_mean"] - base["away_game_points_rank_mean"]
    )
    base.loc[:, "dif_points_rank_l5"] = (
        base["home_game_points_rank_mean_l5"] - base["away_game_points_rank_mean_l5"]
    )

    model_df = base[
        [
            "home_team",
            "away_team",
            "target",
            "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_points_rank",
            "dif_points_rank_l5",
            "is_friendly_0",
            "is_friendly_1",
        ]
    ]
    return model_df


def no_draw(x):
    """

    :param x:
    :return:
    """
    if x == 2:
        return 1
    else:
        return x