Chapter 41: Online Learning

import numpy as np

from kaggle_companion.activities._common import clean

SEED = 41
REPLICATES = 20         # independent streams per delay; results are means over them
STEPS, WARM, SELECT = 2000, 300, 500   # prediction origins run from WARM; the first SELECT origins choose each policy's setting
FEATURES, RIDGE = 3, 1.0
DRIFT_NOISE, DRIFT_PULL = 0.12, 0.02   # the drifting part wanders and is pulled back toward zero
WINDOWS = (30, 60, 120, 240)           # candidate rolling-window lengths, in released rows
DISCOUNTS = (0.95, 0.98, 0.99, 0.995)  # candidate forgetting factors for the incremental policy
POINTS = 40                            # resolution of the cumulative error curves


def make_stream(rng):
    """Row t has features x_t and label y_t = w_t . x_t + noise, where w_t = a stable part + a drifting part."""
    X = rng.normal(size=(STEPS, FEATURES))
    stable = rng.normal(0, 1, FEATURES)                    # the part of the signal a frozen model can keep
    spread = DRIFT_NOISE / np.sqrt(1 - (1 - DRIFT_PULL) ** 2)   # long-run spread of the drifting part
    drift = np.zeros((STEPS, FEATURES))
    drift[0] = rng.normal(0, spread, FEATURES)
    for t in range(1, STEPS):
        drift[t] = drift[t - 1] * (1 - DRIFT_PULL) + DRIFT_NOISE * rng.normal(size=FEATURES)
    w = stable + drift
    return X, (w * X).sum(axis=1) + rng.normal(0, 1, STEPS)


def ridge(X, y):
    return np.linalg.solve(X.T @ X + RIDGE * np.eye(FEATURES), X.T @ y)


def replay(X, y, delay):
    """Squared error at every origin t >= WARM for each policy. Row i is released at i + delay and usable only at origins t > i + delay."""
    origins = range(WARM, STEPS)
    errors = {"frozen": None}
    frozen = ridge(X[:WARM - delay], y[:WARM - delay])           # fitted once, on rows released before the first origin
    errors["frozen"] = (y[WARM:] - X[WARM:] @ frozen) ** 2
    for window in WINDOWS:
        eligible, ignoring_delay = [], []
        for t in origins:
            hi = t - delay                                         # rows 0 .. hi-1 are released before origin t
            w = ridge(X[max(0, hi - window):hi], y[max(0, hi - window):hi])
            eligible.append((y[t] - X[t] @ w) ** 2)
            w = ridge(X[max(0, t - window):t], y[max(0, t - window):t])   # INVALID: also uses labels not yet released
            ignoring_delay.append((y[t] - X[t] @ w) ** 2)
        errors[("rolling", window)], errors[("invalid", window)] = np.array(eligible), np.array(ignoring_delay)
    for discount in DISCOUNTS:                                     # incremental: discounted ridge, one released row at a time
        A, b = np.zeros((FEATURES, FEATURES)), np.zeros(FEATURES)
        for i in range(WARM - delay):
            A, b = discount * A + np.outer(X[i], X[i]), discount * b + X[i] * y[i]
        errs = []
        for t in origins:
            i = t - delay - 1                                      # the row released since the previous origin
            if i >= WARM - delay:
                A, b = discount * A + np.outer(X[i], X[i]), discount * b + X[i] * y[i]
            errs.append((y[t] - X[t] @ np.linalg.solve(A + RIDGE * np.eye(FEATURES), b)) ** 2)
        errors[("incremental", discount)] = np.array(errs)
    return errors


def pick(errors, kind, options):
    """Choose the setting with the lowest error on the first SELECT origins; report it on the later origins only."""
    best = min(options, key=lambda o: errors[(kind, o)][:SELECT].mean())
    return best, errors[(kind, best)][SELECT:]


def run(delay):
    scores = {k: [] for k in ("frozen", "rolling", "incremental", "invalid")}
    settings = {"rolling": [], "incremental": []}
    curves = {k: [] for k in scores}
    for r in range(REPLICATES):
        X, y = make_stream(np.random.default_rng(SEED * 100 + r))
        errors = replay(X, y, delay)
        later = {"frozen": errors["frozen"][SELECT:]}
        for kind, options in (("rolling", WINDOWS), ("incremental", DISCOUNTS), ("invalid", WINDOWS)):
            chosen, later[kind] = pick(errors, kind, options)
            if kind in settings:
                settings[kind].append(chosen)
        for k, e in later.items():
            scores[k].append(e.mean())
            running = np.cumsum(e) / np.arange(1, len(e) + 1)
            curves[k].append(running[np.linspace(0, len(e) - 1, POINTS).astype(int)])
    mean = {k: float(np.mean(v)) for k, v in scores.items()}
    return clean({
        "delay": delay, "mse": mean,
        "gain_rolling": mean["frozen"] - mean["rolling"], "gain_incremental": mean["frozen"] - mean["incremental"],
        "gain_invalid": mean["frozen"] - mean["invalid"],
        "selected_window": float(np.mean(settings["rolling"])), "selected_discount": float(np.mean(settings["incremental"])),
        "curve_steps": np.linspace(1, STEPS - WARM - SELECT, POINTS), "curves": {k: np.mean(v, axis=0) for k, v in curves.items()},
        "noise_floor": 1.0, "replicates": REPLICATES, "scored_origins": STEPS - WARM - SELECT,
    })

import pandas as pd
from kaggle_companion.activities.ch41 import explain

VALUES = [0, 10, 40, 100]
results = {value: run(value) for value in VALUES}
table = pd.DataFrame({str(v): explain(results[v], v)['metrics'] for v in VALUES})
print(table.to_string())

import matplotlib.pyplot as plt
from kaggle_companion.activities.ch41 import draw, NCOLS, HEIGHT

DEFAULT = 40
fig, axes = plt.subplots(1, NCOLS, figsize=(6.6 if NCOLS == 1 else 10.4, HEIGHT), layout='constrained')
draw(axes, results[DEFAULT], DEFAULT)
plt.show()