| """Does a US-market-holiday factor improve the live HAR price range? |
| |
| expiry_holiday_study.py found BTC calmer on weekdays the NYSE is closed |
| (-0.62 at the US-session hour, -0.28 over the rest of the day). The dates are |
| known years ahead, so they qualify for the calendar. This tests whether |
| adding them earns their place: the live HAR range with and without a variance |
| factor for the 24 UTC hours of each holiday, fitted on the training block |
| like the Fed/CPI factor, on the same walk-forward quarters and scoring as |
| backtest_range_upgrades.py. |
| |
| Usage: |
| python scripts/backtest_holiday_factor.py |
| """ |
| from __future__ import annotations |
|
|
| import json |
| import sys |
| from pathlib import Path |
|
|
| import numpy as np |
| import pandas as pd |
|
|
| sys.path.insert(0, str(Path(__file__).resolve().parent)) |
| from backtest_improvements import GAP, STEP, TEST_START, WINDOW, fit_factors, prepare |
| from backtest_price_range import HORIZONS, LEVELS |
| from backtest_range_upgrades import dm, fwd_sum, log_ols, scores |
| from expiry_holiday_study import holidays |
|
|
| ROOT = Path(__file__).resolve().parents[1] |
| RULES = ["HAR (live)", "HAR + US holidays"] |
|
|
|
|
| def main() -> None: |
| base = prepare() |
| idx = base["ret"].index |
| days = {t.floor("D") for t in holidays(idx.min(), idx.max())} |
| hol = pd.Series(idx.floor("D").isin(list(days)), index=idx) |
| ret2 = base["ret"] ** 2 |
| logp = base["logp"] |
| moves = {h: (logp.shift(-h) - logp).to_numpy() for h in HORIZONS} |
| rows = {(h, r): [] for h in HORIZONS for r in RULES} |
| k_log = [] |
|
|
| start = pd.Timestamp(TEST_START, tz="UTC") |
| while start < idx.max(): |
| end = start + STEP |
| train = (idx >= start - WINDOW) & (idx <= start - GAP) |
| test = (idx >= start) & (idx < end) |
| f = fit_factors(base, train) |
| season0 = f["var"]["+ Fed/CPI calendar"] |
| |
| k = float((base["u2"] / season0)[train & hol.to_numpy()].mean()) |
| k_log.append({"quarter": f"{start:%Y-%m}", "k_holiday": k}) |
| for rule, season in (("HAR (live)", season0), |
| ("HAR + US holidays", season0 * np.where(hol, k, 1.0))): |
| e2 = ret2 / season |
| X = np.column_stack([e2.rolling(w, min_periods=int(w * .9)).mean().to_numpy() for w in (24, 168, 720)]) |
| for h in HORIZONS: |
| fs = fwd_sum(season, h) |
| sc = np.sqrt(log_ols(fwd_sum(e2, h) / h, X, train)(X) * fs) |
| z = moves[h] / sc |
| ok_tr, ok_te = train & np.isfinite(z), test & np.isfinite(z) |
| q = {l: float(np.quantile(np.abs(z[ok_tr]), l)) for l in LEVELS} |
| rows[(h, rule)].append(pd.DataFrame({ |
| "z": z[ok_te], "scale": sc[ok_te], "move": moves[h][ok_te], |
| "hol": fwd_sum(pd.Series(hol.astype(float).to_numpy(), index=idx), h)[ok_te] > 0, |
| **{f"q{l}": q[l] for l in LEVELS}}, index=idx[ok_te])) |
| start = end |
|
|
| out = {} |
| print(f"holiday variance factor per refit: {np.mean([r['k_holiday'] for r in k_log]):.2f} on average " |
| f"({min(r['k_holiday'] for r in k_log):.2f}-{max(r['k_holiday'] for r in k_log):.2f})\n") |
| print(f"{'':4}{'rule':20s} {'80% all':>8} {'80% hol':>8} {'n hol':>6} {'score':>7} {'score hol':>10} {'vs live':>8} {'p':>6}") |
| for h in HORIZONS: |
| fr = {r: pd.concat(rows[(h, r)]) for r in RULES} |
| common = fr[RULES[0]].index.intersection(fr[RULES[1]].index) |
| res = {} |
| for r in RULES: |
| d = fr[r].loc[common] |
| q = {l: d[f"q{l}"].to_numpy() for l in LEVELS} |
| s = scores(None, None, d["scale"].to_numpy(), d["move"].to_numpy(), |
| {l: -q[l] for l in LEVELS}, q) |
| m = d["hol"].to_numpy() |
| res[r] = {"cov80": float(s["in80"].mean()), "cov80_holiday": float(s["in80"][m].mean()), |
| "n_holiday": int(m.sum()), "score": s["score"], |
| "score_pct": float(s["score"].mean() * 100), |
| "score_holiday_pct": float(s["score"][m].mean() * 100)} |
| test = dm(res[RULES[1]]["score"], res[RULES[0]]["score"], h) |
| for r in RULES: |
| t = res[r] |
| print(f"{h:>2}h {r:20s} {t['cov80']:>8.1%} {t['cov80_holiday']:>8.1%} {t['n_holiday']:>6} " |
| f"{t['score_pct']:>7.3f} {t['score_holiday_pct']:>10.3f}" |
| + (f" {t['score_pct'] / res[RULES[0]]['score_pct'] - 1:>+8.2%} {test['p']:>6.3f}" |
| if r == RULES[1] else "")) |
| out[f"{h}h | {r}"] = {k: v for k, v in t.items() if k != "score"} |
| out[f"{h}h | dm_holidays_vs_live"] = test |
| (ROOT / "outputs" / "holiday_factor_backtest.json").write_text(json.dumps( |
| {"k_holiday_by_quarter": k_log, "results": out}, indent=2)) |
|
|
|
|
| if __name__ == "__main__": |
| main() |
|
|
