GeneratedGenerated from the built extension by scripts/generate_python_api_reference.py; CI fails if it is stale.

openquant._core.cross_validation

Compiled from Rust (crates/pyopenquant). Signatures and types are from the shipped .pyi stubs; descriptions are the functions’ own docstrings, as help() shows them. All Python modules.

Functions

`count_train_test_overlaps`

count_train_test_overlaps(
    t0: Sequence[int],
    t1: Sequence[int],
    train_indices: Sequence[int],
    test_indices: Sequence[int],
) -> int

Count the training samples whose label span overlaps some test sample's span.

Spans are compared as closed intervals, so a label that ends exactly when a test label starts counts as an overlap. Use it with naive_kfold_splits to measure the leak that purging removes (AFML section 7.3); on a purged split it is always 0.

Parameters

t0 : list[int]: Start of each sample's label span, as int64 nanoseconds since the epoch (plain integers such as bar positions also work; only the order of the values matters). One per sample, in time order.
t1 : list[int]: End of each sample's label span, in the same units as t0.
train_indices : list[int]: Positions of the training samples in t0/t1.
test_indices : list[int]: Positions of the test samples in t0/t1.

Returns

int: Number of entries of train_indices that overlap at least one test span.

Raises

ValueError: If t0 and t1 differ in length, or if an index in train_indices or test_indices is not a valid position in them.

`cpcv_paths`

cpcv_paths(n_splits: int, n_test_splits: int) -> list[list[int]]

The CPCV backtest paths for n_splits folds tested n_test_splits at a time (AFML 12.4).

There are phi = n_test_splits / n_splits * C(n_splits, n_test_splits) paths, and each fold is tested in exactly phi splits of cpcv_splits. Path j takes, for every fold, the j-th split (in split_id order) that tests it; stitching those predictions together gives one out-of-sample prediction for every sample. Paths depend on the fold counts alone, so no label spans are needed.

Parameters

n_splits : int: Number of contiguous folds; at least 2.
n_test_splits : int: Number of folds tested together in each split; 1 <= n_test_splits < n_splits.

Returns

list[list[int]]: One list per path, in path order; element g is the split_id of the cpcv_splits split whose predictions for fold g the path uses.

Raises

ValueError: If the core rejects the input (e.g. n_splits below 2, n_test_splits not in [1, n_splits), or a split count too large for the platform's integers).

`cpcv_splits`

cpcv_splits(
    t0: Sequence[int],
    t1: Sequence[int],
    n_splits: int,
    n_test_splits: int,
    pct_embargo: float,
) -> list[dict[str, Any]]

Combinatorial purged cross-validation splits (AFML section 12.4).

One split is built for each of the C(n_splits, n_test_splits) ways to choose the test folds, in lexicographic order of the chosen folds. Each contiguous run of test folds is purged and embargoed exactly like a purged_kfold_splits fold. All splits are materialised at once: n_splits=10, n_test_splits=5 is already 252 splits. The split_id numbering matches openquant.backtesting_engine.run_cpcv, which expects one return array per split in this order.

Parameters

t0 : list[int]: Start of each sample's label span, as int64 nanoseconds since the epoch (plain integers such as bar positions also work; only the order of the values matters). One per sample, in time order.
t1 : list[int]: End of each sample's label span, in the same units as t0. Must be >= t0.
n_splits : int: Number of contiguous test folds; 2 <= n_splits <= len(t0).
pct_embargo : float: Embargo as a fraction of the whole sample count, in [0, 1); rounded up to a number of samples, so any positive value embargoes at least one. AFML suggests about 0.01.
n_test_splits : int: Number of folds tested together in each split; 1 <= n_test_splits < n_splits.

Returns

list[dict[str, Any]]

One dict per split, in split_id order, with the keys of split_with_diagnostics:

split_id (int): position of the split in the list.
train_indices (list[int]): sorted training indices after purging and embargo.
test_indices (list[int]): sorted test indices.
test_ranges (list[tuple[int, int]]): the test set as half-open [start, stop) blocks.
purged_indices (list[int]): non-test samples removed by purging.
embargo_indices (list[int]): non-test samples inside an embargo window (a sample can be both purged and embargoed).
overlap_count_after_purge (int): training samples whose span still overlaps a test span; always 0, reported so callers can assert it.

plus test_fold_ids (list[int]): the folds tested in this split, ascending.

Raises

ValueError: If t0 and t1 differ in length, or if the core rejects the input (e.g. no samples, n_splits below 2 or above the sample count, pct_embargo not a finite number in [0, 1), a span that ends before it starts, n_test_splits not in [1, n_splits), or a split count too large for the platform's integers).

`naive_kfold_splits`

naive_kfold_splits(n_samples: int, n_splits: int) -> list[tuple[list[int], list[int]]]

Unpurged k-fold splits: contiguous test folds, every other sample trains.

This is the baseline AFML section 7.3 warns against. It exists to measure leakage (for example with count_train_test_overlaps), not to validate models. Folds are contiguous and never shuffled; the first n_samples % n_splits folds hold one extra sample.

Parameters

n_samples : int: Number of samples.
n_splits : int: Number of folds; 2 <= n_splits <= n_samples.

Returns

list[tuple[list[int], list[int]]]: One (train_indices, test_indices) pair per fold, in fold order; both lists sorted.

Raises

ValueError: If the core rejects the input (n_splits below 2 or above n_samples).

`purged_kfold_splits`

purged_kfold_splits(
    t0: Sequence[int],
    t1: Sequence[int],
    n_splits: int,
    pct_embargo: float,
) -> list[tuple[list[int], list[int]]]

Purged k-fold splits with an embargo (AFML Snippet 7.3), as index lists.

Folds are blocks of consecutive samples, never shuffled; the first n % n_splits folds hold one extra sample. Purging (AFML Snippet 7.1) drops every non-test sample whose span intersects, as closed intervals, the window from a test block's first start to its latest end. The embargo (Snippet 7.3) then drops ceil(pct_embargo * n) more samples after each test block, counting from where the purge ends; nothing before a block is embargoed. The purge compares timestamps but the embargo counts positions, so the samples must be sorted by start (this is not checked).

Purging can empty a training set when labels are long relative to the folds; check the length of each training list.

Parameters

t0 : list[int]: Start of each sample's label span, as int64 nanoseconds since the epoch (plain integers such as bar positions also work; only the order of the values matters). One per sample, in time order.
t1 : list[int]: End of each sample's label span, in the same units as t0. Must be >= t0.
n_splits : int: Number of contiguous test folds; 2 <= n_splits <= len(t0).
pct_embargo : float: Embargo as a fraction of the whole sample count, in [0, 1); rounded up to a number of samples, so any positive value embargoes at least one. AFML suggests about 0.01.

Returns

list[tuple[list[int], list[int]]]: One (train_indices, test_indices) pair per fold, in fold order; both lists sorted.

Raises

ValueError: If t0 and t1 differ in length, or if the core rejects the input (e.g. no samples, n_splits below 2 or above the sample count, pct_embargo not a finite number in [0, 1), or a span that ends before it starts).

`split_with_diagnostics`

split_with_diagnostics(
    t0: Sequence[int],
    t1: Sequence[int],
    n_splits: int,
    pct_embargo: float,
) -> list[dict[str, Any]]

Purged k-fold splits together with the indices purging and the embargo removed.

The folds are those of purged_kfold_splits (AFML Snippet 7.3); each also reports why every non-training, non-test sample was left out.

Parameters

t0 : list[int]: Start of each sample's label span, as int64 nanoseconds since the epoch (plain integers such as bar positions also work; only the order of the values matters). One per sample, in time order.
t1 : list[int]: End of each sample's label span, in the same units as t0. Must be >= t0.
n_splits : int: Number of contiguous test folds; 2 <= n_splits <= len(t0).
pct_embargo : float: Embargo as a fraction of the whole sample count, in [0, 1); rounded up to a number of samples, so any positive value embargoes at least one. AFML suggests about 0.01.

Returns

list[dict[str, Any]]

One dict per fold, in fold order, with keys:

split_id (int): position of the split in the list.
train_indices (list[int]): sorted training indices after purging and embargo.
test_indices (list[int]): sorted test indices.
test_ranges (list[tuple[int, int]]): the test set as half-open [start, stop) blocks.
purged_indices (list[int]): non-test samples removed by purging.
embargo_indices (list[int]): non-test samples inside an embargo window (a sample can be both purged and embargoed).
overlap_count_after_purge (int): training samples whose span still overlaps a test span; always 0, reported so callers can assert it.

Raises

ValueError: If t0 and t1 differ in length, or if the core rejects the input (e.g. no samples, n_splits below 2 or above the sample count, pct_embargo not a finite number in [0, 1), or a span that ends before it starts).

`walk_forward_splits`

walk_forward_splits(
    t0: Sequence[int],
    t1: Sequence[int],
    n_splits: int,
    pct_embargo: float,
    min_train_folds: int,
) -> list[dict[str, Any]]

Walk-forward splits over the purged k-fold folds, each trained only on earlier samples.

The folds are those of purged_kfold_splits. Fold g, for g >= min_train_folds, is tested; its training set is every sample *before* the fold whose span does not overlap the fold's window (from its first start to its latest end, closed intervals). Nothing after the fold is trained on, so no model sees the future of its test fold, and the embargo (which only removes samples after a test block) removes nothing. The first min_train_folds folds are never tested.

Purged k-fold and CPCV also fit the model for a middle fold on the samples after it. With a feature that carries memory of the price level this inflates the score even on random walks (issue #217), which purging does not prevent; walk-forward does.

Parameters

t0 : list[int]: Start of each sample's label span, as int64 nanoseconds since the epoch (plain integers such as bar positions also work; only the order of the values matters). One per sample, in time order.
t1 : list[int]: End of each sample's label span, in the same units as t0. Must be >= t0.
n_splits : int: Number of contiguous folds; 2 <= n_splits <= len(t0).
pct_embargo : float: Embargo fraction in [0, 1), validated as for purged_kfold_splits; it removes nothing in walk-forward splits.
min_train_folds : int: Number of leading folds that only train; 1 <= min_train_folds < n_splits.

Returns

list[dict[str, Any]]

One dict per tested fold, in fold order, with the keys of split_with_diagnostics:

split_id (int): position of the split in the list.
train_indices (list[int]): sorted training indices, all before the test fold.
test_indices (list[int]): sorted test indices.
test_ranges (list[tuple[int, int]]): the test fold as one half-open block.
purged_indices (list[int]): earlier samples removed by purging.
embargo_indices (list[int]): always empty.
overlap_count_after_purge (int): always 0.

plus test_fold_id (int): the fold tested, numbered as in purged_kfold_splits.

Raises

ValueError: If t0 and t1 differ in length, or if the core rejects the input (e.g. no samples, n_splits below 2 or above the sample count, pct_embargo not a finite number in [0, 1), a span that ends before it starts, or min_train_folds not in [1, n_splits)).