"""SKH2_ASYM_LS — long/short RISK ASYMMETRY family (Skyhook DD-cut wave). Hypothesis: shorts are essential (prior finding) but they carry the standalone draw-down — in crypto a short gets steamrolled by a vol-up move. Keep the verified V2-winner risk on the LONG side, but put TIGHTER risk on the SHORT side: a shorter time-stop (uscitashort) and/or a tighter SL (smaller sl_atr, or a fixed 'pct' SL), and a leaner TP so shorts take profit fast instead of bleeding into a reversal. SkyhookParams has uscitalong/uscitashort but a SINGLE sl_atr/tp_atr, so direction-asymmetric STOPS require CUSTOM entries. We reuse the engine's regime+pattern signal (htf_features + merge_htf_to_ltf) UNCHANGED — only the per-direction (sl, tp, max_bars) differ. This is causal: the only thing that depends on direction is the offset magnitude applied to close[i]; the SIGNAL (comp_long/comp_short) is computed exactly as the verified winner. Causality: proven by truncated-prefix recompute on the CUSTOM entries (same scheme as sk.causality): an entry emitted on a prefix must match the full-run entry at that index. """ 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 import altlib as al from src.backtest.harness import backtest_signals from src.strategies import skyhook as S from src.strategies.skyhook import SkyhookParams HOLDOUT = sk.HOLDOUT FEE = sk.FEE_RT # Verified V2 winner signal config (the regime/pattern gate we keep). WIN = dict(ptn_n=45, vola_lo=35.0, vola_hi=95.0, vol_lo=0.0) # Winner's symmetric risk (used for longs, and as the symmetric reference): WIN_RISK = dict(sl_atr=2.5, tp_atr=7.0, uscitalong=24, uscitashort=16) # --------------------------------------------------------------------------- # Custom asymmetric entries. Longs keep `long_risk`; shorts use `short_risk`. # Risk dicts: {'mode':'atr'|'pct', 'sl':..., 'tp':..., 'mb':int} # atr -> sl/tp are ATR multiples ; pct -> sl/tp are fractions of close. # --------------------------------------------------------------------------- def asym_entries(ltf, htf, base_p: SkyhookParams, long_risk: dict, short_risk: dict) -> list: feat = S.htf_features(htf, base_p) m = S.merge_htf_to_ltf(ltf, feat) c = m["close"].values.astype(float) a = S.atr(m, base_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 = [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) >= base_p.max_per_day: continue if comp_long[i]: direction, rk = 1, long_risk elif comp_short[i]: direction, rk = -1, short_risk else: continue if rk["mode"] == "atr": sl_off, tp_off = rk["sl"] * a[i], rk["tp"] * a[i] else: sl_off, tp_off = rk["sl"] * c[i], rk["tp"] * 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(rk["mb"])} count_today[day] = count_today.get(day, 0) + 1 return entries # --------------------------------------------------------------------------- # Causality on the CUSTOM entries: prefix recompute must match the full run. # --------------------------------------------------------------------------- def causality_struct(base_p, long_risk, short_risk, asset="BTC", tail=200) -> dict: ltf, htf = sk.frames(asset) full = asym_entries(ltf, htf, base_p, long_risk, short_risk) 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 = asym_entries(ltf.iloc[:cut].reset_index(drop=True), htf_cut, base_p, long_risk, short_risk) for i in range(max(0, cut - tail), cut): checked += 1 x, y = full[i], sub[i] if (x is None) != (y is None): bad += 1 elif x is not None and (x["dir"] != y["dir"] or abs(x["sl"] - y["sl"]) > 1e-6 or abs(x["tp"] - y["tp"]) > 1e-6 or x["max_bars"] != y["max_bars"]): bad += 1 return dict(ok=bool(bad == 0), mismatches=int(bad), checked=int(checked)) def _split(eq, idx, mask): e = eq[mask] if len(e) < 5: return dict(sharpe=0.0, ret=0.0, maxdd=0.0) 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)) def study_asym(name, base_p, long_risk, short_risk): per_asset = {} fee_ok_all = True for a in ("BTC", "ETH"): ltf, htf = sk.frames(a) ent = asym_entries(ltf, htf, base_p, long_risk, short_risk) 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) sweep = {} for f in (0.0, 0.001, 0.002, 0.003): mf = backtest_signals(ltf, ent, fee_rt=f, leverage=1.0, asset=a, tf="230m") sweep[f"{f*100:.2f}%"] = round(mf.sharpe, 3) fee_ok_all = fee_ok_all and (sweep["0.30%"] > 0) # short-only vs long-only DD diagnostic per_asset[a] = dict(full=full, hold=hold, fee_sweep=sweep, yearly={int(y): round(v, 4) for y, v in m.yearly.items()}) mf = min(per_asset[a]["full"]["sharpe"] for a in per_asset) mh = min(per_asset[a]["hold"]["sharpe"] for a in per_asset) mt = min(per_asset[a]["full"]["n_trades"] for a in per_asset) mdd = max(per_asset[a]["full"]["maxdd"] for a in per_asset) grade = "PASS" if (mt >= 20 and mf >= 0.5 and mh >= 0.2 and fee_ok_all) else \ ("WEAK" if (mt >= 20 and mf >= 0.3 and mh >= 0.0) else "FAIL") return dict(grade=grade, minFull=mf, minHold=mh, minTr=mt, maxDD=mdd, fee_ok=fee_ok_all, per_asset=per_asset) def daily_5050(base_p, long_risk, short_risk): series = {} for a in ("BTC", "ETH"): ltf, htf = sk.frames(a) ent = asym_entries(ltf, htf, base_p, long_risk, short_risk) 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))) series[a] = s.resample("1D").last().ffill().pct_change().dropna() J = pd.concat(series, axis=1, join="inner").fillna(0.0) return 0.5 * J["BTC"] + 0.5 * J["ETH"] def marginal_asym(base_p, long_risk, short_risk): return al.marginal_vs_tp01(daily_5050(base_p, long_risk, short_risk)) def print_study(name, r): print(f"\n=== {name} -> {r['grade']} (minFull={r['minFull']:+.2f} minHold={r['minHold']:+.2f}" f" minTr={r['minTr']} maxDD={r['maxDD']*100:.0f}% feeOK={r['fee_ok']})") for a, pa in r["per_asset"].items(): yr = " ".join(f"{y}:{v*100:+.0f}%" for y, v in pa["yearly"].items()) print(f" {a}: FULL Sh={pa['full']['sharpe']:+.2f} ret={pa['full']['ret']*100:+.0f}%" f" DD={pa['full']['maxdd']*100:.0f}% n={pa['full']['n_trades']} wr={pa['full']['win_rate']:.0f}%" f" | HOLD Sh={pa['hold']['sharpe']:+.2f} ret={pa['hold']['ret']*100:+.0f}% DD={pa['hold']['maxdd']*100:.0f}%") print(f" fee: " + " ".join(f"{k}={v:+.2f}" for k, v in pa["fee_sweep"].items())) print(f" yr: {yr}") if __name__ == "__main__": base_p = SkyhookParams(**WIN, **WIN_RISK) # signal + winner risk (used for shape only) # ---- 0) REFERENCE: rebuild the verified symmetric winner via our custom path ------- long_winner = dict(mode="atr", sl=2.5, tp=7.0, mb=24) sym_short = dict(mode="atr", sl=2.5, tp=7.0, mb=16) rREF = study_asym("REF symmetric-winner (rebuilt)", base_p, long_winner, sym_short) print_study("REF symmetric-winner (rebuilt)", rREF) # Long side: ROUND 1 showed pct-SL shorts lift everything but ETH DD sticks ~30.5%. # The standalone DD comes from BOTH directions, so we also tighten the LONG pct-SL a # touch to bring the combined DD under 30 while keeping the winner's long TP behaviour. # We test two long variants: the verified winner (atr) AND a pct long. long_variants = [ ("Latr", dict(mode="atr", sl=2.5, tp=7.0, mb=24)), ("Lpct", dict(mode="pct", sl=0.04, tp=0.10, mb=24)), ] # ---- 1) GRID over asymmetric SHORT risk: pct-SL family is the winner; push SL tighter # to knock ETH DD under 30. Keep the strong tp=0.08 and a couple of mb / SL choices. short_grid = [] for mb_s in (12, 14, 16): for slp in (0.02, 0.025, 0.03): for tpp in (0.06, 0.08): short_grid.append((f"Spct_mb{mb_s}_sl{slp}_tp{tpp}", dict(mode="pct", sl=slp, tp=tpp, mb=mb_s))) # a few tight-ATR shorts for completeness for mb_s in (12, 14): for sl_s in (1.5, 2.0): short_grid.append((f"Satr_mb{mb_s}_sl{sl_s}_tp5.0", dict(mode="atr", sl=sl_s, tp=5.0, mb=mb_s))) candidates = [] for lname, lr in long_variants: for sname, sr in short_grid: candidates.append((f"{lname}|{sname}", lr, sr)) results = [] for name, lr, sr in candidates: r = study_asym(name, base_p, lr, sr) results.append((name, lr, sr, r)) # Rank: feasible (grade != FAIL, fee ok) by lowest DD, then highest minHold. feas = [(n, lr, sr, r) for n, lr, sr, r in results if r["grade"] != "FAIL"] feas.sort(key=lambda t: (t[3]["maxDD"], -t[3]["minHold"])) print("\n\n##### GRID RANK (feasible, by lowest standalone maxDD) #####") for n, lr, sr, r in feas[:16]: print(f" {n:28s} DD={r['maxDD']*100:4.0f}% minFull={r['minFull']:+.2f}" f" minHold={r['minHold']:+.2f} minTr={r['minTr']} grade={r['grade']}") # ---- 2) Detailed study + marginal on the top DD-cutters that keep hold-out --------- # pick best candidates: DD<30 with decent hold-out qualifying = [t for t in feas if t[3]["maxDD"] < 0.30 and t[3]["minHold"] >= 0.50] qualifying.sort(key=lambda t: (t[3]["maxDD"], -t[3]["minHold"])) probe = qualifying[:5] if qualifying else feas[:5] print("\n\n##### DETAIL + MARGINAL on top probes #####") best = None for n, long_risk, sr, r in probe: print_study(n, r) caus = causality_struct(base_p, long_risk, sr, "BTC") caus_e = causality_struct(base_p, long_risk, sr, "ETH") mg = marginal_asym(base_p, long_risk, sr) w25 = mg.get("blends", {}).get("w25", {}) w50 = mg.get("blends", {}).get("w50", {}) earns = (r["grade"] != "FAIL" and mg.get("marginal_verdict") == "ADDS" and mg.get("robust_oos") and not mg.get("is_hedge")) beats = bool(earns and r["maxDD"] < 0.30 and (w25.get("uplift_hold") or -9) >= 0.55 and r["minHold"] >= 0.65) print(f" CAUSALITY BTC={caus} ETH={caus_e}") print(f" MARGINAL: verdict={mg.get('marginal_verdict')} corr_full={mg.get('corr_full')}" f" insample_edge={mg.get('has_insample_edge')} cand_is_sh={mg.get('cand_insample_sharpe')}" 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" 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={earns} beats_winner={beats}") cand = dict(name=n, long_risk=long_risk, short_risk=sr, study=r, caus=caus, caus_e=caus_e, mg=mg, earns=earns, beats=beats) # prefer beats; else lowest-DD earns; else lowest-DD feasible if best is None: best = cand else: key = lambda x: (x["beats"], x["earns"], -x["study"]["maxDD"], x["study"]["minHold"]) if key(cand) > key(best): best = cand print("\n\n##### FINAL BEST #####") b = best r = b["study"] mg = b["mg"] w25 = mg.get("blends", {}).get("w25", {}) w50 = mg.get("blends", {}).get("w50", {}) print(f"BEST CONFIG: signal={WIN} long_risk={b['long_risk']} short_risk={b['short_risk']}") print(f" minFull={r['minFull']:+.2f} minHold={r['minHold']:+.2f} maxDD={r['maxDD']*100:.0f}%" f" minTrades={r['minTr']} fee@0.30%_ok={r['fee_ok']}") print(f" causality BTC={b['caus']} ETH={b['caus_e']}") print(f" marginal: verdict={mg.get('marginal_verdict')} corr_full={mg.get('corr_full')}" f" insample_edge={mg.get('has_insample_edge')} hedge={mg.get('is_hedge')}" f" robust_oos={mg.get('robust_oos')} multicut={mg.get('multicut_persistent')}" f" cleanYr={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={b['earns']} beats_winner={b['beats']}") print(f" BTC_DD={r['per_asset']['BTC']['full']['maxdd']} ETH_DD={r['per_asset']['ETH']['full']['maxdd']}")