"""SKH2_TPSL_DD: RR / stop fine grid around the V2 winner to push standalone maxDD < 30% while holding min-asset HOLD-OUT >= ~0.70 and earns_slot True. Winner baseline: 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) -> minFull +0.83, minHold +0.81, DD BTC 34% / ETH 31% (THE PROBLEM), earns_slot True. Family task: sl_atr in {1.75,2.0,2.25,2.5}, tp_atr in {5,6,7,8}, exit_mode 'pct' vs 'atr'. Tighter SL cuts DD but can lower hold-out. Find DD<30 cell + minHold>=0.7 + plateau. """ from __future__ import annotations import sys sys.path.insert(0, "/opt/docker/PythagorasGoal/scripts/research/skyhook") import numpy as np # noqa: E402 import skyhooklib as sk # noqa: E402 from src.strategies.skyhook import SkyhookParams # noqa: E402 # Winner fixed (non-RR) fields: BASE = dict(ptn_n=45, uscitalong=24, uscitashort=16, vola_lo=35.0, vola_hi=95.0, vol_lo=0.0) def mk(sl_atr=None, tp_atr=None, exit_mode="atr", sl_pct=None, tp_pct=None): kw = dict(BASE) kw["exit_mode"] = exit_mode if exit_mode == "atr": kw["sl_atr"] = sl_atr kw["tp_atr"] = tp_atr else: kw["sl_pct"] = sl_pct kw["tp_pct"] = tp_pct return SkyhookParams(**kw) def metrics(p): """FULL/HOLD/DD min-asset + fee@0.30 + trades, both assets.""" pa = {} fee_ok = True for a in ("BTC", "ETH"): r = sk.run_asset(a, p) # fee sweep at 0.30% rf = sk.run_asset(a, p, fee_rt=0.003) fee_ok = fee_ok and (rf["full"]["sharpe"] > 0) pa[a] = dict(full=r["full"], hold=r["holdout"], yearly=r["yearly"], fee30=rf["full"]["sharpe"]) minFull = min(pa[a]["full"]["sharpe"] for a in pa) minHold = min(pa[a]["hold"]["sharpe"] for a in pa) minTr = min(pa[a]["full"]["n_trades"] for a in pa) maxDD = max(pa[a]["full"]["maxdd"] for a in pa) return dict(pa=pa, minFull=minFull, minHold=minHold, minTr=minTr, maxDD=maxDD, fee_ok=fee_ok) def grade_of(m): enough = m["minTr"] >= 20 if enough and m["minFull"] >= 0.5 and m["minHold"] >= 0.2 and m["fee_ok"]: return "PASS" if enough and m["minFull"] >= 0.3 and m["minHold"] >= 0.0: return "WEAK" return "FAIL" if __name__ == "__main__": # ---- STAGE 1: coarse ATR grid (the family core) ---- sl_grid = [1.75, 2.0, 2.25, 2.5] tp_grid = [5.0, 6.0, 7.0, 8.0] rows = [] print("##### STAGE 1: ATR grid (sl_atr x tp_atr) #####") print(f"{'sl':>5} {'tp':>4} | {'minFull':>8} {'minHold':>8} {'maxDD%':>7} {'minTr':>6} " f"{'BTC_DD':>7} {'ETH_DD':>7} {'feeOK':>5} {'grade':>5}") for sl in sl_grid: for tp in tp_grid: p = mk(sl_atr=sl, tp_atr=tp, exit_mode="atr") m = metrics(p) g = grade_of(m) bdd = m["pa"]["BTC"]["full"]["maxdd"] edd = m["pa"]["ETH"]["full"]["maxdd"] rows.append((sl, tp, "atr", m, g)) print(f"{sl:>5} {tp:>4} | {m['minFull']:>+8.2f} {m['minHold']:>+8.2f} " f"{m['maxDD']*100:>7.1f} {m['minTr']:>6} {bdd*100:>7.1f} {edd*100:>7.1f} " f"{str(m['fee_ok']):>5} {g:>5}") # ---- STAGE 2: pct exit grid (a few sensible RR pairs ~ matching ATR ratios) ---- # ATR LTF ~ a few % of price; pct exit gives a HARD dollar cap on DD per trade. print("\n##### STAGE 2: PCT exit grid (sl_pct x tp_pct) #####") print(f"{'slP':>6} {'tpP':>6} | {'minFull':>8} {'minHold':>8} {'maxDD%':>7} {'minTr':>6} " f"{'BTC_DD':>7} {'ETH_DD':>7} {'feeOK':>5} {'grade':>5}") pct_pairs = [(0.02, 0.06), (0.025, 0.07), (0.03, 0.075), (0.03, 0.09), (0.035, 0.10), (0.04, 0.10), # dense neighbourhood around the DD<30 winner (0.025,0.07) to prove a plateau: (0.0225, 0.065), (0.0225, 0.07), (0.0225, 0.075), (0.025, 0.0625), (0.025, 0.065), (0.025, 0.075), (0.025, 0.08), (0.0275, 0.065), (0.0275, 0.07), (0.0275, 0.075)] for slp, tpp in pct_pairs: p = mk(exit_mode="pct", sl_pct=slp, tp_pct=tpp) m = metrics(p) g = grade_of(m) bdd = m["pa"]["BTC"]["full"]["maxdd"] edd = m["pa"]["ETH"]["full"]["maxdd"] rows.append((slp, tpp, "pct", m, g)) print(f"{slp:>6} {tpp:>6} | {m['minFull']:>+8.2f} {m['minHold']:>+8.2f} " f"{m['maxDD']*100:>7.1f} {m['minTr']:>6} {bdd*100:>7.1f} {edd*100:>7.1f} " f"{str(m['fee_ok']):>5} {g:>5}") # ---- Pick best: DD<30, minHold>=0.7, grade!=FAIL; tie-break by minHold then minFull ---- def ok_dd(r): return r[3]["maxDD"] < 0.30 and r[3]["minHold"] >= 0.70 and r[4] != "FAIL" cands = [r for r in rows if ok_dd(r)] if not cands: # relax: DD<30 and minHold>=0.65 cands = [r for r in rows if r[3]["maxDD"] < 0.30 and r[3]["minHold"] >= 0.65 and r[4] != "FAIL"] relaxed = True else: relaxed = False if not cands: # fall back to lowest DD among non-FAIL with decent hold cands = [r for r in rows if r[4] != "FAIL"] # rank: among DD<30 cells, maximize a balanced score (minHold + minFull) so we don't pick a # low-DD-but-weak-Sharpe corner. DD is already gated < 0.30 above, so optimise value next. cands_sorted = sorted(cands, key=lambda r: -(r[3]["minHold"] + r[3]["minFull"])) best = cands_sorted[0] print(f"\n##### BEST PICK (relaxed={relaxed if cands else 'fallback'}): " f"{'sl/tp' if best[2]=='atr' else 'slP/tpP'}=({best[0]},{best[1]}) mode={best[2]} #####") # Build best params if best[2] == "atr": bp = mk(sl_atr=best[0], tp_atr=best[1], exit_mode="atr") best_cfg = dict(ptn_n=45, sl_atr=best[0], tp_atr=best[1], uscitalong=24, uscitashort=16, vola_lo=35.0, vola_hi=95.0, vol_lo=0.0, exit_mode="atr") else: bp = mk(exit_mode="pct", sl_pct=best[0], tp_pct=best[1]) best_cfg = dict(ptn_n=45, sl_pct=best[0], tp_pct=best[1], uscitalong=24, uscitashort=16, vola_lo=35.0, vola_hi=95.0, vol_lo=0.0, exit_mode="pct") # ---- Full verification of best: study + causality + marginal ---- print("\n##### FULL STUDY of BEST #####") rep = sk.study("SKH2_TPSL_DD-BEST", bp) print(sk.fmt(rep)) caus = sk.causality(bp, "BTC") caus_eth = sk.causality(bp, "ETH") print(f"\ncausality BTC: {caus}") print(f"causality ETH: {caus_eth}") mg = sk.marginal(bp) m = best[3] g = rep["verdict"]["grade"] earns = (g != "FAIL" and mg.get("marginal_verdict") == "ADDS" and bool(mg.get("robust_oos")) and not bool(mg.get("is_hedge"))) w25 = mg.get("blends", {}).get("w25", {}) w50 = mg.get("blends", {}).get("w50", {}) beats = (earns and m["maxDD"] < 0.30 and (w25.get("uplift_hold") or -9) >= 0.55 and m["minHold"] >= 0.65) print("\n========== FINAL BLOCK ==========") print(f"best_cfg = {best_cfg}") print(f"exit_mode = {best[2]}") print(f"minFull = {m['minFull']:+.3f}") print(f"minHold = {m['minHold']:+.3f}") print(f"max_dd (BTC/ETH) = {m['maxDD']:.4f} (BTC {m['pa']['BTC']['full']['maxdd']:.4f} / " f"ETH {m['pa']['ETH']['full']['maxdd']:.4f})") print(f"n_trades_min = {m['minTr']}") print(f"fee@0.30 OK = {m['fee_ok']} (BTC {m['pa']['BTC']['fee30']:+.2f} / " f"ETH {m['pa']['ETH']['fee30']:+.2f})") print(f"causality_ok = {caus['ok'] and caus_eth['ok']} " f"(BTC mism={caus['mismatches']} ETH mism={caus_eth['mismatches']})") print(f"grade = {g}") print("--- marginal vs TP01 ---") print(f"corr_full = {mg.get('corr_full')}") print(f"corr_hold = {mg.get('corr_hold')}") print(f"marginal_verdict = {mg.get('marginal_verdict')}") print(f"has_insample_edge = {mg.get('has_insample_edge')}") print(f"is_hedge = {mg.get('is_hedge')}") print(f"robust_oos = {mg.get('robust_oos')}") print(f"multicut_persist = {mg.get('multicut_persistent')}") print(f"clean_year_uplift = {mg.get('clean_year_uplift')}") print(f"jackknife_min_upl = {mg.get('jackknife_min_uplift')}") print(f"cand_insample_sh = {mg.get('cand_insample_sharpe')}") print(f"blend w25 = {w25}") print(f"blend w50 = {w50}") print(f"earns_slot = {earns}") print(f"BEATS_WINNER = {beats}") # ---- plateau report: neighbors of best in the same mode ---- print("\n##### PLATEAU (neighbors of best) #####") nbrs = [r for r in rows if r[2] == best[2]] nbrs_sorted = sorted(nbrs, key=lambda r: (r[3]["maxDD"])) for r in nbrs_sorted[:8]: tag = f"({r[0]},{r[1]})" print(f" {r[2]} {tag:>14}: DD={r[3]['maxDD']*100:5.1f}% minFull={r[3]['minFull']:+.2f} " f"minHold={r[3]['minHold']:+.2f} grade={r[4]}")