Chapter 33: Data Augmentation for Tabular Data

import numpy as np
from sklearn.ensemble import HistGradientBoostingClassifier
from sklearn.metrics import average_precision_score
from sklearn.model_selection import StratifiedKFold
from sklearn.neighbors import NearestNeighbors

from kaggle_companion.activities._common import clean

SEED = 33
REPLICATES = 8          # independent constructed datasets; every estimate is their mean
N_TRAIN, N_NEW, FEATURES = 4000, 8000, 10


def generate(n, prevalence, rng):
    """Ten standard-normal features; the minority class is shifted by 1.0, 0.8 and 0.6 standard deviations on the first three."""
    y = (rng.random(n) < prevalence).astype(int)
    X = rng.normal(size=(n, FEATURES))
    X[:, :3] += y[:, None] * np.array([1.0, 0.8, 0.6])
    return X, y


def smote(X, y, rng, k=5):
    """Interpolate between each minority row and one of its k nearest minority neighbours until the classes are balanced."""
    minority = X[y == 1]
    need = int((y == 0).sum() - len(minority))
    if need <= 0 or len(minority) < 2:
        return X, y
    k = min(k, len(minority) - 1)
    neighbours = NearestNeighbors(n_neighbors=k + 1).fit(minority).kneighbors(minority)[1][:, 1:]   # column 0 is the row itself
    start = rng.integers(0, len(minority), need)
    partner = neighbours[start, rng.integers(0, k, need)]
    synthetic = minority[start] + rng.random((need, 1)) * (minority[partner] - minority[start])
    return np.vstack([X, synthetic]), np.concatenate([y, np.ones(need, int)])


def model(class_weight=None):
    return HistGradientBoostingClassifier(max_iter=30, max_leaf_nodes=8, max_bins=32, learning_rate=0.1,
                                          class_weight=class_weight, random_state=0)


def one_replicate(prevalence, seed):
    rng = np.random.default_rng(seed)
    X, y = generate(N_TRAIN, prevalence, rng)
    X_new, y_new = generate(N_NEW, prevalence, rng)             # new rows from the same generator: the "truth"
    folds = list(StratifiedKFold(3, shuffle=True, random_state=1).split(X, y))

    def pooled_cv(fit_predict):
        """Pool the held-out predictions of all folds and compute one average precision over them."""
        pooled = np.zeros(N_TRAIN)
        for train, valid in folds:
            pooled[valid] = fit_predict(train, valid)
        return average_precision_score(y, pooled)

    out = {}
    out["none"] = (pooled_cv(lambda a, b: model().fit(X[a], y[a]).predict_proba(X[b])[:, 1]),
                   average_precision_score(y_new, model().fit(X, y).predict_proba(X_new)[:, 1]))
    out["weight"] = (pooled_cv(lambda a, b: model("balanced").fit(X[a], y[a]).predict_proba(X[b])[:, 1]),
                     average_precision_score(y_new, model("balanced").fit(X, y).predict_proba(X_new)[:, 1]))

    def smote_in_fold(train, valid):                            # resample the training fold only; validation rows stay untouched
        Xr, yr = smote(X[train], y[train], np.random.default_rng(seed + int(train[0])))
        return model().fit(Xr, yr).predict_proba(X[valid])[:, 1]

    X_all, y_all = smote(X, y, np.random.default_rng(seed + 1))   # SMOTE on the whole training set: the shortcut
    deployed = average_precision_score(y_new, model().fit(X_all, y_all).predict_proba(X_new)[:, 1])
    out["inside"] = (pooled_cv(smote_in_fold), deployed)

    # SMOTE before the split: cross-validate the resampled table, so synthetic rows built from a validation row's neighbours
    # sit in the training folds. Scoring only the original rows keeps the class balance honest; scoring everything is worse.
    pooled = np.zeros(len(y_all))
    for train, valid in StratifiedKFold(3, shuffle=True, random_state=1).split(X_all, y_all):
        pooled[valid] = model().fit(X_all[train], y_all[train]).predict_proba(X_all[valid])[:, 1]
    original = np.arange(len(y_all)) < N_TRAIN
    out["before"] = (average_precision_score(y_all[original], pooled[original]), deployed)
    out["before_all_rows"] = (average_precision_score(y_all, pooled), deployed)
    return out


def run(prevalence):
    reps = [one_replicate(prevalence, SEED * 100 + i) for i in range(REPLICATES)]
    result = {"prevalence": prevalence, "replicates": REPLICATES, "training_rows": N_TRAIN, "new_rows": N_NEW,
              "training_positives": round(N_TRAIN * prevalence)}
    for name in ("none", "weight", "inside", "before", "before_all_rows"):
        result[name] = {"cv": float(np.mean([r[name][0] for r in reps])), "new_rows": float(np.mean([r[name][1] for r in reps]))}
    return clean(result)

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

VALUES = [0.02, 0.05, 0.15, 0.3]
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.ch33 import draw, NCOLS, HEIGHT

DEFAULT = 0.05
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()