File size: 3,422 Bytes
3ea2315 | 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 | """Daily multi-asset strategy families for the panel sweep.
All signals use data through t-1: `panel.shift(1)` everywhere.
"""
from __future__ import annotations
import numpy as np
import pandas as pd
def _vol_scale(panel: pd.DataFrame, target: float, window: int) -> pd.DataFrame:
ret = panel.pct_change()
rv = ret.rolling(window).std().shift(1) * np.sqrt(365)
return (target / rv).clip(upper=1.0).fillna(0.0)
def tsmom_panel(panel: pd.DataFrame, lookback: int, long_only: bool = False,
vol_target: float | None = None,
vol_window: int = 60) -> pd.DataFrame:
c = panel.shift(1)
mom = c / c.shift(lookback) - 1
sig = pd.DataFrame(np.where(mom > 0, 1.0, -1.0),
index=panel.index, columns=panel.columns)
if long_only:
sig = sig.clip(lower=0.0)
if vol_target is not None:
sig = sig * _vol_scale(panel, vol_target, vol_window)
# pre-listing: no data -> flat
return sig.where(panel.notna(), 0.0).fillna(0.0)
def donchian_panel(panel: pd.DataFrame, window: int, long_only: bool = False,
vol_target: float | None = None) -> pd.DataFrame:
# channel uses highs/lows we don't have on the index panel; approximate
# with rolling max/min of closes through t-2 (close-based breakout)
c = panel.shift(1)
upper = panel.rolling(window).max().shift(2)
lower = panel.rolling(window).min().shift(2)
pos = pd.DataFrame(np.nan, index=panel.index, columns=panel.columns)
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(panel, vol_target, 60)
return pos.where(panel.notna(), 0.0)
def xs_momentum(panel: pd.DataFrame, lookback: int, top_k: int = 2,
short_bottom: bool = False, rebalance: str = "W") -> pd.DataFrame:
"""Cross-sectional momentum: long top-k by trailing return, optional
short bottom-k, rebalanced weekly (Mondays) or monthly."""
c = panel.shift(1)
mom = c / c.shift(lookback) - 1
ranks = mom.rank(axis=1, ascending=False)
n = panel.shape[1]
tgt = pd.DataFrame(0.0, index=panel.index, columns=panel.columns)
tgt[ranks <= top_k] = 1.0 / top_k
if short_bottom:
tgt[ranks > n - top_k] = -1.0 / top_k
idx = panel.index
if rebalance == "W":
mask = pd.Series(idx.dayofweek == 0, index=idx)
else: # monthly: first listed trading day of each month
mask = idx.to_series().groupby(idx.to_period("M")).transform("min").eq(idx)
pos = tgt.where(mask, np.nan).ffill().fillna(0.0)
return pos.where(panel.notna(), 0.0)
def xs_momentum_tsmom_filter(panel: pd.DataFrame, lookback: int, top_k: int = 2,
trend_lookback: int = 100,
rebalance: str = "W") -> pd.DataFrame:
"""XS momentum gated by per-asset trend: only hold top-k names that are
also above their own trend_lookback MA."""
pos = xs_momentum(panel, lookback, top_k, False, rebalance)
c = panel.shift(1)
above = c > c.rolling(trend_lookback).mean()
return pos.where(above, 0.0).fillna(0.0).where(panel.notna(), 0.0)
def buy_and_hold_btc(panel: pd.DataFrame) -> pd.DataFrame:
pos = pd.DataFrame(0.0, index=panel.index, columns=panel.columns)
pos["BTC"] = 1.0
return pos
|