"""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))