parameterized strategy grid (donchian breakout fix)

crypto_eval/grid.py

@@ -0,0 +1,130 @@
+"""Parameterized strategy families for the sweep.
+
+Every function returns positions (units of base asset) indexed by bar
+start time, and uses ONLY information available at decision time t:
+closes/highs/lows shifted by 1 (the closed bar is t-1), funding ticks
+shifted by 1. The harness audit + engine catch peeking downstream.
+
+Families (grounded in the literature survey, 2026-10):
+  tsmom       time-series momentum, lookbacks 1d-200d, long/short or
+              long-only, optional vol targeting, optional rebalance band
+  donchian    breakout channels 20d/55d, same overlays
+  carry       funding-level harvest on the perp (expected to fail post-2024)
+  funding_extreme  asymmetric long event on extreme negative funding
+  meanrev     1h mean reversion (documented cost-loser; sweep as control)
+"""
+
+from __future__ import annotations
+
+import numpy as np
+import pandas as pd
+
+BARS_PER_YEAR = 24 * 365
+
+
+def _ann_vol(close: pd.Series, window: int) -> pd.Series:
+    return close.pct_change().rolling(window).std() * np.sqrt(BARS_PER_YEAR)
+
+
+def _vol_scale(close: pd.Series, target: float, window: int) -> pd.Series:
+    rv = _ann_vol(close, window).shift(1)  # vol realized through t-1
+    return (target / rv).clip(upper=1.0).fillna(0.0)
+
+
+def _apply_band(target: pd.Series, band: float) -> pd.Series:
+    """Hold the previous position until the target moves away by > band."""
+    arr = target.to_numpy()
+    pos = np.empty_like(arr)
+    cur = 0.0
+    for i in range(len(arr)):
+        if abs(arr[i] - cur) > band:
+            cur = arr[i]
+        pos[i] = cur
+    return pd.Series(pos, index=target.index)
+
+
+def tsmom(bars: pd.DataFrame, funding: pd.Series, lookback: int,
+          long_only: bool = False, vol_target: float | None = None,
+          vol_window: int = 720, band: float | None = None) -> pd.Series:
+    c = bars["close"].shift(1)
+    mom = c / c.shift(lookback) - 1
+    target = pd.Series(np.where(mom > 0, 1.0, -1.0), index=bars.index)
+    if long_only:
+        target = target.clip(lower=0.0)
+    if vol_target is not None:
+        target = target * _vol_scale(bars["close"], vol_target, vol_window)
+    if band is not None:
+        target = _apply_band(target, band)
+    return target.fillna(0.0)
+
+
+def donchian(bars: pd.DataFrame, funding: pd.Series, window: int,
+             long_only: bool = False, vol_target: float | None = None,
+             vol_window: int = 720) -> pd.Series:
+    c = bars["close"].shift(1)
+    # channel through bar t-2 (rolling then shift): the close of bar t-1
+    # can actually exceed it; including t-1's own high/low makes breakout
+    # impossible (close <= high of the same bar, always).
+    upper = bars["high"].rolling(window).max().shift(2)
+    lower = bars["low"].rolling(window).min().shift(2)
+    pos = pd.Series(np.nan, index=bars.index)
+    pos[c > upper] = 1.0
+    if long_only:
+        pos[c < lower] = 0.0
+    else:
+        pos[c < lower] = -1.0
+    pos = pos.ffill().fillna(0.0)
+    if vol_target is not None:
+        pos = pos * _vol_scale(bars["close"], vol_target, vol_window)
+    return pos
+
+
+def carry(bars: pd.DataFrame, funding: pd.Series, threshold_mult: float = 1.0,
+          base: float = 1e-4, vol_target: float | None = None) -> pd.Series:
+    """Short the perp when trailing funding > threshold*base, long when < -."""
+    if isinstance(funding, pd.DataFrame):
+        funding = funding["funding_rate"]
+    f = funding.shift(1)  # last announced tick at decision time
+    signal = f.rolling(3, min_periods=1).mean()
+    tick = signal.reindex(bars.index, method="ffill")
+    pos = pd.Series(0.0, index=bars.index)
+    pos[tick > threshold_mult * base] = -1.0
+    pos[tick < -threshold_mult * base] = 1.0
+    if vol_target is not None:
+        pos = pos * _vol_scale(bars["close"], vol_target, vol_window)
+    return pos
+
+
+def funding_extreme(bars: pd.DataFrame, funding: pd.Series, k: float = 3.0,
+                    hold_bars: int = 24) -> pd.Series:
+    """Long event trade: extreme NEGATIVE funding (documented asymmetric)."""
+    if isinstance(funding, pd.DataFrame):
+        funding = funding["funding_rate"]
+    f = funding.shift(1)
+    sigma = f.rolling(90, min_periods=30).std()  # ~30d of 8h ticks
+    trigger = (f < -k * sigma).reindex(bars.index, method="ffill").fillna(False)
+    pos = pd.Series(0.0, index=bars.index)
+    # hold for hold_bars after each trigger
+    trig = trigger.to_numpy()
+    out = np.zeros(len(trig))
+    remaining = 0
+    for i in range(len(trig)):
+        if trig[i]:
+            remaining = hold_bars
+        if remaining > 0:
+            out[i] = 1.0
+            remaining -= 1
+    return pd.Series(out, index=bars.index)
+
+
+def meanrev(bars: pd.DataFrame, funding: pd.Series, window: int,
+            k: float = 2.5) -> pd.Series:
+    """1h mean reversion control: fade >k-sigma moves (expected to lose)."""
+    c = bars["close"].shift(1)
+    ma = c.rolling(window).mean()
+    sd = c.rolling(window).std()
+    z = (c - ma) / sd
+    pos = pd.Series(0.0, index=bars.index)
+    pos[z < -k] = 1.0
+    pos[z > k] = -1.0
+    return pos