"""SKH2_REGIME_TIGHT — DD-reduction wave, family: tighter regime selectivity. Hypothesis: make the regime band MORE selective (narrow vola band, add a volume floor) so only the cleanest setups trade -> fewer, higher-quality entries -> lower standalone DD, while keeping the winner's asymmetric exits (sl 2.5 / tp 7.0, uscitalong 24 / short 16). Baseline winner to beat (V2): SkyhookParams(ptn_n=45, sl_atr=2.5, tp_atr=7.0, uscitalong=24, uscitashort=16, vola_lo=35, vola_hi=95, vol_lo=0) minFull +0.83 minHold +0.81 ; DD BTC 34% / ETH 31% (the unmet goal: DD<30%) ; marginal ADDS, corr_full 0.05, blend w25 uplift_hold +0.58, w50 full 1.59 / hold 1.04 / DD 12.5%. BEATS-THE-WINNER iff: earns_slot AND max_dd<0.30 AND blend_w25_uplift_hold>=0.55 AND min_hold>=0.65. Everything is param-based (no new feature) -> sk.study/sk.marginal/sk.causality are directly valid. """ from __future__ import annotations import sys sys.path.insert(0, "/opt/docker/PythagorasGoal/scripts/research/skyhook") import itertools import skyhooklib as sk from src.strategies.skyhook import SkyhookParams # --- the winner's exits, frozen ---------------------------------------------------------- EXITS = dict(ptn_n=45, sl_atr=2.5, tp_atr=7.0, uscitalong=24, uscitashort=16) def mk(vola_lo, vola_hi, vol_lo, **extra): return SkyhookParams(vola_lo=vola_lo, vola_hi=vola_hi, vol_lo=vol_lo, **EXITS, **extra) def earns_slot(rep, mg): return (rep["verdict"]["grade"] != "FAIL" and mg.get("marginal_verdict") == "ADDS" and bool(mg.get("robust_oos")) and not bool(mg.get("is_hedge"))) def beats(rep, mg, max_dd): es = earns_slot(rep, mg) upl = mg.get("blends", {}).get("w25", {}).get("uplift_hold") mh = rep["verdict"]["min_asset_holdout_sharpe"] return (es and max_dd < 0.30 and (upl is not None and upl >= 0.55) and mh >= 0.65) def summarize(name, p): rep = sk.study(name, p) v = rep["verdict"] dd = max(rep["per_asset"][a]["full"]["maxdd"] for a in rep["per_asset"]) nt = v["min_trades"] print(sk.fmt(rep)) print(f" >> maxDD(BTC,ETH) = {dd*100:.1f}% minTrades={nt}") return rep, dd if __name__ == "__main__": # ---- STAGE 1: coarse grid over the family levers ----------------------------------- vola_los = [35.0, 40.0, 45.0, 50.0] # 35 = winner; 40/45/50 = tighter floor vola_his = [85.0, 90.0, 95.0] # 95 = winner; 85/90 = clip blow-off harder vol_los = [0.0, 30.0, 40.0, 50.0] # 0 = winner; floor = require live volume rows = [] print("########## STAGE 1: coarse DD scan (study, both assets) ##########") for vlo, vhi, vol in itertools.product(vola_los, vola_his, vol_los): p = mk(vlo, vhi, vol) rep = sk.study(f"vlo{vlo:.0f}_vhi{vhi:.0f}_vol{vol:.0f}", p) v = rep["verdict"] dd = max(rep["per_asset"][a]["full"]["maxdd"] for a in rep["per_asset"]) nt = v["min_trades"] ddb = rep["per_asset"]["BTC"]["full"]["maxdd"] dde = rep["per_asset"]["ETH"]["full"]["maxdd"] rows.append(dict(vlo=vlo, vhi=vhi, vol=vol, grade=v["grade"], mf=v["min_asset_full_sharpe"], mh=v["min_asset_holdout_sharpe"], dd=dd, ddb=ddb, dde=dde, nt=nt, fee=v["fee_survives"])) print(f" vlo{vlo:.0f} vhi{vhi:.0f} vol{vol:.0f}: {v['grade']:4s} " f"mF={v['min_asset_full_sharpe']:+.2f} mH={v['min_asset_holdout_sharpe']:+.2f} " f"DD={dd*100:4.1f}% (B{ddb*100:.0f}/E{dde*100:.0f}) nT={nt:3d} feeOK={v['fee_survives']}") # ---- rank: candidates with DD<30%, minTrades>=20, fee OK, holdout>=0.65 ------------- print("\n########## STAGE 1 RANK (filter DD<30, nT>=20, fee OK, mH>=0.65, not FAIL) ##########") cand = [r for r in rows if r["dd"] < 0.30 and r["nt"] >= 20 and r["fee"] and r["mh"] >= 0.65 and r["grade"] != "FAIL"] cand.sort(key=lambda r: (-r["mh"], r["dd"])) # prefer high holdout then low DD if not cand: print(" (none met all hard filters; falling back to lowest-DD with nT>=20 & not FAIL)") cand = [r for r in rows if r["nt"] >= 20 and r["grade"] != "FAIL"] cand.sort(key=lambda r: (r["dd"], -r["mh"])) for r in cand[:8]: print(f" vlo{r['vlo']:.0f} vhi{r['vhi']:.0f} vol{r['vol']:.0f}: {r['grade']} " f"mF={r['mf']:+.2f} mH={r['mh']:+.2f} DD={r['dd']*100:.1f}% nT={r['nt']}") # ---- STAGE 2: full diligence (causality + marginal) on top few --------------------- print("\n########## STAGE 2: causality + marginal on top candidates ##########") best = None top = cand[:4] for r in top: p = mk(r["vlo"], r["vhi"], r["vol"]) name = f"TIGHT vlo{r['vlo']:.0f}_vhi{r['vhi']:.0f}_vol{r['vol']:.0f}" print(f"\n----- {name} -----") rep, dd = summarize(name, p) cau = sk.causality(p, "BTC") cau_e = sk.causality(p, "ETH") cau_ok = bool(cau["ok"] and cau_e["ok"]) mg = sk.marginal(p) es = earns_slot(rep, mg) bw = beats(rep, mg, dd) and cau_ok w25 = mg.get("blends", {}).get("w25", {}) w50 = mg.get("blends", {}).get("w50", {}) print(f" causality BTC={cau} ETH={cau_e} -> ok={cau_ok}") print(f" marginal: verdict={mg.get('marginal_verdict')} corr_full={mg.get('corr_full')} " f"corr_hold={mg.get('corr_hold')} insample_edge={mg.get('has_insample_edge')} " f"(cand_is_sh={mg.get('cand_insample_sharpe')}) hedge={mg.get('is_hedge')} " f"robust_oos={mg.get('robust_oos')} multicut={mg.get('multicut_persistent')} " f"clean_year_uplift={mg.get('clean_year_uplift')}") print(f" blend w25: uplift_hold={w25.get('uplift_hold')} uplift_full={w25.get('uplift_full')} " f"| w50: full={w50.get('full')} hold={w50.get('hold')} dd={w50.get('dd')}") print(f" >> earns_slot={es} beats_winner={bw}") cand_obj = dict(name=name, p=p, rep=rep, dd=dd, mg=mg, cau_ok=cau_ok, es=es, bw=bw, w25=w25, w50=w50, mf=rep["verdict"]["min_asset_full_sharpe"], mh=rep["verdict"]["min_asset_holdout_sharpe"], nt=rep["verdict"]["min_trades"], fee=rep["verdict"]["fee_survives"]) # pick best: prefer beats_winner, then earns_slot+lowDD, then lowDD def keyf(c): return (c["bw"], c["es"], -c["dd"], c["mh"]) if best is None or keyf(cand_obj) > keyf(best): best = cand_obj # ---- FINAL BLOCK ------------------------------------------------------------------- print("\n\n==================== FINAL — REGIME_TIGHT BEST ====================") if best is None: print("NO CANDIDATE — no config passed even the soft filter.") sys.exit(0) b = best pf = {k: getattr(b["p"], k) for k in b["p"].__dataclass_fields__} mg = b["mg"] print(f"config: vola_lo={b['p'].vola_lo} vola_hi={b['p'].vola_hi} vol_lo={b['p'].vol_lo} " f"ptn_n={b['p'].ptn_n} sl_atr={b['p'].sl_atr} tp_atr={b['p'].tp_atr} " f"uscitalong={b['p'].uscitalong} uscitashort={b['p'].uscitashort}") print(f"minFull={b['mf']:+.3f} minHold={b['mh']:+.3f} max_dd={b['dd']*100:.1f}% " f"n_trades_min={b['nt']} fee@0.30%OK={b['fee']} causality_ok={b['cau_ok']}") print(f"earns_slot={b['es']} beats_winner={b['bw']}") print("FULL marginal dict:") for k in ("corr_full", "corr_hold", "marginal_verdict", "has_insample_edge", "cand_insample_sharpe", "is_hedge", "robust_oos", "multicut_persistent", "clean_year_uplift", "jackknife_min_uplift", "multicut_uplift"): print(f" {k} = {mg.get(k)}") print(f" blend w25: {mg.get('blends', {}).get('w25')}") print(f" blend w50: {mg.get('blends', {}).get('w50')}") print("PARAMS:", pf) # machine-readable tail for me to parse import json print("\nRESULT_JSON " + json.dumps(dict( family="REGIME_TIGHT", params=pf, minFull=b["mf"], minHold=b["mh"], max_dd=b["dd"], n_trades_min=b["nt"], fee_ok=b["fee"], causality_ok=b["cau_ok"], earns_slot=b["es"], beats=b["bw"], corr_full=mg.get("corr_full"), marginal_verdict=mg.get("marginal_verdict"), has_insample_edge=mg.get("has_insample_edge"), is_hedge=mg.get("is_hedge"), robust_oos=mg.get("robust_oos"), multicut_persistent=mg.get("multicut_persistent"), clean_year_uplift=mg.get("clean_year_uplift"), w25_uplift_hold=mg.get("blends", {}).get("w25", {}).get("uplift_hold"), ), default=str))