Skills › Scientific Agent Skills › scikit-learn › classification_pipeline.py
classification_pipeline.py
290 lines · skills/scikit-learn/scripts/classification_pipeline.py at 5f2bc03
1"""
2Complete classification pipeline example with preprocessing, model training,
3hyperparameter tuning, and evaluation.
4"""
6import numpy as np
7import pandas as pd
8from sklearn.model_selection import train_test_split, GridSearchCV, cross_val_score, StratifiedKFold
9from sklearn.base import clone
10from sklearn.preprocessing import StandardScaler, OneHotEncoder, FunctionTransformer
11from sklearn.impute import SimpleImputer
12from sklearn.compose import ColumnTransformer
13from sklearn.pipeline import Pipeline
14from sklearn.ensemble import RandomForestClassifier, GradientBoostingClassifier
15from sklearn.linear_model import LogisticRegression
16from sklearn.metrics import (
17 classification_report, confusion_matrix, roc_auc_score,
18 accuracy_score, precision_score, recall_score, f1_score, balanced_accuracy_score
19)
22def normalize_categorical_missing(X):
23 """Map pandas/None missing sentinels to the imputer's np.nan sentinel."""
24 frame = pd.DataFrame(X).astype(object)
25 return frame.where(frame.notna(), np.nan).to_numpy()
28def create_preprocessing_pipeline(numeric_features, categorical_features):
29 """
30 Create a preprocessing pipeline for mixed data types.
32 Parameters:
33 -----------
34 numeric_features : list
35 List of numeric feature column names
36 categorical_features : list
37 List of categorical feature column names
39 Returns:
40 --------
41 ColumnTransformer
42 Preprocessing pipeline
43 """
44 # Numeric preprocessing
45 numeric_transformer = Pipeline(steps=[
46 ('imputer', SimpleImputer(strategy='median', keep_empty_features=True)),
47 ('scaler', StandardScaler())
48 ])
50 # Categorical preprocessing
51 categorical_transformer = Pipeline(steps=[
52 ('missing', FunctionTransformer(normalize_categorical_missing,
53 feature_names_out='one-to-one')),
54 ('imputer', SimpleImputer(strategy='constant', fill_value='missing',
55 keep_empty_features=True)),
56 ('onehot', OneHotEncoder(handle_unknown='ignore', sparse_output=False))
57 ])
59 # Combine transformers
60 preprocessor = ColumnTransformer(
61 transformers=[
62 ('num', numeric_transformer, numeric_features),
63 ('cat', categorical_transformer, categorical_features)
64 ]
65 )
67 return preprocessor
70def train_and_evaluate_model(X, y, numeric_features, categorical_features,
71 test_size=0.2, random_state=42, cv=5,
72 scoring="balanced_accuracy", n_jobs=1):
73 """
74 Preprocess, select, tune, and evaluate independent-row classification data.
76 This random-split demonstration is not appropriate for grouped or temporal
77 observations. Selection CV scores are optimistic; only the untouched test
78 partition evaluates the selected procedure. Dense one-hot encoding is for
79 small/moderate categorical cardinalities.
81 Parameters:
82 -----------
83 X : DataFrame
84 Feature matrix (feature selectors below are column names)
85 y : Series or array
86 Target variable
87 numeric_features : list
88 List of numeric feature names
89 categorical_features : list
90 List of categorical feature names
91 test_size : float
92 Proportion of data for testing
93 random_state : int
94 Random seed
96 cv : int
97 Number of stratified training folds; every class must have this many rows
98 scoring : str or callable
99 Selection metric passed unchanged to cross_val_score and GridSearchCV
100 n_jobs : int
101 Search workers; default 1 avoids nested parallelism
103 Returns:
104 --------
105 dict
106 Trained model, predictions, selection scores, class order and test metrics
107 """
108 # Split data with stratification
109 X_train, X_test, y_train, y_test = train_test_split(
110 X, y, test_size=test_size, stratify=y, random_state=random_state
111 )
113 print(f"Training set size: {len(X_train)}")
114 print(f"Test set size: {len(X_test)}")
115 print(f"Class distribution in training: {pd.Series(y_train).value_counts().to_dict()}")
117 if pd.Series(y_train).value_counts().min() < cv:
118 raise ValueError("Each training class needs at least cv samples; lower cv or add data")
119 splitter = StratifiedKFold(n_splits=cv, shuffle=True, random_state=random_state)
121 # Create preprocessor
122 preprocessor = create_preprocessing_pipeline(numeric_features, categorical_features)
124 # Define models to compare
125 models = {
126 'Logistic Regression': Pipeline([
127 ('preprocessor', clone(preprocessor)),
128 ('classifier', LogisticRegression(max_iter=1000, random_state=random_state))
129 ]),
130 'Random Forest': Pipeline([
131 ('preprocessor', clone(preprocessor)),
132 ('classifier', RandomForestClassifier(n_estimators=100, random_state=random_state))
133 ]),
134 'Gradient Boosting': Pipeline([
135 ('preprocessor', clone(preprocessor)),
136 ('classifier', GradientBoostingClassifier(n_estimators=100, random_state=random_state))
137 ])
138 }
140 # Compare models using cross-validation
141 print("\n" + "="*60)
142 print(f"Model Comparison ({cv}-Fold Cross-Validation; {scoring})")
143 print("="*60)
145 cv_results = {}
146 for name, model in models.items():
147 scores = cross_val_score(model, X_train, y_train, cv=splitter, scoring=scoring, error_score='raise')
148 cv_results[name] = scores.mean()
149 print(f"{name:20s}: {scores.mean():.4f} (fold SD {scores.std():.4f})")
151 # Select best model based on CV
152 best_model_name = max(cv_results, key=cv_results.get)
153 best_model = models[best_model_name]
155 print(f"\nBest model: {best_model_name}")
157 # Hyperparameter tuning for best model
158 if best_model_name == 'Random Forest':
159 param_grid = {
160 'classifier__n_estimators': [100, 200],
161 'classifier__max_depth': [10, 20, None],
162 'classifier__min_samples_split': [2, 5]
163 }
164 elif best_model_name == 'Gradient Boosting':
165 param_grid = {
166 'classifier__n_estimators': [100, 200],
167 'classifier__learning_rate': [0.01, 0.1],
168 'classifier__max_depth': [3, 5]
169 }
170 else: # Logistic Regression
171 param_grid = {
172 'classifier__C': [0.1, 1.0, 10.0],
173 'classifier__l1_ratio': [0.0]
174 }
176 print("\n" + "="*60)
177 print("Hyperparameter Tuning")
178 print("="*60)
180 grid_search = GridSearchCV(
181 best_model, param_grid, cv=splitter, scoring=scoring,
182 n_jobs=n_jobs, verbose=0, error_score="raise"
183 )
185 grid_search.fit(X_train, y_train)
187 print(f"Best parameters: {grid_search.best_params_}")
188 print(f"Best CV score: {grid_search.best_score_:.4f}")
190 # Evaluate on test set
191 tuned_model = grid_search.best_estimator_
192 y_pred = tuned_model.predict(X_test)
193 y_pred_proba = tuned_model.predict_proba(X_test)
195 print("\n" + "="*60)
196 print("Test Set Evaluation")
197 print("="*60)
199 # Calculate metrics
200 accuracy = accuracy_score(y_test, y_pred)
201 precision = precision_score(y_test, y_pred, average='weighted', zero_division=0)
202 recall = recall_score(y_test, y_pred, average='weighted', zero_division=0)
203 f1 = f1_score(y_test, y_pred, average='weighted', zero_division=0)
205 balanced_accuracy = balanced_accuracy_score(y_test, y_pred)
206 macro_f1 = f1_score(y_test, y_pred, average='macro', zero_division=0)
207 metrics = dict(accuracy=accuracy, balanced_accuracy=balanced_accuracy,
208 precision=precision, recall=recall, f1=f1, macro_f1=macro_f1)
210 print(f"Balanced accuracy: {balanced_accuracy:.4f}")
211 print(f"Macro F1: {macro_f1:.4f}")
212 print(f"Accuracy: {accuracy:.4f}")
213 print(f"Precision: {precision:.4f}")
214 print(f"Recall: {recall:.4f}")
215 print(f"F1-Score: {f1:.4f}")
217 # ROC AUC (if binary classification)
218 if len(tuned_model.classes_) == 2:
219 # Column 1 corresponds to classes_[1], which need not mean disease/case.
220 roc_auc = roc_auc_score(y_test == tuned_model.classes_[1], y_pred_proba[:, 1])
221 metrics['roc_auc'] = roc_auc
222 print(f"ROC AUC: {roc_auc:.4f}")
224 print("\n" + "="*60)
225 print("Classification Report")
226 print("="*60)
227 print(classification_report(y_test, y_pred, zero_division=0))
229 print("\n" + "="*60)
230 print("Confusion Matrix")
231 print("="*60)
232 print(confusion_matrix(y_test, y_pred))
234 # Feature importance (if available)
235 if hasattr(tuned_model.named_steps['classifier'], 'feature_importances_'):
236 print("\n" + "="*60)
237 print("Top 10 Impurity Importances (biased toward high-cardinality features)")
238 print("="*60)
240 feature_names = tuned_model.named_steps['preprocessor'].get_feature_names_out()
241 importances = tuned_model.named_steps['classifier'].feature_importances_
243 feature_importance_df = pd.DataFrame({
244 'feature': feature_names,
245 'importance': importances
246 }).sort_values('importance', ascending=False).head(10)
248 print(feature_importance_df.to_string(index=False))
250 return {
251 'model': tuned_model,
252 'y_test': y_test,
253 'y_pred': y_pred,
254 'y_pred_proba': y_pred_proba,
255 'classes': tuned_model.classes_,
256 'positive_class': tuned_model.classes_[1] if len(tuned_model.classes_) == 2 else None,
257 'selection_scoring': scoring,
258 'cv_scores': cv_results,
259 'metrics': metrics
260 }
263# Example usage
264if __name__ == "__main__":
265 # Load example dataset
266 from sklearn.datasets import load_breast_cancer
268 # Load data
269 data = load_breast_cancer()
270 X = pd.DataFrame(data.data, columns=data.feature_names)
271 y = data.target
273 # For demonstration, treat all features as numeric
274 numeric_features = X.columns.tolist()
275 categorical_features = []
277 print("="*60)
278 print("Classification Pipeline Example")
279 print("Dataset: Breast Cancer Wisconsin")
280 print("="*60)
282 # Run complete pipeline
283 results = train_and_evaluate_model(
284 X, y, numeric_features, categorical_features,
285 test_size=0.2, random_state=42
286 )
288 print("\n" + "="*60)
289 print("Pipeline Complete!")
290 print("="*60)