feat(skyhook): SKH01 dual-TF regime+breakout engine + honest eval harness
Porting onesto del sistema ES Skyhook su BTC/ETH certificati: - src/strategies/skyhook.py: 690m(segnale)+230m(exec) da 5m; BuzVola/BuzVolume Chande 0-100 (ancore demo verificate); Donchian breakout HTF; regime gate; composer; entries asimmetrici (uscitalong/short + stop/profit ATR) per backtest_signals. - scripts/research/skyhook/skyhooklib.py: study (FULL/HOLD/fee-sweep/per-anno BTCÐ), causality guard (0 mismatch), marginal-vs-TP01. Baseline: BTC FULL Sh +0.91/+581%, ETH +0.64/+255%, fee-surviving, ma HOLD-OUT debole -> da migliorare. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Adriano Dal Pastro
@@ -0,0 +1,217 @@
|
||||
"""skyhooklib — SHARED HONEST EVAL for the Skyhook (SKH01) multi-agent improvement wave.
|
||||
|
||||
Every agent imports THIS so results are comparable and leak-free:
|
||||
* data builders: certified 5m BTC/ETH -> 230m (exec) + 690m (signal), cached.
|
||||
* study(): FULL + HOLD-OUT (2025-01-01+) + fee sweep + per-year, on BOTH assets, via the
|
||||
project's honest intrabar engine (backtest_signals: TP/SL/max_bars, non-overlap).
|
||||
* causality(): truncated-prefix guard (a Skyhook entry on a prefix must match the full run).
|
||||
* marginal(): does Skyhook ADD to the existing TP01 portfolio? (altlib.marginal_vs_tp01).
|
||||
* verdict(): conservative PASS/WEAK/FAIL on min-asset FULL & HOLD-OUT + fee survival.
|
||||
|
||||
Quick start (inside an agent script):
|
||||
import sys; sys.path.insert(0, "/opt/docker/PythagorasGoal/scripts/research/skyhook")
|
||||
import skyhooklib as sk
|
||||
from src.strategies.skyhook import SkyhookParams
|
||||
rep = sk.study("MY-VARIANT", SkyhookParams(ptn_n=20, sl_atr=2.5))
|
||||
print(sk.fmt(rep)); print(sk.as_json(rep))
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import sys
|
||||
from functools import lru_cache
|
||||
from pathlib import Path
|
||||
|
||||
import numpy as np
|
||||
import pandas as pd
|
||||
|
||||
_ROOT = Path(__file__).resolve().parents[3]
|
||||
if str(_ROOT) not in sys.path:
|
||||
sys.path.insert(0, str(_ROOT))
|
||||
sys.path.insert(0, str(_ROOT / "scripts" / "research" / "alt"))
|
||||
|
||||
from src.backtest.harness import backtest_signals # noqa: E402
|
||||
from src.data.downloader import load_data # noqa: E402
|
||||
from src.strategies.skyhook import ( # noqa: E402
|
||||
SkyhookParams, build_frames, skyhook_entries, signal_counts)
|
||||
|
||||
HOLDOUT = pd.Timestamp("2025-01-01", tz="UTC")
|
||||
FEE_RT = 0.001 # 0.10% round-trip (Deribit taker)
|
||||
FEE_SWEEP = (0.0, 0.001, 0.002, 0.003) # round-trip fee grid
|
||||
CERTIFIED = ("BTC", "ETH")
|
||||
|
||||
|
||||
@lru_cache(maxsize=4)
|
||||
def _frames(asset: str):
|
||||
return build_frames(load_data(asset, "5m"))
|
||||
|
||||
|
||||
def frames(asset: str):
|
||||
"""(ltf 230m, htf 690m) certificati e cached."""
|
||||
return _frames(asset.upper())
|
||||
|
||||
|
||||
def _split_metrics(eq: np.ndarray, idx: pd.DatetimeIndex, mask: np.ndarray) -> dict:
|
||||
e = eq[mask]
|
||||
if len(e) < 5:
|
||||
return dict(sharpe=0.0, ret=0.0, maxdd=0.0, n=int(len(e)))
|
||||
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
|
||||
sharpe = 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(sharpe, 3), ret=round(float(e[-1] / e[0] - 1), 4),
|
||||
maxdd=round(dd, 4), n=int(len(e)))
|
||||
|
||||
|
||||
def run_asset(asset: str, p: SkyhookParams, fee_rt: float = FEE_RT) -> dict:
|
||||
"""Backtest Skyhook su un asset (230m exec). Ritorna FULL+HOLDOUT+per-anno+diagnostica."""
|
||||
ltf, htf = frames(asset)
|
||||
entries = skyhook_entries(ltf, htf, p)
|
||||
m = backtest_signals(ltf, entries, fee_rt=fee_rt, leverage=1.0, asset=asset, 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), cagr=round(m.cagr, 4), maxdd=round(m.max_dd, 4),
|
||||
ret=round(m.net_return, 4), n_trades=int(m.n_trades), win_rate=round(m.win_rate, 1))
|
||||
hold = _split_metrics(eq, idx, hmask)
|
||||
counts = signal_counts(ltf, htf, p)
|
||||
return dict(asset=asset, full=full, holdout=hold,
|
||||
yearly={int(y): round(v, 4) for y, v in m.yearly.items()},
|
||||
counts=counts, _eq=eq, _idx=idx)
|
||||
|
||||
|
||||
def daily_returns(asset: str, p: SkyhookParams, fee_rt: float = FEE_RT) -> pd.Series:
|
||||
"""Rendimenti GIORNALIERI dell'equity Skyhook (per il lens marginal-vs-TP01).
|
||||
NB approssimazione: l'equity di backtest_signals e' marcata a fine-trade (a gradini),
|
||||
quindi i daily sono grezzi -> usalo SOLO per corr/uplift, non come headline Sharpe."""
|
||||
r = run_asset(asset, p, fee_rt)
|
||||
s = pd.Series(r["_eq"], index=r["_idx"])
|
||||
return (s.resample("1D").last().ffill().pct_change().dropna())
|
||||
|
||||
|
||||
def skyhook_daily_5050(p: SkyhookParams, fee_rt: float = FEE_RT) -> pd.Series:
|
||||
"""Serie giornaliera 50/50 BTC+ETH (stessa convenzione di altlib.tp01_baseline_daily)."""
|
||||
series = {a: daily_returns(a, p, fee_rt) for a in CERTIFIED}
|
||||
J = pd.concat(series, axis=1, join="inner").fillna(0.0)
|
||||
return 0.5 * J[CERTIFIED[0]] + 0.5 * J[CERTIFIED[1]]
|
||||
|
||||
|
||||
def marginal(p: SkyhookParams, fee_rt: float = FEE_RT) -> dict:
|
||||
"""Skyhook MIGLIORA il portafoglio TP01 esistente? (altlib.marginal_vs_tp01)."""
|
||||
import altlib as al
|
||||
return al.marginal_vs_tp01(skyhook_daily_5050(p, fee_rt))
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Causality guard (truncated-prefix): un ingresso emesso su un prefisso deve coincidere
|
||||
# con lo stesso indice della run completa (nessuna feature guarda il futuro).
|
||||
# ---------------------------------------------------------------------------
|
||||
def causality(p: SkyhookParams, asset: str = "BTC", tail: int = 200) -> dict:
|
||||
ltf, htf = frames(asset)
|
||||
full = skyhook_entries(ltf, htf, p)
|
||||
n = len(ltf)
|
||||
bad = 0; checked = 0
|
||||
for frac in (0.80, 0.92):
|
||||
cut = int(n * frac)
|
||||
# taglia anche l'HTF alla stessa data di chiusura del prefisso LTF
|
||||
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(max(0, cut - tail), cut):
|
||||
checked += 1
|
||||
a, b = full[i], sub[i]
|
||||
if (a is None) != (b is None):
|
||||
bad += 1
|
||||
elif a is not None and (a["dir"] != b["dir"]
|
||||
or abs(a["sl"] - b["sl"]) > 1e-6
|
||||
or abs(a["tp"] - b["tp"]) > 1e-6):
|
||||
bad += 1
|
||||
return dict(ok=bool(bad == 0), mismatches=int(bad), checked=int(checked))
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Verdict + drivers
|
||||
# ---------------------------------------------------------------------------
|
||||
def _verdict(per_asset: dict, fee_survives: bool) -> dict:
|
||||
min_full = min(per_asset[a]["full"]["sharpe"] for a in per_asset)
|
||||
min_hold = min(per_asset[a]["holdout"]["sharpe"] for a in per_asset)
|
||||
min_trades = min(per_asset[a]["full"]["n_trades"] for a in per_asset)
|
||||
enough = min_trades >= 20
|
||||
pass_ = enough and min_full >= 0.5 and min_hold >= 0.2 and fee_survives
|
||||
weak = enough and min_full >= 0.3 and min_hold >= 0.0
|
||||
grade = "PASS" if pass_ else ("WEAK" if weak else "FAIL")
|
||||
return dict(grade=grade, min_asset_full_sharpe=round(min_full, 3),
|
||||
min_asset_holdout_sharpe=round(min_hold, 3),
|
||||
min_trades=int(min_trades), fee_survives=bool(fee_survives))
|
||||
|
||||
|
||||
def study(name: str, p: SkyhookParams | None = None, assets=CERTIFIED,
|
||||
fee_sweep=FEE_SWEEP) -> dict:
|
||||
"""Run completo: FULL+HOLDOUT+fee-sweep+per-anno su BTCÐ + verdict conservativo."""
|
||||
p = p or SkyhookParams()
|
||||
per_asset = {}
|
||||
fee_ok_all = True
|
||||
for a in assets:
|
||||
r = run_asset(a, p, FEE_RT)
|
||||
sweep = {}
|
||||
for f in fee_sweep:
|
||||
rf = run_asset(a, p, f)
|
||||
sweep[f"{f*100:.2f}%RT"] = rf["full"]["sharpe"]
|
||||
fee_ok = sweep.get("0.30%RT", -9) > 0
|
||||
fee_ok_all = fee_ok_all and fee_ok
|
||||
per_asset[a] = dict(full=r["full"], holdout=r["holdout"], yearly=r["yearly"],
|
||||
counts=r["counts"], fee_sweep=sweep)
|
||||
return dict(name=name, params=_params_dict(p), per_asset=per_asset,
|
||||
verdict=_verdict(per_asset, fee_ok_all))
|
||||
|
||||
|
||||
def _params_dict(p: SkyhookParams) -> dict:
|
||||
return {k: getattr(p, k) for k in p.__dataclass_fields__}
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Output
|
||||
# ---------------------------------------------------------------------------
|
||||
def _clean(o):
|
||||
if isinstance(o, dict):
|
||||
return {k: _clean(v) for k, v in o.items() if not k.startswith("_")}
|
||||
if isinstance(o, (list, tuple)):
|
||||
return [_clean(x) for x in o]
|
||||
if isinstance(o, (np.floating,)):
|
||||
return round(float(o), 4)
|
||||
if isinstance(o, (np.integer,)):
|
||||
return int(o)
|
||||
if isinstance(o, (np.bool_,)):
|
||||
return bool(o)
|
||||
return o
|
||||
|
||||
|
||||
def as_json(rep: dict) -> str:
|
||||
return json.dumps(_clean(rep), default=str)
|
||||
|
||||
|
||||
def fmt(rep: dict) -> str:
|
||||
v = rep["verdict"]
|
||||
lines = [f"=== {rep['name']} -> {v['grade']} "
|
||||
f"(minFull={v['min_asset_full_sharpe']:+.2f} minHold={v['min_asset_holdout_sharpe']:+.2f} "
|
||||
f"minTrades={v['min_trades']} feeOK={v['fee_survives']})"]
|
||||
for a, pa in rep["per_asset"].items():
|
||||
f, h, c = pa["full"], pa["holdout"], pa["counts"]
|
||||
yr = " ".join(f"{y}:{r*100:+.0f}%" for y, r in pa["yearly"].items())
|
||||
lines.append(f" {a}: FULL Sh={f['sharpe']:+.2f} ret={f['ret']*100:+.0f}% DD={f['maxdd']*100:.0f}% "
|
||||
f"n={f['n_trades']} wr={f['win_rate']:.0f}% HOLD Sh={h['sharpe']:+.2f} ret={h['ret']*100:+.0f}% "
|
||||
f"| entries={c['entries']} (L{c['comp_long']}/S{c['comp_short']})")
|
||||
lines.append(f" fee sweep: " + " ".join(f"{k}={val:+.2f}" for k, val in pa["fee_sweep"].items()))
|
||||
lines.append(f" per-anno: {yr}")
|
||||
return "\n".join(lines)
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
print("--- SMOKE skyhooklib: baseline SkyhookParams() ---")
|
||||
rep = study("SKH01-BASELINE", SkyhookParams())
|
||||
print(fmt(rep))
|
||||
print("\ncausality:", causality(SkyhookParams()))
|
||||
@@ -0,0 +1,233 @@
|
||||
"""SKYHOOK (SKH01) — dual-timeframe regime+breakout system, ported to BTC/ETH (2026-06-23).
|
||||
|
||||
NON e' un trend-follower: entra SOLO quando coincidono (a) un REGIME di volatilita'/volume e
|
||||
(b) un PATTERN di breakout/momentum. Porting onesto su BTC/ETH certificati (Deribit mainnet)
|
||||
di un sistema ES (E-mini S&P) genetico a doppio timeframe.
|
||||
|
||||
Architettura (dal brief):
|
||||
* data2 = HTF 690 min (genera il SEGNALE: regime + pattern)
|
||||
* data1 = LTF 230 min (ESEGUE: ingressi/uscite) NB 690 = 3 x 230 (HTF = 3x LTF)
|
||||
Entrambi resampled dal feed 5m certificato con origin='epoch' -> i confini 690 sono un
|
||||
SOTTOINSIEME dei confini 230, quindi una barra HTF chiude esattamente su una chiusura LTF.
|
||||
|
||||
Pipeline per barra (evaluate_bar): barre -> indicatori -> fasce regime -> pattern -> composer
|
||||
-> ingresso/uscita -> SkyhookDecision
|
||||
1. INDICATORI (sul HTF, tipo-Chande, normalizzati 0-100):
|
||||
BuzVola = chande01(ATR) -> dove sei nel CICLO di volatilita' (flat -> 50)
|
||||
BuzVolume= chande01(volume) -> dove sei nel CICLO di volume (rampa -> 100)
|
||||
Ancore della demo del brief (trend lineare): ATR costante -> BuzVola=50 (neutro);
|
||||
volume in rampa -> BuzVolume=100. Entrambe RICOSTRUITE esattamente da chande01.
|
||||
2. FASCE REGIME (Vola, Volume): trade ammesso solo se BuzVola in [vola_lo,vola_hi] E
|
||||
BuzVolume in [vol_lo,vol_hi]. (Le "fasce 4/3/2 - 4/2/2" del sistema originale sono
|
||||
ricostruite come bande-soglia tunabili: i magici interi non sono nel brief.)
|
||||
3. PATTERN (breakout su data2/HTF): Donchian leak-free a `ptn_n` barre (default 13, da 13/13/1).
|
||||
ptn_long = close_htf rompe il massimo delle ptn_n barre PRECEDENTI
|
||||
ptn_short = close_htf rompe il minimo delle ptn_n barre PRECEDENTI
|
||||
4. COMPOSER: contenitore_long = regime_ok AND ptn_long ; contenitore_short = regime_ok AND ptn_short
|
||||
5. INGRESSO (max 1 al giorno): se il composer e' attivo -> OPEN_LONG / OPEN_SHORT alla
|
||||
chiusura LTF. (stop-and-reverse: non-overlap nell'engine -> il rovescio entra alla prima
|
||||
barra utile dopo l'uscita se il segnale persiste.)
|
||||
6. USCITE: time-based ASIMMETRICO (uscitalong=24, uscitashort=18 barre LTF) + hard stop/profit.
|
||||
Lo "stop 2000 / profit 5000" in $ del sistema ES e' tradotto in CRYPTO come multipli di ATR
|
||||
LTF (scale-free): sl = k_sl*ATR, tp = k_tp*ATR (default 2.0/5.0 ~ il rapporto 40:100 pt ES),
|
||||
con modalita' 'pct' alternativa (stop/profit in percentuale).
|
||||
|
||||
CAUSALITA': ogni feature usa dati <= close della barra (HTF: donchian con shift(1), chande01
|
||||
rolling causale). Il merge HTF->LTF e' merge_asof BACKWARD sulla CHIUSURA HTF (<= chiusura LTF):
|
||||
una barra HTF e' usata solo quando e' realmente chiusa. backtest_signals apre a close[i].
|
||||
|
||||
API:
|
||||
from src.strategies.skyhook import SkyhookParams, build_frames, skyhook_entries
|
||||
ltf, htf = build_frames(load_data("BTC","5m")) # resample 5m -> 230m + 690m
|
||||
entries = skyhook_entries(ltf, htf, SkyhookParams()) # list[dict|None] len(ltf), per backtest_signals
|
||||
from src.backtest.harness import backtest_signals
|
||||
m = backtest_signals(ltf, entries, fee_rt=0.001); m.print_summary("SKH01 BTC")
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
from dataclasses import dataclass, field
|
||||
|
||||
import numpy as np
|
||||
import pandas as pd
|
||||
|
||||
# 690 = 3 x 230 ; entrambi multipli esatti di 5m (138 e 46 barre da 5m)
|
||||
HTF_MIN = 690 # data2 — segnale
|
||||
LTF_MIN = 230 # data1 — esecuzione
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Resample dal feed 5m certificato (origin='epoch' -> confini deterministici e allineati)
|
||||
# ---------------------------------------------------------------------------
|
||||
def resample_5m(df5: pd.DataFrame, minutes: int) -> pd.DataFrame:
|
||||
"""5m -> `minutes` barre (origin epoch). Schema con 'datetime' + 'timestamp' (open-labeled)."""
|
||||
g = df5[["timestamp", "open", "high", "low", "close", "volume"]].copy()
|
||||
g.index = pd.to_datetime(g["timestamp"], unit="ms", utc=True)
|
||||
out = (g.resample(f"{minutes}min", label="left", closed="left", origin="epoch")
|
||||
.agg({"open": "first", "high": "max", "low": "min", "close": "last", "volume": "sum"})
|
||||
.dropna(subset=["open"]))
|
||||
out["datetime"] = out.index
|
||||
epoch = pd.Timestamp("1970-01-01", tz="UTC")
|
||||
out["timestamp"] = ((out.index - epoch) // pd.Timedelta(milliseconds=1)).astype("int64")
|
||||
return out.reset_index(drop=True)[["timestamp", "open", "high", "low", "close", "volume", "datetime"]]
|
||||
|
||||
|
||||
def build_frames(df5: pd.DataFrame) -> tuple[pd.DataFrame, pd.DataFrame]:
|
||||
"""Da un feed 5m certificato -> (ltf 230m exec, htf 690m signal)."""
|
||||
return resample_5m(df5, LTF_MIN), resample_5m(df5, HTF_MIN)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Indicatori causali
|
||||
# ---------------------------------------------------------------------------
|
||||
def atr(df: pd.DataFrame, win: int = 14) -> np.ndarray:
|
||||
h, l, c = df["high"].values, df["low"].values, df["close"].values
|
||||
pc = np.roll(c, 1); pc[0] = c[0]
|
||||
tr = np.maximum(h - l, np.maximum(np.abs(h - pc), np.abs(l - pc)))
|
||||
return pd.Series(tr).ewm(alpha=1.0 / win, adjust=False).mean().values
|
||||
|
||||
|
||||
def chande01(x: np.ndarray, n: int) -> np.ndarray:
|
||||
"""Chande Momentum Oscillator su `x`, normalizzato 0-100 (tipo-Chande).
|
||||
CMO = (Su - Sd)/(Su + Sd) in [-1,1] sulle n variazioni; mappato (1+CMO)*50 -> [0,100].
|
||||
Serie piatta (variazioni nulle) -> 50 (neutro). Causale (rolling fino a i)."""
|
||||
x = np.asarray(x, float)
|
||||
d = np.diff(x, prepend=x[0])
|
||||
up = np.where(d > 0, d, 0.0)
|
||||
dn = np.where(d < 0, -d, 0.0)
|
||||
su = pd.Series(up).rolling(n, min_periods=n).sum().values
|
||||
sd = pd.Series(dn).rolling(n, min_periods=n).sum().values
|
||||
denom = su + sd
|
||||
cmo = np.divide(su - sd, denom, out=np.zeros_like(denom), where=denom > 0)
|
||||
out = 50.0 * (1.0 + cmo)
|
||||
out[~np.isfinite(out)] = 50.0
|
||||
return out
|
||||
|
||||
|
||||
def donchian_breakout(df: pd.DataFrame, n: int) -> tuple[np.ndarray, np.ndarray]:
|
||||
"""Breakout leak-free: close[i] rompe il max/min delle n barre STRETTAMENTE precedenti."""
|
||||
hi = pd.Series(df["high"].values).rolling(n, min_periods=n).max().shift(1).values
|
||||
lo = pd.Series(df["low"].values).rolling(n, min_periods=n).min().shift(1).values
|
||||
c = df["close"].values.astype(float)
|
||||
return (c > hi), (c < lo)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Parametri
|
||||
# ---------------------------------------------------------------------------
|
||||
@dataclass
|
||||
class SkyhookParams:
|
||||
# indicatori (HTF)
|
||||
atr_win: int = 14
|
||||
n_vola: int = 13 # finestra Chande su ATR (da PtnL 13)
|
||||
n_volume: int = 13 # finestra Chande su volume (da PtnL 13)
|
||||
# fasce regime (bande-soglia su 0-100). Default = "regime di breakout":
|
||||
# volume vivo (BuzVolume alto) + volatilita' presente ma non da blow-off.
|
||||
vola_lo: float = 35.0
|
||||
vola_hi: float = 95.0
|
||||
vol_lo: float = 50.0
|
||||
vol_hi: float = 100.0
|
||||
# pattern (HTF) — Donchian breakout
|
||||
ptn_n: int = 13 # da PtnL 13/13/1
|
||||
# composer / direzione
|
||||
long_only: bool = False # Skyhook e' L/S di natura; True = solo long (stile crypto difensivo)
|
||||
# ingresso
|
||||
max_per_day: int = 1
|
||||
# uscite — time-based asimmetrico (barre LTF)
|
||||
uscitalong: int = 24
|
||||
uscitashort: int = 18
|
||||
# uscite — hard stop/profit
|
||||
exit_mode: str = "atr" # 'atr' = multipli di ATR LTF ; 'pct' = percentuale fissa
|
||||
sl_atr: float = 2.0
|
||||
tp_atr: float = 5.0
|
||||
sl_pct: float = 0.03
|
||||
tp_pct: float = 0.075
|
||||
ltf_atr_win: int = 14
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Feature HTF -> merge causale su LTF
|
||||
# ---------------------------------------------------------------------------
|
||||
def htf_features(htf: pd.DataFrame, p: SkyhookParams) -> pd.DataFrame:
|
||||
"""Calcola regime+pattern sull'HTF e li restituisce indicizzati per CHIUSURA HTF (timestamp
|
||||
di chiusura = open + 690min). Cosi' il merge backward su LTF e' strettamente causale."""
|
||||
buz_vola = chande01(atr(htf, p.atr_win), p.n_vola)
|
||||
buz_volume = chande01(htf["volume"].values, p.n_volume)
|
||||
ptn_long, ptn_short = donchian_breakout(htf, p.ptn_n)
|
||||
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),
|
||||
})
|
||||
|
||||
|
||||
def merge_htf_to_ltf(ltf: pd.DataFrame, feat: pd.DataFrame) -> pd.DataFrame:
|
||||
"""Attacca a ogni barra LTF l'ultima feature HTF la cui CHIUSURA <= chiusura LTF (causale)."""
|
||||
left = ltf.copy()
|
||||
left["close_ts"] = left["timestamp"].astype("int64").values + LTF_MIN * 60 * 1000
|
||||
m = pd.merge_asof(left.sort_values("close_ts"),
|
||||
feat.sort_values("close_ts"),
|
||||
on="close_ts", direction="backward")
|
||||
return m.sort_index().reset_index(drop=True)
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# Generatore di ingressi per backtest_signals ({'dir','tp','sl','max_bars'})
|
||||
# ---------------------------------------------------------------------------
|
||||
def skyhook_entries(ltf: pd.DataFrame, htf: pd.DataFrame, p: SkyhookParams | None = None) -> list:
|
||||
"""Lista di entry-dict (uno per barra LTF, None = niente segnale), pronta per
|
||||
backtest_signals. Max `max_per_day` ingressi/giorno (prima barra qualificante del giorno).
|
||||
sl/tp e max_bars asimmetrici per direzione. Tutto causale (decide a close[i])."""
|
||||
p = p or SkyhookParams()
|
||||
feat = htf_features(htf, p)
|
||||
m = merge_htf_to_ltf(ltf, feat)
|
||||
|
||||
c = m["close"].values.astype(float)
|
||||
a = atr(m, 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: list = [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) >= p.max_per_day:
|
||||
continue
|
||||
if comp_long[i]:
|
||||
direction, mb = 1, p.uscitalong
|
||||
elif comp_short[i]:
|
||||
direction, mb = -1, p.uscitashort
|
||||
else:
|
||||
continue
|
||||
if p.exit_mode == "atr":
|
||||
sl_off, tp_off = p.sl_atr * a[i], p.tp_atr * a[i]
|
||||
else:
|
||||
sl_off, tp_off = p.sl_pct * c[i], p.tp_pct * 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(mb)}
|
||||
count_today[day] = count_today.get(day, 0) + 1
|
||||
return entries
|
||||
|
||||
|
||||
def signal_counts(ltf: pd.DataFrame, htf: pd.DataFrame, p: SkyhookParams | None = None) -> dict:
|
||||
"""Diagnostica: quante barre passano regime/pattern/composer (prima del cap giornaliero)."""
|
||||
p = p or SkyhookParams()
|
||||
feat = htf_features(htf, p)
|
||||
m = merge_htf_to_ltf(ltf, feat)
|
||||
cl = np.nan_to_num(m["comp_long"].values).astype(bool)
|
||||
cs = np.nan_to_num(m["comp_short"].values).astype(bool)
|
||||
ent = skyhook_entries(ltf, htf, p)
|
||||
return dict(ltf_bars=len(m), comp_long=int(cl.sum()), comp_short=int(cs.sum()),
|
||||
entries=int(sum(e is not None for e in ent)))
|
||||
Reference in New Issue
Block a user