model_eval.py

import pandas as pd
import numpy as np
from sklearn.metrics import (
    accuracy_score, precision_score, recall_score, f1_score, roc_auc_score,
    precision_recall_curve, auc, confusion_matrix, log_loss,
    matthews_corrcoef, cohen_kappa_score
)
from sklearn.model_selection import cross_val_score, StratifiedKFold
from sklearn.linear_model import LogisticRegression
import joblib
import os
import time
from catboost import CatBoostClassifier, Pool
from sklearn.ensemble import HistGradientBoostingClassifier
import xgboost as xgb
from tabulate import tabulate
from sklearn.preprocessing import LabelEncoder, RobustScaler
import warnings
warnings.filterwarnings('ignore')

BASE_PATH = r'E:\ML\Projects\IITH\New folder\Combined_Training_Output'
CATBOOST_PATH = os.path.join(BASE_PATH, 'CatBoostClassifier', 'run_20250717_120744')
HGB_PATH = os.path.join(BASE_PATH, 'HGBClassifier', 'run_20250717_120804')
XGB_PATH = os.path.join(BASE_PATH, 'XGBClassifier', 'run_20250717_121549')
META_MODEL_PATH = os.path.join(BASE_PATH, 'logistic_regression_meta_model.pkl')
STACKED_DATA_PATH = os.path.join(BASE_PATH, 'stacked_train.csv')

CATBOOST_CATEGORICAL = ['income_category', 'education', 'education_level', 'marital_status',
                        'profession', 'house_ownership', 'device_type']

CATEGORICAL_COLUMNS = CATBOOST_CATEGORICAL

def load_test_data(target_column='risk_flag'):
    try:
        test_data = pd.read_csv(r'E:\ML\Projects\IITH\test_data.csv')
        
        if target_column not in test_data.columns:
            print(f"Error: Column '{target_column}' not found in test_data.csv")
            print("Available columns:", test_data.columns.tolist())
            if 'probability_of_default' in test_data.columns:
                print("Using binarized probability_of_default as target")
                y_test = (test_data['probability_of_default'] > 0.5).astype(int)
                test_data = test_data.drop(columns=['probability_of_default'])
            else:
                stacked_data = pd.read_csv(STACKED_DATA_PATH)
                if 'true_label' in stacked_data.columns:
                    print("Using true_label from stacked_train.csv as fallback")
                    y_test = stacked_data['true_label']
                else:
                    raise KeyError("true_label not found in stacked_train.csv")
        else:
            y_test = test_data[target_column]
            test_data = test_data.drop(columns=[target_column])

        X_test_catboost = preprocess_data_catboost(test_data)
        X_test_hgb = preprocess_data_hgb(test_data)
        X_test_xgb = preprocess_data_xgb(test_data)
        return X_test_catboost, X_test_hgb, X_test_xgb, y_test
    except FileNotFoundError:
        print("Error: test_data.csv not found at E:\ML\Projects\IITH")
        stacked_data = pd.read_csv(STACKED_DATA_PATH)
        if 'true_label' in stacked_data.columns:
            print("Using stacked_train.csv for both features and labels")
            X_test = stacked_data[['cat_pred', 'hgb_pred', 'xgb_pred']]
            y_test = stacked_data['true_label']
            return X_test, X_test, X_test, y_test
        else:
            raise KeyError("true_label not found in stacked_train.csv")

def preprocess_data_catboost(data):
    data = data.copy()
    for col in CATBOOST_CATEGORICAL:
        if col in data.columns:
            data[col] = data[col].astype(str)
    return data

def preprocess_data_hgb(data):
    data = data.copy()
    hgb_encoders = joblib.load(os.path.join(HGB_PATH, 'label_encoders.pkl'))
    for col in CATEGORICAL_COLUMNS:
        if col in data.columns:
            le = hgb_encoders[col]
            data[col] = data[col].astype(str).map(lambda s: s if s in le.classes_ else le.classes_[0])
            data[col] = le.transform(data[col])
    data.fillna(0, inplace=True)
    return data

def preprocess_data_xgb(data):
    data = data.copy()
    xgb_encoders = joblib.load(os.path.join(XGB_PATH, 'label_encoders.pkl'))
    xgb_scaler = joblib.load(os.path.join(XGB_PATH, 'scaler.pkl'))
    for col in CATEGORICAL_COLUMNS:
        if col in data.columns:
            le = xgb_encoders[col]
            data[col] = data[col].astype(str).map(lambda s: s if s in le.classes_ else le.classes_[0])
            data[col] = le.transform(data[col])
    data.fillna(data.median(), inplace=True)
    data_scaled = xgb_scaler.transform(data)
    return pd.DataFrame(data_scaled, columns=data.columns, index=data.index)

def load_models():
    try:
        catboost_model = CatBoostClassifier().load_model(os.path.join(CATBOOST_PATH, 'catboost_model.cbm'))
        hgb_model = joblib.load(os.path.join(HGB_PATH, 'hgb_model.pkl'))
        xgb_model = xgb.Booster()
        xgb_model.load_model(os.path.join(XGB_PATH, 'xgb_model.json'))
        meta_model = joblib.load(META_MODEL_PATH)
        return catboost_model, hgb_model, xgb_model, meta_model
    except Exception as e:
        print(f"Error loading models: {e}")
        raise

def generate_stacked_predictions(catboost_model, hgb_model, xgb_model, X_test_catboost, X_test_hgb, X_test_xgb):
    start_time = time.time()
    cat_features_indices = [X_test_catboost.columns.get_loc(col) for col in CATBOOST_CATEGORICAL if col in X_test_catboost.columns]
    cat_pred = catboost_model.predict_proba(Pool(X_test_catboost, cat_features=cat_features_indices))[:, 1]
    cat_time = time.time() - start_time

    start_time = time.time()
    hgb_pred = hgb_model.predict_proba(X_test_hgb)[:, 1]
    hgb_time = time.time() - start_time

    start_time = time.time()
    xgb_threshold = joblib.load(os.path.join(XGB_PATH, 'optimal_threshold.pkl'))
    dtest = xgb.DMatrix(X_test_xgb, feature_names=X_test_xgb.columns.tolist())
    xgb_pred = xgb_model.predict(dtest)
    xgb_time = time.time() - start_time

    stacked_predictions = pd.DataFrame({
        'cat_pred': cat_pred,
        'hgb_pred': hgb_pred,
        'xgb_pred': xgb_pred
    })
    return stacked_predictions, {'CatBoost': cat_time, 'HistGradientBoosting': hgb_time, 'XGBoost': xgb_time}

def evaluate_metrics(model, X, y_true, model_name, is_xgb=False, xgb_model=None, xgb_threshold=0.5, catboost_cat_features=None):
    start_time = time.time()
    if is_xgb:
        dtest = xgb.DMatrix(X, feature_names=X.columns.tolist())
        y_pred = (xgb_model.predict(dtest) > xgb_threshold).astype(int)
        y_pred_proba = xgb_model.predict(dtest)
    else:
        if model_name == "CatBoost":
            y_pred = model.predict(Pool(X, cat_features=catboost_cat_features))
            y_pred_proba = model.predict_proba(Pool(X, cat_features=catboost_cat_features))[:, 1]
        else:
            y_pred = model.predict(X)
            y_pred_proba = model.predict_proba(X)[:, 1]
    pred_time = time.time() - start_time

    cm = confusion_matrix(y_true, y_pred)
    tn, fp, fn, tp = cm.ravel()
    specificity = tn / (tn + fp) if (tn + fp) > 0 else 0
    precision, recall, _ = precision_recall_curve(y_true, y_pred_proba)
    auc_pr = auc(recall, precision)

    complexity = "N/A"
    if model_name == "CatBoost":
        complexity = f"Iterations: {model.get_param('iterations')}"
    elif model_name == "HistGradientBoosting":
        complexity = f"Iterations: {model.n_iter_}"
    elif model_name == "XGBoost":
        complexity = f"Iterations: {xgb_model.best_iteration}"
    elif model_name == "Meta-Model (Logistic Regression)":
        complexity = f"Coefficients: {len(model.coef_[0])}"

    metrics = {
        'Model': model_name,
        'Accuracy': f"{accuracy_score(y_true, y_pred):.4f}",
        'Precision': f"{precision_score(y_true, y_pred, average='weighted'):.4f}",
        'Recall': f"{recall_score(y_true, y_pred, average='weighted'):.4f}",
        'F1-Score': f"{f1_score(y_true, y_pred, average='weighted'):.4f}",
        'Specificity': f"{specificity:.4f}",
        'AUC-ROC': f"{roc_auc_score(y_true, y_pred_proba):.4f}",
        'AUC-PR': f"{auc_pr:.4f}",
        'Confusion Matrix': f"TN: {tn} FP: {fp}\nFN: {fn} TP: {tp}",
        'Log Loss': f"{log_loss(y_true, y_pred_proba):.4f}",
        'MCC': f"{matthews_corrcoef(y_true, y_pred):.4f}",
        'Cohen’s Kappa': f"{cohen_kappa_score(y_true, y_pred):.4f}",
        'Prediction Time (s)': f"{pred_time:.4f}",
        'Model Complexity': complexity
    }
    return metrics

def stability_checks(X, y, model, model_name, segments=None):
    cv = StratifiedKFold(n_splits=5, shuffle=True, random_state=42)
    cv_scores = cross_val_score(model, X, y, cv=cv, scoring='recall')
    
    stability_data = [
        {'Metric': 'Mean Recall', 'Value': f"{cv_scores.mean():.4f}"},
        {'Metric': 'Std Recall', 'Value': f"{cv_scores.std():.4f}"}
    ]
    
    segment_data = []
    if segments is not None:
        for segment_name, segment_indices in segments.items():
            X_seg = X.iloc[segment_indices]
            y_seg = y.iloc[segment_indices]
            recall = recall_score(y_seg, model.predict(X_seg))
            segment_data.append({
                'Segment': segment_name,
                'Recall': f"{recall:.4f}"
            })
    
    return stability_data, segment_data

def cost_based_evaluation(y_true, y_pred, fn_cost=10, fp_cost=1):
    cm = confusion_matrix(y_true, y_pred)
    tn, fp, fn, tp = cm.ravel()
    total_cost = fn * fn_cost + fp * fp_cost
    
    cost_data = [
        {'Metric': 'False Negatives (FN)', 'Value': f"{fn}, Cost: {fn * fn_cost}"},
        {'Metric': 'False Positives (FP)', 'Value': f"{fp}, Cost: {fp * fp_cost}"},
        {'Metric': 'Total Cost', 'Value': f"{total_cost}"}
    ]
    return cost_data

def main():
    try:
        X_test_catboost, X_test_hgb, X_test_xgb, y_test = load_test_data(target_column='risk_flag')
    except Exception as e:
        print(f"Failed to load test data: {e}")
        return
    
    try:
        catboost_model, hgb_model, xgb_model, meta_model = load_models()
    except Exception as e:
        print(f"Failed to load models: {e}")
        return
    
    metrics_data = []
    catboost_cat_features = [X_test_catboost.columns.get_loc(col) for col in CATBOOST_CATEGORICAL if col in X_test_catboost.columns]
    metrics_data.append(evaluate_metrics(catboost_model, X_test_catboost, y_test, "CatBoost", catboost_cat_features=catboost_cat_features))
    metrics_data.append(evaluate_metrics(hgb_model, X_test_hgb, y_test, "HistGradientBoosting"))
    xgb_threshold = joblib.load(os.path.join(XGB_PATH, 'optimal_threshold.pkl'))
    metrics_data.append(evaluate_metrics(None, X_test_xgb, y_test, "XGBoost", is_xgb=True, xgb_model=xgb_model, xgb_threshold=xgb_threshold))
    
    stacked_predictions, pred_times = generate_stacked_predictions(catboost_model, hgb_model, xgb_model, X_test_catboost, X_test_hgb, X_test_xgb)
    start_time = time.time()
    meta_metrics = evaluate_metrics(meta_model, stacked_predictions, y_test, "Meta-Model (Logistic Regression)")
    meta_metrics['Prediction Time (s)'] = f"{time.time() - start_time:.4f}"
    metrics_data.append(meta_metrics)
    
    print("\nModel Performance Metrics:")
    print(tabulate(metrics_data, headers='keys', tablefmt='psql', stralign='left', floatfmt='.4f'))

    segments = {
        'Segment 1': X_test_xgb.index[:len(X_test_xgb)//2],
        'Segment 2': X_test_xgb.index[len(X_test_xgb)//2:]
    }
    stability_data, segment_data = stability_checks(stacked_predictions, y_test, meta_model, "Meta-Model")
    
    print("\nMeta-Model Cross-Validation Stability:")
    print(tabulate(stability_data, headers='keys', tablefmt='psql', stralign='left'))
    
    if segment_data:
        print("\nMeta-Model Segment Analysis:")
        print(tabulate(segment_data, headers='keys', tablefmt='psql', stralign='left'))
    
    meta_pred = meta_model.predict(stacked_predictions)
    cost_data = cost_based_evaluation(y_test, meta_pred, fn_cost=10, fp_cost=1)
    
    print("\nMeta-Model Cost-Based Evaluation:")
    print(tabulate(cost_data, headers='keys', tablefmt='psql', stralign='left'))

def get_model_complexity(model, model_name):
    if model_name == "CatBoost":
        return f"Iterations: {model.get_param('iterations')}"
    elif model_name == "HistGradientBoosting":
        return f"Iterations: {model.n_iter_}"
    elif model_name == "XGBoost":
        return f"Iterations: {model.best_iteration}"
    elif model_name == "Meta-Model (Logistic Regression)":
        return f"Coefficients: {len(model.coef_[0])}"
    return "N/A"

if __name__ == "__main__":
    main()