"""SKH_R_PCTL — REGIME variant: replace Chande01 regime with CAUSAL expanding/rolling PERCENTILE-RANK of ATR and volume (0-1), gate on rank bands. Hypothesis: Chande01 measures the *direction/momentum* of the vol/volume cycle (rising vs falling), mapped to 0-100. A percentile-RANK instead measures *where the current level sits* within its own history (is ATR/volume HIGH or LOW relative to the past). This is a more natural "regime" definition: trade only when vol/volume is in a chosen part of its own distribution. We test expanding (full history) and rolling-window percentile ranks. Causality: rank[i] uses only x[0..i] (expanding) or x[i-w+1..i] (rolling), INCLUSIVE of the current bar — this is legitimate because at HTF close[i] the bar's ATR/volume is known. The HTF feature is then merged backward to LTF on HTF-close timestamp (<= LTF close). We verify the truncated-prefix guard ourselves. Pattern/composer/entry/exit reuse the V1 Skyhook building blocks unchanged. """ from __future__ import annotations import sys sys.path.insert(0, "/opt/docker/PythagorasGoal/scripts/research/skyhook") import numpy as np import pandas as pd import skyhooklib as sk from src.backtest.harness import backtest_signals from src.strategies import skyhook as S from src.strategies.skyhook import SkyhookParams, HTF_MIN, LTF_MIN HOLDOUT = pd.Timestamp("2025-01-01", tz="UTC") FEE = sk.FEE_RT # --------------------------------------------------------------------------- # Causal percentile-rank (0-1). Fraction of the window strictly < current value, # computed inclusive of the current bar (legit: bar is closed). min_periods enforced. # --------------------------------------------------------------------------- def pctl_rank(x: np.ndarray, win: int | None, min_periods: int = 30) -> np.ndarray: """Causal percentile rank in [0,1]. win=None -> EXPANDING; else rolling window. rank[i] = (#{x[j] < x[i]} + 0.5*#{x[j]==x[i]}) / count over the (expanding/rolling) window.""" x = np.asarray(x, float) s = pd.Series(x) if win is None: # expanding rank: pandas .expanding().rank(pct=True) gives rank/count INCLUSIVE of i, # which counts <= (so the current bar's own value is included). Use 'average' to break ties. r = s.expanding(min_periods=min_periods).rank(pct=True) else: r = s.rolling(win, min_periods=min(min_periods, win)).rank(pct=True) return r.values # NaN until min_periods reached # --------------------------------------------------------------------------- # HTF feature df with percentile-rank regime gate + Donchian pattern (V1 pattern reused). # --------------------------------------------------------------------------- def pctl_htf_features(htf: pd.DataFrame, p: SkyhookParams, vola_win: int | None, vol_win: int | None, vola_lo: float, vola_hi: float, vol_lo: float, vol_hi: float) -> pd.DataFrame: """Regime via CAUSAL percentile-rank (0-1) of ATR and volume; pattern via Donchian. Bands here are in [0,1] (percentile space), NOT 0-100 like Chande01.""" atr_htf = S.atr(htf, p.atr_win) vola_rank = pctl_rank(atr_htf, vola_win) vol_rank = pctl_rank(htf["volume"].values, vol_win) ptn_long, ptn_short = S.donchian_breakout(htf, p.ptn_n) # regime_ok requires a valid (non-NaN) rank in band; NaN (warmup) -> False vr_ok = np.where(np.isfinite(vola_rank), (vola_rank >= vola_lo) & (vola_rank <= vola_hi), False) vol_ok = np.where(np.isfinite(vol_rank), (vol_rank >= vol_lo) & (vol_rank <= vol_hi), False) regime_ok = vr_ok & vol_ok 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": vola_rank, "buz_volume": vol_rank, "comp_long": comp_long.astype(bool), "comp_short": comp_short.astype(bool), }) def pctl_entries(ltf: pd.DataFrame, htf: pd.DataFrame, p: SkyhookParams, vola_win, vol_win, vola_lo, vola_hi, vol_lo, vol_hi) -> list: """Same entry/exit machinery as S.skyhook_entries, but regime from pctl features.""" feat = pctl_htf_features(htf, p, vola_win, vol_win, vola_lo, vola_hi, vol_lo, vol_hi) 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 # --------------------------------------------------------------------------- # Eval helpers # --------------------------------------------------------------------------- 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 eval_cfg(cfg, p): """Run both assets; return dict per asset with full+holdout.""" out = {} for a in ("BTC", "ETH"): ltf, htf = sk.frames(a) ent = pctl_entries(ltf, htf, p, **cfg) m = backtest_signals(ltf, ent, fee_rt=FEE, leverage=1.0, asset=a, 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), ret=round(m.net_return, 4), maxdd=round(m.max_dd, 4), n_trades=int(m.n_trades), win_rate=round(m.win_rate, 1)) hold = _split(eq, idx, hmask) out[a] = dict(full=full, hold=hold, _eq=eq, _idx=idx) return out def summarize(tag, res): mf = min(res[a]["full"]["sharpe"] for a in res) mh = min(res[a]["hold"]["sharpe"] for a in res) mt = min(res[a]["full"]["n_trades"] for a in res) mdd = max(res[a]["full"]["maxdd"] for a in res) print(f" [{tag}] minFull={mf:+.2f} minHold={mh:+.2f} minTr={mt} maxDD={mdd*100:.0f}%" f" | BTC F{res['BTC']['full']['sharpe']:+.2f}/H{res['BTC']['hold']['sharpe']:+.2f}" f" ETH F{res['ETH']['full']['sharpe']:+.2f}/H{res['ETH']['hold']['sharpe']:+.2f}") return dict(minFull=mf, minHold=mh, minTr=mt, maxDD=mdd, res=res) # Build a SkyhookParams matching V1 non-regime knobs (pattern + exits) def v1_like_params(**kw): base = dict(ptn_n=55, sl_atr=2.5, tp_atr=6.0) # V1 pattern/exit base.update(kw) return SkyhookParams(**base) # --------------------------------------------------------------------------- # Causality self-check on the structural variant # --------------------------------------------------------------------------- def check_causality(cfg, p, asset="BTC", tail=150): ltf, htf = sk.frames(asset) full = pctl_entries(ltf, htf, p, **cfg) 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 = pctl_entries(ltf.iloc[:cut].reset_index(drop=True), htf_cut, p, **cfg) 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)) if __name__ == "__main__": print("=== SKH_R_PCTL: percentile-rank regime ===\n") # --- V1 reference for comparison (Chande01 regime) --- print("--- V1 reference (Chande01 regime, ptn_n=55 sl2.5 tp6 vola35-95 vol0) ---") v1 = SkyhookParams(ptn_n=55, sl_atr=2.5, tp_atr=6.0, vola_lo=35, vola_hi=95, vol_lo=0.0) v1res = {} for a in ("BTC", "ETH"): r = sk.run_asset(a, v1, FEE) v1res[a] = dict(full=dict(sharpe=r["full"]["sharpe"], n_trades=r["full"]["n_trades"], maxdd=r["full"]["maxdd"]), hold=dict(sharpe=r["holdout"]["sharpe"])) print(f" V1 minFull={min(v1res[a]['full']['sharpe'] for a in v1res):+.2f}" f" minHold={min(v1res[a]['hold']['sharpe'] for a in v1res):+.2f}" f" maxDD={max(v1res[a]['full']['maxdd'] for a in v1res)*100:.0f}%" f" | BTC F{v1res['BTC']['full']['sharpe']:+.2f}/H{v1res['BTC']['hold']['sharpe']:+.2f}" f" ETH F{v1res['ETH']['full']['sharpe']:+.2f}/H{v1res['ETH']['hold']['sharpe']:+.2f}\n") p = v1_like_params() # same pattern/exit as V1; only regime changes # --- Sweep: percentile-rank regime bands --- # Two regime intuitions to test: # (A) HIGH-vol/HIGH-volume regime (breakout-friendly): ranks in upper band. # (B) MID regime (avoid blow-off + dead): ranks in a middle band. # vol_lo=0 means "no lower bound on volume" (mirror V1's vol_lo=0). print("--- EXPANDING percentile-rank sweep ---") cfgs = { # vola band, vol band, both expanding (win=None) "exp_volaHi_vol0": dict(vola_win=None, vol_win=None, vola_lo=0.35, vola_hi=0.95, vol_lo=0.0, vol_hi=1.0), "exp_volaMid_vol0": dict(vola_win=None, vol_win=None, vola_lo=0.30, vola_hi=0.80, vol_lo=0.0, vol_hi=1.0), "exp_volaHi_volHi": dict(vola_win=None, vol_win=None, vola_lo=0.35, vola_hi=0.95, vol_lo=0.40, vol_hi=1.0), "exp_volaLo_vol0": dict(vola_win=None, vol_win=None, vola_lo=0.0, vola_hi=0.70, vol_lo=0.0, vol_hi=1.0), "exp_volaWide_vol0": dict(vola_win=None, vol_win=None, vola_lo=0.20, vola_hi=1.0, vol_lo=0.0, vol_hi=1.0), } results = {} for tag, cfg in cfgs.items(): results[tag] = (cfg, summarize(tag, eval_cfg(cfg, p))) print("\n--- ROLLING percentile-rank sweep (win=60 HTF bars ~ recent regime) ---") cfgs_roll = { "roll60_volaHi_vol0": dict(vola_win=60, vol_win=60, vola_lo=0.35, vola_hi=0.95, vol_lo=0.0, vol_hi=1.0), "roll60_volaMid_vol0": dict(vola_win=60, vol_win=60, vola_lo=0.30, vola_hi=0.80, vol_lo=0.0, vol_hi=1.0), "roll120_volaHi_vol0": dict(vola_win=120, vol_win=120, vola_lo=0.35, vola_hi=0.95, vol_lo=0.0, vol_hi=1.0), "roll60_volaHi_volHi": dict(vola_win=60, vol_win=60, vola_lo=0.35, vola_hi=0.95, vol_lo=0.40, vol_hi=1.0), } for tag, cfg in cfgs_roll.items(): results[tag] = (cfg, summarize(tag, eval_cfg(cfg, p))) # --- Pick winner by minHold (subject to minFull>=0.5, minTr>=20) --- elig = {t: v for t, (c, v) in results.items() if v["minFull"] >= 0.5 and v["minTr"] >= 20} print(f"\n--- eligible (minFull>=0.5, minTr>=20): {list(elig.keys())} ---") if elig: win_tag = max(elig, key=lambda t: elig[t]["minHold"]) else: # fall back to best minHold overall to report honestly win_tag = max(results, key=lambda t: results[t][1]["minHold"]) win_cfg, win_v = results[win_tag][0], results[win_tag][1] print(f"\n*** WINNER = {win_tag} minFull={win_v['minFull']:+.2f} minHold={win_v['minHold']:+.2f}" f" minTr={win_v['minTr']} maxDD={win_v['maxDD']*100:.0f}% ***") print(f" cfg = {win_cfg}") # --- Causality on winner --- caus = check_causality(win_cfg, p, "BTC") print(f"\ncausality(BTC) = {caus}") caus_eth = check_causality(win_cfg, p, "ETH") print(f"causality(ETH) = {caus_eth}") # --- Fee sweep on winner (both assets, FULL) --- print("\n--- fee sweep on winner (FULL sharpe) ---") fee_ok_all = True for a in ("BTC", "ETH"): ltf, htf = sk.frames(a) ent_cache = pctl_entries(ltf, htf, p, **win_cfg) row = [] for f in (0.0, 0.001, 0.002, 0.003): m = backtest_signals(ltf, ent_cache, fee_rt=f, leverage=1.0, asset=a, tf="230m") row.append((f, round(m.sharpe, 3))) sh030 = dict(row)[0.003] fee_ok_all = fee_ok_all and (sh030 > 0) print(f" {a}: " + " ".join(f"{f*100:.2f}%={s:+.2f}" for f, s in row)) print(f" fee_survives 0.30%RT (both): {fee_ok_all}") # --- Marginal vs TP01 on winner --- # Build a daily 50/50 series the same way skyhooklib does, but with our entries. def daily_series(a, p, cfg): ltf, htf = sk.frames(a) ent = pctl_entries(ltf, htf, p, **cfg) m = backtest_signals(ltf, ent, fee_rt=FEE, leverage=1.0, asset=a, tf="230m") s = pd.Series(m.equity, index=pd.DatetimeIndex(pd.to_datetime(m.eq_index, utc=True))) return s.resample("1D").last().ffill().pct_change().dropna() import altlib as al sb = daily_series("BTC", p, win_cfg); se = daily_series("ETH", p, win_cfg) J = pd.concat({"BTC": sb, "ETH": se}, axis=1, join="inner").fillna(0.0) cand_daily = 0.5 * J["BTC"] + 0.5 * J["ETH"] marg = al.marginal_vs_tp01(cand_daily) print("\n--- marginal vs TP01 (winner) ---") print(f" corr_full={marg.get('corr_full')} corr_hold={marg.get('corr_hold')}") print(f" blend w25 uplift_hold={marg.get('blends',{}).get('w25',{}).get('uplift_hold')}") print(f" marginal_verdict={marg.get('marginal_verdict')} robust_oos={marg.get('robust_oos')}") print(f" has_insample_edge={marg.get('has_insample_edge')} is_hedge={marg.get('is_hedge')}") print(f" clean_year_uplift={marg.get('clean_year_uplift')} jackknife_min_uplift={marg.get('jackknife_min_uplift')}") # --- Final verdict echo --- print("\n=== SUMMARY ===") print(f"V1: minFull={min(v1res[a]['full']['sharpe'] for a in v1res):+.2f}" f" minHold={min(v1res[a]['hold']['sharpe'] for a in v1res):+.2f}" f" maxDD={max(v1res[a]['full']['maxdd'] for a in v1res)*100:.0f}%") print(f"PCTL {win_tag}: minFull={win_v['minFull']:+.2f} minHold={win_v['minHold']:+.2f}" f" maxDD={win_v['maxDD']*100:.0f}% causalityOK={caus['ok'] and caus_eth['ok']} feeOK={fee_ok_all}")