Adriano Dal Pastro
83 lines
3.8 KiB
Python
83 lines
3.8 KiB
Python
"""SKH_P_PTN (FAMILY=param)
|
|
On the SKH01-V1 base, sweep ptn_n in {34,45,55,70,89,110} x atr_win in {10,14,21}.
|
|
Slower Donchian breakouts may generalize better OOS. Maximize min-asset HOLD-OUT
|
|
subject to minFull>=0.5, fee survives 0.30%RT, >=20 trades BOTH assets, causality ok.
|
|
Note standalone DD. Always compare vs V1 (ptn_n=55, atr_win=14).
|
|
"""
|
|
import sys
|
|
import itertools
|
|
from dataclasses import replace
|
|
|
|
sys.path.insert(0, "/opt/docker/PythagorasGoal/scripts/research/skyhook")
|
|
import skyhooklib as sk
|
|
from src.strategies.skyhook import SkyhookParams
|
|
|
|
# SKH01-V1 reference base
|
|
V1 = SkyhookParams(ptn_n=55, sl_atr=2.5, tp_atr=6.0, vola_lo=35, vola_hi=95, vol_lo=0.0)
|
|
|
|
|
|
def quick(p: SkyhookParams) -> dict:
|
|
rs = {a: sk.run_asset(a, p, sk.FEE_RT) for a in sk.CERTIFIED}
|
|
mf = min(rs[a]["full"]["sharpe"] for a in rs)
|
|
mh = min(rs[a]["holdout"]["sharpe"] for a in rs)
|
|
mt = min(rs[a]["full"]["n_trades"] for a in rs)
|
|
avg_dd = sum(rs[a]["full"]["maxdd"] for a in rs) / 2
|
|
return dict(minFull=mf, minHold=mh, minTr=mt, dd=round(avg_dd, 4),
|
|
btc=rs["BTC"]["full"]["sharpe"], eth=rs["ETH"]["full"]["sharpe"],
|
|
btcH=rs["BTC"]["holdout"]["sharpe"], ethH=rs["ETH"]["holdout"]["sharpe"],
|
|
btcDD=rs["BTC"]["full"]["maxdd"], ethDD=rs["ETH"]["full"]["maxdd"])
|
|
|
|
|
|
PTN_GRID = (34, 45, 55, 70, 89, 110)
|
|
ATR_GRID = (10, 14, 21)
|
|
|
|
print("=== SKH_P_PTN sweep: ptn_n x atr_win on SKH01-V1 base ===")
|
|
qv1 = quick(V1)
|
|
print(f"V1 (ptn55/atr14): minF={qv1['minFull']:+.2f} minH={qv1['minHold']:+.2f} "
|
|
f"btc/eth F={qv1['btc']:+.2f}/{qv1['eth']:+.2f} H={qv1['btcH']:+.2f}/{qv1['ethH']:+.2f} "
|
|
f"tr={qv1['minTr']} dd~{qv1['dd']*100:.0f}% (btc{qv1['btcDD']*100:.0f}/eth{qv1['ethDD']*100:.0f})")
|
|
print("-" * 108)
|
|
print(f"{'ptn':>4s}{'atr':>4s} {'minF':>6s}{'minH':>6s} {'btcF/ethF':>13s} {'btcH/ethH':>13s} "
|
|
f"{'tr':>4s} {'avgDD':>6s} {'btcDD/ethDD':>12s} gate")
|
|
|
|
rows = []
|
|
for ptn_n, atr_win in itertools.product(PTN_GRID, ATR_GRID):
|
|
p = replace(V1, ptn_n=ptn_n, atr_win=atr_win)
|
|
q = quick(p)
|
|
# gate per task: minFull>=0.5 AND minHold>=0.2 AND minTr>=20
|
|
gate = (q["minFull"] >= 0.5 and q["minHold"] >= 0.2 and q["minTr"] >= 20)
|
|
rows.append((q["minHold"], q["minFull"], q["minTr"], q["dd"], ptn_n, atr_win, q, gate))
|
|
tag = "PASS" if gate else ""
|
|
print(f"{ptn_n:>4d}{atr_win:>4d} {q['minFull']:>+6.2f}{q['minHold']:>+6.2f} "
|
|
f"{q['btc']:>+5.2f}/{q['eth']:>+5.2f} {q['btcH']:>+5.2f}/{q['ethH']:>+5.2f} "
|
|
f"{q['minTr']:>4d} {q['dd']*100:>5.0f}% {q['btcDD']*100:>4.0f}/{q['ethDD']*100:>4.0f}% {tag}")
|
|
|
|
# winner = max min-asset HOLD-OUT among gate-passers (minFull>=0.5, minTr>=20); fallback best minHold
|
|
passers = [r for r in rows if r[7]]
|
|
pool = passers if passers else [r for r in rows if r[1] >= 0.5 and r[2] >= 20]
|
|
if not pool:
|
|
pool = rows
|
|
# rank by minHold, tiebreak lower avgDD then higher minFull
|
|
pool.sort(key=lambda r: (r[0], -r[3], r[1]), reverse=True)
|
|
best = pool[0]
|
|
b_ptn, b_atr = best[4], best[5]
|
|
print("-" * 108)
|
|
print(f"WINNER: ptn_n={b_ptn} atr_win={b_atr} minH={best[0]:+.2f} minF={best[1]:+.2f} "
|
|
f"tr={best[2]} avgDD={best[3]*100:.0f}%")
|
|
|
|
# Full study + causality + marginal on winner (and re-confirm V1 alongside)
|
|
WIN = replace(V1, ptn_n=b_ptn, atr_win=b_atr)
|
|
print("\n=== STUDY winner ===")
|
|
rep = sk.study(f"SKH_P_PTN ptn{b_ptn}/atr{b_atr}", WIN)
|
|
print(sk.fmt(rep))
|
|
caus = sk.causality(WIN, "BTC")
|
|
caus_eth = sk.causality(WIN, "ETH")
|
|
print(f"causality BTC: {caus} ETH: {caus_eth}")
|
|
mg = sk.marginal(WIN)
|
|
print(f"marginal: corr_full={mg.get('corr_full')} "
|
|
f"blend_w25_uplift_hold={mg.get('blends', {}).get('w25', {}).get('uplift_hold')} "
|
|
f"verdict={mg.get('marginal_verdict')} has_insample_edge={mg.get('has_insample_edge')} "
|
|
f"is_hedge={mg.get('is_hedge')}")
|
|
print("\nJSON_STUDY:", sk.as_json(rep))
|
|
print("MARGINAL:", mg)
|