"""VOL04 — DVOL momentum de-risk overlay on long-flat trend. IDEA: Base: long-flat trend signal (TSMOM multi-horizon 1-3-6 months, like TP01). Overlay: scale exposure by DVOL momentum factor. - When DVOL is rising over last k days (fear rising), cut exposure (mul < 1). - When DVOL is falling (fear subsiding / calm), restore full exposure (mul = 1). The rationale: rising implied vol signals deteriorating regime — reduce size. Falling DVOL = benign regime — run full trend size. Implementation: dvol_chg[i] = (dvol[i] / sma(dvol, k)[i]) - 1 (deviation from k-day mean) mul[i] = clip(1 - alpha * dvol_chg[i], min_mul, 1.0) When dvol is above its k-day sma by X%, we reduce position by alpha*X%. When dvol is below its k-day mean, mul is clipped to 1.0 (no leverage boost). Grid: k in {10, 20}, alpha in {1.0, 2.0} -> 4 parameter sets x 2 TF = 8 backtests total. Only run on 1d and 12h (DVOL is daily, aligns naturally to >= daily-ish bars). NOTE: DVOL history starts 2021-03, so full period is 2021+ only for DVOL bars; bars before DVOL start will have NaN dvol -> fall back to pure trend (mul=1.0). """ import sys sys.path.insert(0, "/opt/docker/PythagorasGoal/scripts/research/alt") import altlib as al import numpy as np def tsmom_direction(df): """Causal TSMOM long-flat direction (1-3-6 month horizons, majority vote).""" c = df["close"].values.astype(float) bpd = al.bars_per_day(df) d = np.zeros(len(c)) for months in (1, 3, 6): horizon = int(months * 30 * bpd) s = np.full(len(c), 0.0) s[horizon:] = np.sign(c[horizon:] / c[:-horizon] - 1.0) d += s # long if majority (>0), flat if 0 or negative return np.clip(np.sign(d), 0, 1) def make_vol04(k: int, alpha: float): """Returns a target_fn(df) -> position array implementing DVOL de-risk overlay.""" def target_fn(df): c = df["close"].values.astype(float) n = len(c) # Step 1: base trend direction (long-flat) direction = tsmom_direction(df) # Step 2: get DVOL series, aligned causally to df bars dv = al.dvol(df, "BTC") # NOTE: BTC DVOL used for both; per-asset handled via asset param # Actually we need the per-asset DVOL. al.dvol accepts asset name, but # the function takes `df` not asset. We store the asset in a closure below. # For now this is a placeholder — see make_vol04_asset() below. # Step 3: DVOL k-day SMA (causal) dv_sma = al.sma(dv, k) # Step 4: compute dvol change relative to its mean # dvol_chg[i] = dvol[i] / sma[i] - 1: positive = dvol above mean = rising fear with np.errstate(divide='ignore', invalid='ignore'): dvol_chg = np.where((dv_sma > 0) & np.isfinite(dv_sma) & np.isfinite(dv), dv / dv_sma - 1.0, 0.0) # Step 5: exposure multiplier: cut when dvol rising, cap at 1.0 when falling # mul = clip(1 - alpha * dvol_chg, 0.1, 1.0) mul = np.clip(1.0 - alpha * dvol_chg, 0.1, 1.0) # Step 6: vol-targeted position = direction * mul * vol_scaling # First apply mul to direction, then vol-target scaled_dir = direction * mul # vol_target scales to 20% annualized vol with 2x leverage cap pos = al.vol_target(scaled_dir, df, target_vol=0.20, vol_win_days=30, leverage_cap=2.0) return pos return target_fn def make_vol04_asset(k: int, alpha: float, asset: str): """Asset-aware version: uses the correct DVOL for BTC or ETH.""" def target_fn(df): # Base trend direction direction = tsmom_direction(df) # DVOL aligned to df bars (per asset) dv = al.dvol(df, asset) # k-day SMA of DVOL (causal) dv_sma = al.sma(dv, k) # DVOL change relative to its mean (0 if no DVOL data) with np.errstate(divide='ignore', invalid='ignore'): dvol_chg = np.where( (dv_sma > 0) & np.isfinite(dv_sma) & np.isfinite(dv), dv / dv_sma - 1.0, 0.0 # no DVOL -> no de-risk (pure trend) ) # Multiplier: reduce when dvol > mean, clamp [0.1, 1.0] mul = np.clip(1.0 - alpha * dvol_chg, 0.1, 1.0) # Apply mul to direction scaled_dir = direction * mul # Vol-target the final position pos = al.vol_target(scaled_dir, df, target_vol=0.20, vol_win_days=30, leverage_cap=2.0) return pos return target_fn # -------------------------------------------------------------------------- # study_weights requires a single target_fn(df). But our overlay is asset- # specific (BTC DVOL for BTC, ETH DVOL for ETH). We run per-asset manually # using eval_weights, then assemble the report structure. # -------------------------------------------------------------------------- def run_cell(tf: str, k: int, alpha: float): """Evaluate VOL04(k, alpha) on both assets at given TF.""" per_asset = {} for asset in ("BTC", "ETH"): df = al.get(asset, tf) fn = make_vol04_asset(k, alpha, 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, k=k, alpha=alpha, 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(): # Small internal grid: k in {10, 20}, alpha in {1.0, 2.0}, TFs {1d, 12h} # Total: 2 k * 2 alpha * 2 TF = 8 backtests grid = [ (k, alpha) for k in (10, 20) for alpha in (1.0, 2.0) ] tfs = ("1d", "12h") all_cells = [] for tf in tfs: for k, alpha in grid: print(f" Running tf={tf} k={k} alpha={alpha} ...") cell = run_cell(tf, k, alpha) 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_k = best_cell["k"] best_alpha = best_cell["alpha"] print(f"\nBest config: tf={best_tf}, k={best_k}, alpha={best_alpha}") # Assemble report using best config cells for each TF (one per TF) # For the formal report, pick the best-k/alpha cell for each TF report_cells = [] for tf in tfs: tf_cells = [c for c in all_cells if c["tf"] == tf] best_tf_cell = max(tf_cells, key=lambda c: c["min_asset_holdout_sharpe"]) # Rename for al.fmt compatibility report_cells.append(dict( tf=tf, per_asset=best_tf_cell["per_asset"], min_asset_full_sharpe=best_tf_cell["min_asset_full_sharpe"], min_asset_holdout_sharpe=best_tf_cell["min_asset_holdout_sharpe"], full_sharpe=best_tf_cell["full_sharpe"], fee_survives=best_tf_cell["fee_survives"], )) # Build 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_k=best_k, best_alpha=best_alpha, ) rep = dict( name="VOL04-DVOL-DERISK", kind="weights", cells=report_cells, verdict=verdict, note=f"Best config: tf={best_tf}, k={best_k}, alpha={best_alpha}. " "DVOL history starts 2021-03; pre-DVOL bars fall back to pure trend (mul=1). " "Long-flat TSMOM base (1-3-6mo horizons) + DVOL SMA de-risk overlay." ) print("\n" + al.fmt(rep)) print("JSON:", al.as_json(rep)) if __name__ == "__main__": main()