Add Barra-style multi-factor risk model with factor covariance and specific risk

factor_risk_model.py

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