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.
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.
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.
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.
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.
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.
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
FAQ
Why does standard k-fold give inflated results for trading models?
Two mechanisms cause inflation. First, temporal leakage: when future data appears in the training set, the model learns patterns it could not know in real-time. Second, autocorrelation: nearby observations are highly correlated, so when they end up in both train and test, the model memorizes rather than generalizes. The inflation is typically 5-15 percentage points in accuracy, enough to make a random model appear profitable.
How large should the embargo period be?
The embargo period should be at least as long as the feature lookback window or the label horizon, whichever is longer. If your features use 20-day rolling windows and your labels are 5-day forward returns, use an embargo of at least 20 days. In practice, add a safety margin: use 1.5 to 2 times the maximum lookback. Insufficient embargo is the most common cause of inflated cross-validation results.
Should I use expanding or rolling windows for walk-forward analysis?
Expanding windows use all available historical data, giving the model more training data over time. Rolling windows keep a fixed window size, which better adapts to non-stationary markets where old data becomes irrelevant. Use expanding windows if you believe the underlying relationships are stable. Use rolling windows if you expect regime changes. For robust evaluation, run both and compare.
How do I know if my cross-validation results are statistically significant?
Compare your model's performance to a null model that predicts randomly. Use the binomial test: if your model achieves 53% accuracy over 1,000 test predictions, the probability of achieving this by chance is computable. Additionally, check that performance is consistent across folds: a model with 80% accuracy on one fold and 45% on another is likely overfitting to a specific regime.