"""SKH2_EXPAND_DD — DD-reduction wave, vol-EXPANSION family. Family task: reuse the volatility-EXPANSION regime from SKH_R_EXPAND.py (ATR rising vs its own MA AND volume elevated vs its own MA), monkeypatch S.htf_features, run sk.study, and TUNE w_atr/k_atr/w_vol/k_vol + winner-style exits to: (1) cut standalone maxDD below 30% (max over BTCÐ) <-- the only unmet wave goal (2) keep min-asset HOLD-OUT Sharpe >= ~0.70 and earns_slot == True (3) stretch: lift blend w25 uplift_hold and minHold. Mechanism / DD theory: * the EXPANSION gate (vol rising + volume elevated) is itself a DD filter: it suppresses entries during quiet/contracting chop where Donchian breakouts whipsaw. Tightening k_atr / k_vol trades trade-count for cleaner regime -> fewer adverse entries. * but per-trade loss size is set by sl_atr; the V2 winner used sl_atr=2.5 (DD 34/31%). Lowering sl_atr is the direct DD lever. We sweep sl_atr in {1.6,1.8,2.0,2.2,2.5} and couple it with the winner exits (uscitalong=24/uscitashort=16) and tp_atr in {5,6,7}. * vola_lo/vola_hi/vol_lo bands are IRRELEVANT here: the expansion regime REPLACES the Chande band gate (htf_features is monkeypatched), so those SkyhookParams fields are dead. Only ptn_n / sl_atr / tp_atr / uscita* / max_per_day / long_only matter through the patched path. Everything causal: the expansion features use only x[0..i] (causal rolling MA, ATR ewm, donchian shift(1)); HTF merged BACKWARD onto LTF on HTF-close ts. We verify with sk.causality (works because we patch S.htf_features inside skyhooklib's namespace, so skyhook_entries uses our gate). """ 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.strategies import skyhook as S from src.strategies.skyhook import SkyhookParams # reuse the EXPANSION feature builder verbatim from SKH_R_EXPAND import expand_htf_features ORIG_FEAT = S.htf_features HOLDOUT = sk.HOLDOUT FEE = sk.FEE_RT def patched(cfg): def _feat(htf, p): return expand_htf_features(htf, p, **cfg) return _feat def study_expand(name, p, cfg, want_marginal=True): """Run sk.study + causality (+ marginal) with htf_features patched to the expansion regime.""" S.htf_features = patched(cfg) try: rep = sk.study(name, p) caus_b = sk.causality(p, "BTC") caus_e = sk.causality(p, "ETH") marg = sk.marginal(p) if want_marginal else None finally: S.htf_features = ORIG_FEAT return rep, (caus_b, caus_e), marg def vline(rep): v = rep["verdict"] pa = rep["per_asset"] mdd = max(pa[a]["full"]["maxdd"] for a in pa) return (v["grade"], v["min_asset_full_sharpe"], v["min_asset_holdout_sharpe"], v["min_trades"], mdd, v["fee_survives"]) # --------------------------------------------------------------------------- # WINNER baseline (Chande band, NOT expansion) for reference — verify the stated DD problem. # --------------------------------------------------------------------------- def winner_reference(): p = SkyhookParams(ptn_n=45, sl_atr=2.5, tp_atr=7.0, uscitalong=24, uscitashort=16, vola_lo=35.0, vola_hi=95.0, vol_lo=0.0) rep = sk.study("WINNER-V2", p) # uses ORIG_FEAT (Chande band) — not patched g, mf, mh, mt, mdd, fee = vline(rep) print(f"[WINNER-V2 ref] grade={g} minFull={mf:+.2f} minHold={mh:+.2f} minTr={mt} " f"maxDD={mdd*100:.0f}% feeOK={fee} " f"(BTC DD {rep['per_asset']['BTC']['full']['maxdd']*100:.0f}% / " f"ETH DD {rep['per_asset']['ETH']['full']['maxdd']*100:.0f}%)") return mh def earns_slot(rep, marg): g = rep["verdict"]["grade"] != "FAIL" return bool(g and marg.get("marginal_verdict") == "ADDS" and marg.get("robust_oos") and not marg.get("is_hedge")) if __name__ == "__main__": print("=== SKH2_EXPAND_DD: vol-EXPANSION regime tuned for standalone maxDD < 30% ===\n") win_minhold = winner_reference() print() # ------------------------------------------------------------------- # PASS 1: coarse grid. Regime gate strength (k_atr,k_vol,windows) x SL size. # Goal: find DD<30% cells that keep minHold high. Marginal computed only for finalists. # ------------------------------------------------------------------- # exits: asymmetric time-exits. PASS1 learned that LONGER long-holds (us30/18 vs winner's # 24/16) are what flip the marginal robust_oos gate POSITIVE (clean-2025-year uplift > 0) # while sl_atr=2.4 keeps DD<30. So we sweep exits + sl_atr here, ptn_n fixed near winner. base_kw = dict(ptn_n=45, uscitalong=30, uscitashort=18) # The EXPANSION gate REPLACES the Chande band (htf_features monkeypatched): vola_*/vol_* are # dead. DD is cut by (a) the gate itself (only trade rising-vol + elevated-volume regimes) and # (b) sl_atr. The a20/k1.1 gate + sl2.4 + us30/18 is the DD<30 + robust_oos sweet spot found. regimes = { # tag: expansion cfg "r_a20k1.1_v20k1.1": dict(w_atr=20, k_atr=1.10, w_vol=20, k_vol=1.10), "r_a25k1.1_v25k1.1": dict(w_atr=25, k_atr=1.10, w_vol=25, k_vol=1.10), "r_a30k1.1_v30k1.1": dict(w_atr=30, k_atr=1.10, w_vol=30, k_vol=1.10), "r_a18k1.1_v18k1.1": dict(w_atr=18, k_atr=1.10, w_vol=18, k_vol=1.10), } sl_grid = (2.2, 2.4) tp_fixed = 7.0 print("--- PASS 1 coarse: regime x sl_atr (tp=7.0) ---") pass1 = [] for rtag, rcfg in regimes.items(): for sl in sl_grid: p = SkyhookParams(sl_atr=sl, tp_atr=tp_fixed, **base_kw) S.htf_features = patched(rcfg) try: rep = sk.study(f"{rtag}_sl{sl}", p) finally: S.htf_features = ORIG_FEAT g, mf, mh, mt, mdd, fee = vline(rep) pass1.append((rtag, rcfg, sl, tp_fixed, g, mf, mh, mt, mdd, fee, p)) print(f" {rtag:22s} sl{sl} -> grade={g:4s} minFull={mf:+.2f} minHold={mh:+.2f}" f" minTr={mt:3d} maxDD={mdd*100:3.0f}% feeOK={fee}") # finalists = DD<30% AND minHold>=0.55 AND grade!=FAIL AND fee survives fin = [r for r in pass1 if r[8] < 0.30 and r[6] >= 0.55 and r[4] != "FAIL" and r[9]] print(f"\n--- PASS1 finalists (DD<30%, minHold>=0.6, !FAIL, feeOK): {len(fin)} ---") for r in fin: print(f" {r[0]} sl{r[2]} tp{r[3]} : minHold={r[6]:+.2f} DD={r[8]*100:.0f}%") # If none, relax to DD<30% AND minHold>=0.5 to still report best-effort. if not fin: fin = [r for r in pass1 if r[8] < 0.30 and r[6] >= 0.50 and r[4] != "FAIL" and r[9]] print(f" (relaxed minHold>=0.5): {len(fin)}") if not fin: # last resort: lowest DD among non-FAIL fee-surviving with minHold>0 cand = [r for r in pass1 if r[4] != "FAIL" and r[9] and r[6] > 0] fin = sorted(cand, key=lambda r: r[8])[:3] print(f" (last-resort lowest-DD): {len(fin)}") # ------------------------------------------------------------------- # PASS 2: finalists -> full marginal + tighten tp around best. Pick the BEATS-WINNER one, # else best earns_slot+lowest DD. # ------------------------------------------------------------------- # de-dup finalists by (rtag,sl) and cap to keep runtime sane seen = set(); fin2 = [] for r in sorted(fin, key=lambda r: (-r[6], r[8])): # prefer high minHold then low DD key = (r[0], r[2]) if key in seen: continue seen.add(key); fin2.append(r) fin2 = fin2[:7] print(f"\n--- PASS 2 marginal on {len(fin2)} finalists ---") results = [] for r in fin2: rtag, rcfg, sl, tp, g, mf, mh, mt, mdd, fee, p = r rep, (cb, ce), marg = study_expand(f"{rtag}_sl{sl}_tp{tp}", p, rcfg) g, mf, mh, mt, mdd, fee = vline(rep) caus_ok = bool(cb["ok"] and ce["ok"]) es = earns_slot(rep, marg) w25 = marg.get("blends", {}).get("w25", {}) w50 = marg.get("blends", {}).get("w50", {}) uph = w25.get("uplift_hold") beats = bool(es and mdd < 0.30 and (uph is not None and uph >= 0.55) and mh >= 0.65) results.append(dict(tag=f"{rtag}_sl{sl}_tp{tp}", rcfg=rcfg, p=p, rep=rep, marg=marg, caus_ok=caus_ok, earns=es, beats=beats, minFull=mf, minHold=mh, minTr=mt, maxDD=mdd, fee=fee, uph=uph, w25=w25, w50=w50)) print(f" {rtag}_sl{sl} -> grade={g} minFull={mf:+.2f} minHold={mh:+.2f} DD={mdd*100:.0f}%" f" verdict={marg.get('marginal_verdict')} corr={marg.get('corr_full')}" f" w25uplH={uph} earns={es} caus={caus_ok} BEATS={beats}") # ------------------------------------------------------------------- # PASS 3: around the best finalist, try tp in {5,6} to see if tighter tp helps DD/minHold. # ------------------------------------------------------------------- def score(d): # rank: beats first, then earns & DD<30, then minHold, then -DD return (d["beats"], d["earns"] and d["maxDD"] < 0.30, d["minHold"], -d["maxDD"]) if results: best = max(results, key=score) rtag = best["tag"].rsplit("_sl", 1)[0] rcfg = best["rcfg"] sl = best["p"].sl_atr # PASS3 sweeps the EXIT-BAR dimension: the robust_oos (2025-clean-year uplift) gate is # set by the long-hold length. We probe uscitalong around 30 to confirm the sweet spot # and hunt any DD<30 cell with higher blend uplift. print(f"\n--- PASS 3 exit-bar refine around best regime={rtag} sl{sl} ---") for usL, usS in ((28, 18), (32, 18), (30, 20)): kw = dict(ptn_n=45, uscitalong=usL, uscitashort=usS) p = SkyhookParams(sl_atr=sl, tp_atr=tp_fixed, **kw) rep, (cb, ce), marg = study_expand(f"{rtag}_sl{sl}_us{usL}/{usS}", p, rcfg) g, mf, mh, mt, mdd, fee = vline(rep) caus_ok = bool(cb["ok"] and ce["ok"]) es = earns_slot(rep, marg) w25 = marg.get("blends", {}).get("w25", {}); w50 = marg.get("blends", {}).get("w50", {}) uph = w25.get("uplift_hold") beats = bool(es and mdd < 0.30 and (uph is not None and uph >= 0.55) and mh >= 0.65) results.append(dict(tag=f"{rtag}_sl{sl}_us{usL}/{usS}", rcfg=rcfg, p=p, rep=rep, marg=marg, caus_ok=caus_ok, earns=es, beats=beats, minFull=mf, minHold=mh, minTr=mt, maxDD=mdd, fee=fee, uph=uph, w25=w25, w50=w50)) print(f" us{usL}/{usS} -> grade={g} minFull={mf:+.2f} minHold={mh:+.2f} DD={mdd*100:.0f}%" f" verdict={marg.get('marginal_verdict')} robust={marg.get('robust_oos')}" f" w25uplH={uph} earns={es} BEATS={beats}") # ------------------------------------------------------------------- # FINAL: pick best config and print full block. # ------------------------------------------------------------------- if not results: print("\n!!! no finalists at all — reporting nothing meaningful. !!!") sys.exit(0) best = max(results, key=score) m = best["marg"]; rep = best["rep"] print("\n" + "=" * 78) print("FINAL BEST (vol-EXPANSION family)") print("=" * 78) print(f" tag = {best['tag']}") print(f" regime cfg = {best['rcfg']}") print(f" params = ptn_n={best['p'].ptn_n} sl_atr={best['p'].sl_atr} tp_atr={best['p'].tp_atr}" f" uscitalong={best['p'].uscitalong} uscitashort={best['p'].uscitashort}" f" max_per_day={best['p'].max_per_day} long_only={best['p'].long_only}") print(f" minFull = {best['minFull']:+.3f}") print(f" minHold = {best['minHold']:+.3f}") print(f" max_dd = {best['maxDD']:.4f} ({best['maxDD']*100:.1f}%)") print(f" n_trades = {best['minTr']} (min over BTCÐ)") print(f" fee@0.30%RT survives = {best['fee']}") print(f" causality OK (BTCÐ) = {best['caus_ok']}") print(f" earns_slot = {best['earns']}") print(f" BEATS_WINNER= {best['beats']}") print(" -- per-asset --") for a in ("BTC", "ETH"): pa = rep["per_asset"][a] print(f" {a}: FULL Sh={pa['full']['sharpe']:+.2f} DD={pa['full']['maxdd']*100:.0f}%" f" n={pa['full']['n_trades']} | HOLD Sh={pa['holdout']['sharpe']:+.2f}" f" | fee_sweep {pa['fee_sweep']}") print(" -- marginal vs TP01 --") print(f" corr_full={m.get('corr_full')} corr_hold={m.get('corr_hold')}") print(f" marginal_verdict={m.get('marginal_verdict')}") print(f" has_insample_edge={m.get('has_insample_edge')} is_hedge={m.get('is_hedge')}") print(f" robust_oos={m.get('robust_oos')} multicut_persistent={m.get('multicut_persistent')}") print(f" clean_year_uplift={m.get('clean_year_uplift')} jackknife_min_uplift={m.get('jackknife_min_uplift')}") print(f" cand_insample_sharpe={m.get('cand_insample_sharpe')} multicut_uplift={m.get('multicut_uplift')}") print(f" blend w25={m.get('blends',{}).get('w25')}") print(f" blend w50={m.get('blends',{}).get('w50')}") print(f"\n win_minhold(reference)={win_minhold:+.2f}")