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

feat(skyhook): SKH01-V2-DD — asymmetric %-exits cut standalone DD <30% (2-wave agent research)

Browse Source 
Second agent wave (skyhook-improve-v2, 14 DD-reduction families, each adversarially
verified by 2 skeptics) beats the prior winner on the only unmet goal (DD<30%).

Winner = ASYM_LS -> promoted to engine as SKH01_V2_DD:
  same signal (ptn_n=45, vola[35,95], vol_lo=0, exit-bars 24/16) but exits switched
  from ATR to FIXED-PCT ASYMMETRIC — long sl4%/tp10%, short sl2%(tighter)/tp8%.
  The tight short %-SL caps the per-trade loss that forms the maxDD in vol spikes.

Verified (sk.study, independent re-run): standalone maxDD BTC 21.4% / ETH 27.4% (<30%),
minFull +0.99, minHold +1.26, causality 0/400 both assets, fee-surviving to 0.40%RT,
marginal vs TP01 ADDS (corr 0.09, in-sample edge, robust_oos, multicut, clean-year +0.57),
blend 0.75*TP01+0.25*SKH uplift_hold +0.87; blend 50/50 full 1.84/hold 1.59/DD 10.7%.
Plateau (not knife-edge); both skeptics holds_up=high, killer=null.

Engine: per-direction short exit overrides (exit_mode_short/sl_*_short/tp_*_short),
backward-compatible (None -> symmetric, V1/intermediate-winner unchanged). +3 tests (8/8 pass).

Lessons: DD is cut by changing the exit MECHANISM (%-SL, L/S asymmetry, ensembles), NOT by
entry-only kill-switch / vol-target / cadence. PATTERN_CONF killed as overfit (knife-edge).
PCTL_DD unverified (rate-limit) and ENS_PARAM/TPSL_DD recency/hedge-loaded -> forward-monitor.
NOT yet wired to live sleeves: re-verify blend@0.25 + causality on execution code before deploy.

Includes both waves' research scripts (runs/SKH_* wave 1, runs/SKH2_* wave 2).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Adriano Dal Pastro
2026-06-23 16:10:38 +00:00
parent 8e46a62e67
commit de72e3ce1f
31 changed files with 5768 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/research/skyhook/runs/SKH2_KELTNER_PTN.py
+277
View File
@@ -0,0 +1,277 @@
"""SKH2_KELTNER_PTN — KELTNER/ATR-channel breakout pattern (replaces Donchian).
FAMILY: KELTNER_PTN. Goal of this wave = CUT standalone maxDD below 30% while keeping
hold-out Sharpe high and earns_slot True.
Idea: the V1/V2 Skyhook pattern is a Donchian breakout (close > rolling-high of n bars).
Donchian highs/lows are driven by single wicks -> a fast spike can set a fresh extreme that
the next close pokes through, firing a false breakout that mean-reverts -> drawdown. An
ATR-CHANNEL (Keltner) breakout instead requires close to clear EMA(n) +/- k*ATR(n), a
SMOOTHED reference that ignores isolated wicks. Steadier reference -> fewer wick-driven false
entries -> potentially lower DD for similar exposure.
We keep EVERYTHING ELSE identical to the verified V2 winner (regime Chande01 bands
vola_lo=35/vola_hi=95/vol_lo=0, exits sl_atr=2.5/tp_atr=7.0/uscitalong=24/uscitashort=16) and
ONLY swap the pattern from Donchian to Keltner. We do this by monkeypatching S.htf_features
inside skyhooklib's namespace (same safe technique as SKH_R_EXPAND_study.py) so sk.study /
sk.causality / sk.marginal run the EXACT honest machinery unchanged.
CAUSALITY: EMA and ATR are causal ewm (use x[0..i] inclusive of the current, already-closed
HTF bar); the channel for breakout-comparison is shift(1) (strictly prior bar's channel) so
close[i] is compared against a band known BEFORE bar i closes -> leak-free. We verify with
sk.causality (truncated-prefix guard) on BOTH assets.
"""
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
from src.strategies import skyhook as S
from src.strategies.skyhook import SkyhookParams, HTF_MIN
ORIG_FEAT = S.htf_features
FEE = sk.FEE_RT
# ---------------------------------------------------------------------------
# Keltner channel breakout on HTF (causal, shift-safe).
# mid = EMA(close, n)
# width = k * ATR(n) (ATR over the same n window, ewm)
# upper = mid + width ; lower = mid - width
# ptn_long = close[i] > upper[i-1] (clears the PRIOR bar's upper channel)
# ptn_short = close[i] < lower[i-1]
# The shift(1) on the channel makes the comparison strictly causal: the band the close must
# clear is fully determined by bars <= i-1 (Donchian uses shift(1) on the rolling extreme for
# the same reason). EMA/ATR ewm themselves use only past+current data.
# ---------------------------------------------------------------------------
def keltner_breakout(htf: pd.DataFrame, n: int, k: float, atr_win: int) -> tuple[np.ndarray, np.ndarray]:
c = htf["close"].values.astype(float)
mid = pd.Series(c).ewm(span=n, adjust=False, min_periods=n).mean().values
a = S.atr(htf, atr_win)
upper = mid + k * a
lower = mid - k * a
# compare current close vs the PRIOR bar's channel (shift 1) -> strictly causal
upper_prev = pd.Series(upper).shift(1).values
lower_prev = pd.Series(lower).shift(1).values
ptn_long = np.where(np.isfinite(upper_prev), c > upper_prev, False)
ptn_short = np.where(np.isfinite(lower_prev), c < lower_prev, False)
return ptn_long.astype(bool), ptn_short.astype(bool)
def make_keltner_features(n: int, k: float, kelt_atr_win: int):
"""Return an htf_features replacement: V1 Chande01 regime + Keltner pattern."""
def _feat(htf: pd.DataFrame, p: SkyhookParams) -> pd.DataFrame:
buz_vola = S.chande01(S.atr(htf, p.atr_win), p.n_vola)
buz_volume = S.chande01(htf["volume"].values, p.n_volume)
ptn_long, ptn_short = keltner_breakout(htf, n, k, kelt_atr_win)
regime_ok = ((buz_vola >= p.vola_lo) & (buz_vola <= p.vola_hi)
& (buz_volume >= p.vol_lo) & (buz_volume <= p.vol_hi))
comp_long = regime_ok & ptn_long
comp_short = regime_ok & ptn_short
if p.long_only:
comp_short = np.zeros_like(comp_short, dtype=bool)
close_ts = htf["timestamp"].astype("int64").values + HTF_MIN * 60 * 1000
return pd.DataFrame({
"close_ts": close_ts,
"buz_vola": buz_vola, "buz_volume": buz_volume,
"comp_long": comp_long.astype(bool), "comp_short": comp_short.astype(bool),
})
return _feat
def study_keltner(name, p, n, k, kelt_atr_win):
"""sk.study + causality + marginal with htf_features patched to Keltner."""
S.htf_features = make_keltner_features(n, k, kelt_atr_win)
try:
rep = sk.study(name, p)
caus = sk.causality(p, "BTC")
caus_eth = sk.causality(p, "ETH")
marg = sk.marginal(p)
finally:
S.htf_features = ORIG_FEAT
return rep, (caus, caus_eth), marg
def earns_slot(rep, marg):
grade_ok = rep["verdict"]["grade"] != "FAIL"
return bool(grade_ok and marg.get("marginal_verdict") == "ADDS"
and marg.get("robust_oos") is True and marg.get("is_hedge") is False)
if __name__ == "__main__":
# Winner exits/regime (the verified V2 winner) — only the pattern changes to Keltner.
p = SkyhookParams(sl_atr=2.5, tp_atr=7.0, uscitalong=24, uscitashort=16,
vola_lo=35.0, vola_hi=95.0, vol_lo=0.0)
# n in {13,20,34}, k in {1.0,1.5,2.0}; ATR window for the channel width = winner default 14.
grid = []
for n in (13, 20, 34):
for k in (1.0, 1.5, 2.0):
grid.append((n, k))
KELT_ATR = 14
print("=== SKH2_KELTNER_PTN: ATR-channel (Keltner) breakout sweep (regime+exits = V2 winner) ===\n")
print(f"grid n x k = {grid} kelt_atr_win={KELT_ATR}\n")
# ---- Sweep (cheap pass: FULL/HOLD/DD/trades + fee survival via study) ----
rows = []
for (n, k) in grid:
tag = f"KELT_n{n}_k{k}"
rep, (cb, ce), marg = study_keltner(tag, p, n, k, KELT_ATR)
v = rep["verdict"]
# standalone DD = max over BTC&ETH FULL maxdd
dd = max(rep["per_asset"][a]["full"]["maxdd"] for a in rep["per_asset"])
es = earns_slot(rep, marg)
w25 = marg.get("blends", {}).get("w25", {}).get("uplift_hold")
rows.append(dict(tag=tag, n=n, k=k, rep=rep, caus=(cb, ce), marg=marg,
minFull=v["min_asset_full_sharpe"], minHold=v["min_asset_holdout_sharpe"],
minTr=v["min_trades"], feeOK=v["fee_survives"], grade=v["grade"],
dd=dd, es=es, w25=w25, caus_ok=(cb["ok"] and ce["ok"])))
beats = (es and dd < 0.30 and (w25 is not None and w25 >= 0.55) and v["min_asset_holdout_sharpe"] >= 0.65)
print(f" [{tag}] grade={v['grade']} minFull={v['min_asset_full_sharpe']:+.2f}"
f" minHold={v['min_asset_holdout_sharpe']:+.2f} minTr={v['min_trades']}"
f" DD={dd*100:.0f}% feeOK={v['fee_survives']} caus={cb['ok'] and ce['ok']}"
f" | verdict={marg.get('marginal_verdict')} corr={marg.get('corr_full')}"
f" w25={w25} robust={marg.get('robust_oos')} hedge={marg.get('is_hedge')}"
f" earns_slot={es} BEATS={beats}")
# =======================================================================
# REFINEMENT PASS: the plain swap keeps DD>30% (too many entries / wick pokes).
# DD is driven by (a) wide vola band letting in blow-off breakouts, (b) loose SL,
# (c) shorts bleeding in a structural bull. Sweep regime-tightening + SL + long_only
# around the best earns_slot region (n13/n20, k1.5-2.0) to push DD under 30%.
# =======================================================================
print("\n--- REFINEMENT: tighten regime / SL / long_only to cut DD<30% ---")
refine = [
# (n, k, sl_atr, tp_atr, vola_lo, vola_hi, vol_lo, long_only, tag)
(13, 2.0, 2.0, 7.0, 35.0, 95.0, 0.0, False, "n13k2_sl2.0"),
(13, 2.0, 2.5, 7.0, 45.0, 90.0, 0.0, False, "n13k2_vola45-90"),
(13, 2.0, 2.5, 7.0, 35.0, 85.0, 0.0, False, "n13k2_volaHi85"),
(13, 2.0, 2.5, 7.0, 35.0, 95.0, 40.0, False, "n13k2_volFloor40"),
(13, 2.0, 2.5, 7.0, 35.0, 95.0, 0.0, True, "n13k2_longOnly"),
(13, 2.0, 2.0, 7.0, 45.0, 90.0, 40.0, False, "n13k2_tight_all"),
(20, 2.0, 2.0, 7.0, 35.0, 95.0, 0.0, False, "n20k2_sl2.0"),
(20, 2.0, 2.5, 7.0, 45.0, 90.0, 40.0, False, "n20k2_tight_all"),
(13, 2.5, 2.5, 7.0, 35.0, 95.0, 0.0, False, "n13k2.5"),
(13, 2.5, 2.0, 7.0, 45.0, 90.0, 0.0, False, "n13k2.5_sl2_vola45-90"),
# ---- pass 3: sl2.0 was the DD/hold winner; push SL tighter + lower TP (cut tail) ----
(13, 2.0, 1.5, 6.0, 35.0, 95.0, 0.0, False, "n13k2_sl1.5_tp6"),
(13, 2.0, 1.5, 7.0, 35.0, 95.0, 40.0, False, "n13k2_sl1.5_volFloor40"),
(13, 2.0, 2.0, 5.0, 35.0, 95.0, 0.0, False, "n13k2_sl2_tp5"),
(13, 2.0, 2.0, 6.0, 35.0, 95.0, 40.0, False, "n13k2_sl2_tp6_volFloor40"),
(13, 2.0, 1.5, 5.0, 35.0, 95.0, 0.0, False, "n13k2_sl1.5_tp5"),
(13, 1.5, 2.0, 7.0, 35.0, 95.0, 0.0, False, "n13k1.5_sl2.0"),
]
for (n, k, sl, tp, vlo, vhi, vol_lo, lo, tag) in refine:
pr = SkyhookParams(sl_atr=sl, tp_atr=tp, uscitalong=24, uscitashort=16,
vola_lo=vlo, vola_hi=vhi, vol_lo=vol_lo, long_only=lo)
rep, (cb, ce), marg = study_keltner(tag, pr, n, k, KELT_ATR)
v = rep["verdict"]
dd = max(rep["per_asset"][a]["full"]["maxdd"] for a in rep["per_asset"])
es = earns_slot(rep, marg)
w25 = marg.get("blends", {}).get("w25", {}).get("uplift_hold")
rows.append(dict(tag=tag, n=n, k=k, sl=sl, tp=tp, vlo=vlo, vhi=vhi, vol_lo=vol_lo, lo=lo,
rep=rep, caus=(cb, ce), marg=marg,
minFull=v["min_asset_full_sharpe"], minHold=v["min_asset_holdout_sharpe"],
minTr=v["min_trades"], feeOK=v["fee_survives"], grade=v["grade"],
dd=dd, es=es, w25=w25, caus_ok=(cb["ok"] and ce["ok"])))
b2 = (es and dd < 0.30 and (w25 is not None and w25 >= 0.55) and v["min_asset_holdout_sharpe"] >= 0.65)
print(f" [{tag}] grade={v['grade']} minFull={v['min_asset_full_sharpe']:+.2f}"
f" minHold={v['min_asset_holdout_sharpe']:+.2f} minTr={v['min_trades']}"
f" DD={dd*100:.0f}% feeOK={v['fee_survives']} caus={cb['ok'] and ce['ok']}"
f" | verdict={marg.get('marginal_verdict')} w25={w25} robust={marg.get('robust_oos')}"
f" hedge={marg.get('is_hedge')} earns_slot={es} BEATS={b2}")
# ---- Pick best: prefer DD<30% with earns_slot, then by (minHold, then w25) ----
def beats_winner(r):
return bool(r["es"] and r["dd"] < 0.30
and (r["w25"] is not None and r["w25"] >= 0.55)
and r["minHold"] >= 0.65)
winners = [r for r in rows if beats_winner(r)]
if winners:
best = max(winners, key=lambda r: (r["minHold"], r["w25"] or -9))
pool = "BEATS-WINNER"
else:
# objective priority: DD<30 + earns_slot first; else best DD among earns_slot;
# else best DD among fee-surviving non-FAIL; else lowest DD overall.
cand1 = [r for r in rows if r["dd"] < 0.30 and r["es"]]
# secondary-quality: earns_slot AND meets the two NON-DD beats gates (w25>=0.55, minHold>=0.65)
candQ = [r for r in rows if r["es"] and (r["w25"] is not None and r["w25"] >= 0.55)
and r["minHold"] >= 0.65]
cand2 = [r for r in rows if r["es"]]
cand3 = [r for r in rows if r["grade"] != "FAIL" and r["feeOK"]]
if cand1:
best = max(cand1, key=lambda r: (r["minHold"], r["w25"] or -9)); pool = "DD<30+earns_slot"
elif candQ:
# best DD among configs that already clear the other two beats gates
best = min(candQ, key=lambda r: r["dd"]); pool = "earns_slot+w25>=.55+minHold>=.65 (DD>=30)"
elif cand2:
best = min(cand2, key=lambda r: r["dd"]); pool = "earns_slot (DD>=30)"
elif cand3:
best = min(cand3, key=lambda r: r["dd"]); pool = "fee-surviving non-FAIL"
else:
best = min(rows, key=lambda r: r["dd"]); pool = "lowest-DD overall"
rep, marg = best["rep"], best["marg"]
cb, ce = best["caus"]
v = rep["verdict"]
bl = marg.get("blends", {})
w25 = bl.get("w25", {})
w50 = bl.get("w50", {})
print("\n" + "=" * 78)
print(f"BEST CONFIG ({pool}): {best['tag']} (n={best['n']}, k={best['k']}, kelt_atr_win={KELT_ATR})")
print("=" * 78)
print(sk.fmt(rep))
print(f"\nstandalone max_dd (max BTC&ETH FULL) = {best['dd']:.4f} ({best['dd']*100:.1f}%)")
print(f"causality BTC={cb} ETH={ce} -> ok={cb['ok'] and ce['ok']}")
print(f"minFull={v['min_asset_full_sharpe']:+.3f} minHold={v['min_asset_holdout_sharpe']:+.3f}"
f" minTrades={v['min_trades']} fee_survives_0.30%={v['fee_survives']}")
print("\n--- MARGINAL vs TP01 ---")
print(f" marginal_verdict = {marg.get('marginal_verdict')}")
print(f" corr_full = {marg.get('corr_full')}")
print(f" corr_hold = {marg.get('corr_hold')}")
print(f" has_insample_edge = {marg.get('has_insample_edge')}")
print(f" is_hedge = {marg.get('is_hedge')}")
print(f" robust_oos = {marg.get('robust_oos')}")
print(f" multicut_persistent= {marg.get('multicut_persistent')}")
print(f" clean_year_uplift = {marg.get('clean_year_uplift')}")
print(f" jackknife_min_uplift= {marg.get('jackknife_min_uplift')}")
print(f" multicut_uplift = {marg.get('multicut_uplift')}")
print(f" cand_insample_sharpe= {marg.get('cand_insample_sharpe')}")
print(f" blend w25 uplift_hold = {w25.get('uplift_hold')} uplift_full={w25.get('uplift_full')}")
print(f" blend w50 = full={w50.get('full')} hold={w50.get('hold')}"
f" uplift_hold={w50.get('uplift_hold')} dd={w50.get('dd')}")
es = best["es"]
beats = beats_winner(best)
print(f"\n earns_slot = {es}")
print(f" BEATS_WINNER = {beats} "
f"(need: earns_slot AND max_dd<0.30 AND w25_uplift_hold>=0.55 AND minHold>=0.65)")
# ---- machine-readable final line for the orchestrator/agent to parse ----
import json
out = dict(
family="KELTNER_PTN", tag=best["tag"],
best_config=dict(ptn_kind="keltner", n=best["n"], k=best["k"], kelt_atr_win=KELT_ATR,
sl_atr=best.get("sl", 2.5), tp_atr=best.get("tp", 7.0),
uscitalong=24, uscitashort=16,
vola_lo=best.get("vlo", 35.0), vola_hi=best.get("vhi", 95.0),
vol_lo=best.get("vol_lo", 0.0), long_only=best.get("lo", False)),
min_full_sharpe=v["min_asset_full_sharpe"], min_hold_sharpe=v["min_asset_holdout_sharpe"],
max_dd=best["dd"], n_trades_min=v["min_trades"], fee_survives_030=bool(v["fee_survives"]),
causality_ok=bool(cb["ok"] and ce["ok"]),
marginal_verdict=marg.get("marginal_verdict"),
has_insample_edge=bool(marg.get("has_insample_edge")),
is_hedge=bool(marg.get("is_hedge")), robust_oos=bool(marg.get("robust_oos")),
multicut_persistent=bool(marg.get("multicut_persistent")),
clean_year_uplift=marg.get("clean_year_uplift"), corr_full=marg.get("corr_full"),
blend_w25_uplift_hold=w25.get("uplift_hold"),
earns_slot=bool(es), beats_winner=bool(beats),
)
print("\nFINAL_JSON=" + json.dumps(out, default=str))