| """Reproducible bank + telecom churn training; no customer row data is published.""" |
| from __future__ import annotations |
| import argparse |
| import sys |
| import hashlib |
| import json |
| from datetime import datetime, timezone |
| from pathlib import Path |
|
|
| import joblib |
| import matplotlib |
| matplotlib.use('Agg') |
| import matplotlib.pyplot as plt |
| import numpy as np |
| import pandas as pd |
| import sklearn |
| import xgboost |
| from sklearn.compose import ColumnTransformer |
| from sklearn.ensemble import RandomForestClassifier |
| from sklearn.impute import SimpleImputer |
| from sklearn.inspection import permutation_importance |
| from sklearn.linear_model import LogisticRegression |
| from sklearn.metrics import (average_precision_score, balanced_accuracy_score, |
| classification_report, confusion_matrix, ConfusionMatrixDisplay, f1_score, |
| fbeta_score, precision_score, recall_score, roc_auc_score, roc_curve) |
| from sklearn.model_selection import RandomizedSearchCV, StratifiedKFold, train_test_split |
| from sklearn.pipeline import Pipeline |
| from sklearn.preprocessing import OneHotEncoder, StandardScaler |
| from xgboost import XGBClassifier |
|
|
| SEED = 42 |
| SPECS = { |
| 'bank': { |
| 'filename': 'bank_customer_churn.csv', 'target': 'Exited', |
| 'numeric': ['CreditScore','Age','Tenure','Balance','NumOfProducts','HasCrCard','IsActiveMember','EstimatedSalary'], |
| 'categorical': ['Geography'], 'excluded': ['CustomerId','Surname'], |
| 'display': 'Bank customer churn', |
| }, |
| 'telecom': { |
| 'filename': 'customer_churn_data.csv', 'target': 'Churn', |
| 'numeric': ['SeniorCitizen','tenure','MonthlyCharges','TotalCharges'], |
| 'categorical': ['Partner','Dependents','PhoneService','InternetService','Contract','PaymentMethod'], |
| 'excluded': ['customerID'], 'display': 'Telecom customer churn', |
| }, |
| } |
|
|
| def pyvalue(v): |
| if isinstance(v, np.generic): |
| return v.item() |
| return v |
|
|
| def load_dataset(path, spec): |
| if not path.is_file(): |
| raise FileNotFoundError(f'Required dataset missing: {path}. Upload the original CSV in Colab.') |
| raw = pd.read_csv(path) |
| required = spec['numeric'] + spec['categorical'] + [spec['target']] |
| missing = [c for c in required if c not in raw.columns] |
| if missing: |
| raise ValueError(f'{path.name}: missing columns {missing}. Found {list(raw.columns)}') |
| original_rows = len(raw) |
| frame = raw.drop_duplicates().copy() |
| |
| X = frame[spec['numeric'] + spec['categorical']].copy() |
| for col in spec['numeric']: |
| X[col] = pd.to_numeric(X[col], errors='coerce') |
| for col in spec['categorical']: |
| X[col] = X[col].astype('string').str.strip().replace('', pd.NA) |
| X[col] = X[col].astype(object).where(X[col].notna(), np.nan) |
| if spec['target'] == 'Churn': |
| y = frame[spec['target']].astype('string').str.strip().str.lower().map({'yes': 1, 'no': 0}) |
| else: |
| y = pd.to_numeric(frame[spec['target']], errors='coerce') |
| if y.isna().any() or not set(y.unique()).issubset({0, 1}): |
| raise ValueError(f'{path.name}: target must be fully labeled binary 0/1 or Yes/No') |
| y = y.astype('int64') |
| if y.value_counts().min() < 20: |
| raise ValueError(f'{path.name}: insufficient positive/negative examples') |
| audit = { |
| 'original_rows': int(original_rows), |
| 'duplicate_rows_removed': int(original_rows-len(frame)), |
| 'rows_after_deduplication': int(len(frame)), |
| 'positive_rate': round(float(y.mean()), 5), |
| 'feature_missing_counts': {col: int(X[col].isna().sum()) for col in X.columns}, |
| 'sha256_first12': hashlib.sha256(path.read_bytes()).hexdigest()[:12], |
| } |
| return X, y, audit |
|
|
| def preprocessing(spec): |
| numerical = Pipeline([('impute', SimpleImputer(strategy='median')), |
| ('scale', StandardScaler())]) |
| categorical = Pipeline([('impute', SimpleImputer(strategy='most_frequent')), |
| ('onehot', OneHotEncoder(handle_unknown='ignore', sparse_output=False))]) |
| return ColumnTransformer([('numeric', numerical, spec['numeric']), |
| ('categorical', categorical, spec['categorical'])], |
| remainder='drop') |
|
|
| def score_proba(model, X): |
| return model.predict_proba(X)[:, 1] |
|
|
| def metrics(y, p, threshold): |
| pred = (p >= threshold).astype(int) |
| return { |
| 'roc_auc': round(float(roc_auc_score(y, p)), 5), |
| 'pr_auc': round(float(average_precision_score(y, p)), 5), |
| 'balanced_accuracy': round(float(balanced_accuracy_score(y, pred)), 5), |
| 'precision': round(float(precision_score(y, pred, zero_division=0)), 5), |
| 'recall': round(float(recall_score(y, pred, zero_division=0)), 5), |
| 'f1': round(float(f1_score(y, pred, zero_division=0)), 5), |
| 'f2': round(float(fbeta_score(y, pred, beta=2, zero_division=0)), 5), |
| 'confusion_matrix': confusion_matrix(y, pred, labels=[0, 1]).tolist(), |
| 'classification_report': classification_report(y, pred, labels=[0, 1], |
| output_dict=True, zero_division=0), |
| } |
|
|
| def tune_threshold(y, p): |
| thresholds = np.round(np.arange(0.10, 0.901, 0.025), 3) |
| values = [fbeta_score(y, (p >= t).astype(int), beta=2, zero_division=0) for t in thresholds] |
| return float(thresholds[int(np.argmax(values))]) |
|
|
| def plot_diagnostics(project, key, ytest, ptest, threshold, importance_df): |
| report_dir = project / 'reports' |
| pred = (ptest >= threshold).astype(int) |
| fig, ax = plt.subplots(figsize=(5, 4)) |
| ConfusionMatrixDisplay.from_predictions(ytest, pred, display_labels=['Retained', 'Churned'], |
| cmap='Blues', colorbar=False, ax=ax) |
| ax.set_title(f'{key.title()} — untouched test data') |
| fig.tight_layout() |
| fig.savefig(report_dir / f'{key}_confusion_matrix.png', dpi=140) |
| plt.close(fig) |
| fpr, tpr, _ = roc_curve(ytest, ptest) |
| fig, ax = plt.subplots(figsize=(5.5, 4)) |
| ax.plot(fpr, tpr, label=f'ROC AUC = {roc_auc_score(ytest, ptest):.3f}') |
| ax.plot([0, 1], [0, 1], linestyle='--', label='Chance') |
| ax.set(xlabel='False positive rate', ylabel='True positive rate', title=f'{key.title()} — test ROC') |
| ax.legend(loc='lower right') |
| fig.tight_layout() |
| fig.savefig(report_dir / f'{key}_roc.png', dpi=140) |
| plt.close(fig) |
| top = importance_df.head(10).sort_values('importance_mean') |
| fig, ax = plt.subplots(figsize=(7, 4.5)) |
| ax.barh(top['feature'], top['importance_mean']) |
| ax.set(xlabel='Change in validation PR AUC after shuffling', title=f'{key.title()} — permutation importance') |
| fig.tight_layout() |
| fig.savefig(report_dir / f'{key}_feature_importance.png', dpi=140) |
| plt.close(fig) |
|
|
| def train_one(project, source_dir, key, tuning_iterations): |
| spec = SPECS[key] |
| path = source_dir / spec['filename'] |
| X, y, audit = load_dataset(path, spec) |
| X_dev, X_test, y_dev, y_test = train_test_split(X, y, test_size=0.20, |
| random_state=SEED, stratify=y) |
| X_train, X_val, y_train, y_val = train_test_split(X_dev, y_dev, test_size=0.20, |
| random_state=SEED, stratify=y_dev) |
| minority_weight = float((y_train == 0).sum() / (y_train == 1).sum()) |
| models = { |
| 'Logistic regression': LogisticRegression(max_iter=1500, class_weight='balanced', random_state=SEED), |
| 'Random forest': RandomForestClassifier(n_estimators=170, min_samples_leaf=3, |
| class_weight='balanced_subsample', n_jobs=-1, random_state=SEED), |
| } |
| candidates = {} |
| leaderboard = [] |
| for name, estimator in models.items(): |
| pipeline = Pipeline([('preprocessor', preprocessing(spec)), ('model', estimator)]) |
| pipeline.fit(X_train, y_train) |
| candidates[name] = pipeline |
| row = metrics(y_val, score_proba(pipeline, X_val), 0.5) |
| leaderboard.append({'dataset': key, 'model': name, 'validation_pr_auc': row['pr_auc'], |
| 'validation_roc_auc': row['roc_auc'], 'tuning': 'none'}) |
| xgb_pipe = Pipeline([('preprocessor', preprocessing(spec)), |
| ('model', XGBClassifier(n_estimators=200, max_depth=3, learning_rate=0.05, |
| subsample=0.9, colsample_bytree=0.9, scale_pos_weight=minority_weight, |
| eval_metric='logloss', tree_method='hist', random_state=SEED, n_jobs=1))]) |
| if tuning_iterations: |
| params = { |
| 'model__n_estimators': [100, 180, 260], |
| 'model__max_depth': [2, 3, 4], |
| 'model__learning_rate': [0.025, 0.05, 0.1], |
| 'model__subsample': [0.8, 1.0], |
| 'model__colsample_bytree': [0.8, 1.0], |
| 'model__min_child_weight': [1, 3], |
| } |
| search = RandomizedSearchCV(xgb_pipe, params, n_iter=tuning_iterations, |
| scoring='average_precision', cv=StratifiedKFold(3, shuffle=True, random_state=SEED), |
| random_state=SEED, n_jobs=2, refit=True, verbose=0, error_score='raise') |
| search.fit(X_train, y_train) |
| xgb_pipe = search.best_estimator_ |
| print(f'[{key}] XGBoost best 3-fold training CV PR AUC: {search.best_score_:.4f}') |
| else: |
| xgb_pipe.fit(X_train, y_train) |
| candidates['XGBoost'] = xgb_pipe |
| val_xgb = metrics(y_val, score_proba(xgb_pipe, X_val), 0.5) |
| leaderboard.append({'dataset': key, 'model': 'XGBoost', |
| 'validation_pr_auc': val_xgb['pr_auc'], 'validation_roc_auc': val_xgb['roc_auc'], |
| 'tuning': f'{tuning_iterations}x3 CV' if tuning_iterations else 'off'}) |
| leaderboard.sort(key=lambda r: (r['validation_pr_auc'], r['validation_roc_auc']), reverse=True) |
| winner = leaderboard[0]['model'] |
| champion = candidates[winner] |
| threshold = tune_threshold(y_val, score_proba(champion, X_val)) |
| |
| test_scores = score_proba(champion, X_test) |
| test = metrics(y_test, test_scores, threshold) |
| |
| |
| reliable = (test['roc_auc'] >= 0.60 and test['pr_auc'] >= float(y_test.mean()) + 0.03) |
| deployment_status = 'demonstration' if reliable else 'research_only_low_signal' |
| if not reliable: |
| print(f'[{key}] WARNING: near-chance model; API will not issue actionable churn labels.') |
| val = metrics(y_val, score_proba(champion, X_val), threshold) |
| defaults = {c: float(X_train[c].median()) for c in spec['numeric']} |
| categories = {} |
| for col in spec['categorical']: |
| vals = X_train[col].dropna().astype(str) |
| categories[col] = sorted(vals.unique().tolist()) |
| defaults[col] = str(vals.mode().iloc[0]) |
| importance = permutation_importance(champion, X_val, y_val, scoring='average_precision', |
| n_repeats=3, random_state=SEED, n_jobs=1) |
| importance_df = pd.DataFrame({'feature': spec['numeric'] + spec['categorical'], |
| 'importance_mean': importance.importances_mean, 'importance_std': importance.importances_std}) |
| importance_df = importance_df.sort_values('importance_mean', ascending=False) |
| importance_df.to_csv(project / 'reports' / f'{key}_feature_importance.csv', index=False) |
| plot_diagnostics(project, key, y_test, test_scores, threshold, importance_df) |
| bundle = {'pipeline': champion, 'threshold': threshold, |
| 'numeric': spec['numeric'], 'categorical': spec['categorical'], 'features': spec['numeric'] + spec['categorical']} |
| path_model = project / 'models' / f'{key}_churn.joblib' |
| joblib.dump(bundle, path_model, compress=3) |
| reloaded = joblib.load(path_model) |
| assert np.allclose(reloaded['pipeline'].predict_proba(X_test.iloc[:5])[:,1], test_scores[:5]) |
| metadata = { |
| 'label': spec['display'], 'selected_model': winner, |
| 'numeric': spec['numeric'], 'categorical': spec['categorical'], |
| 'features': spec['numeric'] + spec['categorical'], |
| 'excluded_columns': spec['excluded'], 'categorical_choices': categories, |
| 'defaults': defaults, 'threshold': threshold, |
| 'training_rows': len(X_train), 'validation_rows': len(X_val), 'test_rows': len(X_test), |
| 'audit': audit, 'validation_metrics': val, 'test_metrics': test, |
| 'deployment_status': deployment_status, |
| 'model_size_mb': round(path_model.stat().st_size / 1024**2, 3), |
| 'validation_model_comparison': leaderboard, |
| } |
| print(f'[{key}] Winner: {winner}, threshold: {threshold:.3f}, ' |
| f'test PR AUC: {test["pr_auc"]:.4f}, ROC AUC: {test["roc_auc"]:.4f}, recall: {test["recall"]:.4f}') |
| return metadata, leaderboard |
|
|
| def main(data_dir='.', project_dir='.', tuning_iterations=6): |
| project = Path(project_dir).resolve() |
| data_dir = Path(data_dir).resolve() |
| (project / 'models').mkdir(parents=True, exist_ok=True) |
| (project / 'reports').mkdir(parents=True, exist_ok=True) |
| results, leaderboard = {}, [] |
| for key in SPECS: |
| print(f'\n=== TRAINING {key.upper()} ===') |
| results[key], comparison = train_one(project, data_dir, key, tuning_iterations) |
| leaderboard.extend(comparison) |
| summary = {'generated_utc': datetime.now(timezone.utc).isoformat(), |
| 'random_seed': SEED, 'sklearn_version': sklearn.__version__, |
| 'xgboost_version': xgboost.__version__, 'python_version': sys.version.split()[0], 'datasets': results, |
| 'decision_rule': 'maximize validation F2 (recall-weighted), after selecting by validation PR AUC', |
| 'probability_warning': 'Scores are uncalibrated model probabilities; not validated as true churn probabilities.'} |
| (project / 'model_metadata.json').write_text(json.dumps(summary, indent=2, default=pyvalue), encoding='utf-8') |
| pd.DataFrame(leaderboard).to_csv(project / 'reports' / 'model_comparison.csv', index=False) |
| print('\nSaved trained pipelines, evaluation, charts, and audit metadata to', project) |
| return summary |
|
|
| if __name__ == '__main__': |
| parser = argparse.ArgumentParser() |
| parser.add_argument('--data-dir', default='.') |
| parser.add_argument('--project-dir', default='.') |
| parser.add_argument('--tuning-iterations', type=int, default=6) |
| args = parser.parse_args() |
| main(args.data_dir, args.project_dir, args.tuning_iterations) |
|
|
