diff --git a/src/portfolio/sleeves.py b/src/portfolio/sleeves.py index fd44186..f59284d 100644 --- a/src/portfolio/sleeves.py +++ b/src/portfolio/sleeves.py @@ -236,8 +236,41 @@ def _skyhook_returns() -> pd.Series: return pd.Series(0.5 * J["BTC"].values + 0.5 * J["ETH"].values, index=J.index) +def _skyhook_positions() -> dict: + """Stato corrente del book Skyhook per asset (introspezione live): se c'e' un trade APERTO ORA + -> dir/entry/sl/tp/barre-trascorse; altrimenti 'flat'. Replica la logica non-overlap di + entry+exit (TP/SL/max_bars) fino all'ultima barra 230m chiusa. Causale: usa solo barre chiuse.""" + out = {} + for a in ASSETS: + ltf, htf = build_frames(load_data(a, "5m")) + ent = skyhook_entries(ltf, htf, SKH01_V2_DD) + H = ltf["high"].values; L = ltf["low"].values; Cc = ltf["close"].values + n = len(ltf); i = 0; open_pos = "flat" + while i < n: + e = ent[i] + if e is None: + i += 1; continue + d, sl, tp, mb = e["dir"], e["sl"], e["tp"], e["max_bars"] + exit_idx = None + for s in range(1, mb + 1): + j = i + s + if j >= n: # non ancora uscito: posizione APERTA ora + break + hit = (L[j] <= sl or H[j] >= tp) if d == 1 else (H[j] >= sl or L[j] <= tp) + if hit or s == mb: + exit_idx = j; break + if exit_idx is None: + open_pos = dict(dir="LONG" if d == 1 else "SHORT", entry=round(float(Cc[i]), 2), + sl=round(float(sl), 2), tp=round(float(tp), 2), + bars_in=int((n - 1) - i), max_bars=int(mb)) + break + i = exit_idx + 1 + out[a] = open_pos + return out + + def skyhook_sleeve(weight: float = 0.25) -> Sleeve: - return Sleeve("SKH01_skyhook", weight, _skyhook_returns) + return Sleeve("SKH01_skyhook", weight, _skyhook_returns, pos_fn=_skyhook_positions) # ----------------------------- REGISTRY ----------------------------- diff --git a/tests/test_skyhook.py b/tests/test_skyhook.py index 17ea8eb..aa5db6b 100644 --- a/tests/test_skyhook.py +++ b/tests/test_skyhook.py @@ -129,6 +129,16 @@ def test_short_override_changes_only_shorts(): assert longs_same > 0 and shorts_diff > 0 +def test_skyhook_pos_fn_structure(): + """pos_fn introspettiva: dict BTC/ETH, ciascuno 'flat' o dict con dir/entry/sl/tp coerenti.""" + from src.portfolio.sleeves import _skyhook_positions + pos = _skyhook_positions() + assert set(pos.keys()) == {"BTC", "ETH"} + for a, p in pos.items(): + assert p == "flat" or (isinstance(p, dict) and p["dir"] in ("LONG", "SHORT") + and p["sl"] > 0 and p["tp"] > 0 and 0 <= p["bars_in"] <= p["max_bars"]) + + def test_v2dd_robust_both_assets(): """SKH01-V2-DD: PASS netto fee su BTCÐ, hold-out forte, e maxDD standalone <30%.""" import skyhooklib as sk