utils.py

import pandas as pd
import numpy as np
from sklearn.impute import SimpleImputer
from sklearn.preprocessing import LabelEncoder, StandardScaler, PowerTransformer, KBinsDiscretizer, OneHotEncoder, OrdinalEncoder
from sklearn.model_selection import train_test_split
from sklearn.neighbors import KNeighborsClassifier
from sklearn.naive_bayes import GaussianNB
from sklearn.tree import DecisionTreeClassifier
from sklearn.linear_model import LinearRegression
from sklearn.ensemble import RandomForestRegressor
from sklearn.metrics import (
    accuracy_score, precision_score, recall_score, f1_score, roc_auc_score,
    classification_report, roc_curve, precision_recall_curve,
    mean_squared_error, mean_absolute_error, r2_score
)
import matplotlib.pyplot as plt
import seaborn as sns
import streamlit as st
import io
from fpdf import FPDF
import base64

# ---------------------- Data Loading ----------------------

def load_data(uploaded_file):
    if uploaded_file is not None:
        try:
            if uploaded_file.name.endswith('.csv'):
                df = pd.read_csv(uploaded_file)
            elif uploaded_file.name.endswith('.xlsx'):
                df = pd.read_excel(uploaded_file)
            else:
                return None, "Unsupported file format. Please upload a CSV or Excel file."
            return df, None
        except Exception as e:
            return None, f"Error loading file: {str(e)}"
    else:
        return None, "No file uploaded."

# ---------------------- Problem Type Detection ----------------------

def detect_problem_type(df):
    target = df.columns[-1]
    if df[target].dtype == 'object' or len(df[target].unique()) < 10:
        return 'classification'
    return 'regression'

# ---------------------- Encoding Options ----------------------

def encode_categorical(df, strategy="label"):
    encoded_cols = []
    if strategy == "label":
        le = LabelEncoder()
        for col in df.columns:
            if df[col].dtype == 'object' or df[col].dtype.name == 'category':
                try:
                    df[col] = le.fit_transform(df[col])
                    encoded_cols.append(col)
                except:
                    pass
    elif strategy == "onehot":
        df = pd.get_dummies(df)
        encoded_cols = "All categorical columns (One-Hot)"
    elif strategy == "ordinal":
        ordinal = OrdinalEncoder()
        obj_cols = df.select_dtypes(include=['object', 'category']).columns
        df[obj_cols] = ordinal.fit_transform(df[obj_cols])
        encoded_cols = list(obj_cols)
    return df, encoded_cols

# ---------------------- Preprocessing ----------------------

def handle_missing_values(df):
    missing_info = df.isnull().sum()
    imputer = SimpleImputer(strategy='most_frequent')
    df_imputed = pd.DataFrame(imputer.fit_transform(df), columns=df.columns)
    return df_imputed, missing_info

def remove_duplicates(df):
    duplicates = df.duplicated().sum()
    df_cleaned = df.drop_duplicates()
    return df_cleaned, duplicates

def scale_data(df):
    scaler = StandardScaler()
    return pd.DataFrame(scaler.fit_transform(df), columns=df.columns)

def transform_features(df):
    pt = PowerTransformer()
    return pd.DataFrame(pt.fit_transform(df), columns=df.columns)

def bin_features(df, n_bins=5):
    binner = KBinsDiscretizer(n_bins=n_bins, encode='ordinal', strategy='quantile')
    return pd.DataFrame(binner.fit_transform(df), columns=df.columns)

def preprocess_data(df, explain_mode=False, problem_type='auto', encoding_strategy="label"):
    steps_info = []

    df, missing_info = handle_missing_values(df)
    steps_info.append({"Step": "Missing Values", "Details": missing_info.to_dict()})

    df, duplicates = remove_duplicates(df)
    steps_info.append({"Step": "Duplicates Removed", "Count": duplicates})

    df, encoded_cols = encode_categorical(df, strategy=encoding_strategy)
    steps_info.append({"Step": f"Encoding ({encoding_strategy})", "Columns": encoded_cols})

    # Outlier removal
    outliers_removed = 0
    numeric_cols = df.select_dtypes(include=[np.number]).columns
    for col in numeric_cols:
        Q1 = df[col].quantile(0.25)
        Q3 = df[col].quantile(0.75)
        IQR = Q3 - Q1
        mask = ~((df[col] < (Q1 - 1.5 * IQR)) | (df[col] > (Q3 + 1.5 * IQR)))
        outliers_removed += (~mask).sum()
        df = df[mask]
    steps_info.append({"Step": "Outlier Removal (IQR)", "Removed": int(outliers_removed)})

    df = scale_data(df)
    steps_info.append({"Step": "Feature Scaling", "Method": "StandardScaler"})

    df = transform_features(df)
    steps_info.append({"Step": "Feature Transformation", "Method": "PowerTransformer"})

    if problem_type == 'classification':
        df = bin_features(df)
        steps_info.append({"Step": "Binning", "Strategy": "Quantile", "Bins": 5})

    steps_info.append({"Step": "Feature Engineering", "Details": "(Placeholder)"})
    steps_info.append({"Step": "Feature Extraction", "Details": "(Not applied – consider PCA, SVD)"})
    steps_info.append({"Step": "Noise Handling", "Details": "(Manual detection recommended)"})

    return df, pd.DataFrame(steps_info)

# ---------------------- Model Training ----------------------

def train_models(X, y):
    X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.3, random_state=42)
    models = {
        "KNN": KNeighborsClassifier(),
        "Naive Bayes": GaussianNB(),
        "Decision Tree": DecisionTreeClassifier(random_state=42)
    }
    results = {}
    for name, model in models.items():
        model.fit(X_train, y_train)
        y_pred = model.predict(X_test)
        y_proba = model.predict_proba(X_test)[:, 1] if hasattr(model, "predict_proba") and len(np.unique(y)) == 2 else None
        metrics = {
            "Accuracy": accuracy_score(y_test, y_pred),
            "Precision": precision_score(y_test, y_pred, average='weighted', zero_division=0),
            "Recall": recall_score(y_test, y_pred, average='weighted', zero_division=0),
            "F1 Score": f1_score(y_test, y_pred, average='weighted', zero_division=0),
            "AUC": roc_auc_score(y_test, y_proba) if y_proba is not None else None,
            "Classification Report": classification_report(y_test, y_pred, output_dict=True)
        }
        results[name] = {"model": model, "metrics": metrics, "y_test": y_test, "y_pred": y_pred, "y_proba": y_proba}
    return results

def train_regressors(X, y):
    X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.3, random_state=42)
    models = {
        "Linear Regression": LinearRegression(),
        "Random Forest Regressor": RandomForestRegressor(random_state=42)
    }
    results = {}
    for name, model in models.items():
        model.fit(X_train, y_train)
        y_pred = model.predict(X_test)
        metrics = {
            "RMSE": np.sqrt(mean_squared_error(y_test, y_pred)),
            "MAE": mean_absolute_error(y_test, y_pred),
            "R2 Score": r2_score(y_test, y_pred)
            }

        results[name] = {"model": model, "metrics": metrics, "y_test": y_test, "y_pred": y_pred}
    return results

# ---------------------- Explainable AI ----------------------

def model_supports_feature_importance(model):
    return hasattr(model, 'feature_importances_') or hasattr(model, 'coef_')

def explain_model(model, df, problem_type):
    st.subheader("📊 Explainable AI - Feature Importance")
    if model_supports_feature_importance(model):
        if hasattr(model, 'feature_importances_'):
            importances = model.feature_importances_
        elif hasattr(model, 'coef_'):
            importances = model.coef_[0]
        else:
            st.warning("⚠️ Model does not support extractable feature importance.")
            return
        features = df.drop(columns=[df.columns[-1]]).columns
        importance_df = pd.DataFrame({"Feature": features, "Importance": importances})
        fig = plt.figure(figsize=(10, 6))
        sns.barplot(x="Importance", y="Feature", data=importance_df.sort_values("Importance", ascending=False))
        plt.title("Feature Importances")
        st.pyplot(fig)
    else:
        st.warning("⚠️ This model does not support feature importance explanation.")

# ---------------------- Visualization ----------------------

def visualize_results(results, stage="model_eval", problem_type=None):
    if stage == "model_eval":
        st.subheader("📈 Model Performance Comparison")
        metrics_df = pd.DataFrame({model: info["metrics"] for model, info in results.items()}).T
        st.dataframe(metrics_df.round(3))

        fig, ax = plt.subplots(figsize=(10, 5))
        if problem_type == "regression":
            metrics_df[["RMSE", "MAE", "R2 Score"]].plot(kind='bar', ax=ax)
        else:
            metrics_df[["Accuracy", "Precision", "Recall", "F1 Score"]].plot(kind='bar', ax=ax)
        plt.title("Model Performance Metrics")
        plt.xticks(rotation=0)
        st.pyplot(fig)