utils/ml_utils.py

import streamlit as st
import pandas as pd
import numpy as np
from sklearn.preprocessing import (
    StandardScaler, MinMaxScaler, RobustScaler, MaxAbsScaler,
    OneHotEncoder, OrdinalEncoder, LabelEncoder
)
from sklearn.compose import ColumnTransformer
from sklearn.pipeline import Pipeline
from sklearn.impute import SimpleImputer
from sklearn.model_selection import train_test_split
from sklearn.linear_model import LogisticRegression, LinearRegression, Ridge, Lasso
from sklearn.ensemble import RandomForestClassifier, RandomForestRegressor
from sklearn.neighbors import KNeighborsClassifier, KNeighborsRegressor
from sklearn.naive_bayes import GaussianNB
from sklearn.tree import DecisionTreeClassifier, DecisionTreeRegressor
from sklearn.svm import SVC, SVR
import joblib
import logging
from typing import Dict, Tuple, Callable, Any

from .data_utils import (
    get_numeric_columns,
    get_categorical_columns,
    get_datetime_columns,
)
import warnings

warnings.filterwarnings("ignore")

logger = logging.getLogger(__name__)

# -------- Optional dependency availability flags --------
try:
    import shap  # noqa: F401
    SHAP_AVAILABLE = True
except Exception:
    SHAP_AVAILABLE = False

try:
    from xgboost import XGBClassifier, XGBRegressor  # noqa: F401
    XGB_AVAILABLE = True
except Exception:
    XGB_AVAILABLE = False

try:
    from lightgbm import LGBMClassifier, LGBMRegressor  # noqa: F401
    LGBM_AVAILABLE = True
except Exception:
    LGBM_AVAILABLE = False

try:
    from catboost import CatBoostClassifier, CatBoostRegressor  # noqa: F401
    CATBOOST_AVAILABLE = True
except Exception:
    CATBOOST_AVAILABLE = False

try:
    from imblearn.over_sampling import SMOTE  # noqa: F401
    IMBLEARN_AVAILABLE = True
except Exception:
    IMBLEARN_AVAILABLE = False

try:
    import optuna  # noqa: F401
    OPTUNA_AVAILABLE = True
except Exception:
    OPTUNA_AVAILABLE = False

# ml_utils.py — ML utilities migrated to services/ml_service.py
# This file is kept only for backward compatibility with existing imports.
# Import from services.ml_service instead.

from services.ml_service import *  # noqa: F401, F403

def model_key_to_estimator(model_key: str, problem_type: str):
    """Convert model key string to estimator instance."""
    models = {
        "RandomForest": RandomForestClassifier(random_state=42),
        "LogisticRegression": LogisticRegression(random_state=42, max_iter=1000),
        "DecisionTree": DecisionTreeClassifier(random_state=42),
        "KNN": KNeighborsClassifier(),
        "SVM": SVC(random_state=42),
        "NaiveBayes": GaussianNB(),
        "RandomForestReg": RandomForestRegressor(random_state=42),
        "LinearRegression": LinearRegression(),
        "Ridge": Ridge(random_state=42),
        "Lasso": Lasso(random_state=42),
        "DecisionTreeReg": DecisionTreeRegressor(random_state=42),
        "KNNReg": KNeighborsRegressor(),
        "SVR": SVR(),
    }

    if model_key in models:
        return models[model_key]

    # Optional boosters
    if model_key == "XGBoost":
        if not XGB_AVAILABLE:
            raise ValueError("XGBoost not available. Install xgboost.")
        return XGBClassifier(random_state=42, eval_metric="logloss")
    if model_key == "XGBoostReg":
        if not XGB_AVAILABLE:
            raise ValueError("XGBoost not available. Install xgboost.")
        return XGBRegressor(random_state=42)

    if model_key == "LightGBM":
        if not LGBM_AVAILABLE:
            raise ValueError("LightGBM not available. Install lightgbm.")
        return LGBMClassifier(random_state=42)
    if model_key == "LightGBMReg":
        if not LGBM_AVAILABLE:
            raise ValueError("LightGBM not available. Install lightgbm.")
        return LGBMRegressor(random_state=42)

    if model_key == "CatBoost":
        if not CATBOOST_AVAILABLE:
            raise ValueError("CatBoost not available. Install catboost.")
        return CatBoostClassifier(random_state=42, verbose=False)
    if model_key == "CatBoostReg":
        if not CATBOOST_AVAILABLE:
            raise ValueError("CatBoost not available. Install catboost.")
        return CatBoostRegressor(random_state=42, verbose=False)

    raise ValueError(f"Unknown model key: {model_key}")

def build_preprocessor(
    df: pd.DataFrame,
    target_col: str,
    encoding_strategy: str = "onehot",
    scaling_method: str = "standard",
    poly_degree: int = 1,
    include_interactions: bool = False
) -> Tuple[ColumnTransformer, list, list, Callable]:
    """
    Build a preprocessor for ML pipelines.
    
    Args:
        df: Input dataframe
        target_col: Name of target column to exclude from features
        encoding_strategy: How to encode categorical variables ('onehot', 'ordinal', 'target_encoding')
        scaling_method: How to scale numerical variables ('standard', 'minmax', 'robust', 'maxabs', 'none')
        poly_degree: Degree of polynomial features to generate
        include_interactions: Whether to include interaction features
    
    Returns:
        Tuple of (preprocessor, numeric_cols, categorical_cols, feature_names_getter)
    """
    # Identify feature columns
    feature_cols = [c for c in df.columns if c != target_col]
    X = df[feature_cols]
    
    numeric_cols = get_numeric_columns(X)
    cat_cols = get_categorical_columns(X)
    
    # Handle case where both are empty
    if not numeric_cols and not cat_cols:
        # If no numeric or categorical columns, treat everything as numeric (for cases like datetime)
        numeric_cols = feature_cols
        cat_cols = []
    
    # Preprocessing transformers
    transformers = []
    
    if numeric_cols:
        # Build numeric pipeline
        numeric_steps = []
        
        # Imputation
        numeric_steps.append(('imputer', SimpleImputer(strategy='median')))
        
        # Scaling
        if scaling_method == "standard":
            numeric_steps.append(('scaler', StandardScaler()))
        elif scaling_method == "minmax":
            numeric_steps.append(('scaler', MinMaxScaler()))
        elif scaling_method == "robust":
            numeric_steps.append(('scaler', RobustScaler()))
        elif scaling_method == "maxabs":
            numeric_steps.append(('scaler', MaxAbsScaler()))
        
        # Polynomial features
        if poly_degree > 1:
            from sklearn.preprocessing import PolynomialFeatures
            numeric_steps.append(('poly', PolynomialFeatures(degree=poly_degree, include_bias=False)))
        
        # Interaction features
        if include_interactions and len(numeric_cols) > 1:
            from sklearn.preprocessing import PolynomialFeatures
            numeric_steps.append(('interactions', PolynomialFeatures(
                degree=2, 
                interaction_only=True, 
                include_bias=False
            )))
        
        transformers.append(('num', Pipeline(numeric_steps), numeric_cols))
    
    if cat_cols:
        # Build categorical pipeline
        cat_steps = []
        
        # Imputation (fill with mode)
        cat_steps.append(('imputer', SimpleImputer(strategy='constant', fill_value='missing')))
        
        # Encoding
        if encoding_strategy == "onehot":
            cat_steps.append(('encoder', OneHotEncoder(handle_unknown='ignore', sparse_output=False)))
        elif encoding_strategy == "ordinal":
            cat_steps.append(('encoder', OrdinalEncoder(handle_unknown='use_encoded_value', unknown_value=-1)))
        # Note: Target encoding is handled separately
        
        transformers.append(('cat', Pipeline(cat_steps), cat_cols))
    
    # Create the preprocessor
    preprocessor = ColumnTransformer(
        transformers=transformers,
        remainder='drop'  # Drop columns not specified in transformers
    )
    
    # Helper function to get feature names after transformation
    def get_feature_names(preprocessor):
        """Get feature names after preprocessing."""
        feature_names = []
        
        for name, transformer, columns in preprocessor.transformers_:
            if name == 'num':
                # For numeric features, names stay the same initially
                # But if polynomial features are used, they get transformed
                if hasattr(transformer.named_steps.get('poly'), 'get_feature_names_out'):
                    poly_features = transformer.named_steps['poly'].get_feature_names_out(columns)
                    feature_names.extend(poly_features)
                elif hasattr(transformer.named_steps.get('interactions'), 'get_feature_names_out'):
                    interaction_features = transformer.named_steps['interactions'].get_feature_names_out(columns)
                    feature_names.extend(interaction_features)
                else:
                    feature_names.extend(columns)
            elif name == 'cat':
                encoder = transformer.named_steps.get('encoder')
                if encoder and hasattr(encoder, 'get_feature_names_out'):
                    encoded_features = encoder.get_feature_names_out(columns)
                    feature_names.extend(encoded_features)
                else:
                    feature_names.extend(columns)
        
        return feature_names
    
    return preprocessor, numeric_cols, cat_cols, get_feature_names

def target_encode_column(series: pd.Series, target: pd.Series, smoothing: float = 1.0):
    """
    Perform target encoding on a categorical series with smoothing.
    
    Args:
        series: Categorical series to encode
        target: Target series for encoding
        smoothing: Smoothing factor (higher values = more smoothing toward global mean)
    
    Returns:
        Encoded series
    """
    # Calculate global mean
    global_mean = target.mean()
    
    # Calculate category means and counts
    category_stats = pd.DataFrame({
        'mean': target.groupby(series).mean(),
        'count': target.groupby(series).count()
    })
    
    # Apply smoothing: weighted average of category mean and global mean
    def smoothed_encoding(row):
        if row.name in category_stats.index:
            cat_mean = category_stats.loc[row.name, 'mean']
            cat_count = category_stats.loc[row.name, 'count']
            # Weighted average: more weight to category mean as count increases
            weight = cat_count / (cat_count + smoothing)
            return weight * cat_mean + (1 - weight) * global_mean
        else:
            # For unseen categories, use global mean
            return global_mean
    
    return series.map(smoothed_encoding)

def generate_code_snippet(pipeline, X, y, problem_type: str, feature_names: list, target_name: str):
    """
    Generate a Python code snippet that reproduces the trained pipeline.
    
    Args:
        pipeline: Trained sklearn pipeline
        X: Feature data used for training
        y: Target data used for training
        problem_type: 'classification' or 'regression'
        feature_names: Names of features
        target_name: Name of target column
    
    Returns:
        String containing Python code
    """
    # This is a simplified version - in a real implementation, we would extract
    # the actual steps and parameters from the pipeline
    code = f"""
# Generated code to reproduce the model
import pandas as pd
import numpy as np
from sklearn.model_selection import train_test_split
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import StandardScaler, OneHotEncoder
from sklearn.compose import ColumnTransformer
from sklearn.ensemble import RandomForest{'' if problem_type == 'classification' else 'Reg'}ator

# Load your data
# df = pd.read_csv('your_data.csv')

# Define features and target
X = df[{feature_names}]  # Your feature columns
y = df['{target_name}']  # Your target column

# Split the data
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)

# Define the model (you'll need to reconstruct the actual pipeline)
model = RandomForest{'' if problem_type == 'classification' else 'Reg'}ator(random_state=42)
model.fit(X_train, y_train)

# Make predictions
predictions = model.predict(X_test)

# Evaluate
from sklearn.metrics import {f"accuracy_score, classification_report" if problem_type == 'classification' else 'r2_score, mean_squared_error'}
"""
    if problem_type == 'classification':
        code += """
accuracy = accuracy_score(y_test, predictions)
print(f"Accuracy: {{accuracy}}")
"""
    else:
        code += """
r2 = r2_score(y_test, predictions)
print(f"R2 Score: {r2}")
"""
    
    return code