""" Tactical All-Weather portfolio engine. This module replaces the previous multi-pair risk-parity / kill-switch logic. Core idea --------- Maintain a static baseline allocation to a set of macro sleeves, but use a momentum + trend regime gate to move individual sleeves to cash during bear markets. A "Static All Weather" benchmark remains fully invested at BASE_WEIGHTS. """ from __future__ import annotations from dataclasses import dataclass from typing import Dict import numpy as np import pandas as pd # Baseline macro universe BASE_WEIGHTS: dict[str, float] = { "SPY": 0.30, "TLT": 0.40, "IEF": 0.15, "GLD": 0.075, "DBC": 0.075, } BOND_TICKERS: frozenset[str] = frozenset({"TLT", "IEF"}) @dataclass(frozen=True) class TacticalAWConfig: """ Gate and sizing parameters for :class:`TacticalAllWeatherManager`. Defaults reproduce the original binary SMA(200) + 12-1 momentum rules. """ sma_window: int = 200 mom_skip_days: int = 21 mom_lookback_days: int = 252 mom_threshold: float = 0.0 weight_mode: str = "binary" # binary | partial partial_frac: float = 0.5 risk_on_min_invested: float = 0.0 bond_baseline_mult: float = 1.0 cash_annual_yield: float = 0.04 def _momentum_from_close(close: pd.Series, *, skip: int, lookback: int) -> pd.Series: denom = close.shift(int(lookback)) num = close.shift(int(skip)) return num / denom - 1.0 def _require_columns(df: pd.DataFrame, cols: list[str], *, ticker: str) -> None: missing = [c for c in cols if c not in df.columns] if missing: raise KeyError(f"Missing required columns for {ticker}: {missing}") @dataclass class TacticalAllWeatherManager: """ Dynamic asset allocation engine. build_portfolio expects, for each ticker: - close: price - ret: daily pct change (bar return) SMA and momentum are computed from ``close`` using :class:`TacticalAWConfig`. """ baseline_weights: dict[str, float] | None = None config: TacticalAWConfig = None # type: ignore[assignment] def __post_init__(self) -> None: if self.baseline_weights is None: self.baseline_weights = dict(BASE_WEIGHTS) if self.config is None: self.config = TacticalAWConfig() def build_portfolio( self, data_dict: dict[str, pd.DataFrame], cash_annual_yield: float | None = None, ) -> pd.DataFrame: """ Parameters ---------- data_dict Dict: ticker -> DataFrame with columns: close, ret, sma_200, aqr_mom. All frames will be aligned to a common calendar index. cash_annual_yield Annualized cash yield (e.g. 0.04 for 4%), converted to daily via /252. Returns ------- pd.DataFrame Contains: - portfolio_bar_ret, portfolio_cumulative_ret - static_all_weather_cumulative_ret - spy_cumulative_ret - cash_weight, total_invested_weight - weight_ columns for each sleeve """ if not data_dict: raise ValueError("data_dict cannot be empty") cfg = self.config rf_yield = float(cfg.cash_annual_yield if cash_annual_yield is None else cash_annual_yield) sleeves = list(self.baseline_weights.keys()) missing_sleeves = [t for t in sleeves if t not in data_dict] if missing_sleeves: raise KeyError(f"data_dict missing required baseline tickers: {missing_sleeves}") # Master index = union of all available indices across baseline sleeves. # (We fill via forward-fill and ret via 0.) indices = [pd.to_datetime(data_dict[t].index).tz_localize(None) for t in sleeves] master_index = indices[0] for idx in indices[1:]: master_index = master_index.union(idx) master_index = master_index.sort_values() # Build aligned matrices for close, ret; SMA/mom from config. close_mat: list[pd.Series] = [] ret_mat: list[pd.Series] = [] sma_win = max(20, int(cfg.sma_window)) for t in sleeves: df = data_dict[t] _require_columns(df, ["close", "ret"], ticker=t) dfx = df.copy() dfx.index = pd.to_datetime(dfx.index).tz_localize(None) dfx = dfx.sort_index() dfx = dfx.reindex(master_index) close_s = dfx["close"].ffill() ret_s = dfx["ret"].fillna(0.0) close_mat.append(close_s.rename(t)) ret_mat.append(ret_s.rename(t)) close_df = pd.concat(close_mat, axis=1) ret_df = pd.concat(ret_mat, axis=1) sma_df = close_df.rolling(window=sma_win, min_periods=sma_win).mean() mom_df = pd.DataFrame( { t: _momentum_from_close( close_df[t], skip=int(cfg.mom_skip_days), lookback=int(cfg.mom_lookback_days), ) for t in sleeves }, index=master_index, ) trend = close_df > sma_df mom_ok = mom_df > float(cfg.mom_threshold) mode = str(cfg.weight_mode).strip().lower() base_w = pd.Series(self.baseline_weights, dtype=np.float64).reindex(sleeves) for t in sleeves: if t in BOND_TICKERS: base_w[t] = float(base_w[t]) * float(cfg.bond_baseline_mult) bw = base_w.to_numpy(dtype=np.float64) if mode == "partial": frac = float(cfg.partial_frac) mult = np.where(trend & mom_ok, 1.0, 0.0) mult = mult + np.where(trend & ~mom_ok, frac, 0.0) mult = mult + np.where(~trend & mom_ok, frac, 0.0) mult = np.minimum(mult, 1.0) target_weight_arr = mult * bw else: is_bullish = trend & mom_ok target_weight_arr = np.where( is_bullish.to_numpy(dtype=np.bool_), bw, 0.0, ) target_weight = pd.DataFrame( target_weight_arr, index=master_index, columns=sleeves, ) target_weight = target_weight.shift(1).fillna(0.0) min_inv = float(cfg.risk_on_min_invested) if min_inv > 0.0 and "SPY" in sleeves: spy_ok = ((close_df["SPY"] > sma_df["SPY"]) & (mom_df["SPY"] > cfg.mom_threshold)).shift( 1 ).fillna(False) tot = target_weight.sum(axis=1) shortfall = (min_inv - tot).clip(lower=0.0) active = spy_ok & (shortfall > 1e-9) add_spy = shortfall * 0.70 add_gld = shortfall * 0.30 if "SPY" in target_weight.columns: target_weight.loc[active, "SPY"] = target_weight.loc[active, "SPY"] + add_spy.loc[ active ] if "GLD" in target_weight.columns: target_weight.loc[active, "GLD"] = target_weight.loc[active, "GLD"] + add_gld.loc[ active ] row_sum = target_weight.sum(axis=1) over = row_sum > 1.0 + 1e-9 if over.any(): target_weight.loc[over] = target_weight.loc[over].div(row_sum.loc[over], axis=0) total_invested_weight = target_weight.sum(axis=1) cash_weight = 1.0 - total_invested_weight daily_rf = rf_yield / 252.0 portfolio_bar_ret = (ret_df * target_weight).sum(axis=1) + cash_weight * daily_rf portfolio_cumulative_ret = (1.0 + portfolio_bar_ret).cumprod() - 1.0 # Static All Weather: always fully invested at BASE_WEIGHTS. static_all_weather_ret = (ret_df * base_w).sum(axis=1) static_all_weather_cumulative_ret = (1.0 + static_all_weather_ret).cumprod() - 1.0 # Pure SPY buy & hold (independent of baseline sleeves). spy_ret = pd.Series(np.nan, index=master_index, dtype=np.float64) if "SPY" in data_dict: df_spy = data_dict["SPY"] if "ret" in df_spy.columns: spy_ret = ( df_spy["ret"] .astype(np.float64) .reindex(master_index) .fillna(0.0) ) spy_cumulative_ret = (1.0 + spy_ret).cumprod() - 1.0 out = pd.DataFrame( { "portfolio_bar_ret": portfolio_bar_ret, "portfolio_cumulative_ret": portfolio_cumulative_ret, "static_all_weather_ret": static_all_weather_ret, "static_all_weather_cumulative_ret": static_all_weather_cumulative_ret, "spy_cumulative_ret": spy_cumulative_ret, "cash_weight": cash_weight, "total_invested_weight": total_invested_weight, }, index=master_index, ) # Add sleeve weights for plotting/audit. weight_cols = target_weight.rename(columns={t: f"weight_{t}" for t in sleeves}) out = pd.concat([out, weight_cols], axis=1) return out def plot_tactical_portfolio(portfolio_df: pd.DataFrame, *, save_path: str | None = None) -> None: """ Plot tactical portfolio vs static all-weather + SPY buy-and-hold. Also includes a secondary subplot (stacked area) showing dynamic allocation including cash over time. """ required = [ "portfolio_cumulative_ret", "static_all_weather_cumulative_ret", "spy_cumulative_ret", "cash_weight", ] missing = [c for c in required if c not in portfolio_df.columns] if missing: raise KeyError(f"portfolio_df missing required columns: {missing}") try: import matplotlib.pyplot as plt except ImportError as e: # pragma: no cover raise ImportError("matplotlib is required for plotting.") from e sleeves = list(BASE_WEIGHTS.keys()) weight_cols = [f"weight_{t}" for t in sleeves if f"weight_{t}" in portfolio_df.columns] if len(weight_cols) != len(sleeves): raise KeyError("portfolio_df missing expected weight_ columns for sleeves") fig, (ax_eq, ax_alloc) = plt.subplots( 2, 1, figsize=(12, 8), sharex=True, gridspec_kw={"height_ratios": [2.0, 1.2]}, ) ax_eq.plot( portfolio_df.index, portfolio_df["portfolio_cumulative_ret"].values, label="Tactical All Weather", linewidth=2.0, ) ax_eq.plot( portfolio_df.index, portfolio_df["static_all_weather_cumulative_ret"].values, label="Static All Weather", linewidth=1.5, alpha=0.9, ) ax_eq.plot( portfolio_df.index, portfolio_df["spy_cumulative_ret"].values, label="SPY (buy & hold)", linewidth=1.2, linestyle="--", alpha=0.9, ) ax_eq.axhline(0.0, color="gray", linewidth=0.5) ax_eq.set_ylabel("Cumulative Return") ax_eq.grid(True, alpha=0.3) ax_eq.legend(loc="best") # Allocation stacked area chart: weights + cash. alloc_series = [portfolio_df[col].astype(np.float64) for col in weight_cols] + [ portfolio_df["cash_weight"].astype(np.float64) ] alloc_labels = [col.replace("weight_", "") for col in weight_cols] + ["CASH"] ax_alloc.stackplot(portfolio_df.index, *alloc_series, labels=alloc_labels, alpha=0.9) ax_alloc.set_ylabel("Allocation") ax_alloc.set_ylim(0.0, 1.0) ax_alloc.grid(True, alpha=0.2) ax_alloc.legend(loc="upper left", ncols=2, fontsize=8) fig.tight_layout() if save_path: fig.savefig(save_path, dpi=150) plt.close(fig) else: plt.show() plt.close(fig) # Backwards-compatible names for older imports. PortfolioManager = TacticalAllWeatherManager def plot_portfolio(portfolio_df: pd.DataFrame, *, save_path: str | None = None) -> None: """Compatibility wrapper (old name).""" plot_tactical_portfolio(portfolio_df, save_path=save_path)