| """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) |
| |
| 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: |
| |
| |
| 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: |
| 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 |
|
|
