API reference¶

All public symbols from purgedcv.__all__, auto-rendered from the source docstrings. The constructors of the splitters share a single set of keyword arguments (prediction_times, evaluation_times, purge_horizon, embargo, groups); see BaseTemporalSplitter for the shared contract.

Splitters¶

purgedcv.BaseTemporalSplitter ¶

BaseTemporalSplitter(*, prediction_times: Series, evaluation_times: Series, purge_horizon: HorizonLike | None = None, embargo: HorizonLike | None = None, groups: Series | None = None)

Bases: ABC

Duck-typed sklearn CV splitter with purge + embargo orchestration.

Concrete subclasses implement :meth:_iter_test_indices to yield the raw test-index arrays for each fold. The base class handles purge, embargo, optional group-disjointness, and the sklearn-compatible :meth:split / :meth:get_n_splits protocol.

Times are bound to the splitter at construction. This couples the splitter to a specific dataset's timestamps, which is intentional: a splitter for one dataset is rarely meaningful for another.

.. note:: The subclassing interface (_iter_test_indices, _candidate_train_idx) is not yet covered by the v0.3 stability contract. Subclasses may need adjustments through v1.0.

get_n_splits `abstractmethod` ¶

get_n_splits(X: object = None, y: object = None, groups: object = None) -> int

Return the total number of splits the iterator will yield.

split ¶

split(X: NDArrayAny | DataFrame, y: object = None, groups: object = None) -> Iterator[tuple[NDArrayAny, NDArrayAny]]

Yield (train_idx, test_idx) pairs for each fold.

The y and groups parameters of this method are accepted for sklearn protocol compatibility but ignored — group information must be bound at construction via the groups argument of __init__.

When groups were bound at construction, :func:~purgedcv.diagnostics.assert_groups_disjoint is called on every fold after purge and embargo; a :class:~purgedcv.exceptions.GroupLeakageError is raised if any group identifier appears in both train and test of the same fold.

with_times ¶

with_times(prediction_times: Series, evaluation_times: Series) -> BaseTemporalSplitter

Return a copy of this splitter with new times bound. All other parameters (n_splits, purge_horizon, embargo, groups, and any subclass-specific state such as a cached unique-group list) are preserved unchanged.

To change groups or any other construction parameter, build a fresh splitter via the constructor — this avoids surprising interactions between cached state and rebound inputs.

purgedcv.WalkForwardSplit ¶

WalkForwardSplit(n_splits: int, test_size: int, *, train_size: int | None = None, window: WindowMode = 'expanding', prediction_times: Series, evaluation_times: Series, purge_horizon: HorizonLike | None = None, embargo: HorizonLike | None = None)

Bases: BaseTemporalSplitter

Walk-forward CV with sliding or expanding training window.

For window="expanding" and zero purge/embargo, this matches :class:sklearn.model_selection.TimeSeriesSplit. For window="sliding", the training window has a fixed maximum length (the most recent train_size samples before the test fold).

Examples:

>>> import numpy as np
>>> import pandas as pd
>>> from purgedcv import WalkForwardSplit
>>> pred = pd.Series(pd.date_range("2024-01-01", periods=10, freq="D"))
>>> evalu = pred + pd.Timedelta(days=1)
>>> cv = WalkForwardSplit(
...     n_splits=3, test_size=2,
...     prediction_times=pred, evaluation_times=evalu,
... )
>>> cv.get_n_splits()
3

Configure a walk-forward CV splitter.

Parameters:

Name	Type	Description	Default
`n_splits`	`int`	Number of folds to yield. Test folds tile the END of the dataset; each fold trains only on data strictly before its test indices.	required
`test_size`	`int`	Number of consecutive rows in each test fold. `n_splits * test_size` must be strictly less than `n_samples`.	required
`train_size`	`int \| None`	Maximum number of training rows per fold AFTER purge and embargo. Counts kept rows, not raw indices, so with a 2-day embargo and `train_size=100` the effective training set may use indices spanning more than 100 rows. Must be `None` when `window='expanding'` and a positive integer when `window='sliding'`.	`None`
`window`	`WindowMode`	`"expanding"` (default) uses all pre-test data as training; matches :class:`sklearn.model_selection.TimeSeriesSplit` when purge and embargo are zero. `"sliding"` caps each training set at the most recent `train_size` rows after purge and embargo.	`'expanding'`
`prediction_times`	`Series`	Per-sample prediction times for the dataset. Bound at construction so :meth:`split` matches the sklearn signature.	required
`evaluation_times`	`Series`	Per-sample evaluation times. Required to apply purge and embargo correctly.	required
`purge_horizon`	`HorizonLike \| None`	Symmetric padding around the test fold's label window; training rows whose label horizon overlaps the padded test horizon are dropped. `None` means no purge.	`None`
`embargo`	`HorizonLike \| None`	Post-test embargo duration; training rows whose prediction time falls in the closed window `[test_eval_max, test_eval_max + embargo]` are dropped. `None` means no embargo.	`None`

Raises:

Type	Description
`ValueError`	if `window` is not one of the literal values, if `n_splits < 1` or `test_size < 1`, if `train_size` is supplied with `window='expanding'`, or if `train_size` is `None` with `window='sliding'`.

split ¶

split(X: NDArrayAny | DataFrame, y: object = None, groups: object = None) -> Iterator[tuple[NDArrayAny, NDArrayAny]]

Yield (train_idx, test_idx) pairs for each walk-forward fold.

The base class handles X length validation, purge, and embargo; this override additionally trims each training set to the most recent train_size rows when window="sliding". Training indices are always strictly less than the minimum index of the test fold (the defining walk-forward property).

The number of pairs yielded equals self.n_splits. The y and groups arguments are accepted for sklearn protocol compatibility but ignored.

purgedcv.PurgedKFold ¶

PurgedKFold(n_splits: int, *, prediction_times: Series, evaluation_times: Series, purge_horizon: HorizonLike | None = None, embargo: HorizonLike | None = None)

Bases: BaseTemporalSplitter

K-fold CV with contiguous test folds and purge + embargo applied.

Test folds tile the index space contiguously: fold k holds rows [start_k, start_k + size_k) where size_k differs across folds by at most one due to integer division. Train = complement of the test fold, with D2 purge and D3 embargo applied by the base class.

For zero purge_horizon and embargo the test folds are identical to :class:sklearn.model_selection.KFold(shuffle=False). Purge still drops any training row whose own label horizon [prediction_time, evaluation_time) overlaps the test labels, so the training set equals the full complement only when no two label horizons overlap (for instance when evaluation_times equals prediction_times). The splitter then degrades exactly to KFold(shuffle=False).

The first n_samples % n_splits folds receive n_samples // n_splits + 1 rows; the remaining folds receive n_samples // n_splits. When n_splits > n_samples, the excess folds are empty — the base class purge and embargo handle empty arrays gracefully.

Examples:

>>> import numpy as np
>>> import pandas as pd
>>> from purgedcv import PurgedKFold
>>> pred = pd.Series(pd.date_range("2024-01-01", periods=20, freq="D"))
>>> evalu = pred + pd.Timedelta(days=1)
>>> cv = PurgedKFold(
...     n_splits=5,
...     prediction_times=pred, evaluation_times=evalu,
... )
>>> cv.get_n_splits()
5

Configure a purged k-fold splitter.

Parameters:

Name	Type	Description	Default
`n_splits`	`int`	Number of folds. Must be at least 2.	required
`prediction_times`	`Series`	Per-sample prediction times. Bound at construction so :meth:`split` matches the sklearn signature.	required
`evaluation_times`	`Series`	Per-sample evaluation times.	required
`purge_horizon`	`HorizonLike \| None`	Symmetric padding around the test fold's label window; training rows whose label horizon overlaps the padded test horizon are dropped. `None` means no purge.	`None`
`embargo`	`HorizonLike \| None`	Post-test embargo duration; training rows whose prediction time falls in any closed window `[test_evaluation_time, test_evaluation_time + embargo]` are dropped. `None` means no embargo.	`None`

Raises:

Type	Description
`ValueError`	if `n_splits < 2`.

purgedcv.PurgedGroupKFold ¶

PurgedGroupKFold(n_splits: int, *, prediction_times: Series, evaluation_times: Series, groups: Series, purge_horizon: HorizonLike | None = None, embargo: HorizonLike | None = None)

Bases: BaseTemporalSplitter

Group-aware k-fold variant of :class:PurgedKFold.

Each test fold consists of all rows from a contiguous block of unique group identifiers, so no group_id ever appears in both train and test of the same fold. Purge and embargo still apply ACROSS group boundaries — training rows from other groups whose horizons overlap the test window are dropped by D2/D3 in the base class.

Groups are assigned to fold blocks in first-appearance order within the groups Series (the order returned by pd.Series.unique()). For temporal coherence — patients enrolled in chronological order, assets in time-of-IPO order — ensure the groups Series is sorted by first occurrence time.

The base class's :func:assert_groups_disjoint enforcement runs on every fold automatically because groups is bound at construction.

Useful in clinical ML (no patient appears in both sides of a fold), asset CV (no symbol appears in both sides), and any setting where grouped observations must not leak across the train/test boundary.

Examples:

>>> import numpy as np
>>> import pandas as pd
>>> from purgedcv import PurgedGroupKFold
>>> pred = pd.Series(pd.date_range("2024-01-01", periods=12, freq="D"))
>>> evalu = pred + pd.Timedelta(days=1)
>>> groups = pd.Series(np.repeat([0, 1, 2, 3], 3))
>>> cv = PurgedGroupKFold(
...     n_splits=2,
...     prediction_times=pred, evaluation_times=evalu,
...     groups=groups,
... )
>>> cv.get_n_splits()
2

Configure a group-aware purged k-fold splitter.

Parameters:

Name	Type	Description	Default
`n_splits`	`int`	Number of folds. Must be at least 2 and must not exceed the number of unique group identifiers.	required
`prediction_times`	`Series`	Per-sample prediction times.	required
`evaluation_times`	`Series`	Per-sample evaluation times.	required
`groups`	`Series`	Group identifier per sample. Must have the same length as `prediction_times`. Unique groups are partitioned into `n_splits` contiguous blocks in the order returned by `pd.Series.unique()`.	required
`purge_horizon`	`HorizonLike \| None`	Symmetric padding around the test fold's label window; cross-group training rows whose label horizon overlaps the padded test horizon are dropped. `None` means no purge.	`None`
`embargo`	`HorizonLike \| None`	Post-test embargo duration. `None` means no embargo.	`None`

Raises:

Type	Description
`ValueError`	if `n_splits < 2` or `n_splits > len(groups.unique())`.

purgedcv.CombinatorialPurgedCV ¶

CombinatorialPurgedCV(n_splits: int, n_test_groups: int, *, prediction_times: Series, evaluation_times: Series, purge_horizon: HorizonLike | None = None, embargo: HorizonLike | None = None)

Bases: BaseTemporalSplitter

Combinatorial Purged Cross-Validation (fold enumeration).

Partitions the time-ordered samples into n_splits contiguous group blocks. For each combination of n_test_groups chosen from those blocks, yields one fold whose test indices are the union of the chosen blocks. Total folds: C(n_splits, n_test_groups).

Each group block appears as test in exactly C(n_splits - 1, n_test_groups - 1) folds.

The base class applies D2 purge and D3 embargo to each fold's train set. :meth:backtest_paths then assembles the C(N,K) folds into n_paths time-ordered out-of-sample sequences.

See Advances in Financial Machine Learning (Lopez de Prado, Wiley 2018), chapter 12 section 12.4, for the original method.

Examples:

>>> import numpy as np
>>> import pandas as pd
>>> from purgedcv import CombinatorialPurgedCV
>>> pred = pd.Series(pd.date_range("2024-01-01", periods=24, freq="D"))
>>> evalu = pred + pd.Timedelta(days=1)
>>> cv = CombinatorialPurgedCV(
...     n_splits=6, n_test_groups=2,
...     prediction_times=pred, evaluation_times=evalu,
... )
>>> cv.get_n_splits()
15

Configure a Combinatorial Purged CV splitter.

Parameters:

Name	Type	Description	Default
`n_splits`	`int`	Number of contiguous group blocks to partition the samples into. Must be at least 2.	required
`n_test_groups`	`int`	Number of group blocks chosen as the test set in each fold. Must be in `[1, n_splits - 1]`.	required
`prediction_times`	`Series`	Per-sample prediction times.	required
`evaluation_times`	`Series`	Per-sample evaluation times.	required
`purge_horizon`	`HorizonLike \| None`	Symmetric padding around the test fold's label window; training rows whose label horizon overlaps the padded test horizon are dropped. `None` means no purge.	`None`
`embargo`	`HorizonLike \| None`	Post-test embargo duration; training rows whose prediction time falls in any closed window `[test_evaluation_time, test_evaluation_time + embargo]` are dropped. `None` means no embargo.	`None`

Raises:

Type	Description
`ValueError`	if `n_splits < 2`, if `n_splits` exceeds the number of samples, or if `n_test_groups` is not in `[1, n_splits - 1]`.

backtest_paths ¶

backtest_paths(estimator: object, X: NDArrayAny | DataFrame, y: NDArrayAny | Series) -> NDArrayAny

Fit estimator on each fold and reconstruct the C(N-1, K-1) out-of-sample backtest paths.

For each of the C(N, K) folds:

Clone the estimator (so per-fold fits do not contaminate each other or the original).
Fit on the fold's training set (after purge + embargo).
Predict on the fold's test set.
If the fold has no training rows under an unusually aggressive purge/embargo configuration, the predictions for that fold are NaN.

The per-fold predictions are then handed to :func:reconstruct_paths, which assembles them into an (n_paths, n_samples) matrix where each row is a complete time-ordered out-of-sample prediction sequence.

Parameters:

Name	Type	Description	Default
`estimator`	`object`	A scikit-learn estimator with `fit(X, y)` and `predict(X)` methods.	required
`X`	`NDArrayAny \| DataFrame`	Feature matrix of shape `(n_samples, n_features)`.	required
`y`	`NDArrayAny \| Series`	Target vector of shape `(n_samples,)`.	required

Returns:

Type	Description
`NDArrayAny`	`(n_paths, n_samples)` array of out-of-sample predictions
`NDArrayAny`	with `n_paths = C(n_splits - 1, n_test_groups - 1)`.
`NDArrayAny`	Affected rows contain NaN when an upstream fold could not be
`NDArrayAny`	fit.

Raises:

Type	Description
`AttributeError or TypeError`	if `estimator` lacks `fit` or `predict`.

Examples:

>>> import warnings
>>> import numpy as np
>>> import pandas as pd
>>> from sklearn.dummy import DummyRegressor
>>> from sklearn.exceptions import FitFailedWarning
>>> from purgedcv import CombinatorialPurgedCV
>>> pred = pd.Series(pd.date_range("2024-01-01", periods=16, freq="D"))
>>> evalu = pred + pd.Timedelta(days=1)
>>> cv = CombinatorialPurgedCV(
...     n_splits=4, n_test_groups=2,
...     prediction_times=pred, evaluation_times=evalu,
... )
>>> X = np.arange(16).reshape(-1, 1).astype(float)
>>> y = np.arange(16).astype(float)
>>> with warnings.catch_warnings():
...     warnings.simplefilter("ignore", FitFailedWarning)
...     paths = cv.backtest_paths(DummyRegressor(strategy="mean"), X, y)
>>> paths.shape
(3, 16)

reconstruct_paths ¶

reconstruct_paths(fold_predictions: Sequence[NDArrayAny]) -> NDArrayAny

Assemble per-fold predictions into the C(N-1, K-1) backtest paths.

Ergonomic wrapper around :func:~purgedcv.reconstruct_paths. The splitter already knows n_splits, n_test_groups, the fold test-index layout, and n_samples (from the bound times), so the caller supplies only one prediction array per fold, in :meth:split order. Use this when you ran the fits yourself (for example a per-fold backtest loop) rather than via :meth:backtest_paths.

Parameters:

Name	Type	Description	Default
`fold_predictions`	`Sequence[NDArrayAny]`	One array per fold, in the same order as :meth:`split` yields folds; `fold_predictions[f]` holds the predictions for that fold's test rows, in test-index order. There must be `C(n_splits, n_test_groups)` arrays.	required

Returns:

Type	Description
`NDArrayAny`	`(n_paths, n_samples)` array of out-of-sample predictions with
`NDArrayAny`	`n_paths = C(n_splits - 1, n_test_groups - 1)`.

Raises:

Type	Description
`ValueError`	on a fold-count or fold-prediction length mismatch (propagated from :func:`~purgedcv.reconstruct_paths`).

Examples:

>>> import numpy as np
>>> import pandas as pd
>>> from purgedcv import CombinatorialPurgedCV
>>> pred = pd.Series(pd.date_range("2024-01-01", periods=16, freq="D"))
>>> evalu = pred + pd.Timedelta(days=1)
>>> cv = CombinatorialPurgedCV(
...     n_splits=4, n_test_groups=2,
...     prediction_times=pred, evaluation_times=evalu,
... )
>>> X = np.zeros((16, 1))
>>> fold_preds = [
...     np.full(len(test), float(i))
...     for i, (_, test) in enumerate(cv.split(X))
... ]
>>> cv.reconstruct_paths(fold_preds).shape
(3, 16)

purgedcv.CombinatoriallySymmetricCV ¶

CombinatoriallySymmetricCV(n_splits: int, *, prediction_times: Series, evaluation_times: Series, purge_horizon: HorizonLike | None = None, embargo: HorizonLike | None = None)

Bases: CombinatorialPurgedCV

Combinatorially Symmetric Cross-Validation (CSCV).

The special case of :class:CombinatorialPurgedCV with n_test_groups = n_splits // 2: every fold cuts the timeline into two equal halves, one in-sample and one out-of-sample. CSCV is the substrate of :func:~purgedcv.probability_of_backtest_overfitting; expose it directly when you want the symmetric IS/OOS folds without going through the PBO estimator.

See Advances in Financial Machine Learning (Lopez de Prado, Wiley 2018), chapter 11.

Examples:

>>> import pandas as pd
>>> from purgedcv import CombinatoriallySymmetricCV
>>> pred = pd.Series(pd.date_range("2024-01-01", periods=24, freq="D"))
>>> evalu = pred + pd.Timedelta(days=1)
>>> cv = CombinatoriallySymmetricCV(
...     n_splits=6, prediction_times=pred, evaluation_times=evalu,
... )
>>> cv.get_n_splits()
20

Configure a CSCV splitter.

Parameters:

Name	Type	Description	Default
`n_splits`	`int`	Number of contiguous group blocks. Must be even and at least 2; half are chosen as the test (out-of-sample) groups in each fold.	required
`prediction_times`	`Series`	Per-sample prediction times.	required
`evaluation_times`	`Series`	Per-sample evaluation times.	required
`purge_horizon`	`HorizonLike \| None`	Optional purge horizon applied per fold.	`None`
`embargo`	`HorizonLike \| None`	Optional embargo horizon applied per fold.	`None`

Raises:

Type	Description
`ValueError`	if `n_splits` is odd or below 2.

Backtest paths¶

purgedcv.reconstruct_paths ¶

reconstruct_paths(fold_predictions: Sequence[NDArrayAny], fold_test_indices: Sequence[NDArrayAny], n_splits: int, n_test_groups: int, n_samples: int) -> NDArrayAny

Assemble Combinatorial Purged CV fold predictions into backtest paths.

Given the predictions and test indices for all C(N, K) folds produced by :class:CombinatorialPurgedCV, returns a (n_paths, n_samples) array where each row is a complete time-ordered out-of-sample prediction sequence built from a different combination of the folds.

n_paths = C(n_splits - 1, n_test_groups - 1). Each sample is predicted in every path (no missing entries). The assignment of fold to path for each group is the canonical greedy positional rule: for group g, the k-th fold (in fold-iteration order) that contains g as a test group contributes g's predictions to path k.

NaN predictions in any fold propagate only to the paths that use that fold for the affected group(s).

Parameters:

Name	Type	Description	Default
`fold_predictions`	`Sequence[NDArrayAny]`	One array per fold; `fold_predictions[f]` is the prediction values for the rows in `fold_test_indices[f]`, in the same order.	required
`fold_test_indices`	`Sequence[NDArrayAny]`	One array per fold; the per-fold test_idx as yielded by :meth:`CombinatorialPurgedCV._iter_test_indices`. Used here to recover the group block layout.	required
`n_splits`	`int`	Number of CPCV group blocks.	required
`n_test_groups`	`int`	Number of test groups per fold.	required
`n_samples`	`int`	Total number of samples in the dataset.	required

Returns:

Type	Description
`NDArrayAny`	`(n_paths, n_samples)` array of predictions.

Raises:

Type	Description
`ValueError`	on fold count or fold-prediction length mismatch.

Examples:

>>> import numpy as np
>>> from purgedcv import reconstruct_paths
>>> # Synthetic 4-fold output: 4 samples per fold's test set,
>>> # predictions equal to the fold index.
>>> fold_test = [
...     np.array([0, 1, 2, 3, 4, 5, 6, 7]),     # combo (0, 1)
...     np.array([0, 1, 2, 3, 8, 9, 10, 11]),   # combo (0, 2)
...     np.array([0, 1, 2, 3, 12, 13, 14, 15]), # combo (0, 3)
...     np.array([4, 5, 6, 7, 8, 9, 10, 11]),   # combo (1, 2)
...     np.array([4, 5, 6, 7, 12, 13, 14, 15]), # combo (1, 3)
...     np.array([8, 9, 10, 11, 12, 13, 14, 15]), # combo (2, 3)
... ]
>>> fold_preds = [
...     np.full(len(ti), float(f)) for f, ti in enumerate(fold_test)
... ]
>>> paths = reconstruct_paths(fold_preds, fold_test, 4, 2, 16)
>>> paths.shape
(3, 16)

purgedcv.path_metrics ¶

path_metrics(paths: NDArrayAny, metric_fn: PathMetricFn | None = None, *, bars_per_year: int | None = None) -> pd.DataFrame

Reduce an (n_paths, n_samples) matrix to per-path metrics.

Applies metric_fn to each row and stacks the results into a DataFrame with one row per path and one column per metric.

Parameters:

Name	Type	Description	Default
`paths`	`NDArrayAny`	`(n_paths, n_samples)` array, as returned by :func:`~purgedcv.reconstruct_paths` or :meth:`CombinatorialPurgedCV.backtest_paths`. Each row is one path's per-period return series.	required
`metric_fn`	`PathMetricFn \| None`	Maps a 1-D path to a name -> value mapping. Defaults to :func:`default_backtest_metrics`. A custom function lets you choose your own statistics and return convention.	`None`
`bars_per_year`	`int \| None`	Forwarded to :func:`default_backtest_metrics` for annualisation. Ignored when a custom `metric_fn` is given (bind any such option into the function yourself).	`None`

Returns:

Type	Description
`DataFrame`	DataFrame indexed `0 .. n_paths - 1` (index name `"path"`) with
`DataFrame`	one column per metric. With the default `metric_fn` the
`DataFrame`	`max_drawdown` column is a positive magnitude (`0.30` = a 30%
`DataFrame`	drawdown), so the worst path is `df["max_drawdown"].idxmax()`, not
`DataFrame`	`idxmin()`; see :func:`default_backtest_metrics`.

Raises:

Type	Description
`ValueError`	if `paths` is not 2-D.

Examples:

>>> import numpy as np
>>> from purgedcv import path_metrics
>>> paths = np.array([[0.01, -0.02, 0.03, 0.0], [0.0, 0.01, 0.01, 0.01]])
>>> df = path_metrics(paths)
>>> list(df.columns)
['sharpe', 'calmar', 'max_drawdown', 'total_return']
>>> df.shape
(2, 4)

purgedcv.default_backtest_metrics ¶

default_backtest_metrics(path: NDArrayAny, *, bars_per_year: int | None = None) -> dict[str, float]

Summarise one path of per-period returns.

Non-finite entries (for example NaN from a fold that could not be fit) are dropped before the statistics are computed. A path with fewer than two finite returns yields all-NaN metrics.

Returns are treated as additive (equity curve = cumulative sum), the convention for log returns. For simple returns supply a custom metric.

Parameters:

Name	Type	Description	Default
`path`	`NDArrayAny`	1-D per-period return series.	required
`bars_per_year`	`int \| None`	If given, the Sharpe ratio is annualised by `sqrt(bars_per_year)` and Calmar uses the annualised mean return. If `None`, Sharpe is per-bar and Calmar uses the total return.	`None`

Returns:

Type	Description
`dict[str, float]`	Dict with `sharpe`, `calmar`, `max_drawdown`, and
`dict[str, float]`	`total_return`. `max_drawdown` is a positive magnitude, not a
`dict[str, float]`	signed value: `0.30` means a 30% peak-to-trough drawdown (not
`dict[str, float]`	`-0.30`). So the worst path is the one with the largest
`dict[str, float]`	`max_drawdown`; sort descending, not ascending.

Examples:

>>> import numpy as np
>>> from purgedcv import default_backtest_metrics
>>> m = default_backtest_metrics(np.array([0.01, -0.02, 0.03, 0.0]))
>>> round(m["total_return"], 4)
0.02
>>> round(m["max_drawdown"], 4)
0.02

Row-level primitives¶

purgedcv.purge ¶

purge(train_idx: NDArrayAny, test_idx: NDArrayAny, prediction_times: Series, evaluation_times: Series, purge_horizon: Timedelta | None = None) -> NDArrayAny

Drop training rows whose half-open label horizon overlaps any test horizon.

Each test row contributes a half-open horizon [prediction_time - purge_horizon, evaluation_time + purge_horizon). The intervals are merged before filtering, so disjoint test blocks purge only their local overlap zones rather than the full convex hull between them.

A training row at position i is kept iff its label horizon overlaps none of those merged test horizons.

Parameters:

Name	Type	Description	Default
`train_idx`	`NDArrayAny`	positional indices of training rows.	required
`test_idx`	`NDArrayAny`	positional indices of test rows.	required
`prediction_times`	`Series`	prediction times for all rows.	required
`evaluation_times`	`Series`	evaluation times for all rows.	required
`purge_horizon`	`Timedelta \| None`	optional symmetric padding (`None` ≡ zero).	`None`

Returns:

Type	Description
`NDArrayAny`	The subset of `train_idx` whose horizons do not overlap the test
`NDArrayAny`	window. Input ordering and dtype are preserved.

Examples:

>>> import numpy as np
>>> import pandas as pd
>>> from purgedcv import purge
>>> pred = pd.Series(pd.date_range("2024-01-01", periods=10, freq="D"))
>>> evalu = pred + pd.Timedelta(days=3)
>>> train_idx = np.array([0, 1, 2, 3, 4, 8, 9])
>>> test_idx = np.array([5, 6, 7])
>>> purge(train_idx, test_idx, pred, evalu)
array([0, 1, 2])

purgedcv.apply_embargo ¶

apply_embargo(train_idx: NDArrayAny, test_idx: NDArrayAny, prediction_times: Series, evaluation_times: Series, embargo: Timedelta) -> NDArrayAny

Drop training rows whose prediction_time falls inside any closed embargo window [test_evaluation_time, test_evaluation_time + embargo].

Embargo is asymmetric: rows whose prediction_time is strictly before all test evaluation times are never dropped. embargo == 0 is the identity (the embargo window is logically empty at zero width), avoiding degenerate single-point windows.

Parameters:

Name	Type	Description	Default
`train_idx`	`NDArrayAny`	positional indices of training rows.	required
`test_idx`	`NDArrayAny`	positional indices of test rows.	required
`prediction_times`	`Series`	prediction times for all rows.	required
`evaluation_times`	`Series`	evaluation times for all rows.	required
`embargo`	`Timedelta`	post-test embargo duration.	required

Returns:

Type	Description
`NDArrayAny`	The subset of `train_idx` whose prediction_times fall outside the
`NDArrayAny`	post-test embargo window. Input ordering and dtype are preserved.

Examples:

>>> import numpy as np
>>> import pandas as pd
>>> from purgedcv import apply_embargo
>>> pred = pd.Series(pd.date_range("2024-01-01", periods=20, freq="D"))
>>> evalu = pred + pd.Timedelta(days=1)
>>> train_idx = np.array([11, 12, 13, 14])
>>> test_idx = np.arange(5, 10)
>>> apply_embargo(train_idx, test_idx, pred, evalu, pd.Timedelta(days=1))
array([12, 13, 14])

Time and horizon utilities¶

purgedcv.parse_horizon ¶

parse_horizon(value: HorizonLike) -> pd.Timedelta

Coerce a horizon-like input to a non-negative pd.Timedelta.

Accepts pandas offset strings ("2D", "6h", "30min"), pd.Timedelta, datetime.timedelta, and numpy.timedelta64. Rejects missing/NaT values, negative durations, and calendar-ambiguous offsets such as "M" (month) or "Y" (year), which do not represent a fixed duration in seconds.

Parameters:

Name	Type	Description	Default
`value`	`HorizonLike`	The horizon to parse.	required

Returns:

Type	Description
`Timedelta`	A non-negative `pd.Timedelta`.

Raises:

Type	Description
`ValueError`	if the input is missing/`NaT`, negative, or a calendar-ambiguous string.
`TypeError`	if the input is not one of the supported types.

Examples:

>>> from purgedcv import parse_horizon
>>> parse_horizon("2D")
Timedelta('2 days 00:00:00')
>>> parse_horizon("6h")
Timedelta('0 days 06:00:00')

purgedcv.horizons_overlap ¶

horizons_overlap(a_start: Timestamp, a_end: Timestamp, b_start: Timestamp, b_end: Timestamp) -> bool

Return True iff half-open intervals [a_start, a_end) and [b_start, b_end) overlap.

Touching intervals (a_end == b_start) do NOT overlap. The function is symmetric in its arguments.

Examples:

>>> import pandas as pd
>>> from purgedcv import horizons_overlap
>>> horizons_overlap(
...     pd.Timestamp("2024-01-01"), pd.Timestamp("2024-01-03"),
...     pd.Timestamp("2024-01-02"), pd.Timestamp("2024-01-04"),
... )
True
>>> horizons_overlap(
...     pd.Timestamp("2024-01-01"), pd.Timestamp("2024-01-02"),
...     pd.Timestamp("2024-01-02"), pd.Timestamp("2024-01-03"),
... )
False

purgedcv.validate_times ¶

validate_times(prediction_times: Series, evaluation_times: Series, *, require_monotonic: bool = True) -> None

Validate that prediction_times and evaluation_times are well-formed.

Raises:

Type	Description
`ValueError`	on length mismatch, non-temporal dtype, NaT values, `evaluation_times < prediction_times` at any row, or (when `require_monotonic`) non-monotonic prediction times. The error message names the offending row index when applicable.

Examples:

>>> import pandas as pd
>>> from purgedcv import validate_times
>>> pred = pd.Series(pd.date_range("2024-01-01", periods=5, freq="D"))
>>> evalu = pred + pd.Timedelta(days=1)
>>> validate_times(pred, evalu)

Statistical metrics¶

purgedcv.probabilistic_sharpe_ratio ¶

probabilistic_sharpe_ratio(returns: NDArrayAny, benchmark_skill: float) -> float

Probability that the true Sharpe ratio exceeds benchmark_skill.

Formula (Bailey & Lopez de Prado 2012, Eq. 7):

\[ \text{PSR}(\text{SR}^\ast) = \Phi\!\left( \frac{(\widehat{\text{SR}} - \text{SR}^\ast)\sqrt{n - 1}} {\sqrt{1 - \widehat{\gamma}_3\,\widehat{\text{SR}} + \frac{\widehat{\gamma}_4 - 1}{4}\,\widehat{\text{SR}}^{\,2}}} \right) \]

where \(\widehat{\gamma}_3\) is sample skew, \(\widehat{\gamma}_4\) is sample kurtosis (NOT excess kurtosis), and \(\Phi\) is the standard normal CDF.

Parameters:

Name	Type	Description	Default
`returns`	`NDArrayAny`	1-D array of returns, length >= 2, finite values, non-zero variance.	required
`benchmark_skill`	`float`	The Sharpe-ratio threshold to test against. Use 0 for "is this strategy better than holding cash."	required

Returns:

Type	Description
`float`	Scalar probability in [0, 1].

Raises:

Type	Description
`ValueError`	on length < 2, non-finite values, or zero variance.

Examples:

>>> import numpy as np
>>> from purgedcv import probabilistic_sharpe_ratio
>>> rng = np.random.default_rng(0)
>>> returns = rng.normal(0.001, 0.01, 252)
>>> psr = probabilistic_sharpe_ratio(returns, benchmark_skill=0.0)
>>> 0.0 <= psr <= 1.0
True

purgedcv.deflated_sharpe_ratio ¶

deflated_sharpe_ratio(returns: NDArrayAny, n_trials: int, var_sharpe: float, *, bars_per_year: int | None = None) -> float

Probability that the true Sharpe ratio exceeds the deflated benchmark that accounts for n_trials independent hyperparameter searches under the null.

Formula (Bailey & Lopez de Prado 2014):

\[ \text{SR}^\ast_{n} = \sqrt{V[\text{SR}]} \left[ (1 - \gamma) \Phi^{-1}\!\left(1 - \frac{1}{n_{\text{trials}}}\right) + \gamma \Phi^{-1}\!\left(1 - \frac{1}{n_{\text{trials}} \cdot e}\right) \right] \]

where \(\gamma \approx 0.5772\) is the Euler-Mascheroni constant. DSR is then :func:probabilistic_sharpe_ratio evaluated at the deflated benchmark \(\text{SR}^\ast_n\).

Parameters:

Name	Type	Description	Default
`returns`	`NDArrayAny`	1-D array of returns (passed through to PSR).	required
`n_trials`	`int`	Number of independent hyperparameter searches the user ran before reporting this strategy's Sharpe. Must be >= 1. With `n_trials == 1` there is no correction to apply and DSR reduces to `probabilistic_sharpe_ratio(returns, 0.0)`.	required
`var_sharpe`	`float`	Estimated variance of Sharpe ratios across the `n_trials` candidates. Caller supplies; we do not estimate it because that would require knowing the distribution of submitted strategies, which is private to the caller. :class:`purgedcv.optuna_integration.TrialSharpeRecorder` produces it directly from an Optuna study. UNITS: `var_sharpe` must be in the same Sharpe units as the per-observation Sharpe of `returns` (DSR is intrinsically per-observation). If your trial Sharpes were annualised, pass `bars_per_year` and the conversion is done for you; otherwise `var_sharpe` is taken as already per-observation. Note that :func:`path_metrics` annualises its Sharpe when given `bars_per_year`, so a `var` taken from its output is annualised: pass the same `bars_per_year` here.	required
`bars_per_year`	`int \| None`	If given, `var_sharpe` is interpreted as an annualised Sharpe variance and converted to per-observation (`var_sharpe / bars_per_year`) before deflation. `None` (default) treats `var_sharpe` as already per-observation and leaves prior behaviour unchanged.	`None`

Returns:

Type	Description
`float`	Scalar probability in [0, 1].

Raises:

Type	Description
`TypeError`	if `n_trials` is not an integer.
`ValueError`	on invalid `n_trials`, non-finite/negative `var_sharpe`, or non-positive `bars_per_year`.

Examples:

>>> import numpy as np
>>> from purgedcv import deflated_sharpe_ratio
>>> rng = np.random.default_rng(0)
>>> returns = rng.normal(0.001, 0.01, 252)
>>> dsr = deflated_sharpe_ratio(returns, n_trials=50, var_sharpe=0.01**2)
>>> 0.0 <= dsr <= 1.0
True
>>> # Annualised trial-Sharpe variance: pass bars_per_year to convert.
>>> annual = deflated_sharpe_ratio(returns, 50, 0.5**2, bars_per_year=252)
>>> per_bar = deflated_sharpe_ratio(returns, 50, 0.5**2 / 252)
>>> abs(annual - per_bar) < 1e-12
True

purgedcv.deflated_sharpe_ratio_full ¶

deflated_sharpe_ratio_full(returns: NDArrayAny, n_trials: int, var_sharpe: float, *, bars_per_year: int | None = None) -> DSRDiagnostics

Like :func:deflated_sharpe_ratio but return the intermediate quantities alongside the probability.

When DSR is near 0 the scalar form does not tell you why: was the deflated benchmark high because var_sharpe was large, because n_trials was large, or was the observed Sharpe simply low? This function returns all of those so the deflation can be inspected.

Parameters:

Name	Type	Description	Default
`returns`	`NDArrayAny`	1-D array of returns.	required
`n_trials`	`int`	Number of independent searches (>= 1).	required
`var_sharpe`	`float`	Variance of Sharpe ratios across the candidates. Per observation by default; pass `bars_per_year` if it is annualised (see :func:`deflated_sharpe_ratio` for the unit contract).	required
`bars_per_year`	`int \| None`	If given, `var_sharpe` is annualised and converted to per-observation before deflation. `None` (default) treats it as already per-observation.	`None`

Returns:

Name	Type	Description
`A`	`DSRDiagnostics`	class:`DSRDiagnostics` (frozen dataclass; read fields by
	`DSRDiagnostics`	attribute, e.g. `diag.dsr`, `diag.sr_star`). `dsr` equals
	`DSRDiagnostics`	func:`deflated_sharpe_ratio` for the same arguments. The
	`DSRDiagnostics`	`var_sharpe` field holds the per-observation value actually used
	`DSRDiagnostics`	(after any `bars_per_year` conversion), so
	`DSRDiagnostics`	`sr_star == sqrt(var_sharpe) * expected_max_z` always holds.

Raises:

Type	Description
`TypeError`	if `n_trials` is not an integer.
`ValueError`	on invalid `n_trials`, non-finite/negative `var_sharpe`, non-positive `bars_per_year`, or a degenerate `returns` series.

Examples:

>>> import numpy as np
>>> from purgedcv import deflated_sharpe_ratio, deflated_sharpe_ratio_full
>>> rng = np.random.default_rng(0)
>>> returns = rng.normal(0.001, 0.01, 252)
>>> diag = deflated_sharpe_ratio_full(returns, n_trials=50, var_sharpe=0.01**2)
>>> abs(diag.dsr - deflated_sharpe_ratio(returns, 50, 0.01**2)) < 1e-12
True
>>> diag.n_obs
252
>>> diag.sr_star > 0
True

purgedcv.min_track_record_length ¶

min_track_record_length(observed_sharpe: float, target_sharpe: float, alpha: float, skew: float, kurtosis: float) -> float

Minimum sample size such that PSR(target_sharpe) >= 1 - alpha.

Inverts the :func:probabilistic_sharpe_ratio formula for n:

\[ n^\ast = 1 + \left\lceil \left(\frac{\Phi^{-1}(1 - \alpha) \cdot \sqrt{1 - \gamma_3 \widehat{\text{SR}} + \frac{\gamma_4 - 1}{4} \widehat{\text{SR}}^2}} {\widehat{\text{SR}} - \text{SR}^\ast} \right)^2 \right\rceil \]

Bailey & Lopez de Prado (2012, Eq. 11).

Parameters:

Name	Type	Description	Default
`observed_sharpe`	`float`	The sample Sharpe ratio you actually observed.	required
`target_sharpe`	`float`	The benchmark you want to beat with confidence.	required
`alpha`	`float`	Significance level in (0, 1). PSR must meet 1 - alpha.	required
`skew`	`float`	Sample skew of the return distribution.	required
`kurtosis`	`float`	Sample kurtosis (NOT excess) of the return distribution.	required

Returns:

Name	Type	Description
	`float`	The minimum number of observations, as a `float`. When
	`float`	`observed_sharpe <= target_sharpe` no finite track record can
	`float`	establish the gap, so the answer is `math.inf` rather than an
`error`	`float`	"no length is long enough" is a well-defined result. Wrap
	`float`	in `int(...)` for a count when the value is finite.

Raises:

Type	Description
`ValueError`	if any scalar input is non-finite, if `alpha not in (0, 1)`, or if the higher moments are too extreme for the Gaussian approximation.

Examples:

>>> import math
>>> from purgedcv import min_track_record_length
>>> n = min_track_record_length(
...     observed_sharpe=0.5, target_sharpe=0.2,
...     alpha=0.05, skew=0.0, kurtosis=3.0,
... )
>>> n > 0
True
>>> # No track record proves a Sharpe you have not actually beaten:
>>> math.isinf(min_track_record_length(0.1, 0.2, 0.05, 0.0, 3.0))
True

purgedcv.minimum_backtest_length ¶

minimum_backtest_length(n_trials: int, target_sharpe: float = 1.0) -> float

Minimum backtest length, in years, below which a high in-sample Sharpe is not evidence of skill once n_trials configurations were tried.

Run n_trials independent backtests on a strategy with no real edge (true Sharpe 0) and the best of them shows a positive Sharpe purely by chance. The shorter the backtest, the larger that chance maximum. This returns the length at which the expected maximum annualised Sharpe across n_trials trials under the null equals target_sharpe: a reported annualised Sharpe of target_sharpe over a shorter history is within what selection alone would produce, so it is not yet evidence of skill.

Inverts the Bailey-Lopez de Prado expected-maximum approximation for the length y in years:

\[ \text{MinBTL} = \left( \frac{(1 - \gamma)\,\Phi^{-1}\!\left(1 - \frac{1}{n_{\text{trials}}}\right) + \gamma\,\Phi^{-1}\!\left(1 - \frac{1}{n_{\text{trials}} \cdot e}\right)} {\text{SR}^\ast} \right)^2 \]

where \(\gamma\) is the Euler-Mascheroni constant and \(\text{SR}^\ast\) is target_sharpe. The numerator is the same bracket term that deflates the Sharpe in :func:deflated_sharpe_ratio; MinBTL is its inverse, solving for length instead of for a probability.

Bailey, Borwein, Lopez de Prado & Zhu (2014), "Pseudo-mathematics and financial charlatanism", Notices of the AMS 61(5). AFML Chapter 11.

Parameters:

Name	Type	Description	Default
`n_trials`	`int`	Number of independent configurations tried. Integer >= 1.	required
`target_sharpe`	`float`	The annualised Sharpe threshold to clear. Must be a positive finite number. Defaults to 1.0.	`1.0`

Returns:

Type	Description
`float`	The minimum backtest length in years, as a `float`. Multiply by
`float`	the number of observations per year for a length in observations. A
`float`	single trial (`n_trials == 1`) involves no selection, so the bound
`float`	is `0.0`.

Raises:

Type	Description
`TypeError`	if `n_trials` is not an integer.
`ValueError`	if `n_trials < 1` or `target_sharpe` is not a positive finite number.

Examples:

>>> from purgedcv import minimum_backtest_length
>>> # Trying 10 configs, an annualised Sharpe of 1 is not evidence of
>>> # skill until about 2.5 years of backtest:
>>> round(minimum_backtest_length(10, target_sharpe=1.0), 1)
2.5
>>> # More trials demand a longer backtest for the same Sharpe:
>>> minimum_backtest_length(100) > minimum_backtest_length(10)
True
>>> minimum_backtest_length(1)  # a single trial: no selection
0.0

purgedcv.effective_n_trials ¶

effective_n_trials(trial_sharpes: NDArrayAny, method: str = 'autocorr') -> int

Estimate the number of independent trials behind a correlated search.

:func:deflated_sharpe_ratio assumes the n_trials candidates were drawn independently. Sequential samplers (Optuna's TPE, CMA-ES) draw each trial conditioned on the previous ones, so the trials are autocorrelated and the raw count overstates how many independent bets were really placed. Feeding the raw count inflates the deflated benchmark and makes DSR needlessly conservative (often numerically zero). This returns a smaller effective count to pass to :func:deflated_sharpe_ratio instead.

The estimate is the run length divided by the integrated autocorrelation time of the trial-performance series: n_eff = n / (1 + 2 * sum_k rho_k), summing the autocorrelations rho_k until the first non-positive lag (the initial-positive-sequence truncation, Geyer 1992). It is a heuristic: it depends on the truncation rule and assumes the trial order reflects the sampler's dependence, so treat the result as an order-of-magnitude correction, not an exact figure.

Parameters:

Name	Type	Description	Default
`trial_sharpes`	`NDArrayAny`	1-D array of per-trial performance values (Sharpe or objective), in the order the trials were run. All finite.	required
`method`	`str`	Estimator to use. Only `"autocorr"` is implemented.	`'autocorr'`

Returns:

Type	Description
`int`	Effective trial count, an integer in `[1, n]`. A near-independent
`int`	series returns close to `n`; a strongly autocorrelated one returns
`int`	far fewer. A constant series returns 1 (every trial was the same).
`int`	Fewer than 3 trials returns the raw count (too few to estimate).

Raises:

Type	Description
`ValueError`	on an unknown `method`, an empty or non-1-D array, or non-finite values.

Examples:

>>> import numpy as np
>>> from purgedcv import effective_n_trials
>>> rng = np.random.default_rng(0)
>>> indep = rng.standard_normal(400)            # independent trials
>>> drift = np.cumsum(rng.standard_normal(400))  # strongly autocorrelated
>>> n_indep = effective_n_trials(indep)
>>> n_drift = effective_n_trials(drift)
>>> 1 <= n_drift < n_indep <= 400
True

Backtest overfitting¶

purgedcv.probability_of_backtest_overfitting ¶

probability_of_backtest_overfitting(returns: NDArrayAny, n_splits: int = 16, *, metric: PerformanceMetric = sharpe, prediction_times: Series | None = None, evaluation_times: Series | None = None, purge_horizon: HorizonLike | None = None, embargo: HorizonLike | None = None) -> PBOResult

Estimate the Probability of Backtest Overfitting (PBO) via CSCV.

Given the per-period returns of n_configs candidate configurations, PBO is the probability that the configuration selected as best in-sample performs below the median out-of-sample. The estimate is the relative frequency of that event across all CSCV combinations.

Parameters:

Name	Type	Description	Default
`returns`	`NDArrayAny`	`(n_configs, n_obs)` array; `returns[j]` is the per- period return series of configuration `j` over a shared time axis. At least 2 configurations and `n_obs >= n_splits` are required; all values must be finite.	required
`n_splits`	`int`	Number of contiguous blocks the time axis is cut into (`S` in CSCV). Must be even and >= 2; the number of combinations is `C(n_splits, n_splits // 2)`. The standard choice is 16.	`16`
`metric`	`PerformanceMetric`	Performance metric mapping a 1-D return slice to a scalar (larger is better). Defaults to :func:`sharpe`.	`sharpe`
`prediction_times`	`Series \| None`	Optional per-period prediction times. Supply with `evaluation_times` to enable purge/embargo at IS/OOS boundaries. Length must equal `n_obs`.	`None`
`evaluation_times`	`Series \| None`	Optional per-period evaluation times; see `prediction_times`.	`None`
`purge_horizon`	`HorizonLike \| None`	Optional purge horizon (requires the time series).	`None`
`embargo`	`HorizonLike \| None`	Optional embargo horizon (requires the time series).	`None`

Returns:

Name	Type	Description
`A`	`PBOResult`	class:`PBOResult` (frozen dataclass; read fields by attribute,
	`PBOResult`	e.g. `result.pbo`, `result.slope`).

Raises:

Type	Description
`ValueError`	on a malformed `returns` matrix, an odd or too-small `n_splits`, a time series whose length disagrees with `n_obs`, or purge/embargo requested without times.

Examples:

>>> import numpy as np
>>> from purgedcv import probability_of_backtest_overfitting
>>> rng = np.random.default_rng(0)
>>> returns = rng.standard_normal((10, 240))  # 10 configs, pure noise
>>> result = probability_of_backtest_overfitting(returns, n_splits=8)
>>> 0.0 <= result.pbo <= 1.0
True
>>> result.n_combos
70

Optuna integration¶

purgedcv.optuna_integration.TrialSharpeRecorder ¶

TrialSharpeRecorder(attr: str = 'sharpe')

Optuna study callback that collects per-trial Sharpe ratios.

Pass an instance as a callback to study.optimize and store each trial's Sharpe in a user attribute (default key "sharpe"); the recorder accumulates them and reports the variance for :func:~purgedcv.deflated_sharpe_ratio. If a trial has no such user attribute, the recorder falls back to that trial's objective value. Non-finite and missing values are ignored.

Parameters:

Name	Type	Description	Default
`attr`	`str`	The `trial.user_attrs` key under which the Sharpe ratio is stored. Defaults to `"sharpe"`.	`'sharpe'`

Examples:

>>> from types import SimpleNamespace
>>> from purgedcv.optuna_integration import TrialSharpeRecorder
>>> recorder = TrialSharpeRecorder()
>>> # Optuna calls the recorder with (study, frozen_trial) per trial;
>>> # SimpleNamespace stands in for the trial here.
>>> for s in (1.2, 0.8, 1.5):
...     recorder(study=None, trial=SimpleNamespace(value=s, user_attrs={"sharpe": s}))
>>> recorder.n_trials()
3
>>> round(recorder.var_sharpe(), 4)
0.1233

Usage with a real study::

recorder = TrialSharpeRecorder()
study.optimize(objective, n_trials=300, callbacks=[recorder])
dsr = deflated_sharpe_ratio(
    best_returns, n_trials=recorder.n_trials(), var_sharpe=recorder.var_sharpe()
)

call ¶

__call__(study: object, trial: _TrialLike) -> None

Record one trial's Sharpe ratio. Matches the Optuna callback signature.

Missing, None, non-numeric, and non-finite values are skipped rather than raised: a callback that throws would abort the whole study.optimize run over a single malformed trial.

sharpes ¶

sharpes() -> NDArrayAny

The recorded Sharpe ratios, in trial order, as a 1-D array.

n_trials ¶

n_trials() -> int

Number of trials with a usable (finite) Sharpe ratio recorded.

n_effective ¶

n_effective(method: str = 'autocorr') -> int

Effective number of independent trials, for correlated samplers.

TPE and CMA-ES draw each trial conditioned on the previous ones, so the raw trial count overstates the independent search effort and makes :func:~purgedcv.deflated_sharpe_ratio overly conservative. This applies :func:~purgedcv.effective_n_trials to the recorded Sharpe series; pass the result as n_trials to deflate more realistically. Returns 0 when nothing has been recorded yet. See :func:~purgedcv.effective_n_trials for the heuristic's caveats.

var_sharpe ¶

var_sharpe(ddof: int = 1) -> float

Variance of the recorded Sharpe ratios, for deflated_sharpe_ratio.

Returns nan until more than ddof trials have been recorded.

UNITS: the variance is in the units of whatever Sharpe you stored per trial. :func:~purgedcv.deflated_sharpe_ratio compares against the per-period Sharpe of its returns argument, so record a per-period Sharpe in each trial (the same convention as the returns you will later deflate). Do that and the recorder feeds deflated_sharpe_ratio directly, with no unit conversion.

Diagnostics¶

purgedcv.diagnostics.compute_overlap_fraction ¶

compute_overlap_fraction(train_idx: NDArrayAny, test_idx: NDArrayAny, prediction_times: Series, evaluation_times: Series) -> float

Return the fraction of training rows whose half-open label horizon overlaps any test horizon.

Unlike :func:assert_no_temporal_leakage, this never raises — it returns 0.0 for clean splits and 1.0 when every training row leaks. Useful for logging splitter health metrics or for debugging a splitter that produces unexpected behavior.

Examples:

>>> import numpy as np
>>> import pandas as pd
>>> from purgedcv.diagnostics import compute_overlap_fraction
>>> pred = pd.Series(pd.date_range("2024-01-01", periods=20, freq="D"))
>>> evalu = pred + pd.Timedelta(days=1)
>>> compute_overlap_fraction(np.arange(5), np.arange(10, 15), pred, evalu)
0.0
>>> compute_overlap_fraction(
...     np.arange(10, 15), np.arange(10, 15), pred, evalu
... )
1.0

purgedcv.diagnostics.assert_no_temporal_leakage ¶

assert_no_temporal_leakage(train_idx: NDArrayAny, test_idx: NDArrayAny, prediction_times: Series, evaluation_times: Series, *, purge_horizon: HorizonLike | None = None) -> None

Raise :class:TemporalLeakageError if any training row's label horizon overlaps any test label horizon, optionally padded on both sides by purge_horizon.

Test horizons are checked as a union of half-open intervals, not as one convex hull. This matters for CPCV folds whose test groups are intentionally non-contiguous.

Parameters:

Name	Type	Description	Default
`train_idx`	`NDArrayAny`	positional indices of training rows.	required
`test_idx`	`NDArrayAny`	positional indices of test rows.	required
`prediction_times`	`Series`	prediction times for all rows.	required
`evaluation_times`	`Series`	evaluation times for all rows.	required
`purge_horizon`	`HorizonLike \| None`	optional padding (default: `None` ≡ zero padding).	`None`

Raises:

Type	Description
`TemporalLeakageError`	with the offending training row index and the two overlapping intervals in the message.

Examples:

>>> import numpy as np
>>> import pandas as pd
>>> from purgedcv.diagnostics import assert_no_temporal_leakage
>>> pred = pd.Series(pd.date_range("2024-01-01", periods=20, freq="D"))
>>> evalu = pred + pd.Timedelta(days=1)
>>> assert_no_temporal_leakage(np.arange(5), np.arange(10, 15), pred, evalu)

purgedcv.diagnostics.assert_groups_disjoint ¶

assert_groups_disjoint(train_idx: NDArrayAny, test_idx: NDArrayAny, groups: Series) -> None

Raise :class:GroupLeakageError if any group identifier appears in both train_idx and test_idx.

Used by group-aware splitters to verify that no entity (patient, asset, user, etc.) is represented in both training and test of the same fold. The error message names a representative overlapping group plus the total count of overlapping groups, so the caller can scope follow-up.

Examples:

>>> import numpy as np
>>> import pandas as pd
>>> from purgedcv.diagnostics import assert_groups_disjoint
>>> groups = pd.Series([0, 0, 1, 1, 2, 2])
>>> assert_groups_disjoint(np.array([0, 1]), np.array([2, 3]), groups)

purgedcv.diagnostics.assert_embargo_respected ¶

assert_embargo_respected(train_idx: NDArrayAny, test_idx: NDArrayAny, prediction_times: Series, evaluation_times: Series, embargo: HorizonLike) -> None

Raise :class:EmbargoViolationError if any training row's prediction_time falls inside any closed embargo window [test_evaluation_time, test_evaluation_time + embargo].

Embargo is asymmetric: rows whose prediction_time is strictly before all test evaluation times are never flagged. embargo == 0 is the identity (no rows flagged) — the embargo window is logically empty at zero width.

Examples:

>>> import numpy as np
>>> import pandas as pd
>>> from purgedcv.diagnostics import assert_embargo_respected
>>> pred = pd.Series(pd.date_range("2024-01-01", periods=20, freq="D"))
>>> evalu = pred + pd.Timedelta(days=1)
>>> assert_embargo_respected(
...     np.array([18]), np.arange(5, 10), pred, evalu, embargo="2D"
... )

Exceptions¶

purgedcv.TemporalCVError ¶

Bases: ValueError

Base class for all purged-cross-validation errors.

purgedcv.TemporalLeakageError ¶

Bases: TemporalCVError

Raised when a training row's label horizon overlaps the test horizon.

purgedcv.EmbargoViolationError ¶

Bases: TemporalCVError

Raised when a training row falls inside the post-test embargo window.

purgedcv.GroupLeakageError ¶

Bases: TemporalCVError

Raised when a group_id appears in both training and test of the same fold.

API reference¶

Splitters¶

purgedcv.BaseTemporalSplitter ¶

get_n_splits abstractmethod ¶

split ¶

with_times ¶

purgedcv.WalkForwardSplit ¶

split ¶

purgedcv.PurgedKFold ¶

purgedcv.PurgedGroupKFold ¶

purgedcv.CombinatorialPurgedCV ¶

backtest_paths ¶

reconstruct_paths ¶

purgedcv.CombinatoriallySymmetricCV ¶

Backtest paths¶

purgedcv.reconstruct_paths ¶

purgedcv.path_metrics ¶

purgedcv.default_backtest_metrics ¶

Row-level primitives¶

purgedcv.purge ¶

purgedcv.apply_embargo ¶

Time and horizon utilities¶

purgedcv.parse_horizon ¶

purgedcv.horizons_overlap ¶

purgedcv.validate_times ¶

Statistical metrics¶

purgedcv.probabilistic_sharpe_ratio ¶

purgedcv.deflated_sharpe_ratio ¶

purgedcv.deflated_sharpe_ratio_full ¶

purgedcv.min_track_record_length ¶

purgedcv.minimum_backtest_length ¶

purgedcv.effective_n_trials ¶

Backtest overfitting¶

purgedcv.probability_of_backtest_overfitting ¶

Optuna integration¶

purgedcv.optuna_integration.TrialSharpeRecorder ¶

__call__ ¶

sharpes ¶

n_trials ¶

n_effective ¶

var_sharpe ¶

Diagnostics¶

purgedcv.diagnostics.compute_overlap_fraction ¶

purgedcv.diagnostics.assert_no_temporal_leakage ¶

purgedcv.diagnostics.assert_groups_disjoint ¶

purgedcv.diagnostics.assert_embargo_respected ¶

Exceptions¶

purgedcv.TemporalCVError ¶

purgedcv.TemporalLeakageError ¶

purgedcv.EmbargoViolationError ¶

purgedcv.GroupLeakageError ¶

get_n_splits `abstractmethod` ¶

call ¶