Step 1: A baseline
Wren Parcels shows a "may arrive late" notice when its model gives a parcel a probability above 0.35, and the
customer team plans staff from the predicted late rate. The model went live in week 1. Twelve weeks of production
traffic are in production.csv, and the eight weeks it was trained on are in training.csv. Your job is the
monitoring the model shipped without.
Drift only means something against a fixed reference. The baseline stores, for every input column and for the model's own score, what the training data looked like. Numeric columns are cut into ten quantile bins, so each bin holds about 10% of the training rows. Categorical columns get one share per category, plus a last share for values training never contained.
Do this
1. Write bin_edges(), shares(), category_shares() and build_baseline() in monitor.py. model.py
gives you the data and the shipped model; try_it.py adds each row's score column before calling your code.
2. Run it. items has only four bins: most parcels hold one item, so several quantiles fall on the same value,
and a repeated edge would only add a bin that no value can fall in.
monitor.py, the file you edit112 lines
"""Monitoring for Wren Parcels' late-delivery model. You write the functions marked TODO, one step at a time."""
import numpy as np
import pandas as pd
from sklearn.metrics import roc_auc_score
from model import CATEGORICAL, NUMERIC, LateModel
BINS = 10
EPS = 1e-4 # floor for empty bins, so PSI stays finite
LABEL_DELAY = 2 # outcomes of week w are known at the end of week w + 2
Z_PAGE = 3.0
# ---------- Step 1: a baseline ----------
def bin_edges(values, bins=BINS):
"""The bins - 1 inner quantile cut points of values (duplicates removed), as a list of floats."""
# TODO (Step 1): np.quantile at np.linspace(0, 1, bins + 1)[1:-1], then np.unique, as floats.
raise NotImplementedError("Step 1: write bin_edges()")
def shares(values, edges):
"""The fraction of values in each of the len(edges) + 1 bins. A value equal to an edge counts in the bin above."""
# TODO (Step 1): np.searchsorted(edges, values, side="right") gives each value's bin; count with
# np.bincount(..., minlength=len(edges) + 1) and divide by the number of values.
raise NotImplementedError("Step 1: write shares()")
def category_shares(values, categories):
"""The fraction of values equal to each category, then one last share for values not in categories."""
# TODO (Step 1): count each category, put everything else in one last count, divide by the total.
raise NotImplementedError("Step 1: write category_shares()")
def build_baseline(training):
"""{column: profile} for NUMERIC + ["score"] (profile {"edges", "shares"}) and CATEGORICAL
(profile {"categories": sorted values seen in training, "shares"})."""
# TODO (Step 1): a profile for each numeric column (and "score") and each categorical one.
raise NotImplementedError("Step 1: write build_baseline()")
# ---------- Step 2: drift, week by week ----------
def psi(expected, actual, eps=EPS):
"""Population stability index: sum over bins of (actual - expected) * ln(actual / expected), shares floored at eps."""
raise NotImplementedError("psi() arrives in Step 2")
def column_psi(baseline, column, values):
"""PSI of values against the baseline profile of column, numeric or categorical."""
raise NotImplementedError("column_psi() arrives in Step 2")
def level(value):
""""stable" below 0.1, "moderate" below 0.25, "major" from 0.25."""
raise NotImplementedError("level() arrives in Step 2")
def drift_table(baseline, production):
"""A DataFrame with one row per week (index "week", ascending) and one PSI column per baseline column."""
raise NotImplementedError("drift_table() arrives in Step 2")
# ---------- Step 3: quality, once the labels arrive ----------
def labelled(production, as_of):
"""The production rows whose outcome is known at the end of week as_of."""
raise NotImplementedError("labelled() arrives in Step 3")
def quality(rows):
"""{"n", "late_rate", "predicted_rate", "z", "auc"} for rows with "late" and "score" columns.
z = (late parcels - expected late parcels) / sqrt(sum of score * (1 - score))."""
raise NotImplementedError("quality() arrives in Step 3")
def quality_table(production, as_of):
"""quality() for every week that has labels at the end of week as_of: a DataFrame indexed by "week"."""
raise NotImplementedError("quality_table() arrives in Step 3")
# ---------- Step 4: alerts someone will read ----------
def alerts_at(baseline, production, as_of):
"""The alerts the weekly job raises at the end of week as_of, knowing only what it knows then.
ticket: a column's PSI in week as_of is major and it was not major in week as_of - 1.
page: z of the newest labelled week is >= Z_PAGE and was below it the labelled week before.
Each alert is {"as_of", "severity", "signal", "week", "value"}."""
raise NotImplementedError("alerts_at() arrives in Step 4")
def replay(baseline, production):
"""alerts_at() for every week in production, in order, as one list."""
raise NotImplementedError("replay() arrives in Step 4")
# ---------- Step 5: find the cause, fix it ----------
def segment_report(rows, column):
"""quality() for each value of column in rows: a DataFrame indexed by the value, highest z first."""
raise NotImplementedError("segment_report() arrives in Step 5")
def worst_segment(rows, columns=CATEGORICAL):
"""(column, value, z) of the single segment with the highest z across columns."""
raise NotImplementedError("worst_segment() arrives in Step 5")
def retrain(training, production, as_of):
"""A LateModel fitted at the end of week as_of on training plus every production row labelled by then,
with an indicator for each region in that data."""
raise NotImplementedError("retrain() arrives in Step 5")Provided for you:
model.pyproduction.csvtraining.csvtry_it.py