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

"""VOL03 — DVOL-gated TSMOM
HYPOTHESIS: TP01-style multi-horizon TSMOM (vol-targeted, long-flat) but ONLY active when
DVOL is BELOW its expanding median. When DVOL is elevated (above median), go flat.
Rationale: in calm regimes (low DVOL), trend tends to persist; in high-vol regimes,
momentum can reverse or get choppy. Gating on DVOL below median may improve risk-adjusted returns.

NOTE: DVOL history starts 2021-03, so full backtest (2019+) will have NaN DVOL for early bars.
We handle this by defaulting to ACTIVE (no gate) when DVOL is NaN, so pre-2021 bars
are the same as vanilla TSMOM. This avoids burning early history on a look-ahead free gate.

Internal grid (4 configs, total 2 TFs x 2 configs = 4 backtests within study_weights per TF):
- VOL03-A: multi-horizon (1,3,6 month) TSMOM + DVOL < expanding median
- VOL03-B: multi-horizon (1,3,6 month) TSMOM + DVOL < expanding 40th pctile (stricter gate)
We test on 1d and 12h -> 2 TFs x 2 configs = 4 study_weights calls total (each covers BTC+ETH).
Pick best config by min_asset_holdout_sharpe.
"""
import sys
sys.path.insert(0, "/opt/docker/PythagorasGoal/scripts/research/alt")
import altlib as al
import numpy as np
import pandas as pd


def tsmom_dvol_gated(df: pd.DataFrame, dvol_pctile: float = 0.50) -> np.ndarray:
    """
    Multi-horizon TSMOM (1,3,6 month) long-flat, vol-targeted.
    Gate: position is ZERO when DVOL >= expanding percentile threshold.
    When DVOL is NaN (pre-2021), treat as gate=OFF (keep TSMOM signal).

    dvol_pctile: gate triggers (flat) when DVOL >= this expanding pctile of historical DVOL.
    """
    c = df["close"].values.astype(float)
    bpd = al.bars_per_day(df)
    asset = None
    # Detect asset from data (try BTC first, then ETH)
    # We'll use a closure over the caller's asset name - but since target_fn(df) is called
    # from study_weights which passes df, we need to infer asset from DVOL data availability.
    # Try BTC DVOL first, then ETH.
    dv = None
    for a in ("BTC", "ETH"):
        try:
            dv = al.dvol(df, a)
            asset = a
            break
        except Exception:
            continue

    # Multi-horizon TSMOM signal: sum of sign over 1m, 3m, 6m
    h1 = int(30 * bpd)
    h3 = int(90 * bpd)
    h6 = int(180 * bpd)
    direction = np.zeros(len(c))
    for h in (h1, h3, h6):
        sig = np.full(len(c), np.nan)
        sig[h:] = np.sign(c[h:] / c[:-h] - 1)
        direction += np.nan_to_num(sig, nan=0.0)
    # Long-flat: only go long (direction > 0), else flat
    direction = np.clip(np.sign(direction), 0.0, 1.0)

    # DVOL gate: compute expanding percentile of DVOL causally
    if dv is not None:
        dvol_series = pd.Series(dv)
        # Expanding percentile (causal)
        gate_active = np.zeros(len(c), dtype=bool)  # True = be active (below threshold)
        # Use rolling expanding quantile: pandas expanding().quantile() is causal
        dvol_thresh = dvol_series.expanding(min_periods=30).quantile(dvol_pctile)
        # Gate: active when dvol < threshold (below median = calm regime)
        # NaN dvol (pre-2021): treat as gate=OFF -> still active (no penalty)
        dvol_nan = dvol_series.isna() | dvol_thresh.isna()
        gate_active = dvol_nan | (dvol_series < dvol_thresh)
        direction = direction * gate_active.values.astype(float)

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


def make_target_fn(dvol_pctile: float):
    """Create a target function with given DVOL percentile gate."""
    def target_fn(df: pd.DataFrame) -> np.ndarray:
        return tsmom_dvol_gated(df, dvol_pctile=dvol_pctile)
    return target_fn


# --- Run 4 configs: 2 pctile thresholds x 2 TFs ---
# But study_weights handles 2 TFs internally, so we need 2 separate calls.
# Total: 2 configs x 1 call each (each covers both TFs) = 2 study_weights calls
# Each call tests 2 TFs x 2 assets = 4 backtests per call -> 8 total. OK.

configs = [
    ("VOL03-A-median", 0.50),   # flat when DVOL >= expanding median
    ("VOL03-B-p40",    0.40),   # flat when DVOL >= expanding 40th pctile (stricter gate)
]

reports = []
for name, pctile in configs:
    fn = make_target_fn(pctile)
    rep = al.study_weights(name, fn, tfs=("1d", "12h"))
    print(al.fmt(rep))
    print("JSON:", al.as_json(rep))
    print()
    reports.append((name, pctile, rep))

# Pick best config by min_asset_holdout_sharpe across all cells
best_name, best_pctile, best_rep = max(
    reports,
    key=lambda x: x[2]["verdict"].get("best_holdout_sharpe", -99)
)
print(f"\n=== BEST CONFIG: {best_name} (dvol_pctile={best_pctile}) ===")
print(al.fmt(best_rep))
print("JSON:", al.as_json(best_rep))