Cross-Validation for Trading Models: Avoiding Look-Ahead Bias
Cross-validation is the cornerstone of honest model evaluation. In finance, standard k-fold cross-validation produces wildly overoptimistic results because it ignores the temporal ordering of data, allowing the model to "peek" at future information during training. A single mis-applied cross-validation can make a worthless model appear to generate 20% annual alpha. This guide covers the cross-validation techniques designed specifically for financial time series, including walk-forward analysis, purged k-fold, and combinatorial purged cross-validation. Each method addresses different forms of data leakage while providing reliable estimates of out-of-sample performance. If your labels come from a triple-barrier or meta-labeling scheme, the overlap between adjacent label windows is exactly the kind of leakage purging and embargo periods are designed to remove.Key Takeaways
- Standard k-fold CV is invalid for time series. It randomly mixes past and future data, creating severe look-ahead bias.
- Walk-forward analysis is the gold standard but expensive: each fold requires a full model retraining.
- Purged k-fold with embargo allows traditional CV structure while preventing temporal leakage through purging and embargo periods.
- Multiple CV methods should agree. If walk-forward shows profit but purged k-fold does not, the signal is likely regime-dependent.
The Problem with Standard Cross-Validation
Standard k-fold randomly assigns observations to folds, which creates two forms of leakage in time series. ``python
import numpy as np
import pandas as pd
from sklearn.model_selection import KFold, TimeSeriesSplit
from sklearn.ensemble import GradientBoostingClassifier
from sklearn.metrics import accuracy_score
def demonstrate_cv_bias(
X: pd.DataFrame,
y: pd.Series,
model=None,
) -> dict:
"""
Show the performance gap between standard CV and temporal CV.
Standard CV will show inflated performance.
"""
model = model or GradientBoostingClassifier(
n_estimators=100, max_depth=4, random_state=42
)
# Standard k-fold (WRONG for time series)
kf = KFold(n_splits=5, shuffle=True, random_state=42)
standard_scores = []
for train_idx, test_idx in kf.split(X):
model.fit(X.iloc[train_idx], y.iloc[train_idx])
pred = model.predict(X.iloc[test_idx])
standard_scores.append(accuracy_score(y.iloc[test_idx], pred))
# Time series split (CORRECT)
tscv = TimeSeriesSplit(n_splits=5)
temporal_scores = []
for train_idx, test_idx in tscv.split(X):
model.fit(X.iloc[train_idx], y.iloc[train_idx])
pred = model.predict(X.iloc[test_idx])
temporal_scores.append(accuracy_score(y.iloc[test_idx], pred))
results = {
"standard_cv_mean": np.mean(standard_scores),
"standard_cv_std": np.std(standard_scores),
"temporal_cv_mean": np.mean(temporal_scores),
"temporal_cv_std": np.std(temporal_scores),
"inflation": np.mean(standard_scores) - np.mean(temporal_scores),
}
print(f"Standard k-fold: {results['standard_cv_mean']:.3f} +/- {results['standard_cv_std']:.3f}")
print(f"Temporal CV: {results['temporal_cv_mean']:.3f} +/- {results['temporal_cv_std']:.3f}")
print(f"Inflation: {results['inflation']:.3f} ({results['inflation']/results['temporal_cv_mean']*100:.1f}%)")
return results
`
Walk-Forward Analysis
Walk-forward analysis simulates how the model would perform in production by training on expanding or rolling windows and evaluating on subsequent out-of-sample periods. If you'd rather not hand-roll this walk-forward loop, our Strategy Builder runs out-of-sample backtests directly, no signup required.
`python
class WalkForwardAnalysis:
"""
Walk-forward cross-validation with expanding or rolling windows.
Modes:
- 'expanding': Training window grows over time (anchored)
- 'rolling': Training window slides (fixed size)
"""
def __init__(
self,
mode: str = "expanding",
train_size: int = 504, # 2 years
test_size: int = 63, # 1 quarter
gap: int = 5, # Gap between train/test
step_size: int = 63, # Step between folds
):
self.mode = mode
self.train_size = train_size
self.test_size = test_size
self.gap = gap
self.step_size = step_size
def split(
self, X: pd.DataFrame
) -> list[tuple[np.ndarray, np.ndarray]]:
"""Generate train/test index pairs."""
n = len(X)
splits = []
if self.mode == "expanding":
start = 0
test_start = self.train_size + self.gap
while test_start + self.test_size <= n:
train_idx = np.arange(start, test_start - self.gap)
test_idx = np.arange(test_start, min(test_start + self.test_size, n))
splits.append((train_idx, test_idx))
test_start += self.step_size
elif self.mode == "rolling":
test_start = self.train_size + self.gap
while test_start + self.test_size <= n:
train_start = test_start - self.gap - self.train_size
train_idx = np.arange(max(0, train_start), test_start - self.gap)
test_idx = np.arange(test_start, min(test_start + self.test_size, n))
splits.append((train_idx, test_idx))
test_start += self.step_size
return splits
def evaluate(
self,
X: pd.DataFrame,
y: pd.Series,
model,
returns: pd.Series = None,
) -> pd.DataFrame:
"""
Run walk-forward analysis with comprehensive evaluation.
"""
splits = self.split(X)
results = []
for fold, (train_idx, test_idx) in enumerate(splits):
X_train = X.iloc[train_idx]
y_train = y.iloc[train_idx]
X_test = X.iloc[test_idx]
y_test = y.iloc[test_idx]
# Train
model.fit(X_train, y_train)
# Predict
y_pred = model.predict(X_test)
accuracy = accuracy_score(y_test, y_pred)
fold_result = {
"fold": fold,
"train_start": X.index[train_idx[0]],
"train_end": X.index[train_idx[-1]],
"test_start": X.index[test_idx[0]],
"test_end": X.index[test_idx[-1]],
"train_size": len(train_idx),
"test_size": len(test_idx),
"accuracy": accuracy,
}
# Financial metrics if returns provided
if returns is not None:
test_returns = returns.iloc[test_idx]
strategy_ret = np.where(y_pred == 1, test_returns, 0)
fold_result["total_return"] = (1 + strategy_ret).prod() - 1
fold_result["sharpe"] = (
np.mean(strategy_ret) / np.std(strategy_ret) * np.sqrt(252)
if np.std(strategy_ret) > 0 else 0
)
results.append(fold_result)
results_df = pd.DataFrame(results)
print(f"\nWalk-Forward Results ({self.mode}, {len(splits)} folds):")
print(f" Avg Accuracy: {results_df['accuracy'].mean():.3f} +/- {results_df['accuracy'].std():.3f}")
if "sharpe" in results_df:
print(f" Avg Sharpe: {results_df['sharpe'].mean():.3f} +/- {results_df['sharpe'].std():.3f}")
print(f" Min Accuracy: {results_df['accuracy'].min():.3f}")
print(f" Max Accuracy: {results_df['accuracy'].max():.3f}")
return results_df
`
Purged K-Fold Cross-Validation
Purged k-fold removes observations near fold boundaries to prevent information leakage from overlapping labels. This is especially important when labels span multiple time periods.
`python
class PurgedKFold:
"""
Purged k-fold cross-validation for financial data.
Purging: Removes training samples whose labels overlap with
test samples, preventing leakage through overlapping targets.
Embargo: Removes a buffer of training samples after each test
fold to prevent leakage through autocorrelated features.
Reference: Marcos Lopez de Prado, "Advances in Financial ML"
"""
def __init__(
self,
n_splits: int = 5,
purge_window: int = 5,
embargo_pct: float = 0.01,
):
self.n_splits = n_splits
self.purge_window = purge_window
self.embargo_pct = embargo_pct
def split(
self,
X: pd.DataFrame,
y: pd.Series = None,
label_end_dates: pd.Series = None,
) -> list[tuple[np.ndarray, np.ndarray]]:
"""
Generate purged train/test splits.
Args:
X: feature DataFrame with DatetimeIndex
label_end_dates: for each sample, when its label period ends
(used for precise purging)
"""
n = len(X)
embargo_size = max(1, int(n * self.embargo_pct))
fold_size = n // self.n_splits
indices = np.arange(n)
splits = []
for i in range(self.n_splits):
test_start = i * fold_size
test_end = min((i + 1) * fold_size, n)
test_idx = indices[test_start:test_end]
# Initial train indices (everything not in test)
train_idx = np.concatenate([
indices[:test_start],
indices[test_end:],
])
# Purge: remove train samples whose labels overlap test period
if label_end_dates is not None:
test_start_date = X.index[test_start]
test_end_date = X.index[test_end - 1]
# Remove train samples whose label extends into test period
purge_mask = np.ones(len(train_idx), dtype=bool)
for j, idx in enumerate(train_idx):
if idx < n and label_end_dates.iloc[idx] >= test_start_date:
purge_mask[j] = False
train_idx = train_idx[purge_mask]
else:
# Simple purge: remove samples within purge_window of test boundaries
purge_before = set(range(
max(0, test_start - self.purge_window), test_start
))
purge_after = set(range(
test_end, min(n, test_end + self.purge_window)
))
purge_set = purge_before | purge_after
train_idx = np.array([i for i in train_idx if i not in purge_set])
# Embargo: remove train samples in the embargo period after test
embargo_set = set(range(
test_end, min(n, test_end + embargo_size)
))
train_idx = np.array([i for i in train_idx if i not in embargo_set])
splits.append((train_idx, test_idx))
return splits
def evaluate(
self,
X: pd.DataFrame,
y: pd.Series,
model,
) -> pd.DataFrame:
"""Run purged k-fold evaluation."""
splits = self.split(X, y)
results = []
for fold, (train_idx, test_idx) in enumerate(splits):
model.fit(X.iloc[train_idx], y.iloc[train_idx])
y_pred = model.predict(X.iloc[test_idx])
results.append({
"fold": fold,
"accuracy": accuracy_score(y.iloc[test_idx], y_pred),
"train_size": len(train_idx),
"test_size": len(test_idx),
})
results_df = pd.DataFrame(results)
print(f"Purged K-Fold ({self.n_splits} folds):")
print(f" Accuracy: {results_df['accuracy'].mean():.3f} +/- {results_df['accuracy'].std():.3f}")
return results_df
`
Combinatorial Purged Cross-Validation (CPCV)
CPCV generates many more test paths than traditional CV by combining test folds, providing a better estimate of backtest variance.
`python
from itertools import combinations
class CombinatorialPurgedCV:
"""
Combinatorial Purged Cross-Validation (CPCV).
Generates C(N, k) test paths from N groups,
each using k groups for testing.
This produces many more backtest paths than standard CV,
enabling better estimation of strategy variance.
"""
def __init__(
self,
n_groups: int = 6,
n_test_groups: int = 2,
purge_window: int = 5,
):
self.n_groups = n_groups
self.n_test_groups = n_test_groups
self.purge_window = purge_window
def split(self, X: pd.DataFrame) -> list[tuple[np.ndarray, np.ndarray]]:
"""Generate all combinatorial train/test splits."""
n = len(X)
group_size = n // self.n_groups
groups = []
for i in range(self.n_groups):
start = i * group_size
end = min((i + 1) * group_size, n) if i < self.n_groups - 1 else n
groups.append(np.arange(start, end))
# All combinations of test groups
splits = []
for test_combo in combinations(range(self.n_groups), self.n_test_groups):
test_idx = np.concatenate([groups[i] for i in test_combo])
train_groups = [i for i in range(self.n_groups) if i not in test_combo]
train_idx = np.concatenate([groups[i] for i in train_groups])
# Purge: remove training samples near test boundaries
test_set = set(test_idx)
purge_set = set()
for t in test_idx:
for offset in range(-self.purge_window, self.purge_window + 1):
purge_set.add(t + offset)
purge_set -= test_set
train_idx = np.array([i for i in train_idx if i not in purge_set])
splits.append((train_idx, np.sort(test_idx)))
n_paths = len(splits)
print(f"CPCV: {n_paths} test paths from C({self.n_groups},{self.n_test_groups})")
return splits
def evaluate(
self,
X: pd.DataFrame,
y: pd.Series,
model,
) -> pd.DataFrame:
"""Evaluate across all combinatorial paths."""
splits = self.split(X)
results = []
for path_id, (train_idx, test_idx) in enumerate(splits):
model.fit(X.iloc[train_idx], y.iloc[train_idx])
y_pred = model.predict(X.iloc[test_idx])
results.append({
"path": path_id,
"accuracy": accuracy_score(y.iloc[test_idx], y_pred),
"train_size": len(train_idx),
"test_size": len(test_idx),
})
results_df = pd.DataFrame(results)
print(f"\nCPCV Results ({len(splits)} paths):")
print(f" Accuracy: {results_df['accuracy'].mean():.3f} +/- {results_df['accuracy'].std():.3f}")
print(f" Min: {results_df['accuracy'].min():.3f}")
print(f" Max: {results_df['accuracy'].max():.3f}")
# Probability of loss (assuming 50% is break-even)
prob_loss = (results_df["accuracy"] < 0.50).mean()
print(f" Prob(accuracy < 50%): {prob_loss:.1%}")
return results_df
`
Cross-Validation Comparison Framework
Run all methods and compare results to assess signal robustness.
`python
def comprehensive_cv_evaluation(
X: pd.DataFrame,
y: pd.Series,
model=None,
) -> dict:
"""
Run all CV methods and compare results.
Consistent results across methods = robust signal.
"""
model = model or GradientBoostingClassifier(
n_estimators=100, max_depth=4, random_state=42
)
print("=" * 60)
print("COMPREHENSIVE CROSS-VALIDATION EVALUATION")
print("=" * 60)
results = {}
# 1. Walk-Forward (Expanding)
print("\n1. Walk-Forward (Expanding)")
wfa = WalkForwardAnalysis(mode="expanding", train_size=504, test_size=63)
results["walk_forward_expanding"] = wfa.evaluate(X, y, model)
# 2. Walk-Forward (Rolling)
print("\n2. Walk-Forward (Rolling)")
wfr = WalkForwardAnalysis(mode="rolling", train_size=504, test_size=63)
results["walk_forward_rolling"] = wfr.evaluate(X, y, model)
# 3. Purged K-Fold
print("\n3. Purged K-Fold")
pkf = PurgedKFold(n_splits=5, purge_window=10)
results["purged_kfold"] = pkf.evaluate(X, y, model)
# 4. CPCV
print("\n4. Combinatorial Purged CV")
cpcv = CombinatorialPurgedCV(n_groups=6, n_test_groups=2)
results["cpcv"] = cpcv.evaluate(X, y, model)
# Summary comparison
print("\n" + "=" * 60)
print("SUMMARY")
print("=" * 60)
for name, df in results.items():
mean_acc = df["accuracy"].mean()
std_acc = df["accuracy"].std()
print(f" {name:30s}: {mean_acc:.3f} +/- {std_acc:.3f}")
return results
``