Deployment and MLOps

Week 15 of 16 · Production · 7 days

Full curriculum
Week 15 · Production

Deployment and MLOps

Week 15 · Day 1 of 7

Packaging a Model

The artefact, the contract, pinned versions and a smoke test

By 998 words

A model that lives in a notebook is not a product. Everything from here is about the distance between the two, and almost none of it is about machine learning.

What actually has to be saved

Model artefact: Not the estimator. The estimator plus every transformation applied to the data before it, plus the description of what the inputs are meant to be, plus enough version information to reproduce all of it. Save less and you have saved a puzzle.
import numpy as np
import pandas as pd

df = pd.read_csv('customers.csv').drop_duplicates().copy()
df['contract'] = df['contract'].str.strip().str.title()
df['total_charges'] = pd.to_numeric(df['total_charges'], errors='coerce')

NUM = ['tenure_months', 'support_calls', 'monthly_charges']
CAT = ['contract', 'internet_service', 'payment_method', 'has_dependents']
from sklearn.compose import ColumnTransformer
from sklearn.pipeline import Pipeline
from sklearn.impute import SimpleImputer
from sklearn.preprocessing import StandardScaler, OneHotEncoder
from sklearn.linear_model import LogisticRegression
from sklearn.model_selection import train_test_split

prep = ColumnTransformer([
('num', Pipeline([('i', SimpleImputer(strategy='median')),
('s', StandardScaler())]), NUM),
('cat', Pipeline([('i', SimpleImputer(strategy='most_frequent')),
('o', OneHotEncoder(handle_unknown='ignore'))]), CAT),
])
model = Pipeline([('prep', prep),
('clf', LogisticRegression(max_iter=2000))])

X, y = df[NUM + CAT], df['churned']
X_tr, X_te, y_tr, y_te = train_test_split(X, y, test_size=0.25,
stratify=y, random_state=42)
model.fit(X_tr, y_tr)
import joblib
import json
import sklearn
import sys
from sklearn.metrics import roc_auc_score

joblib.dump(model, 'churn_model.joblib')

contract = {
'model_version': '1.0.0',
'trained_at': '2026-08-08',
'trained_rows': int(len(X_tr)),
'features': {'numeric': NUM, 'categorical': CAT},
'categories': {c: sorted(df[c].dropna().unique().tolist())
for c in CAT},
'ranges': {c: [float(df[c].min()), float(df[c].max())] for c in NUM},
'threshold': 0.35,
'test_auc': round(float(roc_auc_score(
y_te, model.predict_proba(X_te)[:, 1])), 4),
'python': sys.version.split()[0],
'sklearn': sklearn.__version__,
}
json.dump(contract, open('churn_contract.json', 'w'), indent=2)
print(json.dumps(contract, indent=2)[:760])
{
"model_version": "1.0.0",
"trained_at": "2026-08-08",
"trained_rows": 2250,
"features": {
"numeric": [
"tenure_months",
"support_calls",
"monthly_charges"
],
"categorical": [
"contract",
"internet_service",
"payment_method",
"has_dependents"
]
},
"categories": {
"contract": [
"Month-To-Month",
"One Year",
"Two Year"
],
"internet_service": [
"DSL",
"Fibre optic",
"No internet"
],
"payment_method": [
"Bank transfer",
"Credit card",
"Electronic check",
... (13 more lines)

The contract is the part people leave out

The .joblib file knows how to turn a dataframe into a prediction. It does not know that contract has exactly three valid values, that tenure_months was never above 72 in training, that the agreed threshold is 0.35 rather than 0.5, or which version of the model produced a given row in your logs. Six months later, every one of those is a question somebody will ask, and the answer needs to be in a file rather than in your memory.

Pinning the environment

import numpy as np
import pandas as pd

df = pd.read_csv('customers.csv').drop_duplicates().copy()
df['contract'] = df['contract'].str.strip().str.title()
df['total_charges'] = pd.to_numeric(df['total_charges'], errors='coerce')

NUM = ['tenure_months', 'support_calls', 'monthly_charges']
CAT = ['contract', 'internet_service', 'payment_method', 'has_dependents']
from sklearn.compose import ColumnTransformer
from sklearn.pipeline import Pipeline
from sklearn.impute import SimpleImputer
from sklearn.preprocessing import StandardScaler, OneHotEncoder
from sklearn.linear_model import LogisticRegression
from sklearn.model_selection import train_test_split

prep = ColumnTransformer([
('num', Pipeline([('i', SimpleImputer(strategy='median')),
('s', StandardScaler())]), NUM),
('cat', Pipeline([('i', SimpleImputer(strategy='most_frequent')),
('o', OneHotEncoder(handle_unknown='ignore'))]), CAT),
])
model = Pipeline([('prep', prep),
('clf', LogisticRegression(max_iter=2000))])

X, y = df[NUM + CAT], df['churned']
X_tr, X_te, y_tr, y_te = train_test_split(X, y, test_size=0.25,
stratify=y, random_state=42)
model.fit(X_tr, y_tr)
import joblib
import sklearn

loaded = joblib.load('churn_model.joblib')
print('loaded a %s' % type(loaded).__name__)
print('steps: %s' % [name for name, _ in loaded.steps])
print('fitted with sklearn %s' % sklearn.__version__)

same = (loaded.predict_proba(X_te)[:, 1]
== model.predict_proba(X_te)[:, 1]).all()
print('reload predicts identically: %s' % bool(same))
loaded a Pipeline
steps: ['prep', 'clf']
fitted with sklearn 1.8.0
reload predicts identically: True
# The versions are part of the artefact, not a detail of your laptop.
pip freeze > requirements.txt

# Or, better, record only what you asked for and let the tool resolve
# the rest reproducibly:
pip install pip-tools && pip-compile requirements.in

A pickle is not a portable format

joblib uses pickle underneath, which stores references to Python classes. Load it under a different scikit-learn version and you get, in ascending order of unpleasantness: a warning, an exception, or a model that loads cleanly and predicts subtly differently. Pin the version, load in the same environment you trained in, and add a smoke test that checks a handful of known inputs still produce known outputs. That test is the only thing that catches the third case.

import numpy as np
import pandas as pd

df = pd.read_csv('customers.csv').drop_duplicates().copy()
df['contract'] = df['contract'].str.strip().str.title()
df['total_charges'] = pd.to_numeric(df['total_charges'], errors='coerce')

NUM = ['tenure_months', 'support_calls', 'monthly_charges']
CAT = ['contract', 'internet_service', 'payment_method', 'has_dependents']
from sklearn.compose import ColumnTransformer
from sklearn.pipeline import Pipeline
from sklearn.impute import SimpleImputer
from sklearn.preprocessing import StandardScaler, OneHotEncoder
from sklearn.linear_model import LogisticRegression
from sklearn.model_selection import train_test_split

prep = ColumnTransformer([
('num', Pipeline([('i', SimpleImputer(strategy='median')),
('s', StandardScaler())]), NUM),
('cat', Pipeline([('i', SimpleImputer(strategy='most_frequent')),
('o', OneHotEncoder(handle_unknown='ignore'))]), CAT),
])
model = Pipeline([('prep', prep),
('clf', LogisticRegression(max_iter=2000))])

X, y = df[NUM + CAT], df['churned']
X_tr, X_te, y_tr, y_te = train_test_split(X, y, test_size=0.25,
stratify=y, random_state=42)
model.fit(X_tr, y_tr)
import joblib
import json

# The smoke test the warning above asks for: freeze a few predictions
# now, and compare against them on every load, forever.
sample = X_te.head(5)
expected = [round(float(p), 6)
for p in model.predict_proba(sample)[:, 1]]
json.dump({'rows': sample.to_dict('records'), 'expected': expected},
open('smoke_test.json', 'w'), indent=2)

fixture = json.load(open('smoke_test.json'))
reloaded = joblib.load('churn_model.joblib')
got = [round(float(p), 6) for p in
reloaded.predict_proba(pd.DataFrame(fixture['rows']))[:, 1]]
print('expected %s' % fixture['expected'])
print('got %s' % got)
print('smoke test passes: %s' % (got == fixture['expected']))
expected [0.530642, 0.16512, 0.322593, 0.205547, 0.568866]
got [0.530642, 0.16512, 0.322593, 0.205547, 0.568866]
smoke test passes: True

Day 1 takeaway

Ship the whole pipeline in one artefact, a contract file describing the inputs and the agreed threshold, pinned versions, and a smoke test with known inputs and known outputs. The smoke test is what tells you a dependency upgrade changed your predictions.
Week 15 · Day 2 of 7

A Prediction Service

FastAPI, request schemas, and where the latency actually goes

By 879 words

A model behind an HTTP endpoint, which is how most of them are consumed. The service below is written to a file and then genuinely called, so every response on this page came back over a real request cycle.

The service

Save this as service.py. It is loaded and called for real further down the page.

import json
import joblib
import pandas as pd
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field, ValidationError

MODEL = joblib.load("churn_model.joblib")
CONTRACT = json.load(open("churn_contract.json"))


class Customer(BaseModel):
"""The request schema. Pydantic rejects anything that does not match,
before a single row reaches the model."""
tenure_months: int = Field(ge=0, le=120)
support_calls: int = Field(ge=0, le=100)
monthly_charges: float = Field(ge=0, le=500)
contract: str
internet_service: str
payment_method: str
has_dependents: str


app = FastAPI(title="churn-scorer", version=CONTRACT["model_version"])


@app.get("/health")
def health():
return {"status": "ok", "model_version": CONTRACT["model_version"],
"trained_at": CONTRACT["trained_at"]}


@app.post("/predict")
def predict(customer: Customer):
row = pd.DataFrame([customer.model_dump()])
for column in ("contract", "internet_service", "payment_method"):
known = CONTRACT["categories"][column]
if row.loc[0, column] not in known:
raise HTTPException(
status_code=422,
detail="unknown %s: %r (known: %s)"
% (column, row.loc[0, column], known))
p = float(MODEL.predict_proba(row)[0, 1])
return {"churn_probability": round(p, 4),
"decision": "flag" if p >= CONTRACT["threshold"] else "keep",
"threshold": CONTRACT["threshold"],
"model_version": CONTRACT["model_version"]}
Pydantic model: a class whose annotations are enforced at runtime. Field(ge=0, le=120) means a request with tenure_months = -4 is rejected by the framework with a 422 before your code runs. Validation you write yourself is validation you forget to write.

What a request and response look like

SERVICE = '''import json
import joblib
import pandas as pd
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field, ValidationError

MODEL = joblib.load("churn_model.joblib")
CONTRACT = json.load(open("churn_contract.json"))


class Customer(BaseModel):
"""The request schema. Pydantic rejects anything that does not match,
before a single row reaches the model."""
tenure_months: int = Field(ge=0, le=120)
support_calls: int = Field(ge=0, le=100)
monthly_charges: float = Field(ge=0, le=500)
contract: str
internet_service: str
payment_method: str
has_dependents: str


app = FastAPI(title="churn-scorer", version=CONTRACT["model_version"])


@app.get("/health")
def health():
return {"status": "ok", "model_version": CONTRACT["model_version"],
"trained_at": CONTRACT["trained_at"]}


@app.post("/predict")
def predict(customer: Customer):
row = pd.DataFrame([customer.model_dump()])
for column in ("contract", "internet_service", "payment_method"):
known = CONTRACT["categories"][column]
if row.loc[0, column] not in known:
raise HTTPException(
status_code=422,
detail="unknown %s: %r (known: %s)"
% (column, row.loc[0, column], known))
p = float(MODEL.predict_proba(row)[0, 1])
return {"churn_probability": round(p, 4),
"decision": "flag" if p >= CONTRACT["threshold"] else "keep",
"threshold": CONTRACT["threshold"],
"model_version": CONTRACT["model_version"]}
'''
import io, sys, json
io.open('service.py', 'w', encoding='utf-8').write(SERVICE)
sys.path.insert(0, '.')

from fastapi.testclient import TestClient
import service
client = TestClient(service.app)

print('GET /health ->', client.get('/health').status_code)
print(json.dumps(client.get('/health').json(), indent=2))

at_risk = {'tenure_months': 2, 'support_calls': 5,
'monthly_charges': 94.2, 'contract': 'Month-To-Month',
'internet_service': 'Fibre optic',
'payment_method': 'Electronic check',
'has_dependents': 'No'}
settled = dict(at_risk, tenure_months=60, support_calls=0,
contract='Two Year', monthly_charges=42.0)

for label, payload in [('new, unhappy', at_risk),
('long-standing', settled)]:
r = client.post('/predict', json=payload)
print('\n%s -> %d' % (label, r.status_code))
print(json.dumps(r.json(), indent=2))
GET /health -> 200
{
"status": "ok",
"model_version": "1.0.0",
"trained_at": "2026-08-08"
}

new, unhappy -> 200
{
"churn_probability": 0.8896,
"decision": "flag",
"threshold": 0.35,
"model_version": "1.0.0"
}

long-standing -> 200
{
"churn_probability": 0.004,
"decision": "keep",
"threshold": 0.35,
"model_version": "1.0.0"
}

Single or batch?

SERVICE = '''import json
import joblib
import pandas as pd
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field, ValidationError

MODEL = joblib.load("churn_model.joblib")
CONTRACT = json.load(open("churn_contract.json"))


class Customer(BaseModel):
"""The request schema. Pydantic rejects anything that does not match,
before a single row reaches the model."""
tenure_months: int = Field(ge=0, le=120)
support_calls: int = Field(ge=0, le=100)
monthly_charges: float = Field(ge=0, le=500)
contract: str
internet_service: str
payment_method: str
has_dependents: str


app = FastAPI(title="churn-scorer", version=CONTRACT["model_version"])


@app.get("/health")
def health():
return {"status": "ok", "model_version": CONTRACT["model_version"],
"trained_at": CONTRACT["trained_at"]}


@app.post("/predict")
def predict(customer: Customer):
row = pd.DataFrame([customer.model_dump()])
for column in ("contract", "internet_service", "payment_method"):
known = CONTRACT["categories"][column]
if row.loc[0, column] not in known:
raise HTTPException(
status_code=422,
detail="unknown %s: %r (known: %s)"
% (column, row.loc[0, column], known))
p = float(MODEL.predict_proba(row)[0, 1])
return {"churn_probability": round(p, 4),
"decision": "flag" if p >= CONTRACT["threshold"] else "keep",
"threshold": CONTRACT["threshold"],
"model_version": CONTRACT["model_version"]}
'''
import io, sys, json
io.open('service.py', 'w', encoding='utf-8').write(SERVICE)
sys.path.insert(0, '.')

from fastapi.testclient import TestClient
import service
client = TestClient(service.app)
import time

payload = {'tenure_months': 2, 'support_calls': 5,
'monthly_charges': 94.2, 'contract': 'Month-To-Month',
'internet_service': 'Fibre optic',
'payment_method': 'Electronic check',
'has_dependents': 'No'}

client.post('/predict', json=payload) # warm up
t = time.time()
N = 100
for _ in range(N):
client.post('/predict', json=payload)
per_call = (time.time() - t) / N
print('one row at a time: %.1f ms per prediction' % (1000 * per_call))

import pandas as pd
rows = pd.DataFrame([payload] * N)
t = time.time()
service.MODEL.predict_proba(rows)
batched = (time.time() - t) / N
print('%d rows in one call: %.3f ms per prediction' % (N, 1000 * batched))
print('\nratio %.0fx' % (per_call / batched))
one row at a time: 5.3 ms per prediction
100 rows in one call: 0.030 ms per prediction

ratio 176x

Almost none of that time is the model

The per-request cost is HTTP parsing, validation, building a one-row dataframe and serialising a response. The arithmetic the model does is a rounding error next to it. Two consequences: offer a batch endpoint if your caller has many rows, because it is very nearly free; and do not optimise the model for latency until you have measured where the latency is.

Day 2 takeaway

Load the artefact once at import, not per request. Declare the request schema so the framework rejects malformed input before your code sees it. Expose a health endpoint that reports the model version. And offer batch scoring, because the overhead per row collapses when rows are grouped.
Week 15 · Day 3 of 7

Validation and Failing Loudly

The bad inputs that raise, and the dangerous one that does not

By 1048 words

The interesting failures are not the ones that raise. They are the ones where the service returns 200 and a number that means nothing.

The three kinds of bad input

SERVICE = '''import json
import joblib
import pandas as pd
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field, ValidationError

MODEL = joblib.load("churn_model.joblib")
CONTRACT = json.load(open("churn_contract.json"))


class Customer(BaseModel):
"""The request schema. Pydantic rejects anything that does not match,
before a single row reaches the model."""
tenure_months: int = Field(ge=0, le=120)
support_calls: int = Field(ge=0, le=100)
monthly_charges: float = Field(ge=0, le=500)
contract: str
internet_service: str
payment_method: str
has_dependents: str


app = FastAPI(title="churn-scorer", version=CONTRACT["model_version"])


@app.get("/health")
def health():
return {"status": "ok", "model_version": CONTRACT["model_version"],
"trained_at": CONTRACT["trained_at"]}


@app.post("/predict")
def predict(customer: Customer):
row = pd.DataFrame([customer.model_dump()])
for column in ("contract", "internet_service", "payment_method"):
known = CONTRACT["categories"][column]
if row.loc[0, column] not in known:
raise HTTPException(
status_code=422,
detail="unknown %s: %r (known: %s)"
% (column, row.loc[0, column], known))
p = float(MODEL.predict_proba(row)[0, 1])
return {"churn_probability": round(p, 4),
"decision": "flag" if p >= CONTRACT["threshold"] else "keep",
"threshold": CONTRACT["threshold"],
"model_version": CONTRACT["model_version"]}
'''
import io, sys, json
io.open('service.py', 'w', encoding='utf-8').write(SERVICE)
sys.path.insert(0, '.')

from fastapi.testclient import TestClient
import service
client = TestClient(service.app)

good = {'tenure_months': 12, 'support_calls': 1,
'monthly_charges': 70.0, 'contract': 'One Year',
'internet_service': 'DSL',
'payment_method': 'Bank transfer', 'has_dependents': 'Yes'}

cases = [
('valid request', good),
('missing a field', {k: v for k, v in good.items()
if k != 'monthly_charges'}),
('wrong type', dict(good, tenure_months='twelve')),
('out of range', dict(good, tenure_months=-4)),
('unknown category', dict(good, contract='Pay As You Go')),
]
for label, payload in cases:
r = client.post('/predict', json=payload)
body = r.json()
detail = body.get('detail', body)
if isinstance(detail, list):
detail = '%s: %s' % (detail[0].get('loc'), detail[0].get('msg'))
print('%-18s %d %s' % (label, r.status_code, str(detail)[:64]))
valid request 200 {'churn_probability': 0.1575, 'decision': 'keep', 'threshold': 0
missing a field 422 ['body', 'monthly_charges']: Field required
wrong type 422 ['body', 'tenure_months']: Input should be a valid integer, unab
out of range 422 ['body', 'tenure_months']: Input should be greater than or equal
unknown category 422 unknown contract: 'Pay As You Go' (known: ['Month-To-Month', 'On

The one that does not raise

A category the model has never seen is the dangerous case, because OneHotEncoder(handle_unknown='ignore'), which every week of this course has used, for good reasons, turns it silently into a row of zeros and predicts anyway.

import numpy as np
import pandas as pd

df = pd.read_csv('customers.csv').drop_duplicates().copy()
df['contract'] = df['contract'].str.strip().str.title()
df['total_charges'] = pd.to_numeric(df['total_charges'], errors='coerce')

NUM = ['tenure_months', 'support_calls', 'monthly_charges']
CAT = ['contract', 'internet_service', 'payment_method', 'has_dependents']
from sklearn.compose import ColumnTransformer
from sklearn.pipeline import Pipeline
from sklearn.impute import SimpleImputer
from sklearn.preprocessing import StandardScaler, OneHotEncoder
from sklearn.linear_model import LogisticRegression
from sklearn.model_selection import train_test_split

prep = ColumnTransformer([
('num', Pipeline([('i', SimpleImputer(strategy='median')),
('s', StandardScaler())]), NUM),
('cat', Pipeline([('i', SimpleImputer(strategy='most_frequent')),
('o', OneHotEncoder(handle_unknown='ignore'))]), CAT),
])
model = Pipeline([('prep', prep),
('clf', LogisticRegression(max_iter=2000))])

X, y = df[NUM + CAT], df['churned']
X_tr, X_te, y_tr, y_te = train_test_split(X, y, test_size=0.25,
stratify=y, random_state=42)
model.fit(X_tr, y_tr)
import joblib

model = joblib.load('churn_model.joblib')
base = X_te.iloc[[0]].copy()
print('the row as it stands: %.4f'
% model.predict_proba(base)[0, 1])

for value in ['Two Year', 'Pay As You Go', 'PAYG', '']:
row = base.copy()
row['contract'] = value
print('contract=%-16r %.4f (no error raised)'
% (value, model.predict_proba(row)[0, 1]))
the row as it stands: 0.5306
contract='Two Year' 0.0665 (no error raised)
contract='Pay As You Go' 0.2338 (no error raised)
contract='PAYG' 0.2338 (no error raised)
contract='' 0.2338 (no error raised)

Silent is worse than wrong

Every one of those returned a plausible number. If a new contract type is launched and nobody tells the model's owner, the service keeps answering, scoring every customer on that plan as though their contract field were blank, and nothing in any log looks unusual. That is why the service checks the value against the contract file and returns a 422. Rejecting a request is a bad day; answering it wrongly for eight months is a different kind of day.

Range checks are not the same as schema checks

import numpy as np
import pandas as pd

df = pd.read_csv('customers.csv').drop_duplicates().copy()
df['contract'] = df['contract'].str.strip().str.title()
df['total_charges'] = pd.to_numeric(df['total_charges'], errors='coerce')

NUM = ['tenure_months', 'support_calls', 'monthly_charges']
CAT = ['contract', 'internet_service', 'payment_method', 'has_dependents']
from sklearn.compose import ColumnTransformer
from sklearn.pipeline import Pipeline
from sklearn.impute import SimpleImputer
from sklearn.preprocessing import StandardScaler, OneHotEncoder
from sklearn.linear_model import LogisticRegression
from sklearn.model_selection import train_test_split

prep = ColumnTransformer([
('num', Pipeline([('i', SimpleImputer(strategy='median')),
('s', StandardScaler())]), NUM),
('cat', Pipeline([('i', SimpleImputer(strategy='most_frequent')),
('o', OneHotEncoder(handle_unknown='ignore'))]), CAT),
])
model = Pipeline([('prep', prep),
('clf', LogisticRegression(max_iter=2000))])

X, y = df[NUM + CAT], df['churned']
X_tr, X_te, y_tr, y_te = train_test_split(X, y, test_size=0.25,
stratify=y, random_state=42)
model.fit(X_tr, y_tr)
import joblib
import json

model = joblib.load('churn_model.joblib')
contract = json.load(open('churn_contract.json'))
lo, hi = contract['ranges']['tenure_months']
print('training range for tenure_months: %.0f to %.0f' % (lo, hi))

row = X_te.iloc[[0]].copy()
for months in [12, 72, 96, 240]:
row['tenure_months'] = months
p = model.predict_proba(row)[0, 1]
flag = '' if lo <= months <= hi else ' <-- outside training range'
print('tenure=%-5d %.4f%s' % (months, p, flag))
training range for tenure_months: 1 to 72
tenure=12 0.5582
tenure=72 0.0429
tenure=96 0.0117 <-- outside training range
tenure=240 0.0000 <-- outside training range

The schema accepts anything up to 120 months because that is a plausible tenure. The model has never seen beyond 72, so past that it is extrapolating along a straight line in log-odds and there is nothing to stop it. Validate against the schema to reject nonsense; validate against the training ranges to decide whether to trust the answer.

SituationStatusReturn
Field missing or wrong type422Which field, and why
Value outside the schema's range422The permitted range
Category never seen in training422The known values
Inside the schema, outside the training range200The prediction, plus a warning flag
Model file missing or corrupt503Fail the health check, do not answer
Anything unexpected500A request id; log the traceback

Day 3 takeaway

Let the schema reject malformed requests. Check categories against the contract yourself, because the encoder that made your training robust is the same one that makes your service silently wrong. Distinguish input that is invalid from input that is merely unfamiliar: reject the first, answer the second with a flag.
Week 15 · Day 4 of 7

Monitoring and Drift

Logging, PSI, category mix and watching your own output move

By 1093 words

The model is live and answering. The question now is how you find out it has stopped being any good, given that nobody will tell you.

Log the prediction, not just the answer

import numpy as np
import pandas as pd

df = pd.read_csv('customers.csv').drop_duplicates().copy()
df['contract'] = df['contract'].str.strip().str.title()
df['total_charges'] = pd.to_numeric(df['total_charges'], errors='coerce')

NUM = ['tenure_months', 'support_calls', 'monthly_charges']
CAT = ['contract', 'internet_service', 'payment_method', 'has_dependents']
from sklearn.compose import ColumnTransformer
from sklearn.pipeline import Pipeline
from sklearn.impute import SimpleImputer
from sklearn.preprocessing import StandardScaler, OneHotEncoder
from sklearn.linear_model import LogisticRegression
from sklearn.model_selection import train_test_split

prep = ColumnTransformer([
('num', Pipeline([('i', SimpleImputer(strategy='median')),
('s', StandardScaler())]), NUM),
('cat', Pipeline([('i', SimpleImputer(strategy='most_frequent')),
('o', OneHotEncoder(handle_unknown='ignore'))]), CAT),
])
model = Pipeline([('prep', prep),
('clf', LogisticRegression(max_iter=2000))])

X, y = df[NUM + CAT], df['churned']
X_tr, X_te, y_tr, y_te = train_test_split(X, y, test_size=0.25,
stratify=y, random_state=42)
model.fit(X_tr, y_tr)
import hashlib
import joblib
import json

model = joblib.load('churn_model.joblib')
contract = json.load(open('churn_contract.json'))

def log_line(request_id, row, proba):
return json.dumps({
'request_id': request_id,
'model_version': contract['model_version'],
'threshold': contract['threshold'],
'probability': round(float(proba), 4),
'decision': 'flag' if proba >= contract['threshold'] else 'keep',
# the inputs, so drift is measurable later; hash anything
# identifying rather than storing it
'features': {k: (round(v, 2) if isinstance(v, float) else v)
for k, v in row.items()},
})

for i in range(3):
row = X_te.iloc[i].to_dict()
p = model.predict_proba(X_te.iloc[[i]])[0, 1]
print(log_line('req-%03d' % i, row, p))
{"request_id": "req-000", "model_version": "1.0.0", "threshold": 0.35, "probability": 0.5306, "decision": "flag", "features": {"tenure_months": 14, "support_calls": 3, "monthly_charges": 73.35, "contract": "Month-To-Month", "internet_service": "Fibre optic", "payment_method": "Credit card", "has_dependents": "No"}}
{"request_id": "req-001", "model_version": "1.0.0", "threshold": 0.35, "probability": 0.1651, "decision": "keep", "features": {"tenure_months": 26, "support_calls": 1, "monthly_charges": 15.0, "contract": "Month-To-Month", "internet_service": "No internet", "payment_method": "Electronic check", "has_dependents": "No"}}
{"request_id": "req-002", "model_version": "1.0.0", "threshold": 0.35, "probability": 0.3226, "decision": "keep", "features": {"tenure_months": 14, "support_calls": 1, "monthly_charges": 53.11, "contract": "Month-To-Month", "internet_service": "DSL", "payment_method": "Bank transfer", "has_dependents": "No"}}

Log the features or you cannot diagnose anything

A log of probabilities tells you the distribution of your output moved. A log of probabilities and inputs tells you which input moved and therefore what happened in the business. The second costs a few hundred bytes per request and is the difference between "something changed in March" and "fibre pricing changed in March".

Data drift

Population Stability Index: Bucket a feature using the training distribution, compare the proportion of live traffic falling in each bucket against the training proportion, and sum (live − train) × ln(live / train). Conventionally: below 0.1 is stable, 0.1 to 0.25 warrants attention, above 0.25 is a material shift.
import numpy as np
import pandas as pd

df = pd.read_csv('customers.csv').drop_duplicates().copy()
df['contract'] = df['contract'].str.strip().str.title()
df['total_charges'] = pd.to_numeric(df['total_charges'], errors='coerce')

NUM = ['tenure_months', 'support_calls', 'monthly_charges']
CAT = ['contract', 'internet_service', 'payment_method', 'has_dependents']
from sklearn.compose import ColumnTransformer
from sklearn.pipeline import Pipeline
from sklearn.impute import SimpleImputer
from sklearn.preprocessing import StandardScaler, OneHotEncoder
from sklearn.linear_model import LogisticRegression
from sklearn.model_selection import train_test_split

prep = ColumnTransformer([
('num', Pipeline([('i', SimpleImputer(strategy='median')),
('s', StandardScaler())]), NUM),
('cat', Pipeline([('i', SimpleImputer(strategy='most_frequent')),
('o', OneHotEncoder(handle_unknown='ignore'))]), CAT),
])
model = Pipeline([('prep', prep),
('clf', LogisticRegression(max_iter=2000))])

X, y = df[NUM + CAT], df['churned']
X_tr, X_te, y_tr, y_te = train_test_split(X, y, test_size=0.25,
stratify=y, random_state=42)
model.fit(X_tr, y_tr)
import numpy as np

def psi(reference, live, buckets=10):
edges = np.quantile(reference, np.linspace(0, 1, buckets + 1))
edges[0], edges[-1] = -np.inf, np.inf
a = np.histogram(reference, edges)[0] / len(reference)
b = np.histogram(live, edges)[0] / len(live)
a, b = np.clip(a, 1e-6, None), np.clip(b, 1e-6, None)
return float(((b - a) * np.log(b / a)).sum())

rng = np.random.default_rng(0)
reference = X_tr['monthly_charges'].dropna().to_numpy()

print('%-28s %8s %s' % ('live traffic', 'PSI', 'verdict'))
for label, live in [
('same distribution', rng.choice(reference, 800)),
('prices up 5%', rng.choice(reference, 800) * 1.05),
('prices up 20%', rng.choice(reference, 800) * 1.20),
('only premium customers',
rng.choice(reference[reference > np.median(reference)], 800))]:
value = psi(reference, live)
verdict = ('stable' if value < 0.1 else
'investigate' if value < 0.25 else 'material shift')
print('%-28s %8.4f %s' % (label, value, verdict))
live traffic PSI verdict
same distribution 0.0180 stable
prices up 5% 0.0688 stable
prices up 20% 0.5386 material shift
only premium customers 6.1024 material shift

Categorical drift, and drift in the output itself

import numpy as np
import pandas as pd

df = pd.read_csv('customers.csv').drop_duplicates().copy()
df['contract'] = df['contract'].str.strip().str.title()
df['total_charges'] = pd.to_numeric(df['total_charges'], errors='coerce')

NUM = ['tenure_months', 'support_calls', 'monthly_charges']
CAT = ['contract', 'internet_service', 'payment_method', 'has_dependents']
from sklearn.compose import ColumnTransformer
from sklearn.pipeline import Pipeline
from sklearn.impute import SimpleImputer
from sklearn.preprocessing import StandardScaler, OneHotEncoder
from sklearn.linear_model import LogisticRegression
from sklearn.model_selection import train_test_split

prep = ColumnTransformer([
('num', Pipeline([('i', SimpleImputer(strategy='median')),
('s', StandardScaler())]), NUM),
('cat', Pipeline([('i', SimpleImputer(strategy='most_frequent')),
('o', OneHotEncoder(handle_unknown='ignore'))]), CAT),
])
model = Pipeline([('prep', prep),
('clf', LogisticRegression(max_iter=2000))])

X, y = df[NUM + CAT], df['churned']
X_tr, X_te, y_tr, y_te = train_test_split(X, y, test_size=0.25,
stratify=y, random_state=42)
model.fit(X_tr, y_tr)
import numpy as np
import joblib

model = joblib.load('churn_model.joblib')
rng = np.random.default_rng(0)

train_mix = X_tr['contract'].value_counts(normalize=True)
print('training contract mix:')
print(train_mix.round(3).to_string())

# A marketing push sells monthly plans to everybody.
shifted = X_te.copy()
flip = rng.random(len(shifted)) < 0.5
shifted.loc[flip, 'contract'] = 'Month-To-Month'

live_mix = shifted['contract'].value_counts(normalize=True)
print('\nlive contract mix:')
print(live_mix.round(3).to_string())

before = model.predict_proba(X_te)[:, 1]
after = model.predict_proba(shifted)[:, 1]
print('\nmean predicted risk %.4f -> %.4f' % (before.mean(), after.mean()))
print('flagged at 0.35 %.1f%% -> %.1f%%'
% (100 * (before >= 0.35).mean(), 100 * (after >= 0.35).mean()))
training contract mix:
contract
Month-To-Month 0.553
One Year 0.243
Two Year 0.204

live contract mix:
contract
Month-To-Month 0.759
One Year 0.132
Two Year 0.109

mean predicted risk 0.2628 -> 0.3005
flagged at 0.35 33.5% -> 39.9%

Prediction drift is the cheapest alarm you can build, because it needs no labels and no feature analysis, just the mean of today's scores against the mean of last month's. It cannot tell you whether the model is still right. It can tell you that something changed, on the day it changes, which is usually months before the labels arrive.

Day 4 takeaway

Log inputs alongside outputs, or you will be unable to diagnose anything later. Watch PSI per feature, the category mix, and the distribution of your own predictions. All three are available immediately and none of them require knowing whether the model was correct.
Week 15 · Day 5 of 7

Delayed Labels

Why you cannot measure accuracy yet, and what to watch instead

By 701 words

Drift says the inputs changed. It does not say the model got worse, for that you need labels, and labels are the thing you do not have.

The delay is the whole problem

events = [
('Monday', 'model scores 4,000 customers'),
('Monday', 'retention team contacts the 600 flagged'),
('+30 days', 'first contracts come up for renewal'),
('+60 days', 'roughly half of the outcomes are known'),
('+90 days', 'the label for Monday is finally complete'),
]
for when, what in events:
print('%-10s %s' % (when, what))
print('\nAny accuracy you measure today describes the model as it was')
print('three months ago -- on data that has since drifted.')
Monday model scores 4,000 customers
Monday retention team contacts the 600 flagged
+30 days first contracts come up for renewal
+60 days roughly half of the outcomes are known
+90 days the label for Monday is finally complete

Any accuracy you measure today describes the model as it was
three months ago -- on data that has since drifted.

And the intervention destroys the label

Worse than the delay: the retention team acted on the predictions. A customer who was flagged, offered a discount and stayed now counts as a false positive, so a model that worked perfectly and prompted a successful intervention looks, in the data, like a model that was wrong. Measuring a model whose output changes the world requires a holdout that nobody acts on, agreed with the business before you deploy. It is a conversation, not a technique.

What to watch while you wait

SignalAvailableTells you
Prediction distributionImmediatelySomething changed
Feature PSIImmediatelyWhich input changed
Unknown-category rejection rateImmediatelyA new product or a broken upstream field
Missing-value rate per fieldImmediatelyAn upstream pipeline has broken
Latency and error rateImmediatelyThe service itself
Proxy outcomes (support calls, logins)DaysEarly behavioural signal
Actual labelsWeeks to monthsWhether it was any good
Holdout labelsWeeks to monthsWhether it was any good, uncontaminated

An alert that does not cry wolf

import numpy as np
import pandas as pd

df = pd.read_csv('customers.csv').drop_duplicates().copy()
df['contract'] = df['contract'].str.strip().str.title()
df['total_charges'] = pd.to_numeric(df['total_charges'], errors='coerce')

NUM = ['tenure_months', 'support_calls', 'monthly_charges']
CAT = ['contract', 'internet_service', 'payment_method', 'has_dependents']
from sklearn.compose import ColumnTransformer
from sklearn.pipeline import Pipeline
from sklearn.impute import SimpleImputer
from sklearn.preprocessing import StandardScaler, OneHotEncoder
from sklearn.linear_model import LogisticRegression
from sklearn.model_selection import train_test_split

prep = ColumnTransformer([
('num', Pipeline([('i', SimpleImputer(strategy='median')),
('s', StandardScaler())]), NUM),
('cat', Pipeline([('i', SimpleImputer(strategy='most_frequent')),
('o', OneHotEncoder(handle_unknown='ignore'))]), CAT),
])
model = Pipeline([('prep', prep),
('clf', LogisticRegression(max_iter=2000))])

X, y = df[NUM + CAT], df['churned']
X_tr, X_te, y_tr, y_te = train_test_split(X, y, test_size=0.25,
stratify=y, random_state=42)
model.fit(X_tr, y_tr)
import numpy as np
import joblib

model = joblib.load('churn_model.joblib')
rng = np.random.default_rng(1)

baseline = model.predict_proba(X_tr)[:, 1]
mu, sd = baseline.mean(), baseline.std()
print('baseline mean risk %.4f (sd %.4f)' % (mu, sd))

def day_of_traffic(shift):
sample = X_te.sample(400, replace=True, random_state=int(shift * 100))
sample = sample.copy()
sample['monthly_charges'] = sample['monthly_charges'] * (1 + shift)
return model.predict_proba(sample)[:, 1]

print('\n%-8s %12s %12s %s' % ('shift', 'mean risk', 'z-score', 'alert'))
for shift in [0.0, 0.05, 0.10, 0.25, 0.50]:
scores = day_of_traffic(shift)
z = (scores.mean() - mu) / (sd / np.sqrt(len(scores)))
print('%-8.0f%% %12.4f %12.1f %s'
% (100 * shift, scores.mean(), z, 'ALERT' if abs(z) > 4 else ''))
baseline mean risk 0.2680 (sd 0.2247)

shift mean risk z-score alert
0 % 0.2654 -0.2
5 % 0.2696 0.1
10 % 0.2834 1.4
25 % 0.3080 3.6
50 % 0.3490 7.2 ALERT

A threshold on the raw mean would fire on every quiet Tuesday. Comparing against the variability of the baseline, how far today is from normal, in units of how much normal wobbles, is what makes an alert survive contact with a real on-call rota. Set the bar high enough that firing means something, because an alert nobody trusts is worse than no alert at all.

Day 5 takeaway

Labels arrive late and are contaminated by whatever was done with the predictions. Agree an unactioned holdout before launch. In the meantime monitor everything that is available immediately, and scale your alert thresholds by the natural variation of the signal rather than picking round numbers.
Week 15 · Day 6 of 7

Retraining and Rollout

Champion versus challenger with an interval, shadow mode and canaries

By 1055 words

Something has drifted, or three months have passed. Retraining is straightforward; deciding whether the new model is better, and getting it into production without a bad afternoon, is not.

Champion and challenger

import numpy as np
import pandas as pd

df = pd.read_csv('customers.csv').drop_duplicates().copy()
df['contract'] = df['contract'].str.strip().str.title()
df['total_charges'] = pd.to_numeric(df['total_charges'], errors='coerce')

NUM = ['tenure_months', 'support_calls', 'monthly_charges']
CAT = ['contract', 'internet_service', 'payment_method', 'has_dependents']
from sklearn.compose import ColumnTransformer
from sklearn.pipeline import Pipeline
from sklearn.impute import SimpleImputer
from sklearn.preprocessing import StandardScaler, OneHotEncoder
from sklearn.linear_model import LogisticRegression
from sklearn.model_selection import train_test_split

prep = ColumnTransformer([
('num', Pipeline([('i', SimpleImputer(strategy='median')),
('s', StandardScaler())]), NUM),
('cat', Pipeline([('i', SimpleImputer(strategy='most_frequent')),
('o', OneHotEncoder(handle_unknown='ignore'))]), CAT),
])
model = Pipeline([('prep', prep),
('clf', LogisticRegression(max_iter=2000))])

X, y = df[NUM + CAT], df['churned']
X_tr, X_te, y_tr, y_te = train_test_split(X, y, test_size=0.25,
stratify=y, random_state=42)
model.fit(X_tr, y_tr)
import numpy as np
import joblib
from sklearn.ensemble import HistGradientBoostingClassifier
from sklearn.metrics import roc_auc_score, brier_score_loss

champion = joblib.load('churn_model.joblib')

challenger = Pipeline([('prep', prep),
('clf', HistGradientBoostingClassifier(
random_state=42))]).fit(X_tr, y_tr)

print('%-14s %10s %10s %12s' % ('', 'AUC', 'Brier', 'flagged %'))
for name, m in [('champion', champion), ('challenger', challenger)]:
p = m.predict_proba(X_te)[:, 1]
print('%-14s %10.4f %10.4f %11.1f%%'
% (name, roc_auc_score(y_te, p), brier_score_loss(y_te, p),
100 * (p >= 0.35).mean()))

# Is the difference bigger than the noise? Bootstrap the test set.
rng = np.random.default_rng(0)
pc = champion.predict_proba(X_te)[:, 1]
pch = challenger.predict_proba(X_te)[:, 1]
yv = y_te.to_numpy()
diffs = []
for _ in range(400):
idx = rng.integers(0, len(yv), len(yv))
if yv[idx].sum() in (0, len(idx)):
continue
diffs.append(roc_auc_score(yv[idx], pch[idx])
- roc_auc_score(yv[idx], pc[idx]))
lo, hi = np.percentile(diffs, [2.5, 97.5])
print('\nAUC difference (challenger - champion)')
print(' point estimate %+.4f' % (roc_auc_score(yv, pch)
- roc_auc_score(yv, pc)))
print(' 95%% interval %+.4f to %+.4f' % (lo, hi))
AUC Brier flagged %
champion 0.8174 0.1475 33.5%
challenger 0.7797 0.1690 30.3%

AUC difference (challenger - champion)
point estimate -0.0376
95% interval -0.0582 to -0.0141

The challenger loses, and the interval is nowhere near zero, which settles it. This is week 6's result arriving in an operational costume: the data-generating process behind this dataset is close to logistic-linear, so a boosted ensemble has nothing to find that the linear model has not already found, and pays for the search with variance. The correct action here is to keep the champion and write down that the challenger was tried.

"Better" needs an interval, not a point

A challenger that is 0.004 ahead on one test set is not obviously ahead at all. Bootstrap the difference and look at whether the interval clears zero. If it does not, the honest conclusion is that the two models are indistinguishable on the evidence available, at which point you should prefer the one that is simpler, faster, already deployed, or easier to explain. "No change" is a legitimate and frequently correct outcome of a retraining cycle.

Getting it out there

StrategyHowCatchesCosts
ShadowNew model scores every request; output logged, not usedCrashes, latency, wild predictionsCompute, and no outcome data
Canary5% of traffic, then 25%, then allReal-world failures, at 5% of the blast radiusA slower rollout
A/B splitRandomised, both act on their predictionsGenuine causal comparisonWeeks, and real design work
Blue/greenBoth live, switch the routerNothing extra, but rollback is instantTwo environments
Big bangReplace itNothingYour evening
import numpy as np
import pandas as pd

df = pd.read_csv('customers.csv').drop_duplicates().copy()
df['contract'] = df['contract'].str.strip().str.title()
df['total_charges'] = pd.to_numeric(df['total_charges'], errors='coerce')

NUM = ['tenure_months', 'support_calls', 'monthly_charges']
CAT = ['contract', 'internet_service', 'payment_method', 'has_dependents']
from sklearn.compose import ColumnTransformer
from sklearn.pipeline import Pipeline
from sklearn.impute import SimpleImputer
from sklearn.preprocessing import StandardScaler, OneHotEncoder
from sklearn.linear_model import LogisticRegression
from sklearn.model_selection import train_test_split

prep = ColumnTransformer([
('num', Pipeline([('i', SimpleImputer(strategy='median')),
('s', StandardScaler())]), NUM),
('cat', Pipeline([('i', SimpleImputer(strategy='most_frequent')),
('o', OneHotEncoder(handle_unknown='ignore'))]), CAT),
])
model = Pipeline([('prep', prep),
('clf', LogisticRegression(max_iter=2000))])

X, y = df[NUM + CAT], df['churned']
X_tr, X_te, y_tr, y_te = train_test_split(X, y, test_size=0.25,
stratify=y, random_state=42)
model.fit(X_tr, y_tr)
import numpy as np
import joblib
from sklearn.ensemble import HistGradientBoostingClassifier

champion = joblib.load('churn_model.joblib')
challenger = Pipeline([('prep', prep),
('clf', HistGradientBoostingClassifier(
random_state=42))]).fit(X_tr, y_tr)

# Shadow mode: the champion decides, the challenger is recorded.
a = champion.predict_proba(X_te)[:, 1]
b = challenger.predict_proba(X_te)[:, 1]
flag_a, flag_b = a >= 0.35, b >= 0.35

print('champion flags %d' % flag_a.sum())
print('challenger flags %d' % flag_b.sum())
print('they disagree on %d of %d customers (%.1f%%)'
% ((flag_a != flag_b).sum(), len(a),
100 * (flag_a != flag_b).mean()))
print('\nlargest disagreements:')
gap = np.abs(a - b)
for i in np.argsort(-gap)[:4]:
print(' champion %.3f challenger %.3f tenure=%d contract=%s'
% (a[i], b[i], X_te.iloc[i]['tenure_months'],
X_te.iloc[i]['contract']))
champion flags 251
challenger flags 227
they disagree on 112 of 750 customers (14.9%)

largest disagreements:
champion 0.131 challenger 0.665 tenure=33 contract=Month-To-Month
champion 0.392 challenger 0.887 tenure=23 contract=Month-To-Month
champion 0.466 challenger 0.944 tenure=22 contract=Month-To-Month
champion 0.430 challenger 0.889 tenure=22 contract=One Year

Before any traffic moves, this tells you how different the two models actually are in the only terms that matter: how many decisions would change. A challenger that is 0.01 better on AUC but flips a third of your decisions is a much bigger operational event than that number suggests, and the retention team needs to hear about it from you rather than from their workload.

When to retrain

  1. On a schedule, if the world moves steadily. Monthly is a common default; it is a decision, not a law.
  2. On drift, when PSI on an important feature crosses your threshold and stays there.
  3. On performance, when holdout labels show a real fall, the most reliable trigger and the slowest.
  4. On a known event: a new product, a pricing change, a changed upstream field. Do not wait for the monitoring to notice something you already know.
  5. Never automatically without a gate. Automatic retraining plus automatic deployment means a corrupted upstream table becomes a corrupted production model with nobody in the loop.

Day 6 takeaway

Compare challenger against champion with an interval, not a point estimate, and count how many decisions would change rather than only how the metric moved. Roll out through shadow and canary stages. Keep a human gate between retraining and deployment.
Week 15 · Day 7 of 7

The Complete Service

Everything assembled, and the checklist before it takes traffic

By 894 words

The complete service, with everything the previous six days argued for, and a checklist to work through before anything of yours takes real traffic.

The finished thing

import numpy as np
import pandas as pd

df = pd.read_csv('customers.csv').drop_duplicates().copy()
df['contract'] = df['contract'].str.strip().str.title()
df['total_charges'] = pd.to_numeric(df['total_charges'], errors='coerce')

NUM = ['tenure_months', 'support_calls', 'monthly_charges']
CAT = ['contract', 'internet_service', 'payment_method', 'has_dependents']
from sklearn.compose import ColumnTransformer
from sklearn.pipeline import Pipeline
from sklearn.impute import SimpleImputer
from sklearn.preprocessing import StandardScaler, OneHotEncoder
from sklearn.linear_model import LogisticRegression
from sklearn.model_selection import train_test_split

prep = ColumnTransformer([
('num', Pipeline([('i', SimpleImputer(strategy='median')),
('s', StandardScaler())]), NUM),
('cat', Pipeline([('i', SimpleImputer(strategy='most_frequent')),
('o', OneHotEncoder(handle_unknown='ignore'))]), CAT),
])
model = Pipeline([('prep', prep),
('clf', LogisticRegression(max_iter=2000))])

X, y = df[NUM + CAT], df['churned']
X_tr, X_te, y_tr, y_te = train_test_split(X, y, test_size=0.25,
stratify=y, random_state=42)
model.fit(X_tr, y_tr)
import json
import joblib
import numpy as np

class ChurnService:
"""Everything the six days argued for, in one object."""

def __init__(self, model_path, contract_path):
self.model = joblib.load(model_path)
self.contract = json.load(open(contract_path))
self.seen = 0
self.rejected = 0
self.scores = []

def _validate(self, row):
for column, known in self.contract['categories'].items():
if row.get(column) not in known:
return 'unknown %s: %r' % (column, row.get(column))
for column, (lo, hi) in self.contract['ranges'].items():
value = row.get(column)
if value is None or not np.isfinite(value):
return 'missing %s' % column
return None

def predict(self, row):
self.seen += 1
problem = self._validate(row)
if problem:
self.rejected += 1
return {'status': 'rejected', 'reason': problem}
p = float(self.model.predict_proba(pd.DataFrame([row]))[0, 1])
self.scores.append(p)
outside = [c for c, (lo, hi) in self.contract['ranges'].items()
if not lo <= row[c] <= hi]
return {'status': 'ok',
'probability': round(p, 4),
'decision': 'flag' if p >= self.contract['threshold']
else 'keep',
'model_version': self.contract['model_version'],
'extrapolating': outside or None}

def stats(self):
s = np.array(self.scores) if self.scores else np.array([0.0])
return {'requests': self.seen, 'rejected': self.rejected,
'mean_score': round(float(s.mean()), 4),
'flag_rate': round(float(
(s >= self.contract['threshold']).mean()), 4)}

svc = ChurnService('churn_model.joblib', 'churn_contract.json')
for row in X_te.head(200).to_dict('records'):
svc.predict(row)

print(json.dumps(svc.predict(X_te.iloc[0].to_dict()), indent=2))
print(json.dumps(svc.predict(dict(X_te.iloc[0].to_dict(),
contract='Pay As You Go')), indent=2))
print(json.dumps(svc.predict(dict(X_te.iloc[0].to_dict(),
tenure_months=200)), indent=2))
print('\n' + json.dumps(svc.stats(), indent=2))
{
"status": "ok",
"probability": 0.5306,
"decision": "flag",
"model_version": "1.0.0",
"extrapolating": null
}
{
"status": "rejected",
"reason": "unknown contract: 'Pay As You Go'"
}
{
"status": "ok",
"probability": 0.0,
"decision": "keep",
"model_version": "1.0.0",
"extrapolating": [
"tenure_months"
]
}

{
"requests": 203,
"rejected": 28,
"mean_score": 0.2558,
"flag_rate": 0.32
}

Look at the rejection count: 28 of 203 real customers from the test set were refused. They are the rows with a missing categorical value, the roughly six percent that customers.csv has carried since week 1. The training pipeline imputes those without complaining. The service refuses them.

Your service can be stricter than your model, and you must choose

Neither behaviour is wrong, but the disagreement has to be a decision rather than an accident. Rejecting is right when a missing field means an upstream system is broken and a prediction would be misleading. Imputing is right when the field is genuinely optional and the model was trained to cope. What is never right is discovering the difference in production, when one in twenty of your customers cannot be scored and nobody knows why.

Write the rule down in the contract file next to the categories, and test it. Here, the honest fix is to allow missing categoricals (the pipeline handles them, and the training data contained them) while continuing to reject unknown values, which the pipeline silently mishandles.

The checklist

  1. The artefact contains the preprocessing, not just the estimator.
  2. A contract file records the features, valid categories, training ranges, threshold, training date and versions.
  3. A smoke test with frozen inputs and expected outputs runs on every deploy.
  4. Dependencies are pinned and the service runs in the environment it was trained in.
  5. The model loads once at start-up, and a health endpoint reports its version.
  6. Unknown categories are rejected rather than silently zero-encoded.
  7. Inputs outside the training ranges are answered but flagged.
  8. Every prediction is logged with its inputs, the threshold and the model version.
  9. Feature PSI, category mix, missing-value rates and the prediction distribution are monitored, with thresholds set from observed variation.
  10. An unactioned holdout exists, agreed with the business, so performance can be measured at all.
  11. Rollback is a switch somebody can throw without a rebuild.
  12. Somebody's name is against the model, and there is a date to review it.

The part that is not code

Most models that fail in production fail for reasons on that list that have nothing to do with modelling: an upstream field silently changed units, a new product category appeared, nobody agreed who watches the dashboard, the threshold was set by a data scientist and never discussed with the team who work the flagged list. The modelling was the easy part and it was finished twelve weeks ago.

Day 7 takeaway

Ship the pipeline and its contract together, validate at the boundary, log inputs with outputs, monitor what is available immediately, keep a human gate before deployment and a switch for rollback. The hard parts of production machine learning are agreements and plumbing, not algorithms.