File size: 4,481 Bytes
49e3cc7 | 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 | """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 # noqa: E402
from backtest_price_range import HORIZONS, LEVELS # noqa: E402
from backtest_range_upgrades import dm, fwd_sum, log_ols, scores # noqa: E402
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()
|