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)