"""SKH2_FREQ — entry cadence / holding-period family for the SKH01 DD-reduction wave. Goal: cut standalone maxDD below 30% (max over BTC & ETH) while keeping min-asset HOLD-OUT Sharpe >= ~0.70 and earns_slot == True. Lever space (all expressible via SkyhookParams): * max_per_day {1, 2} * uscitalong / uscitashort holding windows {12..30} * atr_win (HTF) / ltf_atr_win (exec) windows Baseline-to-beat (verified V2 winner): 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 maxDD BTC34%/ETH31% earns_slot True blend w25 uplift_hold +0.58, w50 full1.59/hold1.04/DD12.5%. A candidate BEATS the winner iff: earns_slot AND max_dd<0.30 AND blend_w25_uplift_hold>=0.55 AND min_hold_sharpe>=0.65. """ from __future__ import annotations import sys, json sys.path.insert(0, "/opt/docker/PythagorasGoal/scripts/research/skyhook") import skyhooklib as sk from src.strategies.skyhook import SkyhookParams # Winner base (all FREQ variants share the regime/pattern/stop structure of the winner). WINNER = dict(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) def mk(**over) -> SkyhookParams: d = dict(WINNER); d.update(over) return SkyhookParams(**d) def quick(name, p) -> dict: """Fast screen: FULL+HOLD on both assets + standalone maxDD. No fee sweep / marginal yet.""" rb = sk.run_asset("BTC", p) re = sk.run_asset("ETH", p) minF = min(rb["full"]["sharpe"], re["full"]["sharpe"]) minH = min(rb["holdout"]["sharpe"], re["holdout"]["sharpe"]) maxdd = max(rb["full"]["maxdd"], re["full"]["maxdd"]) minTr = min(rb["full"]["n_trades"], re["full"]["n_trades"]) print(f" {name:38s} minF={minF:+.2f} minH={minH:+.2f} maxDD={maxdd*100:4.0f}% " f"(B{rb['full']['maxdd']*100:.0f}/E{re['full']['maxdd']*100:.0f}) " f"nTr={minTr} | Bh={rb['holdout']['sharpe']:+.2f} Eh={re['holdout']['sharpe']:+.2f}") return dict(name=name, p=p, minF=minF, minH=minH, maxdd=maxdd, minTr=minTr, bdd=rb["full"]["maxdd"], edd=re["full"]["maxdd"]) def full_eval(name, p) -> dict: rep = sk.study(name, p) print(sk.fmt(rep)) caus = sk.causality(p, "BTC") causE = sk.causality(p, "ETH") caus_ok = bool(caus["ok"] and causE["ok"]) mg = sk.marginal(p) v = rep["verdict"] maxdd = max(rep["per_asset"][a]["full"]["maxdd"] for a in rep["per_asset"]) w25 = mg.get("blends", {}).get("w25", {}) w50 = mg.get("blends", {}).get("w50", {}) earns_slot = (v["grade"] != "FAIL" and mg.get("marginal_verdict") == "ADDS" and bool(mg.get("robust_oos")) and not bool(mg.get("is_hedge"))) beats = (earns_slot and maxdd < 0.30 and (w25.get("uplift_hold") or -9) >= 0.55 and v["min_asset_holdout_sharpe"] >= 0.65) print(f" causality BTC={caus} ETH={causE} -> ok={caus_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"hedge={mg.get('is_hedge')} robust_oos={mg.get('robust_oos')} " f"multicut={mg.get('multicut_persistent')} cleanYr={mg.get('clean_year_uplift')}") print(f" blends: 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')} " f"uplift_hold={w50.get('uplift_hold')}") print(f" ==> maxDD={maxdd*100:.1f}% earns_slot={earns_slot} BEATS_WINNER={beats}") return dict(name=name, p=p, rep=rep, mg=mg, caus_ok=caus_ok, maxdd=maxdd, earns_slot=earns_slot, beats=beats, w25=w25, w50=w50, v=v) if __name__ == "__main__": print("="*100) print("PHASE 1 — fast screen of cadence / holding / atr-window variants (FULL+HOLD+DD)") print("="*100) screens = [] # 0) reproduce the winner as a sanity anchor screens.append(quick("WINNER(uL24/uS16,mpd1,atr14/14)", mk())) # --- Holding-window grid (the core DD lever): shorter holds cap single-trade risk. print("\n-- holding windows (uscitalong/uscitashort), mpd=1 --") for uL in (12, 16, 20, 24, 28, 30): for uS in (10, 12, 14, 16, 20): screens.append(quick(f"uL{uL}/uS{uS}", mk(uscitalong=uL, uscitashort=uS))) # --- max_per_day: 2 entries/day = more frequent re-entry (more fee, smaller clusters?) print("\n-- max_per_day=2 across a few holds --") for uL in (12, 16, 20, 24): for uS in (10, 12, 16): screens.append(quick(f"mpd2 uL{uL}/uS{uS}", mk(max_per_day=2, uscitalong=uL, uscitashort=uS))) # --- atr windows (HTF signal vola & exec stop sizing), at the WINNER hold (uL24/uS16) # where DD was lowest, not the whipsaw uL16/uS12. print("\n-- atr_win (HTF) x ltf_atr_win (exec), at WINNER hold uL24/uS16 --") for aw in (10, 14, 20): for lw in (10, 14, 20): screens.append(quick(f"atr{aw}/ltf{lw} uL24/uS16", mk(atr_win=aw, ltf_atr_win=lw))) # --- targeted DD-reducers: mpd2 at the winner hold (smaller clusters, keep hold) + # longer ATR for steadier stops; and asymmetric long-bias holds (long crypto = up-drift, # so a longer long-hold + shorter short-hold protects the worst-asset short DD). print("\n-- targeted DD-reducers (mpd2 @ winner hold; long-bias asym holds) --") for cfg in ( dict(max_per_day=2, uscitalong=24, uscitashort=16), dict(max_per_day=2, uscitalong=24, uscitashort=16, ltf_atr_win=20), dict(max_per_day=2, uscitalong=24, uscitashort=16, atr_win=20, ltf_atr_win=20), dict(uscitalong=24, uscitashort=18, ltf_atr_win=20), dict(uscitalong=28, uscitashort=18, ltf_atr_win=20), dict(uscitalong=24, uscitashort=20, ltf_atr_win=20), dict(max_per_day=2, uscitalong=20, uscitashort=16, ltf_atr_win=20), dict(max_per_day=2, uscitalong=20, uscitashort=18, ltf_atr_win=20), ): nm = "DDr " + "/".join(f"{k}={v}" for k, v in cfg.items()) screens.append(quick(nm, mk(**cfg))) # Rank by: meets DD<30 first, then by minH (we need hold >=0.65), then minF. print("\n" + "="*100) print("PHASE 2 — full eval of best DD-reducing candidates (study+causality+marginal)") print("="*100) # candidates: prioritize LOWEST DD with a still-usable hold-out (minH>=0.55), but ALSO # always include the global lowest-DD configs that keep minH>=0.5 (DD is the unmet goal). pool = [s for s in screens if s["minTr"] >= 20] a_set = [s for s in pool if s["maxdd"] < 0.36 and s["minH"] >= 0.55] a_set.sort(key=lambda s: (s["maxdd"], -s["minH"])) b_set = [s for s in pool if s["minH"] >= 0.50] b_set.sort(key=lambda s: s["maxdd"]) # lowest DD overall (usable hold) picked = [] seen = set() for s in a_set[:6] + b_set[:6]: if s["name"] in seen: continue seen.add(s["name"]) picked.append(s) if len(picked) >= 9: break if not picked: pool.sort(key=lambda s: s["maxdd"]) picked = pool[:6] print("Picked for full eval (DD<0.32, minH>=0.55, nTr>=20), sorted by DD:") for s in picked: print(f" {s['name']:38s} maxDD={s['maxdd']*100:.0f}% minH={s['minH']:+.2f} minF={s['minF']:+.2f}") results = [] for s in picked: print("\n" + "-"*90) results.append(full_eval(s["name"], s["p"])) # also full-eval the winner as the reference print("\n" + "-"*90 + "\n[REFERENCE] WINNER full eval:") rwin = full_eval("WINNER", mk()) # ---- pick the best config: prefer beats_winner, else lowest DD with earns_slot & best hold print("\n" + "="*100) print("FINAL RANKING") print("="*100) def score(r): return (not r["beats"], not r["earns_slot"], r["maxdd"], -r["v"]["min_asset_holdout_sharpe"]) allr = results + [rwin] allr.sort(key=score) for r in allr: print(f" {r['name']:38s} beats={r['beats']} earns={r['earns_slot']} maxDD={r['maxdd']*100:.0f}% " f"minF={r['v']['min_asset_full_sharpe']:+.2f} minH={r['v']['min_asset_holdout_sharpe']:+.2f} " f"w25uH={r['w25'].get('uplift_hold')} caus={r['caus_ok']}") best = allr[0] print("\n" + "="*100) print("BEST CONFIG") print("="*100) bp = best["p"] cfg = {k: getattr(bp, k) for k in bp.__dataclass_fields__} print(f"name={best['name']}") print(f"config={json.dumps(cfg)}") print(f"minFull={best['v']['min_asset_full_sharpe']:+.3f}") print(f"minHold={best['v']['min_asset_holdout_sharpe']:+.3f}") print(f"max_dd={best['maxdd']:.4f}") print(f"n_trades_min={best['v']['min_trades']}") print(f"fee_survives_0.30%={best['v']['fee_survives']}") print(f"causality_ok={best['caus_ok']}") mg = best["mg"] print(f"MARGINAL DICT: corr_full={mg.get('corr_full')} corr_hold={mg.get('corr_hold')} " f"verdict={mg.get('marginal_verdict')} has_insample_edge={mg.get('has_insample_edge')} " f"is_hedge={mg.get('is_hedge')} robust_oos={mg.get('robust_oos')} " f"multicut_persistent={mg.get('multicut_persistent')} clean_year_uplift={mg.get('clean_year_uplift')}") print(f"blend w25 uplift_hold={best['w25'].get('uplift_hold')} | " f"w50 full={best['w50'].get('full')} hold={best['w50'].get('hold')} dd={best['w50'].get('dd')}") print(f"earns_slot={best['earns_slot']} BEATS_WINNER={best['beats']}") # dump machine-readable for final structured output print("\nJSON_BEST=" + json.dumps(dict( name=best["name"], config=cfg, minFull=best["v"]["min_asset_full_sharpe"], minHold=best["v"]["min_asset_holdout_sharpe"], max_dd=best["maxdd"], n_trades_min=best["v"]["min_trades"], fee_survives=best["v"]["fee_survives"], causality_ok=best["caus_ok"], earns_slot=best["earns_slot"], beats=best["beats"], 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=best["w25"].get("uplift_hold"))))