""" Assignment 2 — End-to-end ML Pipeline Author: Mohamed Miqdad Date: 2025-08-10 How to use ---------- 1) Put your Kaggle CSV files in: ./datasets/ Examples: - diabetes.csv (target guess: Outcome) - ckd.csv (target guess: classification OR class) - heart.csv (target guess: target) - breast_cancer.csv (target guess: diagnosis OR target) 2) Create or edit datasets_config.json to map each CSV to its target column. A template exists at: datasets_config_example.json Example content: {{ "diabetes.csv": "Outcome", "heart.csv": "target", "ckd.csv": "classification", "breast_cancer.csv": "diagnosis" }} 3) Run this script (Python 3.9+). It will: - Handle missing values - Train baseline ML models with GridSearchCV (no feature selection) - Apply TWO feature selection methods, then re-train/evaluate - Save all metrics, best params, and selected features under ./outputs/ Requirements ------------ pip install pandas numpy scikit-learn Notes ----- - The script tries to auto-detect the target if you don't specify it in the config, by checking common names or using the last column. - Works for binary/multiclass classification; metrics are macro-averaged when needed. """ from __future__ import annotations import os, json, warnings from typing import Dict, List, Tuple, Optional import numpy as np import pandas as pd from sklearn.model_selection import train_test_split, StratifiedKFold, GridSearchCV from sklearn.compose import ColumnTransformer from sklearn.pipeline import Pipeline from sklearn.preprocessing import OneHotEncoder, StandardScaler from sklearn.impute import SimpleImputer from sklearn.metrics import accuracy_score, precision_score, recall_score, f1_score, roc_auc_score, confusion_matrix from sklearn.feature_selection import SelectKBest, mutual_info_classif from sklearn.linear_model import LogisticRegression from sklearn.feature_selection import SelectFromModel from sklearn.svm import SVC from sklearn.ensemble import RandomForestClassifier from sklearn.neighbors import KNeighborsClassifier warnings.filterwarnings("ignore", category=FutureWarning) # ------------------------- Config ------------------------- BASE_DIR = os.path.dirname(os.path.abspath(__file__)) DATASETS_DIR = os.path.join(BASE_DIR, "datasets") OUTPUTS_DIR = os.path.join(BASE_DIR, "outputs") CONFIG_FILE = os.path.join(BASE_DIR, "datasets_config.json") # you can copy the example to this name os.makedirs(OUTPUTS_DIR, exist_ok=True) COMMON_TARGET_NAMES = [ "target","class","label","labels","diagnosis","result","outcome","Class","Outcome","y" ] RANDOM_STATE = 42 N_JOBS = -1 # Models & params for GridSearchCV MODELS: Dict[str, Tuple[object, Dict]] = {{ "logreg": ( LogisticRegression(max_iter=1000), {{ "clf__penalty": ["l2"], "clf__C": [0.01, 0.1, 1.0, 10.0], "clf__solver": ["lbfgs", "liblinear"], }}, ), "svc": ( SVC(probability=True), {{ "clf__C": [0.1, 1, 10], "clf__kernel": ["rbf", "linear"], "clf__gamma": ["scale", "auto"], }}, ), "rf": ( RandomForestClassifier(random_state=RANDOM_STATE), {{ "clf__n_estimators": [100, 300], "clf__max_depth": [None, 5, 10], "clf__min_samples_split": [2, 5], }}, ), "knn": ( KNeighborsClassifier(), {{ "clf__n_neighbors": [3, 5, 11], "clf__weights": ["uniform", "distance"], }}, ), }} # ------------------------- Utilities ------------------------- def read_config(config_path: str) -> Dict[str, str]: if os.path.exists(config_path): with open(config_path, "r", encoding="utf-8") as f: return json.load(f) return {{}} def load_csv(path: str, target_col: Optional[str]) -> Tuple[pd.DataFrame, pd.Series, str]: df = pd.read_csv(path) # Try to auto-detect target if not provided tcol = target_col if tcol is None: lower_cols = {{c.lower(): c for c in df.columns}} for name in COMMON_TARGET_NAMES: if name.lower() in lower_cols: tcol = lower_cols[name.lower()] break if tcol is None: # fallback: last column tcol = df.columns[-1] print(f"[WARN] No target specified for {{os.path.basename(path)}}. Using last column '{{tcol}}'.") y = df[tcol] X = df.drop(columns=[tcol]) return X, y, tcol def split_features(X: pd.DataFrame) -> Tuple[List[str], List[str]]: num_cols = [c for c in X.columns if pd.api.types.is_numeric_dtype(X[c])] cat_cols = [c for c in X.columns if not pd.api.types.is_numeric_dtype(X[c])] return num_cols, cat_cols def build_preprocessor(num_cols: List[str], cat_cols: List[str]) -> ColumnTransformer: numeric_transformer = Pipeline(steps=[ ("imputer", SimpleImputer(strategy="median")), ("scaler", StandardScaler()), ]) categorical_transformer = Pipeline(steps=[ ("imputer", SimpleImputer(strategy="most_frequent")), ("onehot", OneHotEncoder(handle_unknown="ignore")), ]) preprocessor = ColumnTransformer(transformers=[ ("num", numeric_transformer, num_cols), ("cat", categorical_transformer, cat_cols), ]) return preprocessor def macro_auc(y_true, y_proba) -> float: # Handles binary or multiclass try: if isinstance(y_proba, np.ndarray) and (y_proba.ndim == 1 or y_proba.shape[1] == 1): # binary - probabilities for the positive class scores = y_proba if y_proba.ndim == 1 else y_proba[:, -1] return roc_auc_score(y_true, scores) else: return roc_auc_score(y_true, y_proba, multi_class="ovr", average="macro") except Exception: return np.nan def evaluate(y_true, y_pred, y_proba) -> Dict[str, float]: average = "binary" try: # if multiclass, use macro if len(np.unique(y_true)) > 2: average = "macro" except Exception: pass metrics = {{ "accuracy": accuracy_score(y_true, y_pred), "precision": precision_score(y_true, y_pred, average=average, zero_division=0), "recall": recall_score(y_true, y_pred, average=average, zero_division=0), "f1": f1_score(y_true, y_pred, average=average, zero_division=0), }} if y_proba is not None: metrics["roc_auc"] = macro_auc(y_true, y_proba) else: metrics["roc_auc"] = np.nan return metrics def fit_and_search(X, y, preprocessor, model_name: str, model, param_grid, cv_splits=5): pipe = Pipeline(steps=[ ("pre", preprocessor), ("clf", model), ]) cv = StratifiedKFold(n_splits=cv_splits, shuffle=True, random_state=42) gs = GridSearchCV(pipe, param_grid=param_grid, cv=cv, n_jobs=-1, scoring="f1_macro") gs.fit(X, y) return gs def predict_proba_safe(clf, X): if hasattr(clf, "predict_proba"): return clf.predict_proba(X) if hasattr(clf, "decision_function"): df = clf.decision_function(X) # map to [0,1] via logistic for binary; for multiclass, leave as-is if isinstance(df, np.ndarray) and df.ndim == 1: from scipy.special import expit return expit(df) else: return df return None def top_k_for_selectkbest(n_features: int) -> int: if n_features <= 10: return max(1, n_features // 2) return min(20, max(10, n_features // 3)) # ------------------------- Main workflow ------------------------- def run_for_dataset(csv_name: str, target_col: Optional[str], outputs_dir: str): path = os.path.join(DATASETS_DIR, csv_name) if not os.path.exists(path): print(f"[SKIP] {{csv_name}} not found in datasets/.") return None print(f"\n=== Dataset: {csv_name} ===") X, y, tcol = load_csv(path, target_col) num_cols, cat_cols = split_features(X) pre = build_preprocessor(num_cols, cat_cols) X_train, X_test, y_train, y_test = train_test_split( X, y, test_size=0.2, random_state=42, stratify=y if len(np.unique(y))>1 else None ) summary_rows = [] # ----- Step 3: Baseline (no feature selection) ----- for mname, (model, grid) in MODELS.items(): gs = fit_and_search(X_train, y_train, pre, mname, model, grid) best = gs.best_estimator_ y_pred = best.predict(X_test) y_proba = predict_proba_safe(best, X_test) metrics = evaluate(y_test, y_pred, y_proba) row = {{ "dataset": csv_name, "target": tcol, "phase": "baseline", "model": mname, **{{f"best__{{k}}": v for k, v in gs.best_params_.items()}}, **metrics }} summary_rows.append(row) # Save baseline confusion matrix for the best model by F1 baseline_df = pd.DataFrame([r for r in summary_rows if r["phase"]=="baseline"]) if not baseline_df.empty: best_idx = baseline_df["f1"].astype(float).idxmax() best_model_name = baseline_df.loc[best_idx, "model"] print(f"[INFO] Best baseline model for {csv_name}: {best_model_name}") # Refit the best one to get CM best_model, best_grid = MODELS[best_model_name] gs = fit_and_search(X_train, y_train, pre, best_model_name, best_model, best_grid) y_pred = gs.best_estimator_.predict(X_test) cm = confusion_matrix(y_test, y_pred) pd.DataFrame(cm).to_csv(os.path.join(outputs_dir, f"{csv_name}_baseline_confusion_matrix.csv"), index=False) # ----- Step 4/5: Feature Selection #1 — SelectKBest(mutual_info_classif) ----- # choose k based on number of features after preprocessing — approximate using raw n_features k = top_k_for_selectkbest(X.shape[1]) selectk = SelectKBest(score_func=mutual_info_classif, k=k) for mname, (model, grid) in MODELS.items(): pipe = Pipeline(steps=[ ("pre", pre), ("fs", selectk), ("clf", model), ]) cv = StratifiedKFold(n_splits=5, shuffle=True, random_state=42) params = {{}} # Grid params apply to 'clf__*' only params.update(grid) gs = GridSearchCV(pipe, param_grid=params, cv=cv, n_jobs=-1, scoring="f1_macro") gs.fit(X_train, y_train) y_pred = gs.best_estimator_.predict(X_test) y_proba = predict_proba_safe(gs.best_estimator_, X_test) metrics = evaluate(y_test, y_pred, y_proba) row = {{ "dataset": csv_name, "target": tcol, "phase": "FS_SelectKBest", "k": k, "model": mname, **{{f"best__{{k}}": v for k, v in gs.best_params_.items()}}, **metrics }} summary_rows.append(row) # Save selected feature names (indices after preprocessing) try: fs = gs.best_estimator_.named_steps["fs"] support = fs.get_support() selected_idx = np.where(support)[0].tolist() sel_df = pd.DataFrame({{ "selected_feature_index_after_preproc": selected_idx }}) sel_df.to_csv(os.path.join(outputs_dir, f"{csv_name}_SelectKBest_selected_indices.csv"), index=False) except Exception: pass # ----- Feature Selection #2 — SelectFromModel(LogisticRegression L1) ----- l1_selector = SelectFromModel(LogisticRegression(penalty="l1", C=1.0, solver="liblinear", max_iter=1000)) for mname, (model, grid) in MODELS.items(): pipe = Pipeline(steps=[ ("pre", pre), ("fs", l1_selector), ("clf", model), ]) cv = StratifiedKFold(n_splits=5, shuffle=True, random_state=42) params = {{}} params.update(grid) gs = GridSearchCV(pipe, param_grid=params, cv=cv, n_jobs=-1, scoring="f1_macro") gs.fit(X_train, y_train) y_pred = gs.best_estimator_.predict(X_test) y_proba = predict_proba_safe(gs.best_estimator_, X_test) metrics = evaluate(y_test, y_pred, y_proba) row = {{ "dataset": csv_name, "target": tcol, "phase": "FS_SelectFromModel_L1", "model": mname, **{{f"best__{{k}}": v for k, v in gs.best_params_.items()}}, **metrics }} summary_rows.append(row) # Save mask indices try: fs = gs.best_estimator_.named_steps["fs"] support = fs.get_support() selected_idx = np.where(support)[0].tolist() sel_df = pd.DataFrame({{ "selected_feature_index_after_preproc": selected_idx }}) sel_df.to_csv(os.path.join(outputs_dir, f"{csv_name}_SFM_L1_selected_indices.csv"), index=False) except Exception: pass # Save per-dataset summary pd.DataFrame(summary_rows).to_csv(os.path.join(outputs_dir, f"{csv_name}_summary.csv"), index=False) print(f"[DONE] Results saved for {csv_name}.") def main(): cfg = read_config(CONFIG_FILE) # If no config, fall back to any CSVs in datasets/ candidates = [f for f in os.listdir(DATASETS_DIR) if f.lower().endswith(".csv")] if not candidates: print("No CSV files found in ./datasets. Please add your Kaggle datasets.") return for csv_name in candidates: tcol = cfg.get(csv_name) if cfg else None run_for_dataset(csv_name, tcol, OUTPUTS_DIR) # Merge all summaries into one file summaries = [f for f in os.listdir(OUTPUTS_DIR) if f.endswith("_summary.csv")] if summaries: dfs = [pd.read_csv(os.path.join(OUTPUTS_DIR, s)) for s in summaries] big = pd.concat(dfs, ignore_index=True) big.to_csv(os.path.join(OUTPUTS_DIR, "ALL_DATASETS_summary.csv"), index=False) print("[OK] Wrote outputs/ALL_DATASETS_summary.csv") if __name__ == "__main__": main()