"""SKH_R_RV — REGIME family: define BuzVola from REALIZED VOL (rolling std of HTF log-returns, annualized) instead of ATR, then Chande-normalize. Question: does a returns-based vol regime gate better OOS than the ATR-based one? Structural variant: we rebuild htf_features ourselves, swapping ONLY the BuzVola source from chande01(atr) to chande01(realized_vol). Everything else (BuzVolume, Donchian pattern, composer, entries, exits) is IDENTICAL to the engine so the comparison is clean. Causal-only: realized vol uses log-returns up to the HTF close; chande01 is causal rolling; donchian uses shift(1); the HTF->LTF merge is backward on HTF close. We verify causality with a truncated-prefix guard. V1 reference to beat: SkyhookParams(ptn_n=55, sl_atr=2.5, tp_atr=6.0, vola_lo=35, vola_hi=95, vol_lo=0.0) -> minFull +0.69, minHold +0.64 (BTC .64/ETH .64), fee-safe to 0.30%RT, DD ~40-49%. """ from __future__ import annotations import sys sys.path.insert(0, "/opt/docker/PythagorasGoal") sys.path.insert(0, "/opt/docker/PythagorasGoal/scripts/research/skyhook") sys.path.insert(0, "/opt/docker/PythagorasGoal/scripts/research/alt") import numpy as np import pandas as pd import skyhooklib as sk from src.strategies import skyhook as S from src.strategies.skyhook import SkyhookParams, HTF_MIN, LTF_MIN from src.backtest.harness import backtest_signals HOLDOUT = pd.Timestamp("2025-01-01", tz="UTC") FEE = sk.FEE_RT # --------------------------------------------------------------------------- # Realized-vol BuzVola: rolling std of HTF LOG-returns, annualized, Chande-normalized. # rv_win = lookback bars for the realized-vol estimate. # --------------------------------------------------------------------------- def realized_vol(htf: pd.DataFrame, rv_win: int) -> np.ndarray: c = htf["close"].values.astype(float) logret = np.zeros_like(c) logret[1:] = np.log(c[1:] / c[:-1]) # annualization factor: bars per year at 690 min bars_per_year = 365.25 * 24 * 60 / HTF_MIN rv = pd.Series(logret).rolling(rv_win, min_periods=rv_win).std().values * np.sqrt(bars_per_year) return rv def htf_features_rv(htf: pd.DataFrame, p: SkyhookParams, rv_win: int) -> pd.DataFrame: """Same as S.htf_features but BuzVola = chande01(realized_vol) instead of chande01(atr).""" rv = realized_vol(htf, rv_win) buz_vola = S.chande01(rv, p.n_vola) buz_volume = S.chande01(htf["volume"].values, p.n_volume) ptn_long, ptn_short = S.donchian_breakout(htf, p.ptn_n) regime_ok = ((buz_vola >= p.vola_lo) & (buz_vola <= p.vola_hi) & (buz_volume >= p.vol_lo) & (buz_volume <= p.vol_hi)) comp_long = regime_ok & ptn_long comp_short = regime_ok & ptn_short if p.long_only: comp_short = np.zeros_like(comp_short, dtype=bool) close_ts = htf["timestamp"].astype("int64").values + HTF_MIN * 60 * 1000 return pd.DataFrame({ "close_ts": close_ts, "buz_vola": buz_vola, "buz_volume": buz_volume, "comp_long": comp_long.astype(bool), "comp_short": comp_short.astype(bool), }) def rv_entries(ltf: pd.DataFrame, htf: pd.DataFrame, p: SkyhookParams, rv_win: int) -> list: """Identical to S.skyhook_entries but using the RV-based htf_features.""" feat = htf_features_rv(htf, p, rv_win) m = S.merge_htf_to_ltf(ltf, feat) c = m["close"].values.astype(float) a = S.atr(m, p.ltf_atr_win) comp_long = np.nan_to_num(m["comp_long"].values).astype(bool) comp_short = np.nan_to_num(m["comp_short"].values).astype(bool) days = pd.to_datetime(m["datetime"]).dt.floor("D").values entries: list = [None] * len(m) count_today: dict = {} for i in range(len(m)): if not np.isfinite(a[i]) or a[i] <= 0: continue day = days[i] if count_today.get(day, 0) >= p.max_per_day: continue if comp_long[i]: direction, mb = 1, p.uscitalong elif comp_short[i]: direction, mb = -1, p.uscitashort else: continue if p.exit_mode == "atr": sl_off, tp_off = p.sl_atr * a[i], p.tp_atr * a[i] else: sl_off, tp_off = p.sl_pct * c[i], p.tp_pct * c[i] if direction == 1: sl, tp = c[i] - sl_off, c[i] + tp_off else: sl, tp = c[i] + sl_off, c[i] - tp_off entries[i] = {"dir": direction, "tp": float(tp), "sl": float(sl), "max_bars": int(mb)} count_today[day] = count_today.get(day, 0) + 1 return entries # --------------------------------------------------------------------------- # Backtest one asset with RV entries -> FULL + HOLDOUT metrics # --------------------------------------------------------------------------- def _split(eq, idx, mask): e = eq[mask] if len(e) < 5: return dict(sharpe=0.0, ret=0.0, maxdd=0.0, n=int(len(e))) r = np.diff(e) / e[:-1]; r = r[np.isfinite(r)] ix = idx[mask] dt = pd.Series(ix).diff().dt.total_seconds().median() or 86400 bpy = 86400 * 365.25 / dt sh = float(np.mean(r) / np.std(r) * np.sqrt(bpy)) if len(r) and np.std(r) > 0 else 0.0 pk = np.maximum.accumulate(e); dd = float(np.max((pk - e) / pk)) return dict(sharpe=round(sh, 3), ret=round(float(e[-1] / e[0] - 1), 4), maxdd=round(dd, 4), n=int(len(e))) def run_rv(asset: str, p: SkyhookParams, rv_win: int, fee=FEE) -> dict: ltf, htf = sk.frames(asset) ent = rv_entries(ltf, htf, p, rv_win) m = backtest_signals(ltf, ent, fee_rt=fee, leverage=1.0, asset=asset, tf="230m") eq = m.equity idx = pd.DatetimeIndex(pd.to_datetime(m.eq_index, utc=True)) hmask = np.asarray(idx >= HOLDOUT) full = dict(sharpe=round(m.sharpe, 3), maxdd=round(m.max_dd, 4), ret=round(m.net_return, 4), n_trades=int(m.n_trades), win_rate=round(m.win_rate, 1)) hold = _split(eq, idx, hmask) n_ent = int(sum(e is not None for e in ent)) return dict(asset=asset, full=full, holdout=hold, n_entries=n_ent, _eq=eq, _idx=idx) def causality_rv(p: SkyhookParams, rv_win: int, asset="BTC", tail=200) -> dict: ltf, htf = sk.frames(asset) full = rv_entries(ltf, htf, p, rv_win) n = len(ltf); bad = 0; checked = 0 for frac in (0.80, 0.92): cut = int(n * frac) cut_ts = int(ltf["timestamp"].iloc[cut - 1]) htf_cut = htf[htf["timestamp"] <= cut_ts].reset_index(drop=True) sub = rv_entries(ltf.iloc[:cut].reset_index(drop=True), htf_cut, p, rv_win) for i in range(max(0, cut - tail), cut): checked += 1 a, b = full[i], sub[i] if (a is None) != (b is None): bad += 1 elif a is not None and (a["dir"] != b["dir"] or abs(a["sl"] - b["sl"]) > 1e-6 or abs(a["tp"] - b["tp"]) > 1e-6): bad += 1 return dict(ok=bool(bad == 0), mismatches=int(bad), checked=int(checked)) def minrep(label, p, rv_win, fee=FEE): rb = run_rv("BTC", p, rv_win, fee); re = run_rv("ETH", p, rv_win, fee) mnf = min(rb["full"]["sharpe"], re["full"]["sharpe"]) mnh = min(rb["holdout"]["sharpe"], re["holdout"]["sharpe"]) mnt = min(rb["full"]["n_trades"], re["full"]["n_trades"]) print(f" [{label} rv_win={rv_win}] minFull={mnf:+.2f} minHold={mnh:+.2f} minTr={mnt} " f"BTC(F{rb['full']['sharpe']:+.2f}/H{rb['holdout']['sharpe']:+.2f}/DD{rb['full']['maxdd']*100:.0f}%/n{rb['full']['n_trades']}) " f"ETH(F{re['full']['sharpe']:+.2f}/H{re['holdout']['sharpe']:+.2f}/DD{re['full']['maxdd']*100:.0f}%/n{re['full']['n_trades']})") return mnf, mnh, mnt, rb, re if __name__ == "__main__" and "--marginal" not in sys.argv: # V1 geometry as the base (best known config). Sweep rv_win and the vola band. base = dict(ptn_n=55, sl_atr=2.5, tp_atr=6.0, vola_lo=35.0, vola_hi=95.0, vol_lo=0.0) print("=== SKH_R_RV: realized-vol BuzVola gate (V1 geometry) ===") print("-- sweep rv_win (Chande lookback on annualized realized vol), V1 bands --") grid = [] for rv_win in (8, 13, 20, 34): p = SkyhookParams(**base) mnf, mnh, mnt, rb, re = minrep("rvband", p, rv_win) grid.append((rv_win, mnf, mnh, mnt)) # pick best holdout among feasible (minFull>=0.5, minTr>=20) feas = [g for g in grid if g[1] >= 0.5 and g[3] >= 20] pool = feas if feas else grid best = max(pool, key=lambda g: g[2]) best_rv = best[0] print(f"\n-- best rv_win by minHold (feasible): rv_win={best_rv} (minFull={best[1]:+.2f} minHold={best[2]:+.2f}) --") # Around best rv_win, try widening the vola band (RV distribution may differ from ATR's) print("\n-- band variations at best rv_win --") bandvars = [ ("V1band", dict(vola_lo=35.0, vola_hi=95.0)), ("wide", dict(vola_lo=25.0, vola_hi=98.0)), ("midhi", dict(vola_lo=45.0, vola_hi=95.0)), ("nogate", dict(vola_lo=0.0, vola_hi=100.0)), ] cand = [] for nm, bd in bandvars: pp = SkyhookParams(**{**base, **bd}) mnf, mnh, mnt, rb, re = minrep(nm, pp, best_rv) cand.append((nm, bd, mnf, mnh, mnt)) feas2 = [c for c in cand if c[2] >= 0.5 and c[4] >= 20] pool2 = feas2 if feas2 else cand win = max(pool2, key=lambda c: c[3]) win_p = SkyhookParams(**{**base, **win[1]}) print(f"\n=== WINNER: band={win[0]} rv_win={best_rv} minFull={win[2]:+.2f} minHold={win[3]:+.2f} ===") # Causality on winner (both assets) cb = causality_rv(win_p, best_rv, "BTC") ce = causality_rv(win_p, best_rv, "ETH") causal_ok = cb["ok"] and ce["ok"] print(f"causality: BTC={cb} ETH={ce} -> ok={causal_ok}") # Fee sweep on winner (min-asset full sharpe at each fee) print("\n-- fee sweep (min-asset FULL sharpe) --") fee_row = {} for f in (0.0, 0.001, 0.002, 0.003): rb = run_rv("BTC", win_p, best_rv, f); re = run_rv("ETH", win_p, best_rv, f) fee_row[f"{f*100:.2f}%RT"] = round(min(rb["full"]["sharpe"], re["full"]["sharpe"]), 3) print(" ", fee_row) fee_survives = fee_row.get("0.30%RT", -9) > 0 # Marginal vs TP01 on winner. Build daily 50/50 series the same way skyhooklib does. def daily_returns_rv(asset): r = run_rv(asset, win_p, best_rv, FEE) s = pd.Series(r["_eq"], index=r["_idx"]) return s.resample("1D").last().ffill().pct_change().dropna() import altlib as al sb = daily_returns_rv("BTC"); se = daily_returns_rv("ETH") J = pd.concat({"BTC": sb, "ETH": se}, axis=1, join="inner").fillna(0.0) cand_daily = 0.5 * J["BTC"] + 0.5 * J["ETH"] mg = al.marginal_vs_tp01(cand_daily) print("\n-- marginal vs TP01 --") print(f" corr_full={mg.get('corr_full')} verdict={mg.get('marginal_verdict')} " f"w25_uplift_hold={mg.get('blends',{}).get('w25',{}).get('uplift_hold')} " f"clean_year_uplift={mg.get('clean_year_uplift')} has_insample_edge={mg.get('has_insample_edge')} " f"is_hedge={mg.get('is_hedge')} robust_oos={mg.get('robust_oos')}") # Final per-asset detail at winner rb = run_rv("BTC", win_p, best_rv); re = run_rv("ETH", win_p, best_rv) print("\n=== FINAL (winner) ===") for a, r in (("BTC", rb), ("ETH", re)): f, h = r["full"], r["holdout"] print(f" {a}: FULL Sh={f['sharpe']:+.2f} ret={f['ret']*100:+.0f}% DD={f['maxdd']*100:.0f}% " f"n={f['n_trades']} wr={f['win_rate']:.0f}% | HOLD Sh={h['sharpe']:+.2f} ret={h['ret']*100:+.0f}% " f"DD={h['maxdd']*100:.0f}% n_entries={r['n_entries']}") import json summary = dict(label="R_RV", rv_win=best_rv, band=win[0], band_params=win[1], min_full=round(win[2], 3), min_hold=round(win[3], 3), min_trades=int(win[4]), btc_full=rb["full"]["sharpe"], eth_full=re["full"]["sharpe"], btc_hold=rb["holdout"]["sharpe"], eth_hold=re["holdout"]["sharpe"], avg_dd=round((rb["full"]["maxdd"] + re["full"]["maxdd"]) / 2, 4), causality_ok=causal_ok, fee_survives=fee_survives, corr_to_tp01=mg.get("corr_full"), blend_w25_uplift_hold=mg.get("blends", {}).get("w25", {}).get("uplift_hold"), marginal_verdict=mg.get("marginal_verdict")) print("\nJSON " + json.dumps(summary, default=str)) def _full_marginal(): """Re-run winner and dump the COMPLETE marginal dict + per-year, for the final report.""" import json, altlib as al base = dict(ptn_n=55, sl_atr=2.5, tp_atr=6.0, vola_lo=45.0, vola_hi=95.0, vol_lo=0.0) p = SkyhookParams(**base); rv_win = 34 def daily(asset): r = run_rv(asset, p, rv_win, FEE) s = pd.Series(r["_eq"], index=r["_idx"]) return s.resample("1D").last().ffill().pct_change().dropna() J = pd.concat({"BTC": daily("BTC"), "ETH": daily("ETH")}, axis=1, join="inner").fillna(0.0) cd = 0.5 * J["BTC"] + 0.5 * J["ETH"] mg = al.marginal_vs_tp01(cd) print("FULL MARGINAL:", json.dumps({k: mg.get(k) for k in ("corr_full","corr_hold","marginal_verdict","has_insample_edge","is_hedge", "robust_oos","clean_year_uplift","jackknife_min_uplift","multicut_persistent", "multicut_uplift","cand_full_sharpe","cand_hold_sharpe","alpha_ann","resid_sharpe_full", "null_pctl_insample")}, default=str)) print("BLENDS:", json.dumps(mg.get("blends"), default=str)) # per-year via backtest yearly on each asset for a in ("BTC","ETH"): ltf, htf = sk.frames(a); ent = rv_entries(ltf, htf, p, rv_win) m = backtest_signals(ltf, ent, fee_rt=FEE, leverage=1.0, asset=a, tf="230m") yr = " ".join(f"{int(y)}:{v*100:+.0f}%" for y, v in m.yearly.items()) print(f" {a} per-year: {yr}") if __name__ == "__main__" and "--marginal" in sys.argv: _full_marginal()