Adriano/PythagorasGoal
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

test(skyhook): demo anchors + dual-TF alignment + causality + V1 robustness (5 pass)

Browse Source 
Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Adriano Dal Pastro
2026-06-23 14:46:47 +00:00
parent 2d8faf3896
commit c7c07f4c35
5 changed files with 299 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/research/skyhook/check_v1.py
+13
View File
@@ -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)))
scripts/research/skyhook/grid.py
+26
View File
@@ -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}")
scripts/research/skyhook/runs/SKH_P_PTN.py
+82
View File
@@ -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)
scripts/research/skyhook/runs/SKH_P_RR.py
+87
View File
@@ -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.")
tests/test_skyhook.py
+91
View File
@@ -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"]