With a rich feature set from Phase 2, Phase 3 builds the production pipeline. You will apply a two-stage filter-then-wrapper selection to identify the strongest features, wrap the full preprocessing and selection in a scikit-learn Pipeline with all steps named, serialize the fitted pipeline with a model card, and write the pytest test suite. The deliverable of Phase 3 is the production artifact: a tested, versioned, deployable pipeline that takes raw CricketStream subscriber data and delivers calibrated churn predictions.
Analogy🏏Cricket
🏏 Think of it like cricket: Before the match, a seasoned captain walks the pitch, examines the surface, studies the weather, and watches the opposition warm up — all observation, no decisions yet. Only after the complete pre-match inspection does he decide on team composition, batting order, and field settings. Just as the captain's observations drive every match-day decision, Phase 1 observations drive every feature-engineering and cleaning decision. Just as skipping the inspection leads to poor decisions based on assumptions, skipping EDA leads to features built on misunderstood data. The insight is that Phase 1 is the pre-match inspection that makes every subsequent decision deliberate rather than accidental.
Step 1 — Feature Selection
Apply a two-stage selection pipeline: variance threshold to drop near-constants from the engineered feature set, then RFECV with a random forest to identify the optimal subset. Compare the selected features against the feature inventory from Phase 2 to verify that the selection aligns with domain expectations — a feature the domain strongly motivates should be among the selected, and features the domain predicts are irrelevant should be absent.
Analogy🏏Cricket
🏏 Think of it like cricket: two-stage feature selection is picking your final eleven from a bloated squad in two cuts. First, just as you release players who never get a game — the ones with no distinguishing form — a variance threshold drops near-constant features that carry no information. Then, just as a selector runs trial matches and drops the weakest performer each round until the strongest combination remains, RFECV recursively eliminates the least useful features, cross-validating at each step to find the optimal subset. And just as you sanity-check the chosen eleven against scouting reports — a batsman the pitch report strongly favours had better be in the side — you compare selected features against the Phase 2 inventory to confirm the domain-motivated features survived and the noise was cut. The payoff: a lean, validated feature set that keeps every genuinely predictive signal while shedding dead weight, so the model trains faster and generalises better.
python
import numpy as np
import pandas as pd
from sklearn.feature_selection import VarianceThreshold, RFECV
from sklearn.ensemble import RandomForestClassifier
from sklearn.model_selection import StratifiedKFold
from sklearn.pipeline import Pipeline
# (Assuming train, test, y_train, y_test are available from Phase 2)
# Feature columns: drop identifiers, target, and date columns
drop_cols = ["subscriber_id", "churned", "snapshot_date", "signup_date",
"snapshot_month", "last_ticket_text"]
feature_cols = [c for c in train.columns if c not in drop_cols]
X_train = train[feature_cols].fillna(0)
X_test = test[feature_cols].fillna(0)
y_train = train["churned"]
y_test = test["churned"]
# STAGE 1: Variance threshold (no-target filter)
vt = VarianceThreshold(threshold=0.01)
X_train_vt = vt.fit_transform(X_train)
X_test_vt = vt.transform(X_test)
print(f"After variance threshold: {X_train_vt.shape[1]} of {X_train.shape[1]} features")
# STAGE 2: RFECV (model-aware, CV-tuned)
cv = StratifiedKFold(n_splits=5, shuffle=True, random_state=42)
rfecv = RFECV(RandomForestClassifier(n_estimators=50, random_state=42),
cv=cv, scoring="roc_auc", min_features_to_select=5)
rfecv.fit(X_train_vt, y_train)
print(f"RFECV selected {rfecv.n_features_} features")
# Which original features survived?
vt_surviving = np.array(feature_cols)[vt.get_support()]
final_features = vt_surviving[rfecv.get_support()]
print(f"Final features: {list(final_features)}")
print("Verify: domain-motivated features (engagement_decline, plan_utilisation) should appear.")
Step 2 — Assemble the Production Pipeline
Wrap all preprocessing and the final model into a single named scikit-learn Pipeline. Include the custom PercentileClipper from Module 5, the ColumnTransformer for mixed-type preprocessing, and the RFECV selection before the final classifier. Every step is named for GridSearchCV compatibility and documentation clarity.
Analogy🏏Cricket
🏏 Think of it like cricket: assembling the production Pipeline is finalising the full match-day plan as one named, ordered sequence from warm-up to the closing over. Just as the plan runs in strict order — specialist drills, then the batting order, then the strike bowler for the death — the Pipeline chains the custom PercentileClipper, then the ColumnTransformer for mixed-type preprocessing, then RFECV selection, then the final classifier, each step feeding the next. Just as every player wears a numbered, named jersey so the captain can call any of them precisely during play, every step is explicitly named so GridSearchCV can address and tune it and so the plan reads clearly. And just as the whole eleven takes the field as one unit rather than as loose individuals, all preprocessing and the model are wrapped into a single Pipeline object. The payoff: one named, ordered, tuneable artifact that applies the entire strategy identically every time it's called.
python
from sklearn.compose import ColumnTransformer, make_column_selector
from sklearn.preprocessing import StandardScaler, OneHotEncoder, PowerTransformer
from sklearn.impute import SimpleImputer
from sklearn.linear_model import LogisticRegression
# Re-use the PercentileClipper from Lesson 26 / Module 5
from sklearn.base import BaseEstimator, TransformerMixin
class PercentileClipper(BaseEstimator, TransformerMixin):
def __init__(self, lower_pct=2, upper_pct=98):
self.lower_pct = lower_pct; self.upper_pct = upper_pct
def fit(self, X, y=None):
self.lower_ = np.percentile(X, self.lower_pct, axis=0)
self.upper_ = np.percentile(X, self.upper_pct, axis=0)
return self
def transform(self, X):
return np.clip(X, self.lower_, self.upper_)
numeric_transformer = Pipeline([
("clip", PercentileClipper(lower_pct=2, upper_pct=98)),
("impute", SimpleImputer(strategy="median")),
("scale", StandardScaler()),
])
categorical_transformer = Pipeline([
("impute", SimpleImputer(strategy="most_frequent")),
("encode", OneHotEncoder(handle_unknown="ignore", sparse_output=False)),
])
preprocessor = ColumnTransformer([
("num", numeric_transformer, make_column_selector(dtype_include="number")),
("cat", categorical_transformer, make_column_selector(dtype_include="object")),
])
production_pipeline = Pipeline([
("prep", preprocessor),
("select", RFECV(RandomForestClassifier(n_estimators=50, random_state=42),
cv=StratifiedKFold(5), scoring="roc_auc",
min_features_to_select=5)),
("model", LogisticRegression(C=1.0, max_iter=500, random_state=42)),
])
from sklearn.model_selection import cross_val_score
from sklearn.metrics import roc_auc_score
cv_scores = cross_val_score(production_pipeline, X_train, y_train, cv=5, scoring="roc_auc")
production_pipeline.fit(X_train, y_train)
test_auc = roc_auc_score(y_test, production_pipeline.predict_proba(X_test)[:,1])
print(f"CV AUC: {cv_scores.mean():.3f} +/- {cv_scores.std():.3f}")
print(f"Test AUC: {test_auc:.3f}")
print("Target: > 0.75. If below, revisit Phase 2 feature engineering.")
Step 3 — Serialise with Model Card
Save the fitted pipeline and write a model card JSON containing all the information needed to correctly use, validate, and update the pipeline. The model card is not documentation-as-afterthought but the primary delivery artifact alongside the pipeline file.
Analogy🏏Cricket
🏏 Think of it like cricket: serialising the pipeline with a model card is handing over not just the sealed squad kit but the full team dossier that says how to use it. Just as a squad passed to a new coach is useless without a dossier listing each player's role, fitness limits, expected conditions and when to rotate them, the fitted pipeline file is shipped alongside a model card JSON carrying everything needed to use, validate and update it — input schema, training data, metrics, intended use. Just as a professional would never accept a squad with no paperwork and guess the roles, the card is treated as a primary delivery artifact, not an afterthought scribbled later. And just as the dossier is what lets the next coach field the side correctly, the card is what lets the next engineer deploy the pipeline safely. The payoff: a self-documenting deliverable where the model and its instructions travel together, ready to be handed off without loss of knowledge.
python
import joblib, json, sklearn, sys
from pathlib import Path
from datetime import datetime
Path("models").mkdir(exist_ok=True)
# Save fitted pipeline
joblib.dump(production_pipeline, "models/churn_pipeline.pkl", compress=3)
# Write model card
model_card = {
"model_name": "CricketStream Churn Predictor",
"version": "1.0.0",
"saved_at": datetime.now().isoformat(),
"sklearn_version": sklearn.__version__,
"python_version": sys.version.split()[0],
"cv_auc": round(cv_scores.mean(), 4),
"cv_auc_std": round(cv_scores.std(), 4),
"test_auc": round(test_auc, 4),
"n_input_features": X_train.shape[1],
"feature_names": list(X_train.columns),
"prediction_target": "churned within 30 days",
"train_split_boundary": str(train["snapshot_date"].max().date()),
"leakage_prevention": "time-ordered split; all aggregations shifted; transformers fit on train only",
"top_features": list(final_features[:10]) if "final_features" in dir() else [],
}
Path("models/churn_pipeline_card.json").write_text(json.dumps(model_card, indent=2))
print("Saved: models/churn_pipeline.pkl + models/churn_pipeline_card.json")
print(json.dumps(model_card, indent=2))
Step 4 — Write the Test Suite
Write the pytest test suite for the production pipeline, covering the five critical tests: output shape, binary label validity, missing-value handling, unseen-category handling, and a schema-rejection test. Save it to tests/test_churn_pipeline.py and run it to confirm all tests pass before declaring the pipeline production-ready.
Analogy🏏Cricket
🏏 Think of it like cricket: writing the pytest suite is scripting the mandatory pre-match checks that certify the eleven are fit before they can take the field. Just as officials verify each essential condition — the right number of players, valid kit, cleared fitness, correct positions — your five tests verify output shape, binary-label validity, missing-value handling, unseen-category handling, and schema rejection. Just as a schema-rejection check is like the umpire turning away a twelfth man who tries to bat, your test confirms the pipeline refuses malformed input rather than silently misbehaving. And just as no captain declares the side ready until every check clears, you save the suite to tests/test_churn_pipeline.py and run it, declaring the pipeline production-ready only once all tests pass. The payoff: automated proof on the critical dimensions, so the pipeline is certified fit before it ever takes the field in production.
python
# tests/test_churn_pipeline.py
import pytest, numpy as np, pandas as pd, joblib, json
from pathlib import Path
@pytest.fixture(scope="module")
def pipeline():
return joblib.load("models/churn_pipeline.pkl")
@pytest.fixture(scope="module")
def valid_input():
rng = np.random.default_rng(42)
n = 50
return pd.DataFrame({
"weekly_watch_mins": rng.exponential(60, n),
"total_sessions": rng.integers(1, 20, n).astype(float),
"content_categories_watched": rng.integers(1, 8, n).astype(float),
"live_vs_vod_ratio": rng.uniform(0, 1, n),
"plan_tier": rng.choice(["free","standard","premium"], n),
"tenure_days": rng.integers(1, 1000, n).astype(float),
"push_notifications_clicked": rng.integers(0, 10, n).astype(float),
"app_opens": rng.integers(0, 30, n).astype(float),
"support_tickets_open": rng.integers(0, 3, n).astype(float),
"last_ticket_sentiment": rng.choice(["positive","neutral","negative","none"], n),
"region": rng.choice(["North","South","East","West"], n),
})
def test_output_shape(pipeline, valid_input):
preds = pipeline.predict(valid_input)
assert preds.shape == (len(valid_input),)
def test_binary_labels(pipeline, valid_input):
preds = pipeline.predict(valid_input)
assert set(preds).issubset({0, 1})
def test_proba_valid(pipeline, valid_input):
proba = pipeline.predict_proba(valid_input)
assert proba.shape[1] == 2
assert np.allclose(proba.sum(axis=1), 1.0)
def test_missing_numeric_handled(pipeline, valid_input):
data = valid_input.copy()
data.loc[0, "weekly_watch_mins"] = np.nan
assert pipeline.predict(data).shape == (len(data),)
def test_unseen_category(pipeline, valid_input):
data = valid_input.copy()
data.loc[0, "plan_tier"] = "enterprise"
assert pipeline.predict(data).shape == (len(data),)
# Run: pytest tests/test_churn_pipeline.py -v
print("Test suite written to tests/test_churn_pipeline.py")
print("Run: pytest tests/test_churn_pipeline.py -v -> expect 5 PASSED")
Warning: If the test AUC is below 0.65, revisit Phase 2 before proceeding to Phase 4. Low AUC at this stage means either the feature engineering missed important signals or a leakage issue inflated the cross-validation score, making the test set performance appear to regress. Check the engagement_decline feature (did the shift apply correctly?), the text vectoriser (was it fit before or after the split?), and the train-test split boundary (are test snapshots correctly excluded from all fitting?). A sub-0.65 AUC is a signal to debug, not to paper over with model tuning.
- The two-stage filter-then-RFECV selection identifies the strongest features from the engineered set efficiently and honestly.
- The production Pipeline names every step for documentation and GridSearchCV compatibility, wrapping preprocessing, selection, and the model.
- The serialised pipeline and model card are the primary deliverables — the tested, versioned, deployable artifact.
- The pytest suite's five tests verify the pipeline satisfies its critical contracts before it is declared production-ready.
- A test AUC below 0.65 signals a feature-engineering or leakage issue to debug, not a model-architecture problem to tune around.