Shadow deployment

pip install scikit-learn pandas numpy pytest

"""A shadow deployment: the OLD model answers every real request.
The NEW model scores the same request in the background, and its
answer is only logged -- it is never returned to a caller.
"""
import time

import numpy as np
import pandas as pd
from sklearn.linear_model import LogisticRegression
from sklearn.pipeline import make_pipeline
from sklearn.preprocessing import StandardScaler

COLUMNS = ["income", "years", "age"]


def make_training_data(n=400, seed=0):
    rng = np.random.RandomState(seed)
    X = pd.DataFrame({
        "income": rng.uniform(5, 80, n).round(1),
        "years": rng.uniform(0, 10, n).round(1),
        "age": rng.randint(21, 65, n).astype(float),
    })
    score = 0.05 * X["income"] + 0.35 * X["years"] + 0.01 * X["age"] - 3.5
    y = (score + rng.normal(0, 0.8, n) > 0).astype(int)
    return X, y


X, y = make_training_data()
OLD_MODEL = make_pipeline(StandardScaler(), LogisticRegression(max_iter=1000, C=1.0)).fit(X, y)
# "NEW": retrained with different regularisation -- a realistic candidate improvement
NEW_MODEL = make_pipeline(StandardScaler(), LogisticRegression(max_iter=1000, C=0.1)).fit(X, y)

SHADOW_LOG = []


def score_request(applicant: dict) -> dict:
    row = pd.DataFrame([applicant], columns=COLUMNS)

    t0 = time.perf_counter()
    old_answer = int(OLD_MODEL.predict(row)[0])
    old_ms = (time.perf_counter() - t0) * 1000

    t0 = time.perf_counter()
    new_answer = int(NEW_MODEL.predict(row)[0])
    new_ms = (time.perf_counter() - t0) * 1000

    SHADOW_LOG.append({
        "applicant": applicant,
        "old_answer": old_answer,
        "new_answer": new_answer,
        "agree": old_answer == new_answer,
        "old_ms": round(old_ms, 4),
        "new_ms": round(new_ms, 4),
    })

    return {"repaid": old_answer}  # ONLY the old model's answer ever leaves this function

from shadow import *

rng = np.random.RandomState(1)
for _ in range(200):
    applicant = {
        "income": round(rng.uniform(5, 80), 1),
        "years": round(rng.uniform(0, 10), 1),
        "age": float(rng.randint(21, 65)),
    }
    score_request(applicant)

agree_rate = sum(r["agree"] for r in SHADOW_LOG) / len(SHADOW_LOG)
avg_old_ms = sum(r["old_ms"] for r in SHADOW_LOG) / len(SHADOW_LOG)
avg_new_ms = sum(r["new_ms"] for r in SHADOW_LOG) / len(SHADOW_LOG)
print(f"requests scored: {len(SHADOW_LOG)}")
print(f"agreement rate between old and new: {agree_rate:.3f}")
print(f"avg old-model time per request: {avg_old_ms:.4f} ms")
print(f"avg new-model time per request (shadow, never shown to user): {avg_new_ms:.4f} ms")

from shadow import *

rng = np.random.RandomState(1)
for _ in range(200):
    applicant = {
        "income": round(rng.uniform(5, 80), 1),
        "years": round(rng.uniform(0, 10), 1),
        "age": float(rng.randint(21, 65)),
    }
    score_request(applicant)

disagreements = [r for r in SHADOW_LOG if not r["agree"]]
print(f"disagreements: {len(disagreements)} -- first one: {disagreements[0]}")

from shadow import OLD_MODEL, SHADOW_LOG, score_request


def test_the_returned_answer_always_matches_the_old_model():
    SHADOW_LOG.clear()
    applicant = {"income": 60.0, "years": 7.0, "age": 34.0}
    result = score_request(applicant)

    import pandas as pd
    expected = int(OLD_MODEL.predict(pd.DataFrame([applicant], columns=["income", "years", "age"]))[0])
    assert result["repaid"] == expected


def test_the_new_model_is_logged_but_never_returned():
    SHADOW_LOG.clear()
    result = score_request({"income": 60.0, "years": 7.0, "age": 34.0})

    assert "new_answer" not in result          # the response never carries it
    assert "new_answer" in SHADOW_LOG[-1]        # but the log always does


def test_every_scored_request_gets_a_log_entry():
    SHADOW_LOG.clear()
    for i in range(5):
        score_request({"income": 30.0 + i, "years": 2.0, "age": 25.0})
    assert len(SHADOW_LOG) == 5

pytest test_shadow.py -q