From c7c07f4c356eca3bbafb857cc6070d302def2873 Mon Sep 17 00:00:00 2001 From: Adriano Dal Pastro Date: Tue, 23 Jun 2026 14:46:47 +0000 Subject: [PATCH] test(skyhook): demo anchors + dual-TF alignment + causality + V1 robustness (5 pass) Co-Authored-By: Claude Opus 4.8 (1M context) --- scripts/research/skyhook/check_v1.py | 13 ++++ scripts/research/skyhook/grid.py | 26 +++++++ scripts/research/skyhook/runs/SKH_P_PTN.py | 82 +++++++++++++++++++ scripts/research/skyhook/runs/SKH_P_RR.py | 87 +++++++++++++++++++++ tests/test_skyhook.py | 91 ++++++++++++++++++++++ 5 files changed, 299 insertions(+) create mode 100644 scripts/research/skyhook/check_v1.py create mode 100644 scripts/research/skyhook/grid.py create mode 100644 scripts/research/skyhook/runs/SKH_P_PTN.py create mode 100644 scripts/research/skyhook/runs/SKH_P_RR.py create mode 100644 tests/test_skyhook.py diff --git a/scripts/research/skyhook/check_v1.py b/scripts/research/skyhook/check_v1.py new file mode 100644 index 0000000..36eb4fc --- /dev/null +++ b/scripts/research/skyhook/check_v1.py @@ -0,0 +1,13 @@ +import sys +sys.path.insert(0, "/opt/docker/PythagorasGoal/scripts/research/skyhook") +import skyhooklib as sk +from src.strategies.skyhook import SkyhookParams +V1 = SkyhookParams(ptn_n=55, sl_atr=2.5, tp_atr=6.0, vola_lo=35, vola_hi=95, vol_lo=0.0) +rep = sk.study("SKH01-V1", V1) +print(sk.fmt(rep)) +print("causality:", sk.causality(V1)) +print("\n--- marginal vs TP01 (does it ADD as a sleeve?) ---") +import altlib as al +print(al.fmt_marginal(dict(name="SKH01-V1", marginal=sk.marginal(V1), + abs_grade=rep["verdict"]["grade"], marginal_verdict=sk.marginal(V1).get("marginal_verdict"), + earns_slot=False))) diff --git a/scripts/research/skyhook/grid.py b/scripts/research/skyhook/grid.py new file mode 100644 index 0000000..ff48e3a --- /dev/null +++ b/scripts/research/skyhook/grid.py @@ -0,0 +1,26 @@ +"""Combined grid over the scout-winning levers -> rank by min-asset HOLD-OUT (gate minFull>=0.5).""" +import sys, 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 + +base = SkyhookParams() +def quick(p): + rs = {a: sk.run_asset(a, p, sk.FEE_RT) for a in sk.CERTIFIED} + return (min(rs[a]["full"]["sharpe"] for a in rs), + min(rs[a]["holdout"]["sharpe"] for a in rs), + min(rs[a]["full"]["n_trades"] for a in rs), + round(sum(rs[a]["full"]["maxdd"] for a in rs)/2,3)) + +rows=[] +for ptn_n,(sl,tp),vol_lo,(vlo,vhi) in itertools.product( + (8,21,55), ((2.0,5.0),(2.5,6.0),(3.0,8.0)), (0.0,40.0,50.0), ((35.0,95.0),(25.0,95.0))): + p=replace(base, ptn_n=ptn_n, sl_atr=sl, tp_atr=tp, vol_lo=vol_lo, vola_lo=vlo, vola_hi=vhi) + mf,mh,mt,dd=quick(p) + rows.append((mh,mf,mt,dd,ptn_n,sl,tp,vol_lo,vlo,vhi)) +rows.sort(reverse=True) +print(f"{'minH':>6s}{'minF':>6s}{'tr':>5s}{'dd':>5s} ptn sl tp vlo vola") +for mh,mf,mt,dd,ptn_n,sl,tp,vol_lo,vlo,vhi in rows[:18]: + gate = "PASS" if (mf>=0.5 and mh>=0.2 and mt>=20) else "" + print(f"{mh:>+6.2f}{mf:>+6.2f}{mt:>5d}{dd*100:>4.0f}% {ptn_n:>3d} {sl:>3.1f} {tp:>4.1f} {vol_lo:>4.0f} [{vlo:.0f},{vhi:.0f}] {gate}") diff --git a/scripts/research/skyhook/runs/SKH_P_PTN.py b/scripts/research/skyhook/runs/SKH_P_PTN.py new file mode 100644 index 0000000..7c466e1 --- /dev/null +++ b/scripts/research/skyhook/runs/SKH_P_PTN.py @@ -0,0 +1,82 @@ +"""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) diff --git a/scripts/research/skyhook/runs/SKH_P_RR.py b/scripts/research/skyhook/runs/SKH_P_RR.py new file mode 100644 index 0000000..b7d44e1 --- /dev/null +++ b/scripts/research/skyhook/runs/SKH_P_RR.py @@ -0,0 +1,87 @@ +"""SKH_P_RR — fine-sweep reward:risk on the ptn_n=55 V1 base. + +V1 base: SkyhookParams(ptn_n=55, sl_atr=2.5, tp_atr=6.0, vola_lo=35, vola_hi=95, vol_lo=0.0) + -> minFull +0.69, HOLD +0.64 (BTC 0.64 / ETH 0.64), DD ~40-49% (HIGH). + +Sweep: sl_atr in {2.0,2.25,2.5,2.75,3.0,3.5} x tp_atr in {5,6,7,8,9,10}. +Objective: maximize min-asset HOLD-OUT subject to minFull>=0.5, cut DD. Report best + plateau. +""" +from __future__ import annotations +import sys +sys.path.insert(0, "/opt/docker/PythagorasGoal/scripts/research/skyhook") +import skyhooklib as sk +from src.strategies.skyhook import SkyhookParams + +BASE = dict(ptn_n=55, vola_lo=35.0, vola_hi=95.0, vol_lo=0.0) + +SL_GRID = [2.0, 2.25, 2.5, 2.75, 3.0, 3.5] +TP_GRID = [5.0, 6.0, 7.0, 8.0, 9.0, 10.0] + +def cell(sl, tp): + p = SkyhookParams(sl_atr=sl, tp_atr=tp, **BASE) + out = {} + for a in ("BTC", "ETH"): + r = sk.run_asset(a, p, fee_rt=sk.FEE_RT) + out[a] = r + min_full = min(out[a]["full"]["sharpe"] for a in out) + min_hold = min(out[a]["holdout"]["sharpe"] for a in out) + min_tr = min(out[a]["full"]["n_trades"] for a in out) + max_dd = max(out[a]["full"]["maxdd"] for a in out) + return dict(sl=sl, tp=tp, min_full=min_full, min_hold=min_hold, + min_tr=min_tr, max_dd=max_dd, + btc_full=out["BTC"]["full"]["sharpe"], eth_full=out["ETH"]["full"]["sharpe"], + btc_hold=out["BTC"]["holdout"]["sharpe"], eth_hold=out["ETH"]["holdout"]["sharpe"], + btc_dd=out["BTC"]["full"]["maxdd"], eth_dd=out["ETH"]["full"]["maxdd"]) + +print("=== SKH_P_RR sweep (ptn_n=55 base) — fee=0.10%RT ===") +print(f"{'sl':>5} {'tp':>5} | {'minFull':>7} {'minHold':>7} {'minTr':>5} {'maxDD':>6} | " + f"{'btcF':>5} {'ethF':>5} {'btcH':>5} {'ethH':>5} {'btcDD':>5} {'ethDD':>5}") +results = [] +for sl in SL_GRID: + for tp in TP_GRID: + if tp <= sl: # tp must exceed sl for a sensible R:R; skip degenerate + continue + c = cell(sl, tp) + results.append(c) + flag = "" + if c["min_full"] >= 0.5 and c["min_tr"] >= 20: + flag = " *" # eligible + print(f"{sl:>5} {tp:>5} | {c['min_full']:>+7.2f} {c['min_hold']:>+7.2f} " + f"{c['min_tr']:>5} {c['max_dd']*100:>5.0f}% | " + f"{c['btc_full']:>+5.2f} {c['eth_full']:>+5.2f} " + f"{c['btc_hold']:>+5.2f} {c['eth_hold']:>+5.2f} " + f"{c['btc_dd']*100:>4.0f}% {c['eth_dd']*100:>4.0f}%{flag}") + +# Eligible = minFull>=0.5, minTrades>=20. Rank by min_hold, tie-break lower maxDD. +elig = [c for c in results if c["min_full"] >= 0.5 and c["min_tr"] >= 20] +print(f"\nEligible cells (minFull>=0.5, minTr>=20): {len(elig)}") +if elig: + elig_sorted = sorted(elig, key=lambda c: (-round(c["min_hold"], 3), c["max_dd"])) + print("Top by minHold (tie-break lower maxDD):") + for c in elig_sorted[:6]: + print(f" sl={c['sl']} tp={c['tp']}: minHold={c['min_hold']:+.2f} " + f"minFull={c['min_full']:+.2f} maxDD={c['max_dd']*100:.0f}% minTr={c['min_tr']}") + best = elig_sorted[0] + # DD-cutting candidate: best minHold among cells with maxDD < V1-ish (lower DD priority) + dd_cands = sorted(elig, key=lambda c: (c["max_dd"], -round(c["min_hold"], 3))) + print("\nTop by lowest maxDD (DD-cut objective):") + for c in dd_cands[:6]: + print(f" sl={c['sl']} tp={c['tp']}: maxDD={c['max_dd']*100:.0f}% " + f"minHold={c['min_hold']:+.2f} minFull={c['min_full']:+.2f} minTr={c['min_tr']}") + + print("\n=== STUDY on best-by-minHold ===") + pbest = SkyhookParams(sl_atr=best["sl"], tp_atr=best["tp"], **BASE) + rep = sk.study(f"P_RR_sl{best['sl']}_tp{best['tp']}", pbest) + print(sk.fmt(rep)) + print("causality:", sk.causality(pbest)) + print("marginal:", {k: v for k, v in sk.marginal(pbest).items() + if k in ("corr_full","marginal_verdict","has_insample_edge","is_hedge","robust_oos")}) + try: + mg = sk.marginal(pbest) + print("marginal-full-keys:", list(mg.keys())) + print("blend w25 uplift_hold:", mg.get("blends",{}).get("w25",{}).get("uplift_hold")) + except Exception as e: + print("marginal err:", e) + print("\nAS_JSON_STUDY:", sk.as_json(rep)) +else: + print("No eligible cell — V1 base may already be at the frontier.") diff --git a/tests/test_skyhook.py b/tests/test_skyhook.py new file mode 100644 index 0000000..e3efc11 --- /dev/null +++ b/tests/test_skyhook.py @@ -0,0 +1,91 @@ +"""Test della strategia SKH01 (Skyhook) — dual-timeframe regime+breakout su BTC/ETH. + +Coprono: fedelta' al brief (ancore demo BuzVola/BuzVolume), allineamento dual-TF, assenza di +look-ahead (causalita'), e robustezza onesta del config V1 su entrambi gli asset. +""" +import sys +from pathlib import Path + +import numpy as np +import pandas as pd + +PROJECT_ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(PROJECT_ROOT)) +sys.path.insert(0, str(PROJECT_ROOT / "scripts" / "research" / "skyhook")) + +from src.data.downloader import load_data +from src.strategies.skyhook import ( + HTF_MIN, LTF_MIN, SkyhookParams, build_frames, chande01, skyhook_entries) + +# config V1 (vincente del lever-scout/grid; vedi diario 2026-06-23-skyhook) +V1 = dict(ptn_n=55, sl_atr=2.5, tp_atr=6.0, vola_lo=35, vola_hi=95, vol_lo=0.0) + + +# --------------------------------------------------------------------------- +# Fedelta' al brief: indicatori tipo-Chande, normalizzati 0-100. +# --------------------------------------------------------------------------- +def test_chande01_anchors(): + """Semantica del brief: volatilita'/volume STEADY -> 50 (neutro); in RAMPA -> 100; in CALO -> 0.""" + n = 100 + assert abs(chande01(np.full(n, 7.0), 13)[-1] - 50.0) < 1e-9 # costante -> neutro + assert abs(chande01(np.arange(n, dtype=float), 13)[-1] - 100.0) < 1e-9 # rampa su -> 100 + assert abs(chande01(np.arange(n, 0, -1, dtype=float), 13)[-1] - 0.0) < 1e-9 # rampa giu' -> 0 + + +def test_demo_buzvola_buzvolume(): + """Ancore della demo: ATR costante (vol steady) -> BuzVola 50; volume in rampa -> BuzVolume 100.""" + n = 100 + buz_vola = chande01(np.full(n, 2.0), 13) # ATR steady + buz_volume = chande01(np.linspace(1000, 5000, n), 13) # volume in rampa + assert abs(buz_vola[-1] - 50.0) < 1e-9 + assert abs(buz_volume[-1] - 100.0) < 1e-9 + # oscillatori sempre in [0,100] + assert chande01(np.random.default_rng(0).normal(size=500).cumsum() + 100, 13)[20:].min() >= -1e-9 + assert chande01(np.random.default_rng(1).normal(size=500).cumsum() + 100, 13)[20:].max() <= 100 + 1e-9 + + +# --------------------------------------------------------------------------- +# Allineamento dual-timeframe: 690 = 3 x 230, confini HTF subset dei confini LTF. +# --------------------------------------------------------------------------- +def test_dual_tf_alignment(): + assert HTF_MIN == 3 * LTF_MIN + ltf, htf = build_frames(load_data("BTC", "5m")) + # ogni timestamp (open) HTF e' anche un open LTF (stessa griglia epoch) + ltf_opens = set(ltf["timestamp"].astype("int64").tolist()) + htf_opens = htf["timestamp"].astype("int64").tolist() + inside = sum(t in ltf_opens for t in htf_opens) + assert inside / len(htf_opens) > 0.99, "i confini HTF devono essere un sottoinsieme dei confini LTF" + + +# --------------------------------------------------------------------------- +# Causalita': gli ingressi su un prefisso devono coincidere con la run completa. +# --------------------------------------------------------------------------- +def test_no_lookahead_entries(): + p = SkyhookParams(**V1) + ltf, htf = build_frames(load_data("BTC", "5m")) + full = skyhook_entries(ltf, htf, p) + n = len(ltf) + cut = int(n * 0.85) + cut_ts = int(ltf["timestamp"].iloc[cut - 1]) + htf_cut = htf[htf["timestamp"] <= cut_ts].reset_index(drop=True) + sub = skyhook_entries(ltf.iloc[:cut].reset_index(drop=True), htf_cut, p) + for i in range(cut - 200, cut): + a, b = full[i], sub[i] + assert (a is None) == (b is None) + if a is not None: + assert a["dir"] == b["dir"] + assert abs(a["sl"] - b["sl"]) < 1e-6 and abs(a["tp"] - b["tp"]) < 1e-6 + + +# --------------------------------------------------------------------------- +# Robustezza onesta del config V1: PASS su BTC E ETH, netto fee, OOS. +# --------------------------------------------------------------------------- +def test_v1_robust_both_assets(): + import skyhooklib as sk + p = SkyhookParams(**V1) + for a in ("BTC", "ETH"): + r = sk.run_asset(a, p, sk.FEE_RT) + assert r["full"]["sharpe"] >= 0.5, f"{a} FULL Sharpe basso: {r['full']['sharpe']}" + assert r["holdout"]["sharpe"] >= 0.2, f"{a} HOLD-OUT Sharpe basso: {r['holdout']['sharpe']}" + assert r["full"]["n_trades"] >= 20, f"{a} troppo pochi trade: {r['full']['n_trades']}" + assert sk.causality(p, "BTC")["ok"] and sk.causality(p, "ETH")["ok"]