File size: 4,856 Bytes
90809d4 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 | """Walk-forward / CPCV splitting with purge and embargo.
Anchored on López de Prado, Advances in Financial Machine Learning (2018),
ch. 7 and ch. 11-12, implemented on top of the `purgedcv` package (MIT,
pypi purgedcv==0.1.11).
Conventions used by this eval:
- labels carry a horizon h (forward return over h bars); training samples
whose label window [t, t+h] overlaps a test block are PURGED
- EMBARGO of e bars is additionally applied after each test block;
default e = max(label horizon, one funding interval in bars) = 8 bars at 1h
- CPCV: N blocks, K held out per combination -> C(N, K) test sets, which
recombine into backtest paths via purgedcv.reconstruct_paths.
Defaults N=6, K=2: 15 combinations, C(5,1)=5 paths.
"""
from __future__ import annotations
import numpy as np
import pandas as pd
try:
from purgedcv import (
CombinatorialPurgedCV,
reconstruct_paths,
)
except ImportError as e: # pragma: no cover
raise ImportError(
"purgedcv is required: pip install purgedcv==0.1.11"
) from e
__all__ = ["make_cpcv", "make_walkforward", "SplitAudit"]
class SplitAudit:
"""No-lookahead audit for a produced split."""
def __init__(self, n_samples: int, horizon: int, embargo: int):
self.n_samples = n_samples
self.horizon = horizon
self.embargo = embargo
self.violations: list[str] = []
def check(self, train_idx: np.ndarray, test_idx: np.ndarray, label: str) -> None:
if len(test_idx) == 0 or len(train_idx) == 0:
self.violations.append(f"{label}: empty fold")
return
# CPCV test sets are the UNION of K blocks: train samples between
# two test blocks are legitimate. Audit each contiguous test
# segment separately instead of the whole [min, max] range.
breaks = np.flatnonzero(np.diff(np.sort(test_idx)) > 1)
segments = np.split(np.sort(test_idx), breaks + 1)
total_bad = 0
for seg in segments:
t_min, t_max = int(seg.min()), int(seg.max())
# overlap iff i <= t_max+embargo and i+h >= t_min
leak_zone_end = t_max + self.embargo
bad = int(((train_idx <= leak_zone_end)
& (train_idx + self.horizon >= t_min)).sum())
total_bad += bad
if total_bad:
self.violations.append(
f"{label}: {total_bad} training samples overlap test window "
f"(leakage)"
)
@property
def ok(self) -> bool:
return not self.violations
def make_cpcv(index: pd.DatetimeIndex, horizon: int, embargo: int,
freq: str = "1h", n_blocks: int = 6, k: int = 2):
"""Return the purgedcv CPCV splitter and audit it.
index: bar start times (UTC). horizon/embargo are in BARS.
purgedcv 0.1.11 requires prediction/evaluation times and takes
purge/embargo as timedeltas; it purges internally, SplitAudit is
kept as an independent leakage check.
"""
freq_td = pd.Timedelta(freq)
horizon_td = horizon * freq_td
embargo_td = embargo * freq_td
if embargo_td < horizon_td:
raise ValueError(
f"embargo ({embargo} bars) must be >= label horizon ({horizon} bars)"
)
splitter = CombinatorialPurgedCV(
n_splits=n_blocks,
n_test_groups=k,
prediction_times=index.values,
evaluation_times=(index + freq_td).values,
purge_horizon=horizon_td,
embargo=embargo_td,
)
n_samples = len(index)
audit = SplitAudit(n_samples, horizon, embargo)
combos = list(splitter.split(np.arange(n_samples)))
for j, (train_idx, test_idx) in enumerate(combos):
audit.check(np.asarray(train_idx), np.asarray(test_idx), f"cpcv-combo-{j}")
if not audit.ok:
raise RuntimeError(
"no-lookahead audit failed:\n" + "\n".join(audit.violations)
)
return splitter, combos, audit
def make_walkforward(n_samples: int, n_folds: int, horizon: int, embargo: int):
"""Expanding-window walk-forward with purge + embargo, audited."""
if embargo < horizon:
raise ValueError(
f"embargo ({embargo}) must be >= label horizon ({horizon})"
)
fold_bounds = np.linspace(0, n_samples, n_folds + 2, dtype=int)
folds = []
audit = SplitAudit(n_samples, horizon, embargo)
for j in range(n_folds):
test_start, test_end = fold_bounds[j + 1], fold_bounds[j + 2]
train_idx = np.arange(0, test_start - embargo - horizon)
test_idx = np.arange(test_start, test_end)
audit.check(train_idx, test_idx, f"wf-fold-{j}")
folds.append((train_idx, test_idx))
if not audit.ok:
raise RuntimeError(
"no-lookahead audit failed:\n" + "\n".join(audit.violations)
)
return folds, audit
|