Cross-validation that a time series survives

daily = cu.fetch_daily(cu.SP500_EXAMPLES, start="2018-01-01", end="2026-07-01")
db = h5i_db.Database(cu.fresh_db("06_cross_validation_without_leakage"), create=True)
prices = daily.sort_by([("ts", "ascending"), ("symbol", "ascending")])
db.create_table("prices", prices.schema, time_column="ts", sort_key=["ts", "symbol"])
db.append("prices", prices, note="30 large caps, 2018-2026")
db.snapshot("prices-v1", tables=["prices"], note="Every candidate reads this cut")

LOOKBACKS = (21, 42, 63, 126, 189, 252)
pinned = db.table("prices", snapshot="prices-v1")
price_frame = pinned.select(ts=col("ts"), asset=col("symbol"), price=col("adj_close"))

series = {}
for lookback in LOOKBACKS:
    factor = (
        pinned.with_columns(
            past=sql_expr(f"lag(adj_close, {lookback})").over(
                partition_by="symbol", order_by="ts"
            )
        )
        .with_columns(value=col("adj_close") / col("past") - 1)
        .select(ts=col("ts"), asset=col("symbol"), factor=col("value"))
    )
    panel = quant.build_panel(db, factor, price_frame, periods=(HOLD,), quantiles=5)
    spread = panel.spread().to_pandas()[["ts", f"spread_{HOLD}"]]
    series[f"mom_{lookback}"] = spread.set_index("ts")[f"spread_{HOLD}"]

candidates = pd.DataFrame(series).dropna()
candidates.index = pd.DatetimeIndex(candidates.index)
print(f"{len(candidates):,} dates x {candidates.shape[1]} candidates")
candidates.tail(3).round(4)

scores = candidates.mean() / candidates.std() * np.sqrt(252)
print("in-sample Sharpe of each candidate, whole history:")
print(scores.round(3).to_string())
print(f"\nbest in sample: {scores.idxmax()}")

n = len(candidates)
horizons = [HOLD] * n
splits = list(quant.purged_kfold(n, folds=5, horizons=horizons, embargo=0.01))
folds = pd.DataFrame(
    [
        {
            "fold": index,
            "train": split.train_size,
            "test": split.test_size,
            "purged": len(split.purged),
            "purged %": len(split.purged) / (split.train_size + len(split.purged)),
        }
        for index, split in enumerate(splits)
    ]
)
folds.round(4)

fig, ax = plt.subplots(figsize=(9, 3.2))
for index, split in enumerate(splits):
    ax.scatter(split.test, [index] * split.test_size, s=2, color="#e45756", label="test" if index == 0 else None)
    ax.scatter(split.purged, [index] * len(split.purged), s=6, color="#f58518", label="purged" if index == 0 else None)
    ax.scatter(split.train, [index] * split.train_size, s=1, color="#4c78a8", alpha=0.4, label="train" if index == 0 else None)
ax.set_title(f"Purged 5-fold with a {HOLD}-day horizon and a 1% embargo")
ax.set_xlabel("Observation index")
ax.set_ylabel("Fold")
ax.legend(loc="upper right", markerscale=4)
fig.tight_layout()

def sharpe(values) -> float:
    values = np.asarray(values, dtype=float)
    if values.size < 2 or values.std(ddof=1) == 0:
        return float("nan")
    return float(values.mean() / values.std(ddof=1) * np.sqrt(252))


naive = [
    quant.validation.Split(
        train=tuple(i for i in range(n) if not (split.test[0] <= i <= split.test[-1])),
        test=split.test,
    )
    for split in splits
]

comparison = []
for label, chosen in (("naive k-fold", naive), ("purged + embargo", splits)):
    train_scores, test_scores = {}, {}
    for name in candidates.columns:
        column = candidates[name].to_numpy()
        train_scores[name] = np.mean([sharpe(column[list(s.train)]) for s in chosen])
        test_scores[name] = np.mean([sharpe(column[list(s.test)]) for s in chosen])
    winner = max(train_scores, key=train_scores.get)
    comparison.append(
        {
            "procedure": label,
            "train rows": sum(s.train_size for s in chosen),
            "picked": winner,
            "train Sharpe of pick": train_scores[winner],
            "test Sharpe of pick": test_scores[winner],
        }
    )
pd.DataFrame(comparison).round(3)

paths = list(quant.combinatorial_purged(n, groups=6, test_groups=2, horizons=horizons, embargo=0.01))
picked = comparison[-1]["picked"]
column = candidates[picked].to_numpy()
path_scores = np.array([sharpe(column[list(split.test)]) for split in paths])
print(f"{len(paths)} paths from C(6, 2)")
print(f"median   {np.nanmedian(path_scores):.3f}")
print(f"quartiles{np.nanpercentile(path_scores, 25):8.3f} to {np.nanpercentile(path_scores, 75):.3f}")
print(f"worst    {np.nanmin(path_scores):.3f}")
print(f"share of paths below zero: {(path_scores < 0).mean():.0%}")

fig, ax = plt.subplots(figsize=(9, 4))
ax.hist(path_scores, bins=12, color="#4c78a8", alpha=0.8)
ax.axvline(np.nanmedian(path_scores), color="black", linewidth=1.2, label="median")
ax.axvline(0, color="#e45756", linewidth=1.0, linestyle="--")
ax.set_title(f"Out-of-sample Sharpe of {picked} across {len(paths)} CPCV paths")
ax.set_xlabel("Annualized Sharpe on the held-out blocks")
ax.set_ylabel("Paths")
ax.legend()
fig.tight_layout()

rolling = list(quant.walk_forward(n, train_size=500, test_size=125, horizons=horizons, embargo=0.01))
expanding = list(
    quant.walk_forward(n, train_size=500, test_size=125, horizons=horizons, embargo=0.01, expanding=True)
)
def walk_selection(chosen) -> pd.DataFrame:
    """Pick on each fold's training window, score on the block that follows."""
    picks = []
    for index, split in enumerate(chosen):
        train_pick = max(
            candidates.columns,
            key=lambda name: sharpe(candidates[name].to_numpy()[list(split.train)]),
        )
        picks.append(
            {
                "fold": index,
                "picked": train_pick,
                "test Sharpe": sharpe(candidates[train_pick].to_numpy()[list(split.test)]),
            }
        )
    return pd.DataFrame(picks)


walk = pd.DataFrame(
    [
        {
            "window": label,
            "folds": len(chosen),
            "first train size": chosen[0].train_size,
            "last train size": chosen[-1].train_size,
            "changed its mind": walk_selection(chosen)["picked"].nunique(),
            "mean test Sharpe of its pick": walk_selection(chosen)["test Sharpe"].mean(),
        }
        for label, chosen in (("rolling", rolling), ("expanding", expanding))
    ]
)
walk.round(3)

walk_choices = walk_selection(rolling)
print(f"the rolling procedure changed its mind {walk_choices['picked'].nunique()} times")
print(f"mean test Sharpe of what it picked: {walk_choices['test Sharpe'].mean():.3f}")
walk_choices.round(3)

pbo = quant.probability_of_backtest_overfitting(candidates.to_numpy(), partitions=8)
print(f"PBO                  {pbo.pbo:.3f}")
print(f"splits evaluated     {pbo.splits}")
print(f"strategies compared  {pbo.strategies}")

best = candidates[picked].to_numpy()
deflated = quant.deflated_sharpe(best, trials=len(LOOKBACKS), trials_source="counted")
print(f"observed Sharpe (per day)  {deflated.sharpe:.4f}")
print(f"benchmark from {deflated.trials} trials     {deflated.benchmark:.4f}")
print(f"probability it is real     {deflated.probability:.3f}")
print(f"significant at 95%         {deflated.is_significant}")

needed = quant.minimum_track_record_length(best, benchmark=0.0, confidence=0.95)
print(f"\nobservations available     {len(best):,}")
print(f"observations needed        {needed:,.0f}")

verdict = pd.DataFrame(
    [
        {"question": "does selection carry information?", "number": f"PBO {pbo.pbo:.2f}",
         "answer": "yes" if pbo.pbo < 0.5 else "no"},
        {"question": "is the winner's Sharpe real?", "number": f"p {deflated.probability:.2f}",
         "answer": "yes" if deflated.is_significant else "not yet"},
        {"question": "is the track record long enough?",
         "number": f"{len(best):,} of {needed:,.0f}",
         "answer": "yes" if len(best) >= needed else "no"},
    ]
)
verdict