daily multi-asset strategy grid

crypto_eval/grid2.py

@@ -0,0 +1,86 @@
+"""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