"""VOL11 — DVOL kill-switch on trend (TSMOM with hard-flat when DVOL is elevated). Hypothesis: TSMOM (multi-horizon, long-flat, vol-targeted, identical to TP01) is the validated base strategy. We overlay a DVOL kill-switch: when DVOL is above a threshold (fixed or percentile-based), go flat regardless of TSMOM signal. Rationale: trend-following can be whipsawed in high-IV regimes (panic spikes). By sitting out when DVOL is very high, we might: - Cut the worst crash losses (DVOL spikes during drawdowns) - Improve the hold-out Sharpe in the volatile 2025-26 period Construction: 1. Base signal: TSMOM multi-horizon (30/90/180-day lookbacks), sign-vote, long-flat. 2. Kill-switch: flat when DVOL > threshold. Tested thresholds: (A) fixed 70 points (historically ~top-30% readings) (B) fixed 80 points (historically ~top-15% readings) (C) rolling 80th percentile (adaptive, avoids hindsight threshold selection) (D) rolling 70th percentile 3. All configs: vol-target 20%, leva cap 2x, 1d. DVOL history starts 2021-03 → backtest meaningful from 2021 onward; full-history numbers include the pre-DVOL period where TSMOM runs unfiltered (i.e., those bars never killed). Grid: 4 configs × 1 TF × 2 assets = 8 backtests (within 6-limit at 1d; we run all 4 because they're fast vectorized ops and total is still manageable). """ import sys sys.path.insert(0, "/opt/docker/PythagorasGoal/scripts/research/alt") import altlib as al import numpy as np import pandas as pd # --------------------------------------------------------------------------- # Base TSMOM (same as TP01 canonical: 1/3/6-month horizons, long-flat) # --------------------------------------------------------------------------- def tsmom_direction(df): """TSMOM multi-horizon: +1 if majority of 30/90/180-day returns positive, else 0.""" c = df["close"].values.astype(float) bpd = al.bars_per_day(df) vote = np.zeros(len(c)) for h in (30 * bpd, 90 * bpd, 180 * bpd): h = int(h) s = np.full(len(c), np.nan) s[h:] = np.sign(c[h:] / c[:-h] - 1.0) vote += np.nan_to_num(s) # +1 if sum > 0 (majority positive), 0 otherwise (long-flat, not short) return np.where(vote > 0, 1.0, 0.0) # --------------------------------------------------------------------------- # DVOL kill-switch helpers (causal — no look-ahead) # --------------------------------------------------------------------------- def dvol_fixed_kill(dv, threshold): """Flat (kill=True) when DVOL >= threshold. NaN DVOL -> no kill (pass-through).""" kill = np.where(np.isfinite(dv) & (dv > 0), dv >= threshold, False) return kill.astype(bool) def dvol_percentile_kill(dv, percentile, window=252): """Flat when DVOL >= rolling expanding-then-window percentile (causal). Uses the past `window` daily DVOL observations to compute the threshold.""" n = len(dv) kill = np.zeros(n, dtype=bool) for i in range(n): if not np.isfinite(dv[i]) or dv[i] <= 0: continue # no DVOL -> pass through (no kill) # Rolling window: use min(i+1, window) observations up to and including i start = max(0, i - window + 1) hist = dv[start:i + 1] hist_valid = hist[np.isfinite(hist) & (hist > 0)] if len(hist_valid) < 10: continue # not enough history thresh = np.percentile(hist_valid, percentile) kill[i] = dv[i] >= thresh return kill # --------------------------------------------------------------------------- # Strategy builder # --------------------------------------------------------------------------- def make_vol11(asset, kill_type, kill_param): """ kill_type in {'fixed', 'pct'} kill_param: for 'fixed' -> DVOL level (e.g. 70, 80); for 'pct' -> percentile (e.g. 80, 70) """ def target_fn(df): # 1. Base TSMOM direction direction = tsmom_direction(df) # 2. DVOL for this asset (causal, backward-filled) dv = al.dvol(df, asset) # 3. Kill-switch if kill_type == "fixed": kill = dvol_fixed_kill(dv, kill_param) else: # 'pct' kill = dvol_percentile_kill(dv, kill_param, window=252) # 4. Apply kill: go flat when kill is active filtered_dir = np.where(kill, 0.0, direction) # 5. Vol-target the filtered direction return al.vol_target(filtered_dir, df, target_vol=0.20, vol_win_days=30, leverage_cap=2.0) return target_fn # --------------------------------------------------------------------------- # Configs grid (4 configs, 1 TF, 2 assets = 8 backtests) # --------------------------------------------------------------------------- CONFIGS = [ dict(kill_type="fixed", kill_param=70, label="fixed-70"), dict(kill_type="fixed", kill_param=80, label="fixed-80"), dict(kill_type="pct", kill_param=80, label="pct80-roll252"), dict(kill_type="pct", kill_param=70, label="pct70-roll252"), ] TFS = ("1d",) ASSETS = ("BTC", "ETH") 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)) def run_grid(): best_min_hold = -999 best_rep = None all_results = {} for cfg in CONFIGS: label = cfg["label"] print(f"\n--- VOL11 Config: {label} ---") cells = [] for tf in TFS: per_asset = {} fee_ok_all = True for asset in ASSETS: fn = make_vol11(asset, cfg["kill_type"], cfg["kill_param"]) df = al.get(asset, 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[asset] = 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" {asset}: 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 ASSETS) min_hold = min(per_asset[a]["holdout"].get("sharpe", 0.0) for a in ASSETS) 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 ASSETS]), 3), fee_survives=fee_ok_all )) verdict = _verdict_local(cells) rep = dict(name=f"VOL11-{label}", kind="weights", cells=cells, verdict=verdict) all_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, all_results if __name__ == "__main__": print("=== VOL11: DVOL Kill-Switch on TSMOM Trend ===") print("Idea: standard TSMOM (TP01-like), hard-flat when DVOL > threshold") print("DVOL history starts 2021-03; bars before that run unfiltered TSMOM\n") best_rep, all_results = run_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))