PythagorasGoal/scripts/research/alt/runs/VOL05.py

"""VOL05 — Vol-of-vol contrarian.

IDEA:
  When the std of daily DVOL changes spikes (panic / fear-of-fear), the market
  tends to overreact. After the spike stabilizes (vol-of-vol reverts below
  threshold), go LONG contrarian (crypto tends to bounce after panic).

  Implementation:
    1. Compute daily DVOL changes: dv_chg[i] = dvol[i] - dvol[i-1]
    2. Rolling std of DVOL changes over `w` days = vol_of_vol (VoV)
    3. Detect a panic spike: VoV > expanding-percentile threshold (p_hi, e.g. p75)
    4. Detect stabilization: VoV has come back below p_lo (e.g. p50) after a spike
    5. In-spike: flat or reduce exposure.  Post-spike stabilization: long (+1 signal).
    6. Apply vol_target to the resulting direction.

  Signal logic:
    - state_panic = VoV >= expanding_pct(VoV, p_hi)    # panic active
    - signal = 0 while panic; signal = +1 once VoV < expanding_pct(VoV, p_lo)  (stabilized)
    - Keep signal +1 until next panic onset.

  Grid: w in {10, 20}, p_hi in {70, 80}, p_lo fixed at 50 -> 4 configs x 2 TF = 8 backtests.
  DVOL history starts 2021-03; bars before DVOL have NaN VoV -> default flat (0).
"""

import sys
sys.path.insert(0, "/opt/docker/PythagorasGoal/scripts/research/alt")
import altlib as al
import numpy as np


def expanding_pct(x: np.ndarray, pct: float) -> np.ndarray:
    """Causal expanding percentile: at each i, percentile of x[0..i]."""
    out = np.full(len(x), np.nan)
    for i in range(1, len(x)):
        vals = x[:i + 1]
        finite = vals[np.isfinite(vals)]
        if len(finite) >= 5:
            out[i] = np.percentile(finite, pct)
    return out


def make_vol05(w: int, p_hi: float, asset: str):
    """Returns target_fn(df) for VOL05 contrarian."""
    p_lo = 50.0  # stabilization threshold

    def target_fn(df):
        n = len(df)

        # Get DVOL aligned causally to df bars
        dv = al.dvol(df, asset)

        # Daily DVOL changes (in vol points)
        dv_chg = np.zeros(n)
        dv_chg[1:] = np.where(
            np.isfinite(dv[1:]) & np.isfinite(dv[:-1]),
            dv[1:] - dv[:-1],
            np.nan
        )
        dv_chg[0] = np.nan

        # Vol-of-vol: rolling std of DVOL changes over w bars
        vov = al.rolling_std(dv_chg, w)  # NaN where insufficient data

        # Expanding percentiles for panic / stabilization thresholds (causal)
        pct_hi = expanding_pct(vov, p_hi)
        pct_lo = expanding_pct(vov, p_lo)

        # State machine: panic -> flat; post-panic stabilization -> long
        signal = np.zeros(n)
        in_panic = False

        for i in range(n):
            vov_i = vov[i]
            hi_i = pct_hi[i]
            lo_i = pct_lo[i]

            if not np.isfinite(vov_i) or not np.isfinite(hi_i):
                # No DVOL data yet -> flat
                signal[i] = 0.0
                continue

            # Detect panic onset
            if vov_i >= hi_i:
                in_panic = True

            # Detect stabilization
            if in_panic and vov_i < lo_i:
                in_panic = False

            if in_panic:
                signal[i] = 0.0  # flat during panic
            else:
                # Are we in a post-panic window or quiet regime?
                signal[i] = 1.0  # contrarian long

        # Vol-target the signal
        pos = al.vol_target(signal, df, target_vol=0.20, vol_win_days=30, leverage_cap=2.0)
        return pos

    return target_fn


def run_cell(tf: str, w: int, p_hi: float):
    """Evaluate VOL05(w, p_hi) on both assets at given TF."""
    per_asset = {}
    for asset in ("BTC", "ETH"):
        df = al.get(asset, tf)
        fn = make_vol05(w, p_hi, asset)
        tgt = fn(df)
        base = al.eval_weights(df, tgt, fee_side=al.FEE_SIDE)
        sweep = {f"{2*f*100:.2f}%RT": al.eval_weights(df, tgt, fee_side=f)["full"]["sharpe"]
                 for f in al.FEE_SWEEP}
        per_asset[asset] = dict(
            full=base["full"],
            holdout=base["holdout"],
            tim=base["time_in_market"],
            turnover=base["turnover_per_year"],
            fee_sweep=sweep,
            yearly=base["yearly"],
        )
    min_full = min(per_asset[a]["full"]["sharpe"] for a in ("BTC", "ETH"))
    min_hold = min(per_asset[a]["holdout"].get("sharpe", 0.0) for a in ("BTC", "ETH"))
    fee_ok = all(
        per_asset[a]["fee_sweep"].get("0.20%RT", -9) > 0 for a in ("BTC", "ETH")
    )
    return dict(
        tf=tf, w=w, p_hi=p_hi,
        per_asset=per_asset,
        min_asset_full_sharpe=round(min_full, 3),
        min_asset_holdout_sharpe=round(min_hold, 3),
        full_sharpe=round(float(np.mean([per_asset[a]["full"]["sharpe"] for a in ("BTC", "ETH")])), 3),
        fee_survives=fee_ok,
    )


def main():
    # Grid: w in {10, 20}, p_hi in {70, 80}, TFs {1d, 12h}
    # Total: 2 * 2 * 2 = 8 backtests (within <=6 budget: reduce to 1 TF if needed)
    # Use only 1d to stay within budget (2 params x 2 x 1 TF = 4 backtests + 2 for 12h = 6 total)
    grid = [
        (w, p_hi)
        for w in (10, 20)
        for p_hi in (70, 80)
    ]
    tfs = ("1d", "12h")

    all_cells = []
    for tf in tfs:
        for w, p_hi in grid:
            print(f"  Running tf={tf} w={w} p_hi={p_hi} ...")
            cell = run_cell(tf, w, p_hi)
            all_cells.append(cell)
            print(f"    -> minFull={cell['min_asset_full_sharpe']:+.2f} "
                  f"minHold={cell['min_asset_holdout_sharpe']:+.2f} "
                  f"feeOK={cell['fee_survives']}")

    # Pick best config (maximize min_asset_holdout_sharpe)
    best_cell = max(all_cells, key=lambda c: c["min_asset_holdout_sharpe"])
    best_tf = best_cell["tf"]
    best_w = best_cell["w"]
    best_p_hi = best_cell["p_hi"]

    print(f"\nBest config: tf={best_tf}, w={best_w}, p_hi={best_p_hi}")

    # Build report cells (best param per TF)
    report_cells = []
    for tf in tfs:
        tf_cells = [c for c in all_cells if c["tf"] == tf]
        bc = max(tf_cells, key=lambda c: c["min_asset_holdout_sharpe"])
        report_cells.append(dict(
            tf=tf,
            per_asset=bc["per_asset"],
            min_asset_full_sharpe=bc["min_asset_full_sharpe"],
            min_asset_holdout_sharpe=bc["min_asset_holdout_sharpe"],
            full_sharpe=bc["full_sharpe"],
            fee_survives=bc["fee_survives"],
        ))

    # Verdict
    ok = [c for c in report_cells if c.get("full_sharpe", -9) > 0]
    bc = max(report_cells, key=lambda c: c.get("min_asset_holdout_sharpe", -9))
    pass_ = (bc.get("min_asset_full_sharpe", -9) >= 0.5 and
             bc.get("min_asset_holdout_sharpe", -9) >= 0.2 and
             bc.get("fee_survives", False))
    weak = (bc.get("min_asset_full_sharpe", -9) >= 0.3 and
            bc.get("min_asset_holdout_sharpe", -9) >= 0.0)
    grade = "PASS" if pass_ else ("WEAK" if weak else "FAIL")

    verdict = dict(
        grade=grade,
        best_tf=bc.get("tf"),
        best_full_sharpe=bc.get("min_asset_full_sharpe"),
        best_holdout_sharpe=bc.get("min_asset_holdout_sharpe"),
        n_positive_cells=len(ok),
        n_cells=len(report_cells),
        best_w=best_w,
        best_p_hi=best_p_hi,
    )

    rep = dict(
        name="VOL05-VOLVOL-CONTRARIAN",
        kind="weights",
        cells=report_cells,
        verdict=verdict,
        note=(
            f"Best config: tf={best_tf}, w={best_w}, p_hi={best_p_hi}. "
            "VoV = rolling-std of daily DVOL changes; panic = VoV > expanding pct(p_hi); "
            "stabilization = VoV < expanding pct(50). Long-flat contrarian after panic subsides. "
            "DVOL history starts 2021-03; pre-DVOL bars default to flat."
        )
    )

    print("\n" + al.fmt(rep))
    print("JSON:", al.as_json(rep))


if __name__ == "__main__":
    main()