f/fundlab/styletilt.py
Greg Pomerantz 2bbe4e58ec Style-tilt battery + commentary in the fund report pipeline
fundlab/styletilt.py: 22 style/asset sleeves regressed on excess-of-T-bill
returns (full history + 5y); BIC forward selection identifies the tilt
stack; residual-alpha verdict ('factor exposure, not skill' when t<1.75);
data-driven English commentary with sign-specific phrasing. Rendered as a
'Style tilts' block (factor table + prose) in both the app and the HTML
report. All 24 pre-built + 8 ad-hoc fund reports rebuilt.
2026-08-30 20:36:04 -04:00

258 lines
10 KiB
Python

"""Style-tilt battery: mathematical factor identification for one fund.
Regresses the fund's daily excess-of-T-bill returns on a battery of 22
style/asset sleeves (US funds proxying the global style factors; a fund
expresses them in its own universe, so betas are the right sign and
magnitude but approximate), over the full history and the last 5 years:
* plain OLS -> every factor's beta + t-stat (exploratory view)
* BIC forward selection -> the IDENTIFIED tilt stack (the strict gate:
a factor is only in if it measurably improves the model)
* residual alpha after the identified stack -> "is there skill left?"
Plus data-driven English commentary (fundlab.styletilt.commentary).
Caveats stated in the commentary, not hidden: US sleeves proxy global
factors; style sleeves are mutually collinear, so single-OLS t-stats are
indicative and the BIC set is the identification.
"""
from __future__ import annotations
import numpy as np
import pandas as pd
from fundlab import decompose
# factor label -> sleeve ticker (US proxies for the global factors)
BATTERY: dict[str, str] = {
"int-dev (EFA)": "efa", "EM (VWO)": "vwo", "growth (VUG)": "vug",
"value (VTV)": "vtv", "small (IWM)": "iwm", "equalwt (RSP)": "rsp",
"lowvol (USMV)": "usmv", "lowvol (DFLVX)": "dflvx",
"quality (QUAL)": "qual", "momentum (MTUM)": "mtum",
"dividend (HDV)": "hdv", "div-apprec (VYM)": "vym",
"longdur (TLT)": "tlt", "gold (GLD)": "gld",
"EUR (FXE)": "fxe", "JPY (FXY)": "fxy", "AUD (FXU)": "fxu",
"utilities (XLU)": "xlu", "staples (XLP)": "xlp",
"health (XLV)": "xlv", "banks (XLF)": "xlf", "energy (XLE)": "xle",
}
SYMS = list(dict.fromkeys(BATTERY.values()))
INV = {v: k for k, v in BATTERY.items()}
_PANEL: pd.DataFrame | None = None
def _panel() -> pd.DataFrame:
"""Battery sleeves' daily EXCESS-of-T-bill returns, cached per process."""
global _PANEL
if _PANEL is None:
_PANEL = decompose.excess(
decompose.returns_panel(SYMS, start=decompose.FULL_WINDOW))
return _PANEL
def _fund_excess(sym: str) -> pd.Series:
p = decompose.adj_close(sym)
if p is None:
return pd.Series(dtype=float)
y = p.pct_change()
rf = decompose.rf_series()
if rf is not None:
y = y.sub(rf.reindex(y.index).fillna(0.0))
return y
def _fit(y: pd.Series, a: str | None, b: str | None) -> dict:
x = _panel()
idx = x.index
msk = idx >= (a or idx[0])
if b is not None:
msk &= idx <= b
x = x[msk]
y = y.reindex(x.index)
ok = y.notna() & x.notna().all(axis=1)
if ok.sum() < 252:
return {}
X = np.column_stack([np.ones(int(ok.sum()))] +
[x[s][ok].to_numpy() for s in SYMS])
yn = y[ok].to_numpy()
beta, *_ = np.linalg.lstsq(X, yn, rcond=None)
res = yn - X @ beta
dof = max(int(ok.sum()) - len(SYMS) - 1, 1)
sigma2 = float(res @ res) / dof
se = np.sqrt(np.diag(np.linalg.inv(X.T @ X)) * sigma2)
t = np.where(se > 0, beta / se, 0.0)
r2 = 1 - float(res @ res) / float((yn - yn.mean()) @ (yn - yn.mean()))
return {
"n": int(ok.sum()),
"start": str(x.index[ok][0].date()),
"end": str(x.index[ok][-1].date()),
"r2": float(r2),
"alpha_ann": float(beta[0] * 252),
"alpha_t": float(t[0]),
"factors": [
{"name": INV[s], "sym": s, "beta": float(beta[k + 1]),
"t": float(t[k + 1])}
for k, s in enumerate(SYMS)],
}
def _forward(y: pd.Series, a: str | None, b: str | None) -> list[str]:
"""BIC-gated forward selection over the battery -> identified sleeves."""
x = _panel()
idx = x.index
msk = idx >= (a or idx[0])
if b is not None:
msk &= idx <= b
x = x[msk]
y = y.reindex(x.index)
ok = y.notna()
if ok.sum() < 252:
return []
cand = {s: x[s].to_numpy() for s in SYMS}
chosen, _m, _okk = decompose.forward_select(y.to_numpy(), cand, y_ok=ok)
return list(chosen)
# ------------------------------------------------------------- commentary
# human meaning of each sleeve (used for the prose)
MEANING: dict[str, str] = {
"efa": "international developed markets", "vwo": "emerging markets",
"vug": "US growth equities", "vtv": "US value equities",
"iwm": "US small caps", "rsp": "equal-weight (small-tilted) US",
"usmv": "low volatility", "dflvx": "low volatility",
"qual": "quality (high-ROE, low-debt, stable earnings)",
"mtum": "momentum (recent winners)", "hdv": "high dividend",
"vym": "high dividend (appreciation-tilted)",
"tlt": "long-duration Treasuries", "gld": "gold",
"fxe": "the euro", "fxy": "the yen", "fxu": "the Australian dollar",
"xlu": "utilities (defensive bond-proxy)",
"xlp": "consumer staples (defensive)",
"xlv": "healthcare (defensive)", "xlf": "banks/financials",
"xle": "energy",
}
# sign-specific phrasing where the generic 'tilt toward/away from' is wrong
SPECIAL: dict[tuple[str, bool], str] = {
("fxe", False): ("short-euro position that carries the USD/EUR "
"interest-rate differential"),
("fxy", False): ("short-yen position that carries the USD/JPY "
"rate differential"),
("fxu", False): "short-AUD position (USD carry)",
("fxu", True): "long-AUD position (AUD carry)",
("fxe", True): "long-euro position (exposed to EUR moves)",
("fxy", True): "long-yen position (safe-haven/carry)",
("qual", False): ("anti-quality tilt - systematically underweight "
"high-ROE, low-debt names; the mechanical flip side "
"of a high-payout dividend mandate"),
("qual", True): "quality tilt (high-ROE, low-debt, stable earnings)",
("mtum", False): ("anti-momentum character - it does not chase "
"recent winners (low-turnover/contrarian)"),
("mtum", True): "momentum tilt (preference for recent winners)",
("hdv", True): ("high-dividend tilt - a systematic preference for "
"income-paying names"),
("hdv", False): "underweight to high-dividend names",
("vym", True): "dividend-appreciation tilt (growing payers)",
("rsp", False): "cap-weight concentration (tilt away from "
"equal-weight/small)",
("rsp", True): "equal-weight (small-tilted) exposure",
("vwo", False): "developed-only tilt (away from emerging markets)",
("vwo", True): "an emerging-market tilt",
("usmv", True): "low-volatility (defensive beta) tilt",
("dflvx", True): "low-volatility (defensive beta) tilt",
("dflvx", False): "underweight to low-volatility names",
("usmv", False): "underweight to low-volatility names",
}
def _strength(t: float) -> str:
a = abs(t)
return "strong" if a >= 10 else "clear" if a >= 5 else "modest"
def _tilt_sentence(f: dict) -> str:
s = f["sym"]
key = (s, f["beta"] > 0)
if key in SPECIAL:
body = SPECIAL[key]
else:
m = MEANING.get(s, f["name"])
body = (f"tilt toward {m}" if f["beta"] > 0
else f"tilt away from {m}")
return (f"{_strength(f['t'])} {body} "
f"{f['beta']:+.2f}, t {f['t']:+.0f})")
def commentary(sym: str, prof: dict) -> list[str]:
"""English prose for the style-tilt block (data-driven)."""
if not prof or not prof.get("full"):
return []
full, rec5 = prof["full"], prof.get("rec5") or {}
sig_full = {f["sym"]: f for f in full["factors"] if abs(f["t"]) >= 2}
sig_5 = {f["sym"]: f for f in rec5.get("factors", [])
if abs(f["t"]) >= 2}
out = []
# paragraph 1: the identified stack
parts = []
core = sig_full.get("efa")
if core:
parts.append(
f"the core exposure is {abs(core['beta']):.0%} of "
f"international developed equities (t {core['t']:+.0f})")
for f in sorted(sig_full.values(), key=lambda x: -abs(x["t"])):
if f["sym"] == "efa":
continue
parts.append(_tilt_sentence(f))
if parts:
out.append(
f"Style-tilt analysis (22 style/asset sleeves regressed on "
f"the fund's excess returns; US funds proxying the global "
f"factors): over {full['start'][:4]}-to-{full['end'][:4]} the "
f"fit explains {full['r2']:.0%} of the excess return (R² "
f"{full['r2']:.2f}). The statistically identified tilt stack: "
+ "; ".join(parts) + ".")
# factors that only appear in the 5y window
recent_only = [s for s in sig_5
if s not in sig_full and s != "efa"]
if recent_only:
out.append(
"Visible only in the recent 5-year window: "
+ "; ".join(_tilt_sentence(sig_5[s]) for s in recent_only)
+ " - newer behavior, or a factor the longer sample dilutes.")
# paragraph 2: the residual - is there skill left?
a, t = full["alpha_ann"], full["alpha_t"]
if abs(t) < 1.75:
out.append(
f"After stripping the identified tilts, the residual excess "
f"return is {a * 100:+.1f}%/year (t {t:+.1f}) - NOT "
f"statistically significant: the fund's outperformance is "
f"factor exposure, not skill alpha.")
else:
out.append(
f"After stripping the identified tilts, a significant "
f"residual of {a * 100:+.1f}%/year (t {t:+.1f}) remains - "
f"genuine alpha on top of the factor stack.")
out.append(
"Caveat: the sleeves are US funds proxying global factors; the "
"fund expresses them in its own holdings, so the betas are the "
"right sign and magnitude but approximate. An index-matched "
"benchmark (the fund's own published index, when one exists) "
"would absorb part of the residual as well.")
return out
def style_profile(sym: str) -> dict | None:
"""Full + 5y battery fits, the BIC-identified stack, and commentary.
Returns None when the fund has no usable local history."""
y = _fund_excess(sym)
if len(y.dropna()) < 252:
return None
full = _fit(y, None, None)
if not full:
return None
rec5 = _fit(y, "2021-01-01", None)
prof = {"full": full,
"rec5": rec5 or {},
"identified": _forward(y, None, None)}
prof["commentary"] = commentary(sym, prof)
return prof