Machine Learning Workflow

Four lessons showed how a model is built: a ready table, a ready feature set, a ready split and a single measurement. In a real job none of these comes ready. The problem definition changes after measurement, the feature set widens after evaluation, and the flow never runs once from start to finish.

This lesson builds the flow as seven steps and counts not the steps but the returns. The analysis workflow lesson in the Introduction to Data Analytics course had counted returns too; there the reason for a return was that a decision was seen only after the step it belonged to. Here what triggers the return is not a decision but a measurement: the flow goes back because a number stayed under a threshold.

• FC26. The flow is seven steps: problem definition, data preparation, feature selection, model building, evaluation, deployment, monitoring. What is measured is not the number of steps but the number of returns.
• FC27. A gate is the check that ties the decision to advance in the flow to a number. The threshold of a gate is written before the measurement; if it is written afterward, the threshold becomes not a decision but a justification.
• FC28. In this run a return has three separate sources: the measurement failing to reach the threshold, the deployment constraint and drift of the data in monitoring.
• FC29. The deployment constraint can change the problem definition. The output of classification is a set, that of ranking is a queue; what a team working with limited capacity wants is a queue.
• FC30. Monitoring is the only step at which the number can change although the model never changes. What drifts is not the model itself but the data that reaches the model.
• FC31. The decision the model produces after deployment is called inference. The term is distinct from the inference used in the Visualization, Reporting and Analysis at Scale course in the sense of the conclusion a reader draws from a chart: here the one making the inference is not the reader but the model.

Rebuilding the Fictional Dataset

# setup.py — the fictional dataset of lesson 01 is rebuilt with the same seed: the readings
# table, the subscriber summary, the fictional label, the train / held-out split and the tree learner.
import math

SEED, RAW, M32 = 20260218, 1400, 0xFFFFFFFF
REGION = [("north", 0.28, 21), ("south", 0.22, 17), ("east", 0.18, 26),
          ("west", 0.14, 14), ("central", 0.18, 23)]
FEATURES = ["mean", "per_person", "drop", "zero_readings", "reading_count",
            "satisfaction", "household_size", "highest", "lowest"]


def rng(t):
    x = ((t ^ (t >> 16)) * 2246822507) & M32
    x = ((x ^ (x >> 13)) * 3266489909) & M32
    s = [(x ^ (x >> 16)) & M32]

    def next_value():
        s[0] = (s[0] * 1664525 + 1013904223) & M32
        return s[0] / 4294967296
    return next_value


def pick(u, w):
    t = 0.0
    for i, x in enumerate(w):
        t += x
        if u < t:
            return i
    return len(w) - 1


_subscribers, _dropped, _series = [], [], {}
for i in range(RAW):
    r = rng(SEED + i)
    b = REGION[pick(r(), [x[1] for x in REGION])]
    k = {"subscriber_id": 10001 + i, "region": b[0], "base": b[2],
         "household_size": pick(r(), [0.06, 0.24, 0.30, 0.24, 0.11, 0.05]),
         "satisfaction": pick(r(), [0.08, 0.14, 0.27, 0.34, 0.17]) + 1}
    (_dropped if r() < 0.046 else _subscribers).append(k)
_valid = {a["subscriber_id"] for a in _subscribers}
for k in _subscribers + _dropped:
    r = rng(SEED + 7000 + k["subscriber_id"])
    for d in range(pick(r(), [0.05, 0.12, 0.21, 0.62])):
        t = 0.0 if r() < 0.038 else math.floor(k["base"] * math.exp(
            (r() + r() + r() - 1.5) * 0.62) * (1 - d * 0.05) * 100 + 0.5) / 100
        if k["subscriber_id"] in _valid:
            _series.setdefault(k["subscriber_id"], []).append(t)
        r()

SUBSCRIBERS = []
for k in _subscribers:
    t = _series.get(k["subscriber_id"])
    if not t:
        continue
    mean = sum(t) / len(t)
    a = {"subscriber_id": k["subscriber_id"], "region": k["region"], "series": t,
         "household_size": k["household_size"], "satisfaction": k["satisfaction"],
         "reading_count": len(t), "mean": mean, "highest": max(t),
         "lowest": min(t), "zero_readings": sum(1 for v in t if v == 0),
         "per_person": mean / max(1, k["household_size"]),
         "drop": (t[0] - t[-1]) / t[0] if len(t) > 1 and t[0] > 0 else 0.0}
    r = rng(SEED + 31000 + a["subscriber_id"])
    a["suspicion"] = 1 if (1.30 * a["zero_readings"] + 1.25 * (a["drop"] > 0.45)
                           + 1.05 * (a["per_person"] > 12.0) + 0.65 * (a["reading_count"] < 3)
                           + 0.45 * (a["satisfaction"] <= 2) - 1.55
                           + (r() + r() - 1.0) * 1.1) > 0 else 0
    SUBSCRIBERS.append(a)

r = rng(SEED + 90001)
TRAIN, HELD_OUT = [], []
for a in SUBSCRIBERS:
    (TRAIN if r() < 0.70 else HELD_OUT).append(a)
positives = lambda k: sum(a["suspicion"] for a in k)
accuracy = lambda f, k: sum(1 for a in k if f(a) == a["suspicion"]) / len(k)
_impurity = lambda k: 2 * (positives(k) / len(k)) * (1 - positives(k) / len(k)) if k else 0.0


def fit(k, depth=3, min_size=40):
    """The greedy decision tree of lesson 01: the split criterion and stopping rule are there."""
    best, choice = _impurity(k) * len(k), None
    if depth and len(k) >= min_size:
        for feat in FEATURES:
            for threshold in sorted({a[feat] for a in k})[1:]:
                left = [a for a in k if a[feat] < threshold]
                right = [a for a in k if a[feat] >= threshold]
                m = _impurity(left) * len(left) + _impurity(right) * len(right)
                if m < best - 1e-9:
                    best, choice = m, (feat, threshold, left, right)
    if choice is None:
        return 1 if 2 * positives(k) > len(k) else 0
    lt, rt = fit(choice[2], depth - 1, min_size), fit(choice[3], depth - 1, min_size)
    return lt if lt == rt else (choice[0], choice[1], lt, rt)


def apply(d, a):
    while isinstance(d, tuple):
        d = d[2] if a[d[0]] < d[1] else d[3]
    return d


print(f"fictional dataset (seed {SEED}): {len(SUBSCRIBERS)} subscribers, {len(TRAIN)} train, "
      f"{len(HELD_OUT)} held-out, {positives(SUBSCRIBERS)} suspicious")

fictional dataset (seed 20260218): 1260 subscribers, 883 train, 377 held-out, 327 suspicious

Seven Steps and the First Gate

The flow is defined at the first step: the question is “is this subscriber suspicious”, the measure is accuracy, the baseline is the most frequent class. The threshold of the gate at the fifth step is also written here: the model must exceed the baseline on the held-out set by at least 0.0500. This number is an engineering decision, and it is a decision because it is written before the measurement; had it been written after the measurement it would have been a justification that legitimizes the result.

At the second step the data is prepared. In the first pass the summaries of the readings table are not yet ready, so the third step chooses only the three columns that come from the subscriber record.

# workflow.py — seven steps and two gates. The gates look at a number: the gate at the fifth
# step measures the gap over the baseline, the one at the sixth the daily field capacity.
import statistics
from collections import Counter

STEPS = ["1 problem definition", "2 data preparation", "3 feature selection", "4 model building",
         "5 evaluation", "6 deployment", "7 monitoring"]
ACCEPT_GAP, CAPACITY = 0.05, 40
RETURNS, FULL = [], FEATURES[:]
RECORD = ["household_size", "satisfaction", "reading_count"]
most_frequent = Counter(a["suspicion"] for a in TRAIN).most_common(1)[0][0]
BASE_TRAIN = accuracy(lambda a: most_frequent, TRAIN)
BASE_HELD = accuracy(lambda a: most_frequent, HELD_OUT)


def tree_with(feats, subset):
    global FEATURES
    FEATURES = feats
    d = fit(subset)
    FEATURES = FULL
    return d


W = [8, 16, 11, 11, 11, 11, 22]
row = lambda h: str(h[0]).ljust(W[0]) + str(h[1]).ljust(W[1]) + "".join(
    str(v).rjust(W[i + 2]) for i, v in enumerate(h[2:6])) + "  " + str(h[6])
print(row(["pass", "features", "baseline", "train", "held-out", "gap", "gate"]))
TREES = {}
for pass_no, feats in ((1, RECORD), (2, FULL)):
    d = TREES[pass_no] = tree_with(feats, TRAIN)
    inner, outer = accuracy(lambda a: apply(d, a), TRAIN), accuracy(
        lambda a: apply(d, a), HELD_OUT)
    gap = outer - BASE_HELD
    passed = gap >= ACCEPT_GAP
    print(row([pass_no, f"{len(feats)} columns", f"{BASE_HELD:.4f}", f"{inner:.4f}", f"{outer:.4f}",
               f"{gap:+.4f}", "passed" if passed else f"closed, threshold {ACCEPT_GAP:.2f}"]))
    if not passed:
        RETURNS.append([STEPS[4], STEPS[2], "gap below threshold", f"{gap:+.4f}", None])
    else:
        RETURNS[-1][4] = f"{gap:+.4f}"

flagged = sum(1 for a in SUBSCRIBERS if apply(TREES[2], a))
print(f"\nstep 6: the model flags {flagged} of {len(SUBSCRIBERS)} subscribers; "
      f"daily capacity is {CAPACITY} visits, so the list "
      f"spreads over {-(-flagged // CAPACITY)} days and has no order inside")
RETURNS.append([STEPS[5], STEPS[0], "does not fit capacity", f"{flagged} flagged", None])

pass    features           baseline      train   held-out        gap  gate
1       3 columns            0.7533     0.7939     0.7851    +0.0318  closed, threshold 0.05
2       9 columns            0.7533     0.8709     0.8594    +0.1061  passed

step 6: the model flags 240 of 1260 subscribers; daily capacity is 40 visits, so the list spreads over 6 days and has no order inside

The first pass stayed at the gate: 0.0318 is below the threshold. The flow returns from the fifth step to the third, and the feature set grows to nine columns with the summaries of the reading table. In the second pass the gap is 0.1061 and the gate opens. This is the number the return moved: 0.1061 instead of 0.0318.

At the sixth step the second gate comes into play, and the model measures not performance but the constraint of the job. The model flags 240 of 1,260 subscribers. The daily capacity is 40 visits, so the list spreads over six days, and the output of classification does not say who will be visited first within the list. The model passed the gate on performance, it could not pass on the constraint.

The Constraint Changes the Problem, Monitoring the Data

From the sixth step the flow returns to the first. This is the longest return of the flow and it changes not the model but the question: a question that asks for a decision is replaced by a question that asks for an order. The measure changes too — the number of illicit cases found in the first 40 visits instead of accuracy — and the baseline is chosen anew: the random order.

At the seventh step the model is in the field and is monitored. What monitoring measures is not the model itself, because the model does not change. What changes is the data that reaches the model: the network changes the reading policy, zero readings are filled with an estimate and missing periods are completed. The subscribers are the same, the true illicit-use status is the same, only the way it is written to the table has changed.

# ranking_and_monitoring.py — the problem is redefined: an order is wanted, not a decision.
# After deployment the measurement policy changes and the same model is measured again.
r = rng(SEED + 7205)
ENSEMBLE = [fit([TRAIN[int(r() * len(TRAIN))] for _ in TRAIN]) for _ in range(9)]
r = rng(SEED + 7206)
lots = {a["subscriber_id"]: r() for a in SUBSCRIBERS}
found_in_top = lambda f, s: positives(sorted(s, key=f)[:CAPACITY])
ranking = lambda T: (lambda a: (-sum(apply(d, a) for d in T), a["subscriber_id"]))
random_order = lambda a: lots[a["subscriber_id"]]


def drifted(a):
    """The reading policy changed: zero readings are filled with an estimate, missing
    periods are completed. The true illicit-use status, that is the label, does not change."""
    filled = [v for v in a["series"] if v > 0]
    est = round(statistics.fmean(filled), 2) if filled else 0.0
    t = [v if v > 0 else est for v in a["series"]] + [est] * (3 - len(a["series"]))
    avg = sum(t) / len(t)
    return {**a, "series": t, "reading_count": 3, "mean": avg, "highest": max(t),
            "lowest": min(t), "zero_readings": sum(1 for v in t if v == 0),
            "per_person": avg / max(1, a["household_size"]),
            "drop": (t[0] - t[-1]) / t[0] if t[0] > 0 else 0.0}


DRIFT_TRAIN, DRIFT_HELD = [drifted(a) for a in TRAIN], [drifted(a) for a in HELD_OUT]
r = rng(SEED + 7305)
NEW = [fit([DRIFT_TRAIN[int(r() * len(DRIFT_TRAIN))] for _ in DRIFT_TRAIN]) for _ in range(9)]
V = [30, 14, 12, 12]
row2 = lambda h: str(h[0]).ljust(V[0]) + "".join(str(v).rjust(V[i + 1])
                                                 for i, v in enumerate(h[1:]))
print(row2([f"illicit in first {CAPACITY} visits", "data", "train", "held-out"]))
print(row2(["baseline: random order", "old", found_in_top(random_order, TRAIN),
            found_in_top(random_order, HELD_OUT)]))
print(row2(["model: nine-tree order", "old", found_in_top(ranking(ENSEMBLE), TRAIN),
            found_in_top(ranking(ENSEMBLE), HELD_OUT)]))
print(row2(["baseline: random order", "drifted", found_in_top(random_order, DRIFT_TRAIN),
            found_in_top(random_order, DRIFT_HELD)]))
print(row2(["same model", "drifted", found_in_top(ranking(ENSEMBLE), DRIFT_TRAIN),
            found_in_top(ranking(ENSEMBLE), DRIFT_HELD)]))
print(row2(["model built on drifted data", "drifted", found_in_top(ranking(NEW), DRIFT_TRAIN),
            found_in_top(ranking(NEW), DRIFT_HELD)]))
print(f"\nlabels unchanged: old {positives(HELD_OUT)}, drifted {positives(DRIFT_HELD)} suspicious")
print(f"two corrupted features: zero-reading total "
      f"{sum(a['zero_readings'] for a in HELD_OUT)} -> {sum(a['zero_readings'] for a in DRIFT_HELD)}, "
      f"subscribers with missing periods {sum(1 for a in HELD_OUT if a['reading_count'] < 3)} -> "
      f"{sum(1 for a in DRIFT_HELD if a['reading_count'] < 3)}")
RETURNS[1][4] = f"{found_in_top(ranking(ENSEMBLE), HELD_OUT)} illicit in top {CAPACITY}"
RETURNS.append([STEPS[6], STEPS[1], "measurement policy changed",
                f"{found_in_top(ranking(ENSEMBLE), DRIFT_HELD)} illicit",
                f"{found_in_top(ranking(NEW), DRIFT_HELD)} illicit"])

illicit in first 40 visits              data       train    held-out
baseline: random order                   old          10          14
model: nine-tree order                   old          30          30
baseline: random order               drifted          10          14
same model                           drifted          33          23
model built on drifted data          drifted          39          30

labels unchanged: old 93, drifted 93 suspicious
two corrupted features: zero-reading total 39 -> 3, subscribers with missing periods 117 -> 0

The same model, the same subscribers, the same labels and a falling number: 23 illicit cases in the first 40 visits instead of 30. Since the baseline of the random order stayed fixed at 14, the drop belongs to the model, it is not the question getting harder. The cause is written in the last line: the two features the model relied on have been erased. The zero-reading total has gone from 39 to 3, the number of subscribers with missing periods from 117 to 0. The model was deciding with the information carried by these two columns, and that information is no longer written to the table.

From the seventh step the flow returns to the second. The data is prepared anew according to the new policy, the model is rebuilt on the drifted data and the number returns to 30. The loss is closed, but the price of closing it is a return, and a return is seen only if the monitoring step exists. Without monitoring the list in the field would go on producing seven illicit cases too few and nobody would notice.

The Accounting of Returns

# returns.py — the measure of the flow is not the number of steps but the number of returns.
W3 = [16, 24, 30, 16, 22]
row3 = lambda h: "".join(str(v).ljust(W3[i]) for i, v in enumerate(h))
print(row3(["from", "to", "reason", "before return", "after return"]))
for d in RETURNS:
    print(row3(d))
print(f"\n{len(STEPS)}-step flow produced {len(RETURNS)} returns; "
      f"total step visits {sum(int(x[0][0]) - int(x[1][0]) + 1 for x in RETURNS) + len(STEPS)}")

from            to                      reason                        before return   after return          
5 evaluation    3 feature selection     gap below threshold           +0.0318         +0.1061               
6 deployment    1 problem definition    does not fit capacity         240 flagged     30 illicit in top 40  
7 monitoring    2 data preparation      measurement policy changed    23 illicit      30 illicit            

7-step flow produced 3 returns; total step visits 22

All three returns leave from a different step and land on a different step. The return from the fifth step to the third is short and three steps are visited again; the return from the sixth to the first makes six steps be visited again, and the return from the seventh to the second again six steps. The flow, which is seven steps when read linearly, has been completed with 22 step visits, that is, each step has been visited 3.1 times on average.

The accounting itself is a decision too. Here a return was counted as the re-execution of all the steps from the target step to the exit step. Some steps may not really need to be repeated — in the return from the sixth to the first, data preparation has not changed — and in that case the number would have come out smaller. Without writing down which accounting produced it, 22 cannot be read.

The number of returns is not a measure of defects. A flow with zero returns is a flow whose gates test nothing; had there been no gate in the first pass, a model of 0.0318 would have gone out to the field and nobody would have measured it. A return is the proof that the gate works.

Summary

• The flow is seven steps and is not linear; in this run it produced three returns and was completed with 22 step visits.
• The gate at the fifth step tests the gap over the baseline with a threshold of 0.0500: the model with three columns gives 0.0318 and stays at the gate, the model with nine columns gives 0.1061 and passes.
• The gate at the sixth step measures not performance but the constraint: the model flags 240 subscribers, the daily capacity is 40, and the constraint turns the problem from classification into ranking; the ranking finds 30 illicit cases in the first 40 visits, the random order 14.
• At the seventh step the number falls without the model changing: when the measurement policy changes the same model finds 23 illicit cases, because the two features it relied on have been erased; when rebuilt on the drifted data it returns to 30.
• The number of returns is not a measure of defects but proof that the gates work; zero returns means the gates tested nothing.

Next Step

This lesson ran the flow from a single desk: one person wrote the question, chose the features, built the model, set the threshold and monitored the field. In reality these steps are divided among different job definitions, and the division not only shares out the work, it also changes what is delivered. The Introduction to Data Analytics course had separated three analyst roles by their acceptance criteria; the next lesson makes the same separation on the machine learning side. The same illicit-use problem turns into three separate outputs in three roles, and the three choose three separate baselines: for one role the baseline is the field’s current procedure, for another the model already running, for the third the system itself built without a model.