1. Promise

One-sentence promise

After this lecture, you can protect a quant pipeline with unit tests for transforms, dataset invariants, and a smoke test run that catches failures early.

End tag line

Most bugs are not clever. Your tests should not be clever either.

2. Learning objectives

After reading this lecture, you will be able to:

  • Write unit tests for time-series transforms that prevent look-ahead leakage.
  • Enforce invariants: monotonic time, uniqueness, and train-before-test ordering.
  • Detect common pipeline issues: NaNs after feature building, duplicates, non-causal rolling windows.
  • Create a “smoke test” that runs the end-to-end pipeline on a tiny sample in seconds.
  • Produce a minimal test suite that a validator can run.

3. Conceptual core

3.1 Unit tests for transforms

Transform bugs are high impact because they create false performance.

You test transforms for:

  • Causality / no look-ahead
    • Features at time tt must not depend on values at t+1t+1 or later.
  • No NaNs after pipeline
    • After the pipeline’s defined dropna() point, you should not carry NaNs.
  • No silent alignment bugs
    • Feature matrix and target vector must be aligned in time.

Typical transform tests:

  • Lags: lag1[t] == y[t-1].
  • Rolling stats: must be trailing, never centred.
  • Scaling: must be fit on train only, then applied.

Validator sentence

A transform is guilty until proven causal.

3.2 Invariants (dataset integrity)

These invariants must always hold:

  1. Monotonic timestamps
  • index is strictly increasing.
  1. No duplicates
  • no repeated timestamps.
  1. Train < test in time
  • split boundary is time-consistent.
  • no mixing of future observations into training.
  1. No missing required columns
  • pipeline should fail fast.
  1. Stable row counts
  • after each stage, you know what you dropped and why.

Validator sentence

If your index is wrong, everything is wrong.

3.3 Smoke test on small sample

A smoke test is not about correctness in detail. It is about preventing the pipeline from breaking.

A smoke test should:

  • run in < 5 seconds,
  • use a tiny sample (e.g., 200 rows),
  • exercise the full path: data → features → split → fit → predict → score → artefacts,
  • assert that outputs exist and are sane.

Validator sentence

If the smoke test fails, you stop. You do not “try again”.

4. Common failure modes (and what to test)

  • Centred rolling windows (look-ahead).
  • Scaling on full sample (peeking).
  • Label shift wrong sign (target is yt−hy_{t-h} not yt+hy_{t+h}).
  • Duplicate timestamps after joins.
  • Missing days treated as zeros.
  • Feature/target alignment mismatch (off-by-one).

5. Control lens

In control systems: - you validate sensors, - you check monotonic time, - you run a pre-flight check before deployment.

Quant modelling is the same. Testing is your pre-flight checklist.

6. Python lab (mini test suite + smoke test)

This lab uses plain Python assert so it works anywhere. If you use pytest, these functions become test cases directly.

It includes:

  • transform unit tests (lags and trailing rolling),
  • invariants (monotonic, unique, train < test),
  • a smoke test run on a small sample.
import numpy as np
import pandas as pd

# -----------------------------
# Pipeline components (toy)
# -----------------------------
def make_data(T=300, seed=0):
    rng = np.random.default_rng(seed)
    idx = pd.date_range("2024-01-01", periods=T, freq="B")
    y = np.cumsum(rng.normal(0, 1, T))
    return pd.DataFrame({"y": y}, index=idx)

def build_features(df, lags=(1,2,5), roll_window=10):
    d = df.copy()
    for L in lags:
        d[f"lag{L}"] = d["y"].shift(L)

    # trailing rolling mean (LEGAL)
    d[f"roll_mean_{roll_window}"] = d["y"].rolling(roll_window).mean()

    # Example target: 1-step ahead
    d["y_target"] = d["y"].shift(-1)

    # Drop NA from lags/rolling/target
    d = d.dropna().copy()
    return d

def time_split(df_feat, train_frac=0.7):
    n = len(df_feat)
    split = int(train_frac * n)
    train = df_feat.iloc[:split].copy()
    test = df_feat.iloc[split:].copy()
    return train, test

def fit_last_value(train, test):
    # baseline: predict y_target ≈ lag1 (i.e. predict next value = current value)
    y_hat = test["lag1"].values
    y_true = test["y_target"].values
    rmse = float(np.sqrt(np.mean((y_true - y_hat)**2)))
    return rmse

# -----------------------------
# Unit tests: transforms
# -----------------------------
def test_lag_is_causal():
    df = make_data(T=50, seed=1)
    d = build_features(df, lags=(1,2), roll_window=5)

    # pick a few rows and check lag definition
    # For each timestamp in d, lag1 equals y from previous business day in original df
    for ts in d.index[:5]:
        assert np.isclose(d.loc[ts, "lag1"], df.loc[ts - pd.tseries.offsets.BDay(1), "y"])

def test_rolling_is_trailing_not_centered():
    df = make_data(T=60, seed=2)

    # trailing rolling mean (legal)
    d_trail = df["y"].rolling(10).mean()

    # centered rolling mean (ILLEGAL) - uses future values
    d_cent = df["y"].rolling(10, center=True).mean()

    # There exists at least one index where the two differ
    # This is not a proof, but it catches accidental 'center=True' usage.
    diff = (d_trail - d_cent).abs().dropna()
    assert diff.max() > 1e-12

def test_no_nans_after_pipeline():
    df = make_data(T=80, seed=3)
    d = build_features(df, lags=(1,2,5), roll_window=10)
    assert not d.isna().any().any()

# -----------------------------
# Invariants: dataset integrity
# -----------------------------
def test_monotonic_and_unique_index():
    df = make_data(T=100, seed=4)
    assert df.index.is_monotonic_increasing
    assert df.index.is_unique

def test_train_before_test_in_time():
    df = make_data(T=120, seed=5)
    d = build_features(df, lags=(1,2,5), roll_window=10)
    train, test = time_split(d, train_frac=0.7)
    assert train.index.max() < test.index.min()

def test_no_duplicates_after_join_like_operation():
    df = make_data(T=100, seed=6)
    # Simulate a bad join that duplicates timestamps
    bad = pd.concat([df.iloc[:60], df.iloc[50:80]], axis=0)  # overlap creates duplicates
    assert not bad.index.is_unique  # duplicates exist
    # Your pipeline should detect this and fail fast
    try:
        assert bad.index.is_unique
    except AssertionError:
        pass  # expected

# -----------------------------
# Smoke test: end-to-end pipeline
# -----------------------------
def smoke_test_pipeline():
    df = make_data(T=220, seed=7)                 # small sample
    d = build_features(df, lags=(1,2,5), roll_window=10)
    train, test = time_split(d, train_frac=0.7)
    rmse = fit_last_value(train, test)

    # Smoke assertions: pipeline runs and metrics are finite
    assert np.isfinite(rmse)
    assert rmse >= 0.0
    assert len(train) > 0 and len(test) > 0

# -----------------------------
# Run tests (simple runner)
# -----------------------------
tests = [
    test_lag_is_causal,
    test_rolling_is_trailing_not_centered,
    test_no_nans_after_pipeline,
    test_monotonic_and_unique_index,
    test_train_before_test_in_time,
    test_no_duplicates_after_join_like_operation,
    smoke_test_pipeline,
]

for fn in tests:
    fn()

print(f"All tests passed: {len(tests)}")

7. How to use this in a real repo

  • Put tests in tests/test_pipeline.py.
  • Run them on every change (pre-commit or CI).
  • Treat any failure as “stop the line”.

8. Summary

  • Unit tests protect you from leakage and alignment bugs.
  • Invariants protect you from corrupted indices and invalid splits.
  • A smoke test ensures the pipeline runs and produces sane outputs.
  • Testing is a differentiator because it is rare and it prevents expensive mistakes.