| """Does Deribit's DVOL improve the live HAR price range? |
| |
| The live range sizes its variance with a HAR blend of the last day, week and |
| month (backtest_range_upgrades.py). DVOL is the options market's 30-day |
| expectation of volatility: forward-looking information the price series does |
| not contain. Here it enters the same log-linear HAR regression as a fourth |
| input, implied hourly variance (DVOL/100)^2 / 8760, on the same walk-forward |
| quarters and scoring, against HAR without it on identical hours. |
| |
| Usage: |
| python scripts/backtest_range_dvol.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 |
|
|
| ROOT = Path(__file__).resolve().parents[1] |
| RULES = ("HAR (live)", "HAR + DVOL") |
|
|
|
|
| def main() -> None: |
| base = prepare() |
| idx = base["ret"].index |
| dv = pd.read_csv(ROOT / "outputs" / "volatility" / "btc_dvol_1h.csv") |
| dv["timestamp"] = pd.to_datetime(dv["timestamp"], utc=True) |
| iv_var = ((dv.set_index("timestamp")["close"] / 100) ** 2 / (24 * 365)).reindex(idx).to_numpy() |
| ret2, logp = base["ret"] ** 2, 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} |
|
|
| 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) |
| season = fit_factors(base, train)["var"]["+ Fed/CPI calendar"] |
| 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, target = fwd_sum(season, h), fwd_sum(e2, h) / h |
| for rule, XX in (("HAR (live)", X), ("HAR + DVOL", np.column_stack([X, iv_var]))): |
| sc = np.sqrt(log_ols(target, XX, train)(XX) * 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({"scale": sc[ok_te], "move": moves[h][ok_te], |
| **{f"q{l}": q[l] for l in LEVELS}}, index=idx[ok_te])) |
| start = end |
|
|
| out = {} |
| for h in HORIZONS: |
| frames = {r: pd.concat(rows[(h, r)]) for r in RULES} |
| common = frames[RULES[0]].index.intersection(frames[RULES[1]].index) |
| res = {} |
| for r, f in frames.items(): |
| d = f.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) |
| width = {str(int(l * 100)): float((np.exp(q[l] * d["scale"]) - np.exp(-q[l] * d["scale"])).mean() * 100) |
| for l in LEVELS} |
| res[r] = {"coverage": {f"{int(l * 100)}%": float(s[f"in{int(l * 100)}"].mean()) for l in LEVELS}, |
| "mean_width_pct": width, "interval_score_80_pct": float(s["score"].mean() * 100), |
| "n": int(len(d)), "_score": s["score"]} |
| res[RULES[1]]["dm_vs_HAR"] = dm(res[RULES[1]]["_score"], res[RULES[0]]["_score"], h) |
| for r in RULES: |
| t = res[r] |
| out[f"{h}h | {r}"] = {k: v for k, v in t.items() if k != "_score"} |
| c = t["coverage"] |
| print(f"{h:>2}h {r:11s} cover {c['50%']:.1%}/{c['80%']:.1%}/{c['95%']:.1%} " |
| f"50% width {t['mean_width_pct']['50']:.2f}% score {t['interval_score_80_pct']:.3f}" |
| + (f" vs HAR {t['interval_score_80_pct'] / res[RULES[0]]['interval_score_80_pct'] - 1:+.2%} " |
| f"p={t['dm_vs_HAR']['p']:.3f}" if r == RULES[1] else "")) |
| (ROOT / "outputs" / "price_range_dvol.json").write_text(json.dumps({ |
| "protocol": "walk-forward as price_range_upgrades.json; DVOL as a fourth log-HAR input; " |
| "identical test hours for both rules", "results": out}, indent=2)) |
|
|
|
|
| if __name__ == "__main__": |
| main() |
|
|
