quant-trading/strategies/sleeve_portfolio.py

"""
Sleeve portfolio: several strategies held side by side in one account, each
with a fixed share of capital.

``us_three_sleeve`` is the portfolio from the 2026-09-29 US research
(research/lab/us_final.py, report https://claude.ai/artifact/VGhZvKkS9UBYZkLeqGjcNC):
one third each of

  momentum  : overnight_recovery_top10  (S&P 500 stocks)
  defensive : lowrisk_top20             (S&P 500 stocks)
  trend     : trend_rider_v7_vt24       (leveraged / defensive ETFs)

Their monthly return correlations are 0.2-0.4, which is where the portfolio's
higher Sharpe and smaller drawdown come from (2011-2026: Sharpe 1.21 vs
0.88-1.03 for the sleeves alone).

Each sleeve keeps its own strategy's targets. The stock sleeves see only the
stock columns and the trend sleeve only its ETFs. A sleeve's share of the
account drifts with its own return, like a separate account would, and all
shares are reset to the fixed ones at the first session of a month when any
share is more than ``band`` (relative) away from its fixed value. Resetting
every month instead would resize most of the ~30 names each month, which the
live book pays a ticket fee for.

Timing: every sleeve decides on completed bars, so the trend sleeve reads
closes through the previous session here, where the stand-alone
trend_rider_v7_vt24 reads the session's open as its latest price.
"""

from __future__ import annotations

import numpy as np
import pandas as pd

from strategies.base import BarStrategy, Strategy


def decisions(strategy: Strategy, bars: dict[str, pd.DataFrame]) -> pd.DataFrame:
    """Targets decided on each row's own bars.

    A Strategy's weights on row t use data through t-1, so the decision made
    on the last row only appears on a row after it. Padding one copy of the
    last row and shifting back yields it; a causal strategy ignores the value
    of the padded row."""
    close = bars["close"]
    nxt = close.index[-1] + pd.Timedelta(days=1)
    padded = {f: pd.concat([b, b.iloc[[-1]].set_axis([nxt], axis=0)]) for f, b in bars.items()}
    data = padded if isinstance(strategy, BarStrategy) else padded["close"]
    weights = strategy.generate_signals(data)
    weights = weights.reindex(index=padded["close"].index, columns=close.columns).fillna(0.0)
    return weights.shift(-1).iloc[:-1].set_axis(close.index, axis=0)


def drifting_shares(sleeve_weights: list[np.ndarray], returns: np.ndarray, base: np.ndarray,
                    month_start: np.ndarray, band: float) -> np.ndarray:
    """Share of the account in each sleeve per row (rows x sleeves).

    Row k's shares grow with each sleeve's return from k-1 to k (its row k-1
    weights times row k asset returns; the unallocated part of a sleeve is
    cash) and are reset to ``base`` on a month-start row when any share is
    more than ``band`` away from its base, relatively."""
    n = returns.shape[0]
    shares = np.empty((n, len(base)))
    s = base.astype(float).copy()
    for k in range(n):
        if k > 0:
            growth = np.array([1.0 + w[k - 1] @ returns[k] for w in sleeve_weights])
            s = s * growth
            s = s / s.sum() if s.sum() > 0 else base.copy()
            if month_start[k] and np.max(np.abs(s / base - 1.0)) > band:
                s = base.astype(float).copy()
        shares[k] = s
    return shares


class SleevePortfolioStrategy(BarStrategy):
    """Fixed-share portfolio of sleeve strategies in one account.

    ``sleeves`` is a list of ``(name, strategy, share, universe)``; universe
    "stocks" gives the sleeve every column except ``etfs``, "etfs" gives it
    the ``etfs`` columns present."""

    def __init__(self, sleeves: list[tuple[str, Strategy, float, str]], etfs: list[str],
                 band: float = 0.25):
        total = sum(share for _, _, share, _ in sleeves)
        self.sleeves = [(n, s, share / total, u) for n, s, share, u in sleeves]
        self.etfs = list(etfs)
        self.band = band
        fields = {"close"}
        for _, s, _, _ in sleeves:
            fields |= set(s.fields) if isinstance(s, BarStrategy) else set()
        self.fields = tuple(sorted(fields, key=lambda f: (f != "close", f)))

    def _sleeve_bars(self, strategy: Strategy, bars: dict, cols: list[str]) -> dict:
        wanted = strategy.fields if isinstance(strategy, BarStrategy) else ("close",)
        sub = {f: bars[f][cols] for f in wanted}
        if "eligible" in bars:
            sub["eligible"] = bars["eligible"].reindex(columns=cols).fillna(True)
        return sub

    def target_weights(self, bars: dict[str, pd.DataFrame]) -> tuple[pd.DataFrame, pd.DataFrame]:
        """Unshifted combined targets (row k decided on bars through k) and the
        sleeve shares behind them."""
        close = bars["close"]
        etf_cols = [c for c in close.columns if c in self.etfs]
        stock_cols = [c for c in close.columns if c not in self.etfs]
        parts = []
        for _, strategy, _, universe in self.sleeves:
            cols = etf_cols if universe == "etfs" else stock_cols
            d = decisions(strategy, self._sleeve_bars(strategy, bars, cols))
            parts.append(d.reindex(columns=close.columns).fillna(0.0).to_numpy())
        rets = np.nan_to_num(close.ffill().pct_change(fill_method=None).to_numpy(dtype=float), nan=0.0)
        months = close.index.to_period("M")
        month_start = np.r_[False, months[1:] != months[:-1]]
        base = np.array([share for _, _, share, _ in self.sleeves])
        shares = drifting_shares(parts, rets, base, month_start, self.band)
        combined = sum(shares[:, [j]] * parts[j] for j in range(len(parts)))
        names = [n for n, _, _, _ in self.sleeves]
        return (pd.DataFrame(combined, index=close.index, columns=close.columns),
                pd.DataFrame(shares, index=close.index, columns=names))

    def generate_weights(self, bars: dict[str, pd.DataFrame]) -> pd.DataFrame:
        combined, _ = self.target_weights(bars)
        return combined.shift(1).fillna(0.0)


def us_three_sleeve() -> SleevePortfolioStrategy:
    from strategies.permanent import ETF_UNIVERSE
    from strategies.robust_momentum import RobustMomentumStrategy
    from strategies.trend_rider_v7 import TrendRiderV7
    return SleevePortfolioStrategy(
        [
            ("momentum", RobustMomentumStrategy("overnight_recovery", top_n=10, rebal_freq=10,
                                                buffer=2.0), 1.0, "stocks"),
            ("defensive", RobustMomentumStrategy("low_risk", top_n=20, rebal_freq=21,
                                                 buffer=2.0), 1.0, "stocks"),
            ("trend", TrendRiderV7(target_vol=0.24, min_lev=0.5), 1.0, "etfs"),
        ],
        etfs=sorted(set(ETF_UNIVERSE)),
    )