File size: 5,493 Bytes
ea7542e | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 | """factor_risk_model.py — Barra-Style Multi-Factor Risk Model
Decomposes portfolio risk into factor (systematic) and specific (idiosyncratic)
components. Models factor covariance using PCA + exponential weighting, and
estimates specific risk from residuals. Essential for risk budgeting and
attribution.
References:
- Grinold & Kahn 2000: "Active Portfolio Management" (Barra model)
- Menchero et al. 2010: "The Barra US Equity Model (USE4)"
- Connor et al. 2010: "The Structure of Factor Risk Premiums"
"""
import numpy as np, pandas as pd
from scipy.linalg import eigh
class FactorRiskModel:
"""Barra-style factor risk model."""
def __init__(self, n_factors=20, halflife=126):
self.n_factors = n_factors
self.halflife = halflife
self.factor_cov = None
self.factor_loadings = None
self.specific_var = None
self.factor_names = None
def _exp_weights(self, n):
lambda_ = 0.5 ** (1.0 / self.halflife)
w = np.array([lambda_ ** (n - 1 - i) for i in range(n)])
return w / w.sum()
def fit(self, returns):
"""Fit factor model via PCA with exponential weighting."""
r = returns.dropna()
T, N = r.shape
w = self._exp_weights(T)
# Weighted covariance
rw = r.values * np.sqrt(w[:, None])
cov = (rw.T @ rw) / (1 - lambda_ ** T) # normalize
# PCA
eigvals, eigvecs = eigh(cov)
idx = np.argsort(eigvals)[::-1]
eigvals = eigvals[idx]; eigvecs = eigvecs[:, idx]
self.factor_loadings = eigvecs[:, :self.n_factors]
self.factor_cov = np.diag(eigvals[:self.n_factors])
self.factor_names = [f"PC{i+1}" for i in range(self.n_factors)]
# Specific risk from residuals
factor_rets = r.values @ self.factor_loadings
explained = factor_rets @ self.factor_loadings.T
residuals = r.values - explained
self.specific_var = np.var(residuals, axis=0)
return self
def portfolio_risk(self, weights):
"""Decompose portfolio risk into factor + specific."""
w = np.array(weights).reshape(-1)
# Factor risk
factor_exposure = w @ self.factor_loadings
factor_var = factor_exposure @ self.factor_cov @ factor_exposure
# Specific risk
specific_var = np.sum((w ** 2) * self.specific_var)
total_var = factor_var + specific_var
return {
'total_vol': float(np.sqrt(total_var)),
'factor_vol': float(np.sqrt(factor_var)),
'specific_vol': float(np.sqrt(specific_var)),
'factor_pct': float(factor_var / (total_var + 1e-10) * 100),
'specific_pct': float(specific_var / (total_var + 1e-10) * 100),
'factor_exposures': dict(zip(self.factor_names, factor_exposure.tolist()))
}
def marginal_risk_contrib(self, weights):
"""Marginal risk contribution per asset."""
w = np.array(weights).reshape(-1)
total_var = self.portfolio_risk(weights)
sigma = total_var['total_vol']
# Gradient of variance w.r.t weights
cov_total = (self.factor_loadings @ self.factor_cov @ self.factor_loadings.T +
np.diag(self.specific_var))
grad = cov_total @ w
mrc = w * grad / (sigma + 1e-10)
return pd.Series(mrc, index=[f'Asset_{i}' for i in range(len(w))])
def risk_budget(self, weights, target_risk=None):
"""Risk budgeting: find weights such that each asset contributes equally."""
n = len(weights)
w0 = np.ones(n) / n
def risk_parity_objective(w):
mrc = self.marginal_risk_contrib(w)
target = mrc.sum() / n
return np.sum((mrc - target) ** 2)
# Simple iterative approach
for _ in range(100):
mrc = self.marginal_risk_contrib(w0)
w0 = w0 * (1.0 / (mrc.values + 1e-10))
w0 = w0 / w0.sum()
if target_risk:
vol = self.portfolio_risk(w0)['total_vol']
w0 = w0 * (target_risk / (vol + 1e-10))
return w0
def risk_report(self, weights):
"""Human-readable risk decomposition."""
risk = self.portfolio_risk(weights)
mrc = self.marginal_risk_contrib(weights)
report = f"""## Factor Risk Decomposition
| Risk Component | Volatility | % of Total |
|----------------|-----------|------------|
| Total | {risk['total_vol']*100:.2f}% | 100% |
| Factor (Systematic) | {risk['factor_vol']*100:.2f}% | {risk['factor_pct']:.1f}% |
| Specific (Idiosyncratic) | {risk['specific_vol']*100:.2f}% | {risk['specific_pct']:.1f}% |
**Top Factor Exposures:**
"""
top = sorted(risk['factor_exposures'].items(), key=lambda x: abs(x[1]), reverse=True)[:5]
for name, exp in top:
report += f"- {name}: {exp:.3f}\n"
report += f"\n**Top Risk Contributors:**\n"
top_mrc = mrc.sort_values(ascending=False).head(5)
for asset, contrib in top_mrc.items():
report += f"- {asset}: {contrib*100:.2f}%\n"
return report
if __name__ == '__main__':
np.random.seed(42)
returns = pd.DataFrame(np.random.normal(0.0003, 0.015, (500, 10)),
columns=[f'Stock_{i}' for i in range(10)],
index=pd.date_range('2022-01-01', periods=500, freq='B'))
model = FactorRiskModel(n_factors=5).fit(returns)
weights = np.array([0.1]*10)
print(model.risk_report(weights))
|