Blame for scripts/strategy_backtest.py

"""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        # noqa: E402

ROOT = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(ROOT))
from app.forecasting.price_range import releases                  # noqa: E402

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()                  # past week, hourly
    day = pd.DataFrame({
        "close": hourly.resample("1D").last(),
        # value at the END of the previous day = what is known at 00:00 today
        "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)
    # End labels, pushed apart in log space so close finishes do not overlap.
    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()