Cross-Validation for Time Series: Avoiding Data Leakage

Standard cross-validation destroys financial models. Here's how to properly validate time series predictions.

The Problem with Random Splitting

In standard K-fold CV, data is randomly shuffled:

Random Split (WRONG for time series):
Fold 1: [Day 5, Day 12, Day 3] → predict [Day 8, Day 1]
Fold 2: [Day 1, Day 8, Day 15] → predict [Day 3, Day 10]

Problems:

Temporal leakage: Future data informs past predictions
Autocorrelation: Nearby samples are correlated
Overly optimistic: Inflated performance metrics

Walk-Forward Validation

The gold standard for time series: train on past, test on future.

Expanding Window

$\text{Train}: [1, t], \quad \text{Test}: [t+1, t+h]$

As $t$ increases, training set grows.

Rolling Window

$\text{Train}: [t-w, t], \quad \text{Test}: [t+1, t+h]$

Fixed-size training window slides forward.

python

from sklearn.model_selection import TimeSeriesSplit
import numpy as np

class WalkForwardCV:
    """
    Walk-forward cross-validation for time series.
    """
    def __init__(self, n_splits=5, test_size=None, gap=0, expanding=True):
        """
        Parameters:
        n_splits -- number of folds
        test_size -- size of test set (None = 1/n_splits of data)
        gap -- number of samples to skip between train and test (embargo period)
        expanding -- if True, training set grows; if False, rolling window
        """
        self.n_splits = n_splits
        self.test_size = test_size
        self.gap = gap
        self.expanding = expanding

    def split(self, X, y=None, groups=None):
        n_samples = len(X)

        if self.test_size is None:
            test_size = n_samples // (self.n_splits + 1)
        else:
            test_size = self.test_size

        # Minimum training size
        min_train_size = n_samples - (self.n_splits * test_size) - (self.n_splits * self.gap)

        for i in range(self.n_splits):
            if self.expanding:
                train_start = 0
            else:
                train_start = i * test_size

            train_end = min_train_size + i * test_size
            test_start = train_end + self.gap
            test_end = test_start + test_size

            train_indices = np.arange(train_start, train_end)
            test_indices = np.arange(test_start, min(test_end, n_samples))

            yield train_indices, test_indices

    def get_n_splits(self, X=None, y=None, groups=None):
        return self.n_splits


# Usage example
cv = WalkForwardCV(n_splits=5, gap=5, expanding=True)

for fold, (train_idx, test_idx) in enumerate(cv.split(X)):
    X_train, X_test = X.iloc[train_idx], X.iloc[test_idx]
    y_train, y_test = y.iloc[train_idx], y.iloc[test_idx]

    print(f"Fold {fold + 1}:")
    print(f"  Train: {train_idx[0]} to {train_idx[-1]} ({len(train_idx)} samples)")
    print(f"  Test:  {test_idx[0]} to {test_idx[-1]} ({len(test_idx)} samples)")

Purged K-Fold Cross-Validation

When you need K-fold but with temporal structure preserved.

The Embargo Period

To prevent leakage from labels that span multiple days:

$\text{Gap} = \text{Label Horizon} + \text{Safety Buffer}$

Implementation

python

class PurgedKFold:
    """
    Purged K-Fold cross-validation for financial data.

    Removes training samples that could leak information to test set.
    """
    def __init__(self, n_splits=5, embargo_pct=0.01):
        """
        Parameters:
        n_splits -- number of folds
        embargo_pct -- percentage of data to embargo after each test set
        """
        self.n_splits = n_splits
        self.embargo_pct = embargo_pct

    def split(self, X, y=None, groups=None):
        n_samples = len(X)
        indices = np.arange(n_samples)
        embargo = int(n_samples * self.embargo_pct)

        # Sort by time (assuming index is datetime or sequential)
        test_size = n_samples // self.n_splits

        for i in range(self.n_splits):
            test_start = i * test_size
            test_end = (i + 1) * test_size if i < self.n_splits - 1 else n_samples

            test_indices = indices[test_start:test_end]

            # Purge: remove training samples within embargo of test set
            train_indices = np.concatenate([
                indices[:max(0, test_start - embargo)],
                indices[min(n_samples, test_end + embargo):]
            ])

            yield train_indices, test_indices

    def get_n_splits(self, X=None, y=None, groups=None):
        return self.n_splits

Combinatorial Purged Cross-Validation (CPCV)

For more robust validation, combine multiple train/test paths.

python

from itertools import combinations

class CombinatorialPurgedCV:
    """
    Combinatorial Purged Cross-Validation.

    Tests on all possible combinations of folds while purging.
    """
    def __init__(self, n_splits=5, n_test_splits=2, embargo_pct=0.01):
        self.n_splits = n_splits
        self.n_test_splits = n_test_splits
        self.embargo_pct = embargo_pct

    def split(self, X, y=None, groups=None):
        n_samples = len(X)
        indices = np.arange(n_samples)
        embargo = int(n_samples * self.embargo_pct)

        # Split into groups
        group_size = n_samples // self.n_splits
        groups_list = [indices[i*group_size:(i+1)*group_size]
                      for i in range(self.n_splits)]

        # Handle remainder
        remainder = n_samples % self.n_splits
        if remainder:
            groups_list[-1] = np.concatenate([
                groups_list[-1],
                indices[-remainder:]
            ])

        # Generate all combinations of test groups
        for test_group_indices in combinations(range(self.n_splits), self.n_test_splits):
            test_indices = np.concatenate([groups_list[i] for i in test_group_indices])
            test_indices.sort()

            # Purge training indices
            train_mask = np.ones(n_samples, dtype=bool)
            train_mask[test_indices] = False

            # Apply embargo
            for idx in test_indices:
                start = max(0, idx - embargo)
                end = min(n_samples, idx + embargo + 1)
                train_mask[start:end] = False

            train_indices = indices[train_mask]

            yield train_indices, test_indices

    def get_n_splits(self, X=None, y=None, groups=None):
        from math import comb
        return comb(self.n_splits, self.n_test_splits)

Practical Validation Framework

python

def validate_model(model, X, y, cv_method='walk_forward', n_splits=5):
    """
    Complete model validation with proper metrics.
    """
    from sklearn.metrics import accuracy_score, precision_score, recall_score, f1_score

    # Select CV method
    if cv_method == 'walk_forward':
        cv = WalkForwardCV(n_splits=n_splits, gap=5)
    elif cv_method == 'purged':
        cv = PurgedKFold(n_splits=n_splits, embargo_pct=0.02)
    elif cv_method == 'cpcv':
        cv = CombinatorialPurgedCV(n_splits=n_splits, n_test_splits=2)

    results = {
        'accuracy': [],
        'precision': [],
        'recall': [],
        'f1': [],
        'train_size': [],
        'test_size': []
    }

    for fold, (train_idx, test_idx) in enumerate(cv.split(X)):
        X_train, X_test = X.iloc[train_idx], X.iloc[test_idx]
        y_train, y_test = y.iloc[train_idx], y.iloc[test_idx]

        # Train
        model.fit(X_train, y_train)

        # Predict
        y_pred = model.predict(X_test)

        # Metrics
        results['accuracy'].append(accuracy_score(y_test, y_pred))
        results['precision'].append(precision_score(y_test, y_pred, zero_division=0))
        results['recall'].append(recall_score(y_test, y_pred, zero_division=0))
        results['f1'].append(f1_score(y_test, y_pred, zero_division=0))
        results['train_size'].append(len(train_idx))
        results['test_size'].append(len(test_idx))

    # Summary statistics
    print(f"\nCross-Validation Results ({cv_method}, {n_splits} folds):")
    print(f"{'Metric':<12} {'Mean':>8} {'Std':>8} {'Min':>8} {'Max':>8}")
    print("-" * 45)

    for metric in ['accuracy', 'precision', 'recall', 'f1']:
        values = results[metric]
        print(f"{metric:<12} {np.mean(values):>8.4f} {np.std(values):>8.4f} "
              f"{np.min(values):>8.4f} {np.max(values):>8.4f}")

    return results

Key Takeaways

Never randomly shuffle time series data
Walk-forward validation is the gold standard
Embargo periods prevent label leakage
Purged K-fold when you need multiple test periods
CPCV for most robust estimates
Out-of-sample > In-sample performance always

Proper validation is the difference between backtested profits and real losses!

Share this article

Tweet LinkedIn

Cross-Validation for Time Series: Avoiding Data Leakage

Cross-Validation for Time Series: Avoiding Data Leakage

The Problem with Random Splitting

Walk-Forward Validation

Expanding Window

Rolling Window

Purged K-Fold Cross-Validation

The Embargo Period

Implementation

Combinatorial Purged Cross-Validation (CPCV)

Practical Validation Framework

Key Takeaways

Share this article

TheMLTrader

Related Articles

Neural Networks from Scratch: The Complete Mathematical Guide

LSTM Networks for Time Series Forecasting

Feature Engineering for Algorithmic Trading

Ready to Apply These Concepts?