257 lines · 7.9 KB · Python
RawDownload
"""Complete classification pipeline example with preprocessing, model training,hyperparameter tuning, and evaluation.""" import numpy as npimport pandas as pdfrom sklearn.model_selection import train_test_split, GridSearchCV, cross_val_scorefrom sklearn.preprocessing import StandardScaler, OneHotEncoderfrom sklearn.impute import SimpleImputerfrom sklearn.compose import ColumnTransformerfrom sklearn.pipeline import Pipelinefrom sklearn.ensemble import RandomForestClassifier, GradientBoostingClassifierfrom sklearn.linear_model import LogisticRegressionfrom sklearn.metrics import (    classification_report, confusion_matrix, roc_auc_score,    accuracy_score, precision_score, recall_score, f1_score)import warningswarnings.filterwarnings('ignore')  def create_preprocessing_pipeline(numeric_features, categorical_features):    """    Create a preprocessing pipeline for mixed data types.     Parameters:    -----------    numeric_features : list        List of numeric feature column names    categorical_features : list        List of categorical feature column names     Returns:    --------    ColumnTransformer        Preprocessing pipeline    """    # Numeric preprocessing    numeric_transformer = Pipeline(steps=[        ('imputer', SimpleImputer(strategy='median')),        ('scaler', StandardScaler())    ])     # Categorical preprocessing    categorical_transformer = Pipeline(steps=[        ('imputer', SimpleImputer(strategy='constant', fill_value='missing')),        ('onehot', OneHotEncoder(handle_unknown='ignore', sparse_output=False))    ])     # Combine transformers    preprocessor = ColumnTransformer(        transformers=[            ('num', numeric_transformer, numeric_features),            ('cat', categorical_transformer, categorical_features)        ]    )     return preprocessor  def train_and_evaluate_model(X, y, numeric_features, categorical_features,                             test_size=0.2, random_state=42):    """    Complete pipeline: preprocess, train, tune, and evaluate a classifier.     Parameters:    -----------    X : DataFrame or array        Feature matrix    y : Series or array        Target variable    numeric_features : list        List of numeric feature names    categorical_features : list        List of categorical feature names    test_size : float        Proportion of data for testing    random_state : int        Random seed     Returns:    --------    dict        Dictionary containing trained model, predictions, and metrics    """    # Split data with stratification    X_train, X_test, y_train, y_test = train_test_split(        X, y, test_size=test_size, stratify=y, random_state=random_state    )     print(f"Training set size: {len(X_train)}")    print(f"Test set size: {len(X_test)}")    print(f"Class distribution in training: {pd.Series(y_train).value_counts().to_dict()}")     # Create preprocessor    preprocessor = create_preprocessing_pipeline(numeric_features, categorical_features)     # Define models to compare    models = {        'Logistic Regression': Pipeline([            ('preprocessor', preprocessor),            ('classifier', LogisticRegression(max_iter=1000, random_state=random_state))        ]),        'Random Forest': Pipeline([            ('preprocessor', preprocessor),            ('classifier', RandomForestClassifier(n_estimators=100, random_state=random_state))        ]),        'Gradient Boosting': Pipeline([            ('preprocessor', preprocessor),            ('classifier', GradientBoostingClassifier(n_estimators=100, random_state=random_state))        ])    }     # Compare models using cross-validation    print("\n" + "="*60)    print("Model Comparison (5-Fold Cross-Validation)")    print("="*60)     cv_results = {}    for name, model in models.items():        scores = cross_val_score(model, X_train, y_train, cv=5, scoring='accuracy')        cv_results[name] = scores.mean()        print(f"{name:20s}: {scores.mean():.4f} (+/- {scores.std() * 2:.4f})")     # Select best model based on CV    best_model_name = max(cv_results, key=cv_results.get)    best_model = models[best_model_name]     print(f"\nBest model: {best_model_name}")     # Hyperparameter tuning for best model    if best_model_name == 'Random Forest':        param_grid = {            'classifier__n_estimators': [100, 200],            'classifier__max_depth': [10, 20, None],            'classifier__min_samples_split': [2, 5]        }    elif best_model_name == 'Gradient Boosting':        param_grid = {            'classifier__n_estimators': [100, 200],            'classifier__learning_rate': [0.01, 0.1],            'classifier__max_depth': [3, 5]        }    else:  # Logistic Regression        param_grid = {            'classifier__C': [0.1, 1.0, 10.0],            'classifier__penalty': ['l2']        }     print("\n" + "="*60)    print("Hyperparameter Tuning")    print("="*60)     grid_search = GridSearchCV(        best_model, param_grid, cv=5, scoring='accuracy',        n_jobs=-1, verbose=0    )     grid_search.fit(X_train, y_train)     print(f"Best parameters: {grid_search.best_params_}")    print(f"Best CV score: {grid_search.best_score_:.4f}")     # Evaluate on test set    tuned_model = grid_search.best_estimator_    y_pred = tuned_model.predict(X_test)    y_pred_proba = tuned_model.predict_proba(X_test)     print("\n" + "="*60)    print("Test Set Evaluation")    print("="*60)     # Calculate metrics    accuracy = accuracy_score(y_test, y_pred)    precision = precision_score(y_test, y_pred, average='weighted')    recall = recall_score(y_test, y_pred, average='weighted')    f1 = f1_score(y_test, y_pred, average='weighted')     print(f"Accuracy:  {accuracy:.4f}")    print(f"Precision: {precision:.4f}")    print(f"Recall:    {recall:.4f}")    print(f"F1-Score:  {f1:.4f}")     # ROC AUC (if binary classification)    if len(np.unique(y)) == 2:        roc_auc = roc_auc_score(y_test, y_pred_proba[:, 1])        print(f"ROC AUC:   {roc_auc:.4f}")     print("\n" + "="*60)    print("Classification Report")    print("="*60)    print(classification_report(y_test, y_pred))     print("\n" + "="*60)    print("Confusion Matrix")    print("="*60)    print(confusion_matrix(y_test, y_pred))     # Feature importance (if available)    if hasattr(tuned_model.named_steps['classifier'], 'feature_importances_'):        print("\n" + "="*60)        print("Top 10 Most Important Features")        print("="*60)         feature_names = tuned_model.named_steps['preprocessor'].get_feature_names_out()        importances = tuned_model.named_steps['classifier'].feature_importances_         feature_importance_df = pd.DataFrame({            'feature': feature_names,            'importance': importances        }).sort_values('importance', ascending=False).head(10)         print(feature_importance_df.to_string(index=False))     return {        'model': tuned_model,        'y_test': y_test,        'y_pred': y_pred,        'y_pred_proba': y_pred_proba,        'metrics': {            'accuracy': accuracy,            'precision': precision,            'recall': recall,            'f1': f1        }    }  # Example usageif __name__ == "__main__":    # Load example dataset    from sklearn.datasets import load_breast_cancer     # Load data    data = load_breast_cancer()    X = pd.DataFrame(data.data, columns=data.feature_names)    y = data.target     # For demonstration, treat all features as numeric    numeric_features = X.columns.tolist()    categorical_features = []     print("="*60)    print("Classification Pipeline Example")    print("Dataset: Breast Cancer Wisconsin")    print("="*60)     # Run complete pipeline    results = train_and_evaluate_model(        X, y, numeric_features, categorical_features,        test_size=0.2, random_state=42    )     print("\n" + "="*60)    print("Pipeline Complete!")    print("="*60)