| """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) |
| 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) |
| |
| |
| |
| 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, |
| vol_window: int = 720) -> 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) |
| 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() |
| trigger = (f < -k * sigma).reindex(bars.index, method="ffill").fillna(False) |
| pos = pd.Series(0.0, index=bars.index) |
| |
| 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 |
|
|
