Production ML

# A skew test worth running in CI: score a frozen sample through BOTH paths.
import numpy as np, pandas as pd

golden = pd.read_parquet('golden_sample.parquet')       # 1,000 rows captured from prod

offline = pipe.predict_proba(golden)[:, 1]              # the training-time code path
online = np.array([serving_client.score(row) for row in golden.to_dict('records')])

delta = np.abs(offline - online)
print(f'max {delta.max():.6f}  mean {delta.mean():.6f}  over 1e-4: {(delta > 1e-4).sum()}')
assert delta.max() < 1e-4, 'training and serving paths disagree'

# Batch: throughput is all that matters, so vectorise and write results down.
import pandas as pd

frame = pd.read_parquet('active_users.parquet')
frame['p_churn'] = pipe.predict_proba(frame[FEATURES])[:, 1]
frame[['user_id', 'p_churn']].to_parquet('scores/2026-09-19.parquet')

# Online: one row, p99 budget, model loaded once at startup.
MODEL = joblib.load('model.joblib')

def score(payload: dict) -> float:
    row = pd.DataFrame([payload], columns=FEATURES)
    return float(MODEL.predict_proba(row)[0, 1])

# The two reads a feature store exists to provide.

# Offline, point-in-time correct: labels joined to feature values as of the event.
training = store.get_historical_features(
    entity_df=labels[['user_id', 'event_ts', 'churned']],   # event_ts is the cutoff
    features=['user:spend_30d', 'user:tickets_7d', 'user:tenure_months'],
).to_df()

# Online, same definitions, single-row latency.
values = store.get_online_features(
    features=['user:spend_30d', 'user:tickets_7d', 'user:tenure_months'],
    entity_rows=[{'user_id': 'u-7'}],
).to_dict()

import numpy as np, time

def latency_profile(model, X, n=2000):
    rows = X.sample(n, replace=True, random_state=0).to_dict('records')
    times = []
    for r in rows:
        frame = pd.DataFrame([r])
        t0 = time.perf_counter()
        model.predict_proba(frame)
        times.append((time.perf_counter() - t0) * 1000)
    t = np.array(times)
    p50, p95, p99 = np.percentile(t, [50, 95, 99])
    return {'p50': p50, 'p95': p95, 'p99': p99}

for name, m in candidates.items():
    prof = latency_profile(m, X_val)
    print(f"{name:<22} p50 {prof['p50']:5.1f}ms  p99 {prof['p99']:6.1f}ms")

import numpy as np

rng = np.random.default_rng(0)
body = rng.gamma(shape=2.0, scale=3.0, size=100_000)           # the common case
tail = rng.gamma(shape=2.0, scale=3.0, size=100_000) + rng.exponential(60, 100_000)
is_slow = rng.random(100_000) < 0.02                            # 2% hit a slow path
latency = np.where(is_slow, tail, body)

for q in (50, 90, 95, 99, 99.9):
    print(f'p{q:<5} {np.percentile(latency, q):7.1f} ms')
print(f'mean  {latency.mean():7.1f} ms   over 50ms: {(latency > 50).mean():.2%}')

import json, pathlib, shutil, datetime

REGISTRY = pathlib.Path('registry')

def register(pipeline, metrics, data_version, git_sha, owner):
    version = datetime.datetime.now(datetime.timezone.utc).strftime('%Y%m%d%H%M%S')
    target = REGISTRY / version
    target.mkdir(parents=True)
    joblib.dump(pipeline, target / 'model.joblib')
    (target / 'meta.json').write_text(json.dumps({
        'version': version, 'metrics': metrics, 'data_version': data_version,
        'git_sha': git_sha, 'owner': owner, 'sklearn': sklearn.__version__,
    }, indent=2))
    return version

def promote(version):
    link = REGISTRY / 'production'
    if link.exists() or link.is_symlink():
        shutil.move(str(link), str(REGISTRY / 'previous'))
    link.symlink_to(version)              # rollback = promote(previous)

import hashlib

CANARY_SHARE = 0.05

def pick_model(user_id, champion, canary, salt='release-20260919'):
    h = hashlib.sha256(f'{salt}:{user_id}'.encode()).digest()
    in_canary = (int.from_bytes(h[:8], 'big') % 10_000) < CANARY_SHARE * 10_000
    return ('canary', canary) if in_canary else ('champion', champion)

GUARDRAILS = {
    'p99_latency_ms': ('max', 50),
    'error_rate': ('max', 0.001),
    'mean_score': ('within_pct_of_champion', 10),
    'null_feature_rate': ('max', 0.02),
}

def should_roll_back(canary_metrics, champion_metrics):
    breaches = []
    if canary_metrics['p99_latency_ms'] > 50:
        breaches.append('latency')
    if canary_metrics['error_rate'] > 0.001:
        breaches.append('errors')
    drift = abs(canary_metrics['mean_score'] - champion_metrics['mean_score'])
    if drift > 0.10 * champion_metrics['mean_score']:
        breaches.append('score_distribution')
    return breaches

import time, numpy as np

# Caching: key on the feature values AND the model version, with a TTL.
_cache = {}
TTL_S = 300

def score_cached(key, features, model, version):
    now = time.monotonic()
    hit = _cache.get((version, key))
    if hit and now - hit[1] < TTL_S:
        return hit[0]
    value = float(model.predict_proba(features)[0, 1])
    _cache[(version, key)] = (value, now)
    return value

# Batching: per-row cost collapses as the batch grows.
for size in (1, 8, 32, 128, 512):
    rows = X_val.sample(size, replace=True, random_state=0)
    t0 = time.perf_counter()
    pipe.predict_proba(rows)
    per_row = (time.perf_counter() - t0) * 1000 / size
    print(f'batch {size:>4}  {per_row:6.3f} ms/row')

STAGES = [
    ('frame',     'primary metric and guardrails written down, with a decision threshold'),
    ('data',      'versioned snapshot, leakage audit clean, point-in-time joins verified'),
    ('experiment','runs logged: data + code + params + seed + env + metrics'),
    ('package',   'one Pipeline artifact registered with meta.json and an owner'),
    ('skew test', 'golden sample scores identically through offline and serving paths'),
    ('shadow',    'live traffic scored, nothing acted on; distributions and latency checked'),
    ('canary',    '1/5/25/50/100% with automatic rollback on guardrail breach'),
    ('ab test',   'powered per the MDE calculation; primary metric only'),
    ('monitor',   'drift, prediction distribution, latency, delayed-label performance'),
    ('retire',    'successor promoted, artifact archived, alerts and jobs removed'),
]

for name, exit_criterion in STAGES:
    print(f'{name:<10} -> {exit_criterion}')