File size: 5,086 Bytes
eb9cbc3 30d3ad2 eb9cbc3 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 | """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,
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) # 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
|