| """Can the volatility forecast help a BTC holder, without predicting direction? |
| |
| Five ways to hold BTC, Jan 2023 - Sep 2026, decided once a day at 00:00 UTC |
| from information available then, with a 0.1% fee on every unit traded: |
| |
| Buy and hold 100% BTC throughout |
| Weekly DCA cash invested in equal weekly slices (Mondays) |
| Volatility-managed exposure = min(1, target / forecast volatility); the |
| forecast is the past week's hourly volatility (model |
| B's core) and the target its trailing one-year median, |
| so a calm market is held fully and a wild one partly |
| (Moreira & Muir 2017, J. Finance) |
| + Fed/CPI rule as above, halved on a day with a Fed decision or CPI |
| release, whose dates are published ahead |
| 20-week trend 100% BTC when yesterday's close is above its 140-day |
| average, otherwise cash |
| |
| The live VAE+Transformer regime model is NOT used: it was trained on data to |
| 2026, so trading 2023 with it would use the future. The causal inputs above |
| are all known at the moment of each decision. |
| |
| Metrics: total and annualised return, annualised volatility, Sharpe ratio |
| (daily, risk-free 0, x sqrt(365)), maximum drawdown, average exposure. |
| A block bootstrap (30-day blocks) gives a 95% interval for each strategy's |
| Sharpe minus buy-and-hold's. |
| |
| Usage: |
| python scripts/strategy_backtest.py |
| """ |
| from __future__ import annotations |
|
|
| import json |
| import sys |
| from pathlib import Path |
|
|
| import matplotlib |
| matplotlib.use("Agg") |
| import matplotlib.pyplot as plt |
| import numpy as np |
| import pandas as pd |
|
|
| sys.path.insert(0, str(Path(__file__).resolve().parent)) |
| from news_topic_event_study import GRID, INK, INK_2, SURFACE |
|
|
| ROOT = Path(__file__).resolve().parents[1] |
| sys.path.insert(0, str(ROOT)) |
| from app.forecasting.price_range import releases |
|
|
| START, FEE = "2023-01-01", 0.001 |
| COLORS = {"Buy and hold": "#8b8a85", "Weekly DCA": "#c9a227", "Volatility-managed": "#2a78d6", |
| "+ Fed/CPI rule": "#0b3d91", "20-week trend": "#eb6834"} |
|
|
|
|
| def inputs() -> pd.DataFrame: |
| btc = pd.read_csv(ROOT / "outputs" / "volatility" / "btc_1h.csv", usecols=["timestamp", "close"]) |
| btc["timestamp"] = pd.to_datetime(btc["timestamp"], utc=True) |
| hourly = btc.drop_duplicates("timestamp").set_index("timestamp")["close"].sort_index().asfreq("1h") |
| r = np.log(hourly).diff() |
| sigma = r.rolling(168, min_periods=150).std() |
| day = pd.DataFrame({ |
| "close": hourly.resample("1D").last(), |
| |
| "sigma": sigma.resample("1D").last().shift(1), |
| }) |
| day["target"] = day["sigma"].rolling(365, min_periods=180).median() |
| day["sma140"] = day["close"].rolling(140).mean().shift(1) |
| day["prev_close"] = day["close"].shift(1) |
| rel_days = {t.floor("D") for t, _ in releases()} |
| day["release_day"] = day.index.isin(list(rel_days)) |
| day["ret"] = day["close"].pct_change() |
| return day[day.index >= START].dropna(subset=["ret", "sigma", "target", "sma140"]) |
|
|
|
|
| def run(weights: pd.Series, ret: pd.Series) -> pd.Series: |
| """Daily portfolio return: weight held over the day, fee on the change.""" |
| turnover = weights.diff().abs().fillna(weights.iloc[0]) |
| return weights * ret - FEE * turnover |
|
|
|
|
| def dca(day: pd.DataFrame) -> pd.Series: |
| """Account value with cash fed into BTC in equal Monday slices.""" |
| mondays = day.index[day.index.dayofweek == 0] |
| slice_ = 1.0 / len(mondays) |
| cash, coins, values = 1.0, 0.0, [] |
| for t, row in day.iterrows(): |
| if t in mondays: |
| coins += slice_ * (1 - FEE) / row["close"] |
| cash -= slice_ |
| values.append(cash + coins * row["close"]) |
| return pd.Series(values, index=day.index) |
|
|
|
|
| def metrics(value: pd.Series, exposure: float) -> dict: |
| daily = value.pct_change().dropna() |
| years = (value.index[-1] - value.index[0]).days / 365 |
| peak = value.cummax() |
| return { |
| "total_return_pct": float((value.iloc[-1] / value.iloc[0] - 1) * 100), |
| "annual_return_pct": float(((value.iloc[-1] / value.iloc[0]) ** (1 / years) - 1) * 100), |
| "annual_volatility_pct": float(daily.std() * np.sqrt(365) * 100), |
| "sharpe": float(daily.mean() / daily.std() * np.sqrt(365)), |
| "max_drawdown_pct": float(((value / peak) - 1).min() * 100), |
| "average_exposure_pct": float(exposure * 100), |
| } |
|
|
|
|
| def sharpe_gap_ci(a: pd.Series, b: pd.Series, rng, block=30, n=2000): |
| """95% interval for Sharpe(a) - Sharpe(b) by moving-block bootstrap.""" |
| x = pd.concat([a, b], axis=1).dropna().to_numpy() |
| starts = np.arange(len(x) - block) |
| out = [] |
| for _ in range(n): |
| idx = np.concatenate([np.arange(s, s + block) for s in rng.choice(starts, len(x) // block + 1)])[:len(x)] |
| s = x[idx] |
| sh = s.mean(0) / s.std(0) * np.sqrt(365) |
| out.append(sh[0] - sh[1]) |
| return [float(np.quantile(out, 0.025)), float(np.quantile(out, 0.975))] |
|
|
|
|
| def main() -> None: |
| sys.stdout.reconfigure(encoding="utf-8") |
| rng = np.random.default_rng(0) |
| day = inputs() |
| ret = day["ret"] |
| vm = (day["target"] / day["sigma"]).clip(upper=1.0) |
| weights = { |
| "Buy and hold": pd.Series(1.0, index=day.index), |
| "Volatility-managed": vm, |
| "+ Fed/CPI rule": vm.where(~day["release_day"], vm * 0.5), |
| "20-week trend": (day["prev_close"] > day["sma140"]).astype(float), |
| } |
| values = {k: (1 + run(w, ret)).cumprod() for k, w in weights.items()} |
| values["Weekly DCA"] = dca(day) |
| exposure = {k: float(w.mean()) for k, w in weights.items()} |
| dca_btc = (values["Weekly DCA"] - (1 - np.minimum(1, np.cumsum(day.index.dayofweek == 0) / |
| (day.index.dayofweek == 0).sum()))) / values["Weekly DCA"] |
| exposure["Weekly DCA"] = float(dca_btc.mean()) |
|
|
| order = ["Buy and hold", "Weekly DCA", "Volatility-managed", "+ Fed/CPI rule", "20-week trend"] |
| res = {k: metrics(values[k], exposure[k]) for k in order} |
| bh = values["Buy and hold"].pct_change() |
| for k in order[1:]: |
| res[k]["sharpe_minus_buy_hold_ci"] = sharpe_gap_ci(values[k].pct_change(), bh, rng) |
| print(f"{day.index[0].date()} -> {day.index[-1].date()}, {len(day)} days, fee {FEE:.1%} per trade\n") |
| print(f"{'strategy':20s} {'total':>8} {'per yr':>7} {'vol/yr':>7} {'Sharpe':>7} {'max DD':>8} {'in BTC':>7} Sharpe vs B&H (95%)") |
| for k in order: |
| r = res[k] |
| ci = r.get("sharpe_minus_buy_hold_ci") |
| print(f"{k:20s} {r['total_return_pct']:>+7.0f}% {r['annual_return_pct']:>+6.1f}% " |
| f"{r['annual_volatility_pct']:>6.1f}% {r['sharpe']:>7.2f} {r['max_drawdown_pct']:>+7.1f}% " |
| f"{r['average_exposure_pct']:>6.0f}% " + (f"[{ci[0]:+.2f}, {ci[1]:+.2f}]" if ci else "")) |
|
|
| (ROOT / "outputs" / "strategy_backtest.json").write_text(json.dumps({ |
| "period": [str(day.index[0].date()), str(day.index[-1].date())], "fee_per_trade": FEE, |
| "decision": "daily at 00:00 UTC from information known then", |
| "results": res}, indent=2)) |
| draw(values, res, order) |
|
|
|
|
| def draw(values, res, order) -> None: |
| plt.rcParams.update({"font.family": "DejaVu Sans", "text.color": INK, "axes.labelcolor": INK_2}) |
| fig, (a, b) = plt.subplots(2, 1, figsize=(13.5, 8.6), facecolor=SURFACE, sharex=True, |
| gridspec_kw={"height_ratios": [1.6, 1]}) |
| fig.subplots_adjust(left=0.07, right=0.8, top=0.86, bottom=0.07, hspace=0.12) |
| for ax in (a, b): |
| ax.set_facecolor(SURFACE) |
| for s in ("top", "right"): |
| ax.spines[s].set_visible(False) |
| for s in ("left", "bottom"): |
| ax.spines[s].set_color(GRID) |
| ax.tick_params(colors=INK_2, labelsize=8.5, length=0) |
| ax.grid(color=GRID, linewidth=0.8, axis="y") |
| ax.set_axisbelow(True) |
| for k in order: |
| v = values[k] |
| w = 2.6 if k == "+ Fed/CPI rule" else 1.8 |
| a.plot(v.index, v, color=COLORS[k], linewidth=w) |
| b.plot(v.index, (v / v.cummax() - 1) * 100, color=COLORS[k], linewidth=w * 0.75) |
| |
| ends = sorted(order, key=lambda k: values[k].iloc[-1], reverse=True) |
| placed = [] |
| for k in ends: |
| y = values[k].iloc[-1] |
| if placed and np.log(placed[-1]) - np.log(y) < 0.2: |
| y = placed[-1] / np.exp(0.2) |
| placed.append(y) |
| r = res[k] |
| a.annotate(f"{k}\n{r['total_return_pct']:+.0f}%, Sharpe {r['sharpe']:.2f}, max DD {r['max_drawdown_pct']:.0f}%", |
| xy=(values[k].index[-1], values[k].iloc[-1]), xytext=(values[k].index[-1] + pd.Timedelta(days=25), y), |
| va="center", fontsize=8.3, color=INK, annotation_clip=False, |
| arrowprops=dict(arrowstyle="-", color=COLORS[k], linewidth=1)) |
| a.set_yscale("log") |
| ticks = [1, 1.5, 2, 3, 4, 6] |
| a.set_yticks(ticks) |
| a.set_yticklabels([f"${t:g}" for t in ticks]) |
| a.minorticks_off() |
| a.set_ylabel("value of $1 (log scale)", fontsize=8.5) |
| a.set_title("Growth of $1", loc="left", fontsize=11, fontweight="bold") |
| b.set_ylabel("% below previous peak", fontsize=8.5) |
| b.set_title("Drawdown: how far each strategy was below its own best", loc="left", fontsize=11, |
| fontweight="bold") |
| fig.suptitle("Using the volatility forecast to size a BTC position, Jan 2023 – Sep 2026", |
| x=0.07, ha="left", fontsize=14, fontweight="bold", y=0.975) |
| fig.text(0.07, 0.915, "Decided daily at 00:00 UTC from information known then; 0.1% fee per trade. " |
| "No strategy predicts direction: they change how much BTC is held, or when it is bought.", |
| fontsize=9.2, color=INK_2, ha="left") |
| fig.savefig(ROOT / "outputs" / "figures" / "fig18_strategy_backtest.png", dpi=150, facecolor=SURFACE) |
| plt.close(fig) |
| print("\nwritten: outputs/figures/fig18_strategy_backtest.png") |
|
|
|
|
| if __name__ == "__main__": |
| main() |
|
|
