Pipelines

One sentence was repeated over four lessons: this step is learned from the training set only. The region median, the scale mean, the selected column set, the component directions — all four are numbers read from the data, and all four have to be read from the right set. As the sentence was repeated, another thing became visible: all of these steps have the same shape. Each has a fit face — it extracts a number or a mapping from the data — and a transform face — it works that number into the rows it is given.

This shape is an object definition. When the steps are called one by one by hand, the order can get mixed up, a step can learn from the wrong set, a step can be forgotten at test time. The structure that gathers the steps in a single object is called a pipeline: the pipeline too has a fit face and sees only the training set, and it has a transform face and applies the same learned numbers to every set. This lesson writes the pipeline and asks a single question: when the same chain is applied by hand, how many steps leak, and how much does the number inflate.

• FS30. The chain is four steps: missing value imputation, scaling, target encoding, feature selection. The model comes fifth and does not change: a decision tree of depth 6.
• FS31. The missing satisfaction values and the tariff_code column are fictional, generated from the same seed. The tariff code depends on the region and the household size, so it carries a real relation with the label.
• FS32. Target encoding learns the suspicious rate for each level of the tariff code. An unseen category falls to the overall rate.
• FS33. In the leakage measurement the first k steps learn from the whole set and the rest from the training set; the model is trained from the training set only, in every case.
• FS34. The inflation is the difference from the pipeline’s number on the test set. The baseline is 0.7579.
• FS35. The Data Leakage lesson of the Introduction to Machine Learning course counted the four kinds of leakage one by one; the channel here is its first kind — preparation before the split — and the inflation numbers are not repeated.

The Two Faces of a Step

# COMMON — MODEL: the FICTIONAL subscriber table of the Introduction to Machine Learning course is rebuilt with the same seed;
# the model is a decision tree of depth 6. The only new column left from the previous lesson is per_person.
import math

SEED, RAW, M32 = 20260218, 1400, 0xFFFFFFFF
REGION = [("north", 0.28, 21, 0.00), ("south", 0.22, 17, -0.10),
          ("east", 0.18, 26, 0.30), ("west", 0.14, 14, -0.05),
          ("central", 0.18, 23, 0.05)]


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, DATA = [], []
for i in range(RAW):
    r = rng(SEED + i)
    b = REGION[pick(r(), [x[1] for x in REGION])]
    household = 1 + pick(r(), [0.06, 0.24, 0.30, 0.24, 0.11, 0.05])
    rating = 1 + pick(r(), [0.08, 0.14, 0.27, 0.34, 0.17])
    if r() >= 0.046:
        SUBSCRIBERS.append({"id": 10001 + i, "region": b, "household": household,
                            "satisfaction": rating})
for k in SUBSCRIBERS:
    r, v = rng(SEED + 7000 + k["id"]), []
    for n in range(pick(r(), [0.05, 0.12, 0.21, 0.62])):
        v.append(0.0 if r() < 0.038 else math.floor(
            k["region"][2] * math.exp((r() + r() + r() - 1.5) * 0.62)
            * (1 - n * 0.05) * 100 + 0.5) / 100)
        r()
    if not v:
        continue
    mean, vol, r = sum(v) / len(v), max(v) - min(v), rng(SEED + 51000 + k["id"])
    z = (0.052 * (mean - 20) + 0.85 * (min(v) == 0.0) + 0.026 * vol
         + 0.24 * (k["household"] >= 5) - 0.20 * (k["satisfaction"] >= 4) + k["region"][3]
         + 0.9 * (k["id"] - 10001) / 1399 + (r() + r() + r() - 1.5) * 1.30)
    x = {"id": k["id"], "household": k["household"], "satisfaction": k["satisfaction"],
         "period": len(v), "mean_consumption": round(mean, 2), "volatility": round(vol, 2),
         "zero_readings": int(min(v) == 0.0), "suspicious": int(z > 1.35),
         "region_name": k["region"][0]}
    for b in REGION:
        x["area_" + b[0]] = int(k["region"][0] == b[0])
    DATA.append(x)

BASE = ["mean_consumption", "volatility", "household", "satisfaction", "period",
        "zero_readings", "area_north", "area_south", "area_east", "area_west", "area_central"]
BASE_FIELDS = BASE + ["per_person"]
for x in DATA:
    x["per_person"] = round(x["mean_consumption"] / x["household"], 3)


def shuffle(data, seed):
    r, s = rng(seed), list(range(len(data)))
    for i in range(len(s) - 1, 0, -1):
        j = int(r() * (i + 1))
        s[i], s[j] = s[j], s[i]
    return [data[i] for i in s]


K = shuffle(DATA, SEED + 90000)
TRAIN, VAL, TEST = K[:756], K[756:1008], K[1008:]


def gini(rows):
    p = sum(x["suspicious"] for x in rows) / len(rows) if rows else 0.0
    return 2 * p * (1 - p)


def tree(rows, depth, fields, min_size=2):
    p, best = sum(x["suspicious"] for x in rows) / len(rows), None
    if depth and len(rows) >= 2 * min_size and 0.0 < p < 1.0:
        for f in fields:
            values = sorted({x[f] for x in rows})
            for v in (values[1:] if len(values) < 10 else
                      [values[int(i * len(values) / 10)] for i in range(1, 10)]):
                left = [x for x in rows if x[f] < v]
                right = [x for x in rows if x[f] >= v]
                if min(len(left), len(right)) < min_size:
                    continue
                k = gini(rows) - (len(left) * gini(left) + len(right) * gini(right)) / len(rows)
                if best is None or k > best[0]:
                    best = (k, f, v, left, right)
    if best is None or best[0] <= 1e-9:
        return {"prediction": int(p > 0.5)}
    return {"field": best[1], "threshold": best[2],
            "left": tree(best[3], depth - 1, fields, min_size),
            "right": tree(best[4], depth - 1, fields, min_size)}


def predict(node, x):
    while "prediction" not in node:
        node = node["left"] if x[node["field"]] < node["threshold"] else node["right"]
    return node["prediction"]


def accuracy(node, rows):
    return sum(predict(node, x) == x["suspicious"] for x in rows) / len(rows)


def baseline(train, rows):
    c = int(sum(x["suspicious"] for x in train) * 2 > len(train))
    return sum(x["suspicious"] == c for x in rows) / len(rows)

All four steps carry the same two methods. fit looks only at the rows given to it and writes what it has learned to the state field; apply works that state into the rows it is given and learns nothing. The pipeline carries this pair to the whole chain.

for x in DATA:
    r = rng(SEED + 61000 + x["id"])
    r(), r()
    x["kw_seal"] = int(r() < (0.42 if x["suspicious"] else 0.10))
    r = rng(SEED + 21000 + x["id"])
    x["tariff_code"] = (24 * [b[0] for b in REGION].index(x["region_name"])
                        + 4 * (x["household"] - 1) + int(r() * 4))
    x["satisfaction"] = None if r() < 0.08 else x["satisfaction"]
POOL = BASE_FIELDS + ["kw_seal", "tariff_target"]


class Imputer:
    name = "imputation"

    def fit(self, rows):
        v = sorted(x["satisfaction"] for x in rows if x["satisfaction"] is not None)
        self.state = {"median": v[len(v) // 2]}

    def apply(self, rows):
        for x in rows:
            if x["satisfaction"] is None:
                x["satisfaction"] = self.state["median"]


class Scaler:
    name = "scaling"

    def fit(self, rows):
        self.state = {}
        for f in ("mean_consumption", "volatility", "per_person"):
            m = sum(x[f] for x in rows) / len(rows)
            self.state[f] = (m, max(1e-9, (sum((x[f] - m) ** 2 for x in rows)
                                           / len(rows)) ** 0.5))

    def apply(self, rows):
        for x in rows:
            for f, (m, d) in self.state.items():
                x[f] = round((x[f] - m) / d, 4)


class TargetEncoder:
    name = "target encoding"

    def fit(self, rows):
        g, self.overall = {}, sum(x["suspicious"] for x in rows) / len(rows)
        for x in rows:
            g.setdefault(x["tariff_code"], []).append(x["suspicious"])
        self.state = {k: sum(v) / len(v) for k, v in g.items()}

    def apply(self, rows):
        for x in rows:
            x["tariff_target"] = round(self.state.get(x["tariff_code"], self.overall), 4)


class Selector:
    name = "selection"

    def __init__(self, pool):
        self.pool = pool

    def fit(self, rows):
        g0, self.state = gini(rows), []
        for f in self.pool:
            values, top = sorted({x[f] for x in rows}), 0.0
            for v in (values[1:] if len(values) < 10 else
                      [values[int(i * len(values) / 10)] for i in range(1, 10)]):
                left = [x for x in rows if x[f] < v]
                right = [x for x in rows if x[f] >= v]
                if min(len(left), len(right)) >= 20:
                    top = max(top, g0 - (len(left) * gini(left)
                                         + len(right) * gini(right)) / len(rows))
            if top >= 0.004:
                self.state.append(f)

    def apply(self, rows):
        pass


class Pipeline:
    def __init__(self, steps, depth=6):
        self.steps, self.depth = steps, depth

    def fit(self, train, leakage_source=None, leaked=0):
        train = [dict(x) for x in train]
        dirty = [dict(x) for x in (leakage_source or [])]
        for i, step in enumerate(self.steps):
            if i < leaked:
                step.fit(dirty)
                step.apply(dirty)
            else:
                step.fit(train)
            step.apply(train)
        self.model = tree(train, self.depth, self.steps[-1].state)
        self.train_score = accuracy(self.model, train)
        return self

    def measure(self, rows):
        rows = [dict(x) for x in rows]
        for step in self.steps:
            step.apply(rows)
        return accuracy(self.model, rows)


def chain(pool=None):
    return [Imputer(), Scaler(), TargetEncoder(), Selector(pool or POOL)]


PL = Pipeline(chain()).fit(TRAIN)
print(f"{'setup':<26} {'leaked':>6} {'columns':>7} {'train':>9} {'held_out':>9} "
      f"{'inflation':>9}")
print(f"{'pipeline':<26} {0:>6} {len(PL.steps[-1].state):>7} {PL.train_score:>9.4f} "
      f"{PL.measure(TEST):>9.4f}")
for k, label in ((1, "manual: imputation"), (2, "manual: + scaling"),
                 (3, "manual: + target encoding"), (4, "manual: + selection")):
    b = Pipeline(chain()).fit(TRAIN, K, k)
    print(f"{label:<26} {k:>6} {len(b.steps[-1].state):>7} {b.train_score:>9.4f} "
          f"{b.measure(TEST):>9.4f} {b.measure(TEST) - PL.measure(TEST):>+9.4f}")
print(f"\nbaseline {baseline(TRAIN, TEST):.4f}; tariff code "
      f"{len({x['tariff_code'] for x in DATA})} levels, missing satisfaction "
      f"{sum(1 for x in DATA if x['satisfaction'] is None)} rows")

setup                      leaked columns     train  held_out inflation
pipeline                        0       8    0.8955    0.7817
manual: imputation              1       8    0.8955    0.7817   +0.0000
manual: + scaling               2       8    0.8955    0.7817   +0.0000
manual: + target encoding       3       8    0.9021    0.8214   +0.0397
manual: + selection             4       9    0.9008    0.8294   +0.0476

baseline 0.7579; tariff code 119 levels, missing satisfaction 103 rows

When the chain is applied by hand, all four steps can be learned from the whole set, because all four are a fit call, and which list is given to that call is written somewhere else in the code. In the pipeline this decision sits in a single place: the fit method takes the training set and the whole chain learns from that set. Adding a step does not increase the risk of leakage.

The inflation column has to be read step by step. The first two rows are zero: when the imputation value is taken from the whole set the number does not change, and when the scale is taken from the whole set it does not change either. This does not mean those two steps did not leak — they did; the measured number did not change in this set and with this model. The decision tree makes its splits by ranking, and standardizing does not change the ranking; the median imputation affects 103 rows, and the two medians are close to each other. The non-appearance of a leak is not proof of its absence.

The third row is target encoding, and the number inflates by 0.0397: ten subscribers. The reason is clear — this step uses the label directly. The tariff code has 119 levels and on average ten rows fall to each level; when the encoding is learned from the whole set, the own label of a test row makes up one tenth of the number that row will see. In the fourth row the selection is also made from the whole set, the number of selected columns goes from eight to nine, and the total inflation reaches 0.0476, twelve subscribers.

The second rule of the course is completed in this table: every preparation step is learned from the training set only and applied to the held-out set. The pipeline turns this rule into a procedure; the rule stops being a written warning and becomes the interface of the object.

What the Pipeline Measures

The second benefit of the pipeline is not preventing leakage but making the steps measurable. When a step is removed from the chain and the same measurement is repeated, the honest contribution of that step can be read.

NARROW = BASE_FIELDS + ["kw_seal"]
print("objects learned by the pipeline: " + ", ".join(
    f"{s.name} {len(s.state)}" for s in PL.steps))
seen = set(PL.steps[2].state)
print(f"{len(seen)} of the {len({x['tariff_code'] for x in DATA})} levels of the tariff code "
      f"were seen in training; rows in the test set falling into an unseen category: "
      f"{sum(x['tariff_code'] not in seen for x in TEST)}\n")
print(f"{'chain':<26} {'steps':>5} {'columns':>7} {'train':>9} {'held_out':>9} "
      f"{'contrib':>8}")
print(f"{'four steps':<26} {4:>5} {len(PL.steps[-1].state):>7} {PL.train_score:>9.4f} "
      f"{PL.measure(TEST):>9.4f}")
for label, steps in (("scaling removed", [Imputer(), TargetEncoder(), Selector(POOL)]),
                     ("target encoding removed", [Imputer(), Scaler(), Selector(NARROW)]),
                     ("only imputation+selection", [Imputer(), Selector(NARROW)])):
    b = Pipeline(steps).fit(TRAIN)
    print(f"{label:<26} {len(steps):>5} {len(b.steps[-1].state):>7} {b.train_score:>9.4f} "
          f"{b.measure(TEST):>9.4f} {b.measure(TEST) - PL.measure(TEST):>+8.4f}")

objects learned by the pipeline: imputation 1, scaling 3, target encoding 116, selection 8
116 of the 119 levels of the tariff code were seen in training; rows in the test set falling into an unseen category: 1

chain                      steps columns     train  held_out  contrib
four steps                     4       8    0.8955    0.7817
scaling removed                3       8    0.8955    0.7817  +0.0000
target encoding removed        3       7    0.8730    0.8135  +0.0317
only imputation+selection      2       7    0.8730    0.8135  +0.0317

The pipeline learns 131 numbers in total: one median, two numbers each for three columns, one rate each for 116 tariff levels, and a list of eight columns. All of these 131 numbers come from the 756 rows of the training set and none of them looks at the test set. The leakage discussion is a discussion of how much of this count comes from which set.

Of the 119 levels of the tariff code, 116 were seen in the training set; one of the remaining three levels meets us in the test set with a single row and falls to the overall rate as an unseen category. In a hand-written preparation chain this row typically turns into an error or a silent empty value; in the pipeline the transform face of the step makes the decision once and makes the same decision in every set.

The lower table reads the honest contribution of two steps. When scaling is removed the number does not change at all: the contribution is exactly zero, because the decision tree looks at the ranking and a linear scale transformation does not change the ranking. When the step is done wrong it changes nothing either, so it can stay silently wrong. When target encoding is removed, on the other hand, the number rises from 0.7817 to 0.8135: the honest contribution of the step is minus 0.0317. Encoding a high-cardinality column with the target does harm in this set even when it is learned from the right set; when it was learned from the whole set it seemed to bring 0.0397. The same step is the most profitable step when it is leaked and the most harmful when it is applied honestly.

For this reason the before-the-decision and after-the-decision numbers of this lesson are read in opposite directions. The chain applied by hand gives 0.8294, the pipeline 0.7817; the contribution is minus 0.0476. The minus sign here does not show a loss but that the lost number was never learned. What the pipeline adds is not a number but the correctness of the number.

Summary

• The preparation step is an object with two faces: fit sees only the training set and stores what it has learned, apply works those numbers into every set without learning anything.
• When the chain is applied by hand, all four steps can be learned from the whole set; in the pipeline the decision sits in a single place and adding a step does not increase the risk of leakage.
• The inflation comes out as 0.0000, 0.0000, 0.0397 and 0.0476 step by step; the leakage of the first two steps does not change the number, and the non-appearance of a leak is not proof of its absence.
• The pipeline learns 131 numbers and all of them come from the 756 rows of the training set; 116 of the 119 levels of the tariff code were seen, and one row in the test set falls into an unseen category.
• The same step brings 0.0397 when it is leaked and loses 0.0317 when it is applied honestly; the contribution of the pipeline is minus 0.0476, and this minus shows that the lost number was never learned.

Next Step

The pipeline bound the whole chain to learning from a single set, and the measurement was read from a single split: 756 rows of training, 252 rows of testing. Those 252 rows were used once and the number they gave is a single number; another split would have given another number. Taking the measurement over several splits is called cross-validation, and when the preparation chain enters here a new question arises: is the chain learned once outside the folds, or again inside each fold. The two procedures give two different numbers, and the next lesson measures which one comes out closer to the real number on the held-out set.