test_script.py

import os
import pandas as pd
import pickle
import ast
import re
import numpy as np
from sklearn.metrics import mean_squared_error, r2_score, accuracy_score

DIRECTORY = os.path.dirname(os.path.abspath(__file__))
TEST_DATA_PATH = os.path.join(DIRECTORY, 'unseen_test_data')
MODELS_PATH = os.path.join(DIRECTORY, 'output', 'models')
PREPROCESSORS_PATH = os.path.join(DIRECTORY, 'output', 'preprocessors')

DEMOS_PATH = os.path.join(TEST_DATA_PATH, 'demos_test.csv')
DLCS_PATH = os.path.join(TEST_DATA_PATH, 'dlcs_test.csv')
INFO_BASE_PATH = os.path.join(TEST_DATA_PATH, 'info_base_games_test.csv')
REGRESSOR_GAMALYTIC_PATH = os.path.join(TEST_DATA_PATH, 'gamalytic_steam_games_reg.csv')
CLASSIFIER_GAMALYTIC_PATH = os.path.join(TEST_DATA_PATH, 'gamalytic_cls_reviews.csv')

def load_data(milestone):
    if milestone == 1:
        info_base = pd.read_csv(INFO_BASE_PATH)
        gamalytic = pd.read_csv(REGRESSOR_GAMALYTIC_PATH)
        dlcs = pd.read_csv(DLCS_PATH)
        demos = pd.read_csv(DEMOS_PATH)
    elif milestone == 2:
        info_base = pd.read_csv(INFO_BASE_PATH)
        gamalytic = pd.read_csv(CLASSIFIER_GAMALYTIC_PATH)
        dlcs = pd.read_csv(DLCS_PATH)
        demos = pd.read_csv(DEMOS_PATH)
    else:
        raise ValueError(f"Invalid milestone: {milestone}")
    return info_base, gamalytic, dlcs, demos

def load_models(milestone):
    if milestone == 1:
        en = pickle.load(open(os.path.join(MODELS_PATH, 'regressors', 'ensemble_model.pkl'), 'rb'))
        lgbm = pickle.load(open(os.path.join(MODELS_PATH, 'regressors', 'lgbm_model.pkl'), 'rb'))
        rf = pickle.load(open(os.path.join(MODELS_PATH, 'regressors', 'random_forest_model.pkl'), 'rb'))
        xgb = pickle.load(open(os.path.join(MODELS_PATH, 'regressors', 'xgboost_model.pkl'), 'rb'))
    elif milestone == 2:
        en = pickle.load(open(os.path.join(MODELS_PATH, 'classifiers', 'ensemble_classifier_model.pkl'), 'rb'))
        lgbm = pickle.load(open(os.path.join(MODELS_PATH, 'classifiers', 'lgbm_classifier_model.pkl'), 'rb'))
        rf = pickle.load(open(os.path.join(MODELS_PATH, 'classifiers', 'random_forest_classifier_model.pkl'), 'rb'))
        xgb = pickle.load(open(os.path.join(MODELS_PATH, 'classifiers', 'xgboost_classifier_model.pkl'), 'rb'))
    else:
        raise ValueError(f"Invalid milestone: {milestone}")
    return en, lgbm, rf, xgb

def load_preprocessors(milestone):
    if milestone == 1:
        le_publisher = pickle.load(open(os.path.join(PREPROCESSORS_PATH, 'le_publisher_milestone1.pkl'), 'rb'))
        mlb_genres = pickle.load(open(os.path.join(PREPROCESSORS_PATH, 'mlb_genres_milestone1.pkl'), 'rb'))
        mlb_platforms = pickle.load(open(os.path.join(PREPROCESSORS_PATH, 'mlb_platforms_milestone1.pkl'), 'rb'))
        return le_publisher, mlb_genres, mlb_platforms
    elif milestone == 2:
        le_publisher = pickle.load(open(os.path.join(PREPROCESSORS_PATH, 'le_publisher_milestone2.pkl'), 'rb'))
        le_target = pickle.load(open(os.path.join(PREPROCESSORS_PATH, 'le_target_milestone2.pkl'), 'rb'))
        mlb_genres = pickle.load(open(os.path.join(PREPROCESSORS_PATH, 'mlb_genres_milestone2.pkl'), 'rb'))
        mlb_platforms = pickle.load(open(os.path.join(PREPROCESSORS_PATH, 'mlb_platforms_milestone2.pkl'), 'rb'))
        scaler = pickle.load(open(os.path.join(PREPROCESSORS_PATH, 'standard_scaler_milestone2.pkl'), 'rb'))
        selector = pickle.load(open(os.path.join(PREPROCESSORS_PATH, 'select_k_best_milestone2.pkl'), 'rb'))
        return le_publisher, le_target, mlb_genres, mlb_platforms, scaler, selector
    else:
        raise ValueError(f"Invalid milestone: {milestone}")

def clean_info_base(info_base):
    """Clean the info_base dataframe by removing duplicates."""
    info_base = info_base.rename(columns={"steam_appid": "appid"})
    print(f"Initial unique appids: {info_base['appid'].nunique()}")
    print(f"Duplicate appids: {info_base['appid'].duplicated().sum()}")

    index = info_base[info_base["appid"].duplicated()].index
    print(f"Duplicate indices: {index}")
    print(info_base.iloc[index])

    info_base.drop_duplicates(subset=['appid'], inplace=True)
    print(f"Remaining duplicates: {info_base['appid'].duplicated().sum()}")
    return info_base

def prepare_demos_data(demos):
    """Prepare the demos dataframe by cleaning and validating."""
    demos_df = demos.drop(columns=demos.columns[0])

    if 'full_game_appid' in demos_df.columns:
        print(f"Unique full_game_appids: {demos_df['full_game_appid'].nunique()}")
    else:
        print("Column 'full_game_appid' not found in DataFrame")
        print("Available columns:", demos_df.columns.tolist())
    return demos_df

def merge_datasets(info_base, gamalytic, dlcs, demos_df):
    """Merge all datasets into a single dataframe."""
    df = info_base.merge(gamalytic, left_on='appid', right_on='steamId', how='inner')

    # Add DLC flag
    dlc_flag = dlcs[['base_appid']].drop_duplicates()
    dlc_flag['has_dlc'] = 1
    df = df.merge(dlc_flag, left_on='appid', right_on='base_appid', how='left')
    df['has_dlc'] = df['has_dlc'].fillna(0).astype(int)

    # Add demo flag
    demo_flag = demos_df[['full_game_appid']].drop_duplicates()
    demo_flag['has_demo'] = 1
    df = df.merge(demo_flag, left_on='appid', right_on='full_game_appid', how='left')
    df['has_demo'] = df['has_demo'].fillna(0).astype(int)

    return df

def clean_merged_data(df):
    """Clean the merged dataframe by removing unnecessary columns and handling missing values."""
    df.drop(columns=['steamId', 'base_appid', 'full_game_appid'], inplace=True)

    # Convert numeric columns
    df['steam_achievements'] = df['steam_achievements'].astype(int)
    df['steam_trading_cards'] = df['steam_trading_cards'].astype(int)
    df['workshop_support'] = df['workshop_support'].astype(int)

    # Analyze the percentage of missing values in each column
    null_counts = df.isnull().sum()
    print("\nPercentage of missing values in each column:")
    print(null_counts[null_counts > 0] / len(df))

    # Analyze percentage of games with no demo or dlc
    print("\nPercentage of games with no demo or dlc:")
    print(df['has_demo'].value_counts() / len(df))
    print(df['has_dlc'].value_counts() / len(df))

    # Analyze percentage of unique values in each column
    print("\nPercentage of unique values in each column:")
    print(df.nunique() / len(df))

    # Remove unnecessary columns
    df.drop(columns=['aiContent', 'metacritic', 'achievements_total', 'has_dlc', 'has_demo', 'name', 'appid'], inplace=True)

    # Handle missing genres
    df['genres'] = df['genres'].fillna("Unknown")

    # Add free game flag
    df['game_is_free'] = (df['price'] == 0).astype(int)

    return df

def encode_features_regresser(df, mlb_platforms, mlb_genres, le_publisher, milestone):
    """Encode categorical features using appropriate encoders."""
    # Encode platforms
    df['platforms_split'] = df['supported_platforms'].fillna('').apply(lambda x: ast.literal_eval(x))
    platforms_encoded = pd.DataFrame(mlb_platforms.transform(df['platforms_split']),
                                   columns=[f"Platform_{p}" for p in mlb_platforms.classes_],
                                   index=df.index)
    df = pd.concat([df.drop(columns=['supported_platforms', 'platforms_split']), platforms_encoded], axis=1)

    # Encode genres
    df['genres_split'] = df['genres'].fillna('').apply(lambda x: [genre.strip() for genre in x.split(',')])
    genres_encoded = pd.DataFrame(mlb_genres.transform(df['genres_split']),
                                columns=[f"Genre_{g}" for g in mlb_genres.classes_],
                                index=df.index)
    df = pd.concat([df.drop(columns=['genres', 'genres_split']), genres_encoded], axis=1)

    
    df['publisherClass'] = le_publisher.transform(df['publisherClass'])
    return df

def encode_features_classifier(df, mlb_platforms, mlb_genres, le_publisher, le_target, milestone):
    """Encode categorical features using appropriate encoders."""
    # Encode platforms
    df['platforms_split'] = df['supported_platforms'].fillna('').apply(lambda x: ast.literal_eval(x))
    platforms_encoded = pd.DataFrame(mlb_platforms.transform(df['platforms_split']),
                                   columns=[f"Platform_{p}" for p in mlb_platforms.classes_],
                                   index=df.index)
    df = pd.concat([df.drop(columns=['supported_platforms', 'platforms_split']), platforms_encoded], axis=1)

    # Encode genres
    df['genres_split'] = df['genres'].fillna('').apply(lambda x: [genre.strip() for genre in x.split(',')])
    genres_encoded = pd.DataFrame(mlb_genres.transform(df['genres_split']),
                                columns=[f"Genre_{g}" for g in mlb_genres.classes_],
                                index=df.index)
    df = pd.concat([df.drop(columns=['genres', 'genres_split']), genres_encoded], axis=1)

    
    df['publisherClass'] = le_publisher.transform(df['publisherClass'])

    df['reviewScore'] = le_target.transform(df['reviewScore'])
    

    return df

def try_parse_date(x):
    for fmt in ("%b %d, %Y", "%b-%y", "%Y"):
        try:
            return pd.to_datetime(x, format=fmt)
        except (ValueError, TypeError):
            continue
    match = re.match(r"(Q[1-4])\s*(\d{4})", x) or re.match(r"(\d{4})\s*(Q[1-4])", x)
    if match:
        parts = match.groups()
        q, y = (parts[0], parts[1]) if 'Q' in parts[0] else (parts[1], parts[0])
        month = {'Q1': 1, 'Q2': 4, 'Q3': 7, 'Q4': 10}[q]
        return pd.Timestamp(year=int(y), month=month, day=1)
    else:
        timestamp_dict = {
            "To be announced": pd.Timestamp(year=2025, month=12, day=31),
            "Coming soon": pd.Timestamp(year=2025, month=12, day=31),
        }
        return timestamp_dict.get(x, pd.NaT)

def process_dates(df):
    """Process and extract features from release dates."""
    df['release_date'] = df['release_date'].apply(try_parse_date)
    df['release_date_weekday'] = df['release_date'].dt.strftime('%w').astype(int)
    df['release_date_month'] = df['release_date'].dt.month.astype(int)
    df['release_date_year'] = df['release_date'].dt.year.astype(int)
    df.drop(columns=['release_date'], inplace=True)
    return df

def handle_outliers(df, columns, method='winsorize', threshold=0.05):
    """Handle outliers without removing them.
    
    Methods:
        - winsorize: Caps outliers at specified percentiles
        - log: Applies log transformation to reduce impact of outliers
        - robust_scale: Uses robust scaling that is less sensitive to outliers
    """
    df_processed = df.copy()
    
    for col in columns:
        if df_processed[col].dtype in [np.float64, np.int64]:
            if method == 'winsorize':
                # Cap values at percentiles
                lower_limit = df_processed[col].quantile(threshold)
                upper_limit = df_processed[col].quantile(1 - threshold)
                df_processed.loc[df_processed[col] < lower_limit, col] = lower_limit
                df_processed.loc[df_processed[col] > upper_limit, col] = upper_limit
                print(f"Winsorized column {col} at {threshold} and {1-threshold} percentiles")
            
            elif method == 'log':
                # Apply log transformation (adding 1 to handle zeros)
                if (df_processed[col] <= 0).any():
                    min_val = abs(df_processed[col].min()) + 1 if df_processed[col].min() < 0 else 0
                    df_processed[col] = np.log1p(df_processed[col] + min_val)
                else:
                    df_processed[col] = np.log1p(df_processed[col])
                print(f"Applied log transformation to column {col}")
    
    return df_processed

def predict(X_test, y_test, milestone):
    if milestone == 1:
        en, lgbm, rf, xgb = load_models(milestone)
    elif milestone == 2:
        en, lgbm, rf, xgb = load_models(milestone)
    else:
        raise ValueError(f"Invalid milestone: {milestone}")
    model_type = "Regressor" if milestone == 1 else "Classifier"
    models = {
            f"Random Forest {model_type}": rf,
            f"XGBoost {model_type}": xgb,
            f"LGBM {model_type}": lgbm,
            f"Ensemble {model_type}": en
        }
    if milestone == 1:
        results = []
        for name, model in models.items():
            test_pred = model.predict(X_test)
            test_rmse = np.sqrt(mean_squared_error(np.expm1(y_test), np.expm1(test_pred)))
            test_r2 = r2_score(y_test, test_pred)

            results.append({
                'Model': name,
                'Test RMSE': f"{test_rmse:,.2f}",
                'Test R2': f"{test_r2:.4f}"
            })

        return pd.DataFrame(results)
    elif milestone == 2:
        results = []
        for name, model in models.items():
            test_pred = model.predict(X_test)
            test_accuracy = accuracy_score(y_test, test_pred)
            results.append({
                'Model': name,
                'Test Accuracy': f"{test_accuracy:.4f}"
            })

        return pd.DataFrame(results)

def safe_target_transform(series):
    """Transform the target variable safely by capping outliers and applying log transformation."""
    cap = series.quantile(0.999)
    clipped = series.clip(upper=cap)
    return np.log1p(clipped)


def main():
    milestone = int(input("Enter the milestone number (1 or 2): "))
    if milestone == 2:
        le_publisher, le_target, mlb_genres, mlb_platforms, scaler, selector = load_preprocessors(milestone)
    elif milestone == 1:
        le_publisher, mlb_genres, mlb_platforms = load_preprocessors(milestone)
    else:
        raise ValueError(f"Invalid milestone: {milestone}")
    info_base, gamalytic, dlcs, demos = load_data(milestone)
    info_base = clean_info_base(info_base)
    demos_df = prepare_demos_data(demos)
    df = merge_datasets(info_base, gamalytic, dlcs, demos_df)
    df = clean_merged_data(df)
    if milestone == 2:
        df = encode_features_classifier(df, mlb_platforms, mlb_genres, le_publisher, le_target, milestone)
    else:
        df = encode_features_regresser(df, mlb_platforms, mlb_genres, le_publisher, milestone)
    df = process_dates(df)

    numerical_cols = df.select_dtypes(include=[np.float64, np.int64]).columns.tolist()
    numerical_cols = [col for col in numerical_cols if col != 'reviewScore']
    df = handle_outliers(df, numerical_cols, method='winsorize', threshold=0.05)

    if milestone == 1:
        df = df.rename(columns = lambda x:re.sub('[^A-Za-z0-9_]+', '', x))

    X = pd.get_dummies(df.drop(columns=['reviewScore']), drop_first=True)
    y = df['reviewScore']
    
    if milestone == 1:
        y = safe_target_transform(y)

    X_test = X

    if milestone == 2:
        X_test = scaler.transform(X)
        X_test = pd.DataFrame(X_test, columns=X.columns)
        X_test = selector.transform(X_test)
        X_test = pd.DataFrame(X_test, columns=X.columns[selector.get_support()])
    
    results = predict(X_test, y, milestone)
    print(results)

if __name__ == "__main__":
    while True:
        try:
            main()
            break
        except Exception as e:
            print(f"Error: {e}")
            print("Please try again.")