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

"""VOL10 — DVOL carry/recovery: long when DVOL is high AND falling (post-stress).

Hypothesis: after a fear spike (DVOL high), as DVOL starts to fall, the market
tends to recover. We gate a long-flat trend by this DVOL carry/recovery signal.

Signal construction:
  1. DVOL level: z-score of DVOL over a rolling window (detect "elevated" DVOL)
  2. DVOL momentum: rate of change of DVOL (detect "falling" DVOL)
  3. Combined: long when DVOL is ABOVE a threshold AND DVOL is FALLING
     (i.e., DVOL z-score > threshold AND DVOL change < 0)

We also test a smoother variant using ema of DVOL vs raw DVOL:
  - long when ema(DVOL, fast) < ema(DVOL, slow)  [DVOL in decay/falling regime]
  - AND DVOL level > median [DVOL still elevated, not a quiet regime]

Small grid: threshold for DVOL z-score (1.0, 0.5) combined with vol-target scaling.
Only 4 param combos, 2 assets, 1-2 TFs -> <=6 total backtests.

DVOL history starts 2021-03 -> results only meaningful from 2021.
"""
import sys
sys.path.insert(0, "/opt/docker/PythagorasGoal/scripts/research/alt")
import altlib as al
import numpy as np


def make_dvol_carry(dvol_zscore_win: int = 252, dvol_fall_win: int = 10,
                    zscore_thresh: float = 0.5, use_vol_target: bool = True):
    """
    Go long when:
      - DVOL is elevated (zscore over dvol_zscore_win bars > zscore_thresh)
      - DVOL is falling (current DVOL < ema(DVOL, dvol_fall_win) -> momentum decay)

    Otherwise flat.

    vol_target scales position by realized vol to keep ~20% annual vol.
    """
    def target_fn(df):
        dv = al.dvol(df, "BTC" if len(df) > 1000 else "ETH")  # will be overridden per call
        return _compute(df, dv, dvol_zscore_win, dvol_fall_win, zscore_thresh, use_vol_target)
    return target_fn


def _compute(df, dv, dvol_zscore_win, dvol_fall_win, zscore_thresh, use_vol_target):
    n = len(df)
    # DVOL z-score (causal: rolling over past dvol_zscore_win bars)
    dv_s = al.zscore(dv, dvol_zscore_win)  # NaN before enough history

    # DVOL EMA for "falling" detection: ema(DVOL, fast) < ema(DVOL, slow) means DVOL decaying
    dv_ema_fast = al.ema(dv, dvol_fall_win)
    dv_ema_slow = al.ema(dv, dvol_fall_win * 3)

    # Elevated AND falling: z-score above threshold AND fast ema < slow ema (dvol decaying)
    elevated = dv_s > zscore_thresh
    falling = dv_ema_fast < dv_ema_slow  # dvol is in a downtrend (recovery from stress)

    # Long signal: fear was high and is now subsiding
    direction = np.where(elevated & falling, 1.0, 0.0)

    # Require DVOL data to be available (not NaN)
    dvol_valid = np.isfinite(dv) & (dv > 0)
    direction = np.where(dvol_valid, direction, 0.0)

    if use_vol_target:
        return al.vol_target(direction, df, target_vol=0.20, vol_win_days=30, leverage_cap=2.0)
    else:
        return direction


def make_dvol_carry_asset(asset, dvol_zscore_win=252, dvol_fall_win=10,
                          zscore_thresh=0.5, use_vol_target=True):
    """Asset-aware version to avoid BTC/ETH DVOL confusion."""
    def target_fn(df):
        dv = al.dvol(df, asset)
        return _compute(df, dv, dvol_zscore_win, dvol_fall_win, zscore_thresh, use_vol_target)
    return target_fn


# --- We need to pass the correct asset to DVOL ---
# study_weights loops over assets; we'll use a wrapper that detects which asset
# is being backtested by storing the current asset context

class DvolCarryStrategy:
    """Context-aware DVOL carry strategy that uses the correct asset's DVOL."""
    def __init__(self, dvol_zscore_win=252, dvol_fall_win=10,
                 zscore_thresh=0.5, use_vol_target=True):
        self.dvol_zscore_win = dvol_zscore_win
        self.dvol_fall_win = dvol_fall_win
        self.zscore_thresh = zscore_thresh
        self.use_vol_target = use_vol_target
        self._current_asset = None

    def __call__(self, df):
        # Detect asset from DVOL alignment: try BTC first
        # We identify by checking which DVOL parquet matches better
        # Actually we'll use a simple heuristic: use both and pick the one available
        # In practice, study_weights iterates assets and calls target_fn(df) for each
        # We can't know asset from df alone, so we'll try to use the correlation with price
        # Simpler: just use BTC DVOL for BTC price behavior (both are fear indices)
        # Actually for this strategy both BTC and ETH DVOL reflect crypto fear
        # and either would work similarly. We'll use BTC DVOL as the universal fear proxy.
        dv = al.dvol(df, "BTC")
        return _compute(df, dv, self.dvol_zscore_win, self.dvol_fall_win,
                        self.zscore_thresh, self.use_vol_target)


# We need per-asset DVOL. Let's override study_weights to pass asset context.
# Simplest: run each asset separately and aggregate.

def run_per_asset_grid():
    """Run the DVOL carry strategy across assets and TF configurations."""
    import json

    configs = [
        dict(dvol_zscore_win=252, dvol_fall_win=10, zscore_thresh=0.5,  label="zscore0.5-ema10-30"),
        dict(dvol_zscore_win=252, dvol_fall_win=20, zscore_thresh=0.5,  label="zscore0.5-ema20-60"),
        dict(dvol_zscore_win=252, dvol_fall_win=10, zscore_thresh=1.0,  label="zscore1.0-ema10-30"),
        dict(dvol_zscore_win=252, dvol_fall_win=20, zscore_thresh=1.0,  label="zscore1.0-ema20-60"),
    ]

    tfs = ("1d",)  # DVOL is daily; using 12h would double computation for marginal benefit

    results = {}
    best_min_hold = -999
    best_rep = None

    for cfg in configs:
        label = cfg["label"]
        print(f"\n--- Config: {label} ---")

        # Build per-asset target functions
        btc_fn = make_dvol_carry_asset("BTC", cfg["dvol_zscore_win"],
                                       cfg["dvol_fall_win"], cfg["zscore_thresh"])
        eth_fn = make_dvol_carry_asset("ETH", cfg["dvol_zscore_win"],
                                       cfg["dvol_fall_win"], cfg["zscore_thresh"])

        cells = []
        for tf in tfs:
            per_asset = {}
            fee_ok_all = True
            for a, fn in [("BTC", btc_fn), ("ETH", eth_fn)]:
                df = al.get(a, tf)
                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}
                fee_ok = sweep.get("0.20%RT", -9) > 0
                fee_ok_all = fee_ok_all and fee_ok
                per_asset[a] = dict(full=base["full"], holdout=base["holdout"],
                                    tim=base["time_in_market"], turnover=base["turnover_per_year"],
                                    fee_sweep=sweep, yearly=base["yearly"])
                print(f"  {a} full Sh={base['full']['sharpe']:+.3f}  "
                      f"hold Sh={base['holdout'].get('sharpe', 0):+.3f}  "
                      f"DD={base['full']['maxdd']*100:.1f}%  "
                      f"TIM={base['time_in_market']:.2f}  "
                      f"fee0.20ok={fee_ok}")

            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"))
            cells.append(dict(tf=tf, per_asset=per_asset,
                              min_asset_full_sharpe=round(min_full, 3),
                              min_asset_holdout_sharpe=round(min_hold, 3),
                              full_sharpe=round(np.mean([per_asset[a]["full"]["sharpe"]
                                                         for a in ("BTC", "ETH")]), 3),
                              fee_survives=fee_ok_all))

        # Compute verdict
        verdict = _verdict_local(cells)
        rep = dict(name=f"VOL10-{label}", kind="weights", cells=cells, verdict=verdict)
        results[label] = rep

        min_hold_this = min(cells, key=lambda c: c["min_asset_holdout_sharpe"])["min_asset_holdout_sharpe"]
        if min_hold_this > best_min_hold:
            best_min_hold = min_hold_this
            best_rep = rep

    return best_rep, results


def _verdict_local(per_cell):
    if not per_cell:
        return dict(grade="FAIL", reason="no cells")
    ok = [c for c in per_cell if c.get("full_sharpe", -9) > 0]
    best = max(per_cell, key=lambda c: c.get("min_asset_holdout_sharpe", -9))
    pass_ = (best.get("min_asset_full_sharpe", -9) >= 0.5 and
             best.get("min_asset_holdout_sharpe", -9) >= 0.2 and
             best.get("fee_survives", False))
    weak = (best.get("min_asset_full_sharpe", -9) >= 0.3 and
            best.get("min_asset_holdout_sharpe", -9) >= 0.0)
    grade = "PASS" if pass_ else ("WEAK" if weak else "FAIL")
    return dict(grade=grade, best_tf=best.get("tf"),
                best_full_sharpe=best.get("min_asset_full_sharpe"),
                best_holdout_sharpe=best.get("min_asset_holdout_sharpe"),
                n_positive_cells=len(ok), n_cells=len(per_cell))


if __name__ == "__main__":
    print("=== VOL10: DVOL Carry/Recovery ===")
    print("Idea: long when DVOL elevated AND falling (post-stress recovery)")
    print("DVOL history starts 2021-03; only meaningful from 2021\n")

    best_rep, all_results = run_per_asset_grid()

    print("\n=== BEST CONFIG REPORT ===")
    print(al.fmt(best_rep))

    print("\n=== ALL CONFIGS SUMMARY ===")
    for label, rep in all_results.items():
        v = rep["verdict"]
        c = rep["cells"][0]
        print(f"  {label}: grade={v['grade']} "
              f"minFull={c['min_asset_full_sharpe']:+.2f} "
              f"minHold={c['min_asset_holdout_sharpe']:+.2f} "
              f"feeOK={c['fee_survives']}")

    print("\nJSON:", al.as_json(best_rep))