حزم النموذج النهائي وتوثيقه وعرضه

Why Packaging and Documentation Matter

Serialising the Final Pipeline with joblib

import joblib
import json
from datetime import date

# Save the fitted pipeline
model_filename = f'churn_xgboost_{date.today().isoformat()}_auc0883.pkl'
joblib.dump(best_pipeline, model_filename)
print(f'Saved pipeline: {model_filename}')

# Verify round-trip
loaded_pipeline = joblib.load(model_filename)
y_reloaded = loaded_pipeline.predict_proba(X_test[:5])[:, 1]
y_original = best_pipeline.predict_proba(X_test[:5])[:, 1]
print('Predictions match after reload:', all(y_reloaded == y_original))

Writing a JSON Metadata Sidecar

import hashlib, json, sklearn, joblib, os
from datetime import date

def sha256_file(path):
    h = hashlib.sha256()
    with open(path, 'rb') as f:
        for chunk in iter(lambda: f.read(8192), b''):
            h.update(chunk)
    return h.hexdigest()

metadata = {
    'model_type': 'XGBClassifier',
    'task': 'binary_classification',
    'target': 'churn_90d',
    'training_date': date.today().isoformat(),
    'sklearn_version': sklearn.__version__,
    'feature_names': list(feature_names),
    'cv_auc_mean': 0.883,
    'cv_auc_std': 0.007,
    'test_auc': 0.879,
    'sha256': sha256_file(model_filename)
}
with open(model_filename.replace('.pkl', '_metadata.json'), 'w') as f:
    json.dump(metadata, f, indent=2)
print(json.dumps(metadata, indent=2))

Writing a Model Card

model_card_template = '''
# Model Card: Customer Churn Predictor v1.0

## Model Details
- Type: XGBoost binary classifier inside a scikit-learn Pipeline
- Task: Predict 90-day customer churn
- Version: 1.0.0 | Training date: 2026-06-25

## Intended Use
- Primary use: Nightly batch scoring to generate a churn risk score per customer
- Out-of-scope: Real-time scoring, non-B2C segments, churn windows != 90 days

## Metrics
- 5-fold CV AUC: 0.883 +/- 0.007
- Held-out test AUC: 0.879
- Precision@threshold=0.4: 0.67 | Recall@threshold=0.4: 0.82

## Training Data
- Source: orders DB + CRM, 2024-01-01 to 2026-04-30
- Samples: 120,000 customers | Churn rate: 8.3%

## Ethical Considerations
- Age and region features audited for demographic parity (difference < 0.03)
- No direct use of protected attributes

## Caveats
- Performance may degrade if product catalogue changes significantly
- Retraining recommended if test AUC drops below 0.85 in monitoring
'''
print(model_card_template)

Generating Performance Artefacts

from sklearn.metrics import RocCurveDisplay, PrecisionRecallDisplay, ConfusionMatrixDisplay
import matplotlib.pyplot as plt

fig, axes = plt.subplots(1, 3, figsize=(15, 4))

# ROC curve
RocCurveDisplay.from_predictions(y_test, y_proba, ax=axes[0], name='XGBoost')
axes[0].set_title('ROC Curve')

# Precision-Recall curve
PrecisionRecallDisplay.from_predictions(y_test, y_proba, ax=axes[1], name='XGBoost')
axes[1].set_title('Precision-Recall Curve')

# Confusion matrix at threshold 0.4
y_pred_thresh = (y_proba >= 0.4).astype(int)
ConfusionMatrixDisplay.from_predictions(y_test, y_pred_thresh, ax=axes[2], display_labels=['retained', 'churned'])
axes[2].set_title('Confusion Matrix @ threshold=0.4')

plt.tight_layout()
plt.savefig('model_card_performance.png', dpi=150)

Versioning with Git and DVC

# DVC workflow (shell commands — not Python)
# pip install dvc[s3]

# Initialise DVC in the repo
# dvc init

# Add model artifact to DVC tracking
# dvc add churn_xgboost_2026-06-25_auc0883.pkl
# git add churn_xgboost_2026-06-25_auc0883.pkl.dvc .gitignore
# git commit -m 'Add XGBoost churn model v1.0'

# Push artifact to S3
# dvc remote add myremote s3://my-ml-artifacts/churn-model
# dvc push

# To reproduce: checkout a git commit, then:
# dvc pull  # downloads the exact artifact for that commit
print('DVC enables git-compatible versioning of large binary model artifacts.')

Packaging as a Python Module

# churn_model/predictor.py
import joblib
import numpy as np
from pathlib import Path

_MODEL_PATH = Path(__file__).parent / 'artifacts' / 'churn_xgboost_latest.pkl'
_PIPELINE = None

def _load():
    global _PIPELINE
    if _PIPELINE is None:
        _PIPELINE = joblib.load(_MODEL_PATH)
    return _PIPELINE

def predict(features: dict) -> dict:
    '''Return churn probability for a single customer feature dict.'''
    pipeline = _load()
    # Convert dict to 2D array in correct feature order
    feature_order = pipeline.feature_names_in_
    X = np.array([[features[col] for col in feature_order]])
    proba = pipeline.predict_proba(X)[0, 1]
    return {'churn_probability': float(proba), 'churn_flag': proba >= 0.4}

Writing the Five-Minute Demo

demo_outline = [
    ('Slide 1 — Problem', '30s',
     'Company loses $2.4M/year to churn. We predict who will churn 90 days in advance.'),
    ('Slide 2 — Data', '30s',
     '120k customers, 3 data sources, 8.3% churn rate.'),
    ('Slide 3 — Model',  '45s',
     'XGBoost tournament winner. AUC=0.88: catches 82% of churners with 67% precision.'),
    ('Slide 4 — Explanation', '60s',
     'Top 3 signals: days since last purchase, support tickets, plan type. SHAP beeswarm.'),
    ('Slide 5 — Business impact', '45s',
     'At 15% retention rate from targeted offers: estimated $360k annual recovery.'),
    ('Slide 6 — Next steps', '30s',
     'A/B test for 90 days. Weekly retraining. Fairness audit before full rollout.')
]
for slide, timing, content in demo_outline:
    print(f'{slide} [{timing}]: {content}')

Testing the Packaged Model

import pytest
import numpy as np
from churn_model.predictor import predict

def test_output_is_probability():
    sample = {'days_since_last_purchase': 45, 'total_spend_90d': 120.5, 'support_tickets': 2}
    result = predict(sample)
    assert 0.0 <= result['churn_probability'] <= 1.0

def test_missing_feature_raises():
    with pytest.raises(KeyError):
        predict({'days_since_last_purchase': 45})  # missing required features

def test_deterministic():
    sample = {'days_since_last_purchase': 10, 'total_spend_90d': 500.0, 'support_tickets': 0}
    r1 = predict(sample)['churn_probability']
    r2 = predict(sample)['churn_probability']
    assert r1 == r2  # model must not use randomness at inference

print('Run: pytest test_predictor.py -v')

Setting Up Monitoring Before Launch

# Prediction logging middleware
import json
import time
from pathlib import Path

LOG_FILE = Path('/var/log/churn_model/predictions.jsonl')
LOG_FILE.parent.mkdir(parents=True, exist_ok=True)

def predict_and_log(features: dict) -> dict:
    result = predict(features)
    log_entry = {
        'timestamp': time.time(),
        'customer_id': features.get('customer_id'),
        'churn_probability': result['churn_probability'],
        'churn_flag': result['churn_flag']
    }
    with open(LOG_FILE, 'a') as f:
        f.write(json.dumps(log_entry) + '\n')
    return result

print('Prediction logging writes to JSONL; ingest into Grafana or BigQuery for dashboarding.')

The Complete Deliverable Checklist

final_checklist = [
    '[x] Trained pipeline serialised with joblib + SHA-256 verified',
    '[x] JSON metadata sidecar with training date, metrics, and feature names',
    '[x] Model card written and reviewed by domain expert',
    '[x] SHAP global importance plot attached to model card',
    '[x] ROC + PR curve + confusion matrix artefacts generated',
    '[x] Fairness audit: demographic parity difference < 0.05',
    '[x] Prediction module with clean API and input validation',
    '[x] 100% test coverage of prediction module (pytest)',
    '[x] Prediction logging middleware deployed',
    '[x] Alert thresholds configured in monitoring dashboard',
    '[x] Stakeholder demo delivered and recorded'
]
for item in final_checklist:
    print(item)

Quick Check

Lesson Recap