"""Walk Forward cross-validator."""
# Copyright (c) 2023-2026
# Author: Hugo Delatte <hugo.delatte@skfoliolabs.com>
# SPDX-License-Identifier: BSD-3-Clause
# Implementation derived from:
# scikit-portfolio, Copyright (c) 2022, Carlo Nicolini, Licensed under MIT Licence.
# scikit-learn, Copyright (c) 2007-2010 David Cournapeau, Fabian Pedregosa, Olivier
# Grisel Licensed under BSD 3 clause.
from __future__ import annotations
import datetime as dt
from collections.abc import Iterator
import numpy as np
import pandas as pd
import sklearn.model_selection as sks
import sklearn.utils as sku
from skfolio.typing import ArrayLike, IntArray
from skfolio.utils.tools import (
_is_integer_number,
_validate_non_negative_integer,
_validate_positive_integer,
)
[docs]
class WalkForward(sks.BaseCrossValidator):
"""Walk Forward Cross-Validator.
Provides train/test indices to split time series data samples using a walk-forward
logic.
In each split, test indices must be higher than the previous ones. Therefore,
shuffling in cross-validator is inappropriate.
Compared to `sklearn.model_selection.TimeSeriesSplit`, you control the train/test
folds by specifying the number of training and test samples instead of the number
of splits, making it more suitable for portfolio cross-validation.
If your data is a DataFrame indexed with a DatetimeIndex, you can split the data
using specific datetime frequencies and offsets.
Parameters
----------
test_size : int
Length of each test set.
If `freq` is `None` (default), it represents the number of observations.
Otherwise, it represents the number of periods defined by `freq`.
train_size : int | pandas.offsets.DateOffset | datetime.timedelta
Length of each training set.
If `freq` is `None` (default), it represents the number of observations.
Otherwise, for integers, it represents the number of periods defined by `freq`.
For a pandas DateOffset or datetime timedelta, that offset is subtracted
from each test period's start date to determine when training begins.
freq : str | pandas.offsets.DateOffset, optional
If provided, it must be a frequency string or a pandas DateOffset, and the
returns `X` must be a DataFrame with an index of type `DatetimeIndex`.
For a list of pandas frequencies and offsets, see `here <https://pandas.pydata.org/pandas-docs/stable/user_guide/timeseries.html#timeseries-offset-aliases>`_.
The default (`None`) means `test_size` and `train_size` represent the number of
observations. Calendar-based splits with no test observations are omitted.
Below are some common examples:
* Rebalancing : Monthly on the first day
* Test Duration : 1 month
* Train Duration : 6 months
>>> cv = WalkForward(test_size=1, train_size=6, freq="MS")
* Rebalancing : Quarterly on the first day
* Test Duration : 1 quarter
* Train Duration : 2 months
>>> cv = WalkForward(test_size=1, train_size=pd.DateOffset(months=2), freq="QS")
* Rebalancing : Monthly on the third Friday
* Test Duration : 1 month
* Train Duration : 6 weeks
>>> cv = WalkForward(test_size=1, train_size=pd.offsets.Week(6), freq= "WOM-3FRI")
* Rebalancing : Semi-annually on the last day
* Test Duration : 6 months
* Train Duration : 1 year
>>> cv = WalkForward(test_size=1, train_size=2, freq=pd.offsets.SemiMonthEnd())
* Rebalancing : Every 2 months on the second day
* Test Duration : 2 months
* Train Duration : 6 months
>>> cv = WalkForward(test_size=2, train_size=6, freq="MS", freq_offset=dt.timedelta(days=2))
freq_offset : pandas DateOffset | datetime timedelta, optional
Shifts the dates generated by `freq` by this amount.
Only used when `freq` is provided.
previous : bool, default=False
Only used if `freq` is provided. If set to `True`, and if the period start
or period end is not in the `DatetimeIndex`, the previous observation is used.
Otherwise, the next observation is used (default).
expand_train : bool, default=False
If set to `True`, each subsequent training set after the first one will
use all past observations.
The default is `False`.
reduce_test : bool, default=False
If `True`, the final split is included even when there is not enough data
for a full test window. Otherwise, that split is omitted.
The default is `False`.
purged_size : int, default=0
The number of observations to exclude from the end of each training set before
the test set.
The default value is `0`.
.. warning::
**Execution timing and look-ahead control**
With `purged_size=0`:
- Training ends at the current period and testing begins immediately.
- Assumes you can observe, compute, and execute within the same period.
- If observation/computation-to-execution latency is non-negligible
(submission cutoffs, illiquidity, end-of-period finalization, or
markets with no intraday quotation), results may be too optimistic.
With `purged_size=1`:
- One observation is dropped between training and test.
- Decisions made on the current period start affecting performance from
the next period.
Rules of thumb:
- Use `purged_size=0` only when you truly can execute at the same period
with minimal latency.
- Use `purged_size >= 1` when execution is delayed (daily-priced assets,
illiquid markets, end-of-day data that settles after the close).
Examples
--------
Tutorials using `WalkForward`:
* :ref:`sphx_glr_auto_examples_pre_selection_plot_3_custom_pre_selection_volumes.py`
* :ref:`sphx_glr_auto_examples_clustering_plot_3_hrp_vs_herc.py`
* :ref:`sphx_glr_auto_examples_mean_risk_plot_8_regularization.py`
* :ref:`sphx_glr_auto_examples_clustering_plot_5_nco_grid_search.py`
* :ref:`sphx_glr_auto_examples_ensemble_plot_1_stacking.py`
>>> import numpy as np
>>> from skfolio.datasets import load_sp500_dataset, load_factors_dataset
>>> from skfolio.model_selection import WalkForward
>>> from skfolio.preprocessing import prices_to_returns
>>>
>>> X = np.random.randn(6, 2) # 6 observations
>>> cv = WalkForward(test_size=1, train_size=2)
>>> for i, (train_index, test_index) in enumerate(cv.split(X)):
... print(f"Fold {i}:")
... print(f" Train: index={train_index}")
... print(f" Test: index={test_index}")
Fold 0:
Train: index=[0 1]
Test: index=[2]
Fold 1:
Train: index=[1 2]
Test: index=[3]
Fold 2:
Train: index=[2 3]
Test: index=[4]
Fold 3:
Train: index=[3 4]
Test: index=[5]
>>> cv = WalkForward(test_size=1, train_size=2, purged_size=1)
>>> for i, (train_index, test_index) in enumerate(cv.split(X)):
... print(f"Fold {i}:")
... print(f" Train: index={train_index}")
... print(f" Test: index={test_index}")
Fold 0:
Train: index=[0 1]
Test: index=[3]
Fold 1:
Train: index=[1 2]
Test: index=[4]
Fold 2:
Train: index=[2 3]
Test: index=[5]
>>> cv = WalkForward(test_size=2, train_size=3)
>>> for i, (train_index, test_index) in enumerate(cv.split(X)):
... print(f"Fold {i}:")
... print(f" Train: index={train_index}")
... print(f" Test: index={test_index}")
Fold 0:
Train: index=[0 1 2]
Test: index=[3 4]
>>> cv = WalkForward(test_size=2, train_size=3, reduce_test=True)
>>> for i, (train_index, test_index) in enumerate(cv.split(X)):
... print(f"Fold {i}:")
... print(f" Train: index={train_index}")
... print(f" Test: index={test_index}")
Fold 0:
Train: index=[0 1 2]
Test: index=[3 4]
Fold 1:
Train: index=[2 3 4]
Test: index=[5]
>>> cv = WalkForward(test_size=2, train_size=3, expand_train=True, reduce_test=True)
>>> for i, (train_index, test_index) in enumerate(cv.split(X)):
... print(f"Fold {i}:")
... print(f" Train: index={train_index}")
... print(f" Test: index={test_index}")
Fold 0:
Train: index=[0 1 2]
Test: index=[3 4]
Fold 1:
Train: index=[0 1 2 3 4]
Test: index=[5]
>>>
>>> # Time-based (calendar) rebalancing
>>> prices = load_sp500_dataset()
>>> X = prices_to_returns(prices)
>>> X = X["2021":"2022"]
>>> # Rebalance every 3 months on the third Friday, and train on the last 12 months.
>>> cv = WalkForward(test_size=3, train_size=12, freq="WOM-3FRI")
>>>
>>> for i, (train_index, test_index) in enumerate(cv.split(X)):
... print(f"Fold {i}:")
... print(f" Train: size={len(train_index)}")
... print(f" Test: size={len(test_index)}")
Fold 0:
Train: size=256
Test: size=59
Fold 1:
Train: size=253
Test: size=61
Fold 2:
Train: size=251
Test: size=69
"""
def __init__(
self,
test_size: int,
train_size: int | pd.offsets.BaseOffset | dt.timedelta,
freq: str | pd.offsets.BaseOffset | None = None,
freq_offset: pd.offsets.BaseOffset | dt.timedelta | None = None,
previous: bool = False,
expand_train: bool = False,
reduce_test: bool = False,
purged_size: int = 0,
) -> None:
self.test_size = test_size
self.train_size = train_size
self.freq = freq
self.freq_offset = freq_offset
self.previous = previous
self.expand_train = expand_train
self.reduce_test = reduce_test
self.purged_size = purged_size
[docs]
def split(
self, X: ArrayLike, y: None = None, groups: None = None
) -> Iterator[tuple[IntArray, IntArray]]:
"""Generate positional indices for each training and test set.
Parameters
----------
X : array-like of shape (n_observations, n_assets)
Price returns of the assets.
y : None
Always ignored, exists for compatibility.
groups : None
Always ignored, exists for compatibility.
Yields
------
train : ndarray
Positional indices of the training observations.
test : ndarray
Positional indices of the test observations.
Raises
------
ValueError
If a window size has an invalid type, an integer window size is not
positive, or `purged_size` is not a non-negative integer.
When `freq=None`, also raised if training and purging leave no
observations for testing. When `freq` is set, also raised if any split
with test observations has no training observations after applying
`expand_train` and `purged_size`. This is checked before the first
split is yielded.
"""
test_size, train_size = self._validate_window_sizes()
X, y = sku.indexable(X, y)
n_samples = X.shape[0]
if self.freq is None:
_validate_observation_count(
n_samples=n_samples,
train_size=train_size, # ty: ignore[invalid-argument-type]
purged_size=self.purged_size,
)
return _split_by_observation_count(
n_samples=n_samples,
train_size=train_size, # ty: ignore[invalid-argument-type]
test_size=test_size,
purged_size=self.purged_size,
expand_train=self.expand_train,
reduce_test=self.reduce_test,
)
if not hasattr(X, "index") or not isinstance(X.index, pd.DatetimeIndex):
raise ValueError(
"X must be a DataFrame with an index of type DatetimeIndex"
)
return _split_by_calendar(
train_size=train_size,
test_size=test_size,
freq=self.freq,
freq_offset=self.freq_offset,
previous=self.previous,
purged_size=self.purged_size,
expand_train=self.expand_train,
reduce_test=self.reduce_test,
ts_index=X.index,
)
[docs]
def get_n_splits(
self,
X: ArrayLike | None = None,
y: None = None,
groups: None = None,
) -> int:
"""Return the number of train/test splits without generating their indices.
Parameters
----------
X : array-like of shape (n_observations, n_assets)
Price returns of the assets.
y : None
Always ignored, exists for compatibility.
groups : None
Always ignored, exists for compatibility.
Returns
-------
n_folds : int
Number of train/test splits.
Raises
------
ValueError
If `X` is `None`, a window size has an invalid type, an integer window
size is not positive, or `purged_size` is not a non-negative integer.
When `freq=None`, also raised if training and purging leave no
observations for testing. When `freq` is set, also raised if any split
with test observations has no training observations after applying
`expand_train` and `purged_size`.
"""
if X is None:
raise ValueError("The 'X' parameter should not be None.")
test_size, train_size = self._validate_window_sizes()
X, y = sku.indexable(X, y)
n_samples = X.shape[0]
if self.freq is None:
_validate_observation_count(
n_samples=n_samples,
train_size=train_size, # ty: ignore[invalid-argument-type]
purged_size=self.purged_size,
)
n = n_samples - train_size - self.purged_size
if self.reduce_test and n % test_size != 0:
return n // test_size + 1
return n // test_size
if not hasattr(X, "index") or not isinstance(X.index, pd.DatetimeIndex):
raise ValueError(
"X must be a DataFrame with an index of type DatetimeIndex"
)
_, test_starts, _ = _calendar_window_bounds(
train_size=train_size,
test_size=test_size,
freq=self.freq,
freq_offset=self.freq_offset,
previous=self.previous,
purged_size=self.purged_size,
expand_train=self.expand_train,
reduce_test=self.reduce_test,
ts_index=X.index,
)
return len(test_starts)
def _validate_window_sizes(
self,
) -> tuple[int, int | pd.offsets.BaseOffset | dt.timedelta]:
"""Validate window sizes and convert integer sizes to Python `int` values.
Returns
-------
test_size : int
Test size converted to a Python `int`.
train_size : int | pandas.offsets.DateOffset | datetime.timedelta
Training size converted to a Python `int` if it is an integer.
Date offsets and timedeltas are returned unchanged.
Raises
------
ValueError
If a window size has an invalid type, if `test_size` or an integer
`train_size` is not positive, or if `purged_size` is not a non-negative
integer.
"""
_validate_positive_integer(self.test_size, "test_size")
train_size = self.train_size
if _is_integer_number(train_size):
train_size = int(train_size)
_validate_positive_integer(train_size, "train_size")
elif self.freq is None:
raise ValueError(
f"train_size must be an integer when freq is None, got {train_size!r}"
)
elif not isinstance(train_size, (pd.offsets.BaseOffset, dt.timedelta)):
raise ValueError(
"train_size must be an integer, pandas DateOffset, or datetime "
f"timedelta when freq is set, got {train_size!r}"
)
_validate_non_negative_integer(self.purged_size, "purged_size")
return int(self.test_size), train_size
def _validate_observation_count(
n_samples: int, train_size: int, purged_size: int
) -> None:
"""Raise if `train_size` and `purged_size` leave no observations for testing."""
total_size = train_size + purged_size
if total_size >= n_samples:
raise ValueError(
f"The sum of `train_size={train_size}` and `purged_size={purged_size}` "
f"(total={total_size}) must be less than the number of "
f"observations={n_samples}."
)
def _split_by_observation_count(
n_samples: int,
train_size: int,
test_size: int,
purged_size: int,
expand_train: bool,
reduce_test: bool,
) -> Iterator[tuple[IntArray, IntArray]]:
"""Generate positional train/test indices.
`train_size` and `test_size` specify the number of observations in each window.
Parameters
----------
n_samples : int
Number of observations in the dataset.
train_size : int
Number of observations in each rolling training window.
test_size : int
Number of observations in each test window.
purged_size : int
Number of observations removed between the training and test windows.
expand_train : bool
If `True`, training always starts at index 0. Otherwise, a rolling
training window of fixed length is used.
reduce_test : bool
If `True`, keep the final split even when the last test window is
shorter than `test_size`.
Yields
------
train_indices : ndarray
Positional indices of the training observations.
test_indices : ndarray
Positional indices of the test observations.
"""
indices = np.arange(n_samples)
test_start = train_size + purged_size
while True:
if test_start >= n_samples:
return
test_end = test_start + test_size
train_end = test_start - purged_size
if expand_train:
train_start = 0
else:
train_start = train_end - train_size
if test_end > n_samples:
if not reduce_test:
return
test_indices = indices[test_start:]
else:
test_indices = indices[test_start:test_end]
train_indices = indices[train_start:train_end]
yield train_indices, test_indices
test_start = test_end
def _split_by_calendar(
train_size: int | pd.offsets.BaseOffset | dt.timedelta,
test_size: int,
freq: str | pd.offsets.BaseOffset,
freq_offset: pd.offsets.BaseOffset | dt.timedelta | None,
previous: bool,
purged_size: int,
expand_train: bool,
reduce_test: bool,
ts_index: pd.DatetimeIndex,
) -> Iterator[tuple[IntArray, IntArray]]:
"""Generate positional train/test indices using calendar dates.
Integer `train_size` and `test_size` values count periods defined by `freq`.
A date offset or timedelta for `train_size` is subtracted from each test
period's start date to find when training begins.
Parameters
----------
train_size : int | pandas DateOffset | datetime timedelta
If an integer, the number of periods defined by `freq` in each training
window. Otherwise, the amount of time subtracted from the start date of
each test period to determine when training begins.
test_size : int
Number of calendar periods included in each test window.
freq : str | pandas.offsets.DateOffset
Calendar frequency used to define rebalancing dates.
freq_offset : pandas DateOffset or datetime timedelta, optional
Offset applied to each rebalancing date.
previous : bool
If a period's start date is missing from `ts_index`, the preceding
observation is used when `True`, or the following observation otherwise.
purged_size : int
Number of observations removed from the end of each training window.
expand_train : bool
If `True`, training always starts at index 0.
reduce_test : bool
If `True`, the final split is included even when there is not enough data
for a full test window. Otherwise, that split is omitted.
ts_index : DatetimeIndex
Datetime index of the input data.
Yields
------
train_indices : ndarray
Positional indices of the training observations.
test_indices : ndarray
Positional indices of the test observations.
"""
train_starts, test_starts, test_stops = _calendar_window_bounds(
train_size=train_size,
test_size=test_size,
freq=freq,
freq_offset=freq_offset,
previous=previous,
purged_size=purged_size,
expand_train=expand_train,
reduce_test=reduce_test,
ts_index=ts_index,
)
for train_start, test_start, test_stop in zip(
train_starts, test_starts, test_stops, strict=True
):
if expand_train:
train_start = 0
# Purging shortens training without moving the test window's start.
train_indices = np.arange(train_start, test_start - purged_size)
test_indices = np.arange(test_start, test_stop)
yield train_indices, test_indices
def _calendar_window_bounds(
train_size: int | pd.offsets.BaseOffset | dt.timedelta,
test_size: int,
freq: str | pd.offsets.BaseOffset,
freq_offset: pd.offsets.BaseOffset | dt.timedelta | None,
previous: bool,
purged_size: int,
expand_train: bool,
reduce_test: bool,
ts_index: pd.DatetimeIndex,
) -> tuple[IntArray, IntArray, IntArray]:
"""Compute start and stop indices for windows defined by calendar dates.
Each returned array has one entry per split. For split `k`, the training
slice is `train_starts[k]:test_starts[k]` before applying `expand_train` or
`purged_size`. The test slice is `test_starts[k]:test_stops[k]`. As with
Python slices, the stop index is not included. Splits with no test
observations are omitted.
Parameters
----------
train_size : int | pandas DateOffset | datetime timedelta
If an integer, the number of periods defined by `freq` in each training
window. Otherwise, the amount of time subtracted from the start date of
each test period to determine when training begins.
test_size : int
Number of calendar periods included in each test window.
freq : str | pandas.offsets.DateOffset
Calendar frequency used to define rebalancing dates.
freq_offset : pandas DateOffset or datetime timedelta, optional
Offset applied to each rebalancing date.
previous : bool
If a period's start date is missing from `ts_index`, the preceding
observation is used when `True`, or the following observation otherwise.
purged_size : int
Number of observations removed from the end of each training window.
Used to check that at least one training observation remains.
expand_train : bool
If `True`, each training window is validated as starting at index 0.
The returned `train_starts` are unchanged.
reduce_test : bool
If `True`, the final split is included even when there is not enough data
for a full test window. Otherwise, that split is omitted.
ts_index : DatetimeIndex
Datetime index of the input data.
Returns
-------
train_starts : ndarray
Positional index where each training window starts before applying
`expand_train`. The caller applies `expand_train` and `purged_size`
when generating the training indices.
test_starts : ndarray
Positional index of the first test observation in each split.
With `purged_size=0`, training ends immediately before this index.
test_stops : ndarray
Positional index immediately after the final test observation in each
split. Can equal `len(ts_index)` when a test window reaches the end of
the data.
Raises
------
ValueError
If a split with test observations has no training observations after
applying `expand_train` and `purged_size`.
"""
start = ts_index[0]
end = ts_index[-1]
if freq_offset is not None:
start = min(start, start - freq_offset)
date_range = pd.date_range(start=start, end=end, freq=freq)
if freq_offset is not None:
date_range += freq_offset
# A shifted date beyond the data cannot close a full test window.
# The final partial window is handled below when reduce_test is True.
if date_range.is_monotonic_increasing:
date_range = date_range[: date_range.searchsorted(end, side="right")]
else:
date_range = date_range[date_range <= end]
period_idx = ts_index.get_indexer(
date_range, method="ffill" if previous else "bfill"
)
if previous and freq_offset is not None:
# Dates shifted before the data have no previous observation. We start those
# windows at the first observation instead of interpreting -1 as the last.
period_idx[period_idx == -1] = 0
n_periods = len(period_idx)
if isinstance(train_size, int):
first_period = train_size
else:
lookback_idx = ts_index.get_indexer(date_range - train_size, method="ffill")
# We keep missing lookbacks because they identify periods without enough history.
if np.all(lookback_idx == -1):
return period_idx[:0], period_idx[:0], period_idx[:0]
first_period = int(np.argmax(lookback_idx > -1))
# Exclusive calendar-period limit for the start of a test window.
end_period = n_periods if reduce_test else n_periods - test_size
if first_period >= end_period:
return period_idx[:0], period_idx[:0], period_idx[:0]
test_starts = period_idx[first_period:end_period:test_size]
test_stops = period_idx[first_period + test_size :: test_size]
if reduce_test:
# The calendar schedule ends before the final test stop, so we use the data's end.
test_stops = np.concatenate((test_stops, [len(ts_index)]))
if isinstance(train_size, int):
train_starts = period_idx[: end_period - train_size : test_size]
else:
train_starts = lookback_idx[first_period:end_period:test_size]
# Missing periods (weekends, holidays) can share the same observation.
nonempty = test_stops > test_starts
# We check every retained fold without generating its observation indices.
if expand_train:
empty_train = test_starts <= purged_size
else:
empty_train = test_starts - train_starts <= purged_size
empty_train &= nonempty
if np.any(empty_train):
split_idx = int(np.argmax(empty_train))
test_start = ts_index[test_starts[split_idx]]
purge_advice = ", reduce purged_size" if purged_size else ""
raise ValueError(
f"The training window for the test starting {test_start} contains no "
f"observations (train_size={train_size!r}, freq={freq!r}, "
f"expand_train={expand_train}, purged_size={purged_size}). "
f"Increase train_size{purge_advice}, or use freq=None if window sizes "
"should count observations."
)
if np.all(nonempty):
return train_starts, test_starts, test_stops
return train_starts[nonempty], test_starts[nonempty], test_stops[nonempty]
