Upload factor_decomposition.py

factor_decomposition.py

@@ -0,0 +1,81 @@
+"""Factor Decomposition Engine - Break returns into style factors."""
+import numpy as np
+import pandas as pd
+from typing import Dict, Optional
+import warnings
+warnings.filterwarnings('ignore')
+
+
+class FactorDecomposition:
+    """Decompose returns into style factors."""
+    
+    def __init__(self):
+        self.factor_names = ['momentum', 'value', 'size', 'volatility', 'quality', 'market']
+        self.factor_returns = None
+        self.exposures = None
+        
+    def compute_factor_returns(self, returns_df: pd.DataFrame) -> pd.DataFrame:
+        factors = pd.DataFrame(index=returns_df.index)
+        factors['market'] = returns_df.mean(axis=1)
+        
+        momentum_rets = []; value_rets = []; size_rets = []; vol_rets = []; quality_rets = []
+        
+        for date in returns_df.index:
+            day_returns = returns_df.loc[date].dropna()
+            if len(day_returns) < 5:
+                for lst in [momentum_rets, value_rets, size_rets, vol_rets, quality_rets]:
+                    lst.append(0)
+                continue
+            
+            past_returns = returns_df.loc[:date].iloc[-21:-1].mean().reindex(day_returns.index).fillna(0) if date in returns_df.index else pd.Series(0, index=day_returns.index)
+            value_score = 1.0 / (1 + day_returns.abs())
+            size_score = 1.0 / (day_returns.rolling(21).std().reindex(day_returns.index).fillna(0.01) + 0.01)
+            vol_score = -day_returns.rolling(21).std().reindex(day_returns.index).fillna(0.01)
+            quality_score = (day_returns > 0).astype(float)
+            
+            X = pd.DataFrame({'momentum': past_returns, 'value': value_score, 'size': size_score,
+                              'volatility': vol_score, 'quality': quality_score}).fillna(0)
+            X = (X - X.mean()) / (X.std() + 1e-8)
+            y = day_returns.reindex(X.index).fillna(0)
+            
+            try:
+                coefs = np.linalg.lstsq(X.values, y.values, rcond=None)[0]
+            except:
+                coefs = np.zeros(5)
+            
+            momentum_rets.append(coefs[0]); value_rets.append(coefs[1]); size_rets.append(coefs[2])
+            vol_rets.append(coefs[3]); quality_rets.append(coefs[4])
+        
+        factors['momentum'] = momentum_rets; factors['value'] = value_rets
+        factors['size'] = size_rets; factors['volatility'] = vol_rets; factors['quality'] = quality_rets
+        
+        self.factor_returns = factors
+        return factors
+    
+    def compute_exposures(self, asset_returns: pd.Series, factor_returns: pd.DataFrame, window: int = 63) -> pd.DataFrame:
+        from sklearn.linear_model import Ridge
+        exposures = pd.DataFrame(index=asset_returns.index, columns=factor_returns.columns)
+        
+        for factor in factor_returns.columns:
+            for i in range(len(asset_returns)):
+                if i < window: exposures.iloc[i][factor] = 0; continue
+                y = asset_returns.iloc[i-window:i].values
+                X = factor_returns[factor].iloc[i-window:i].values.reshape(-1, 1)
+                try:
+                    model = Ridge(alpha=1.0).fit(X, y)
+                    exposures.iloc[i][factor] = model.coef_[0]
+                except:
+                    exposures.iloc[i][factor] = 0
+        
+        self.exposures = exposures
+        return exposures
+    
+    def attribution(self, portfolio_returns: pd.Series, factor_returns: pd.DataFrame, exposures: pd.DataFrame) -> Dict:
+        factor_contrib = exposures.multiply(factor_returns.reindex(exposures.index).fillna(0)).sum(axis=1)
+        residual = portfolio_returns - factor_contrib
+        return {
+            'total_return': portfolio_returns.sum(),
+            'factor_return': factor_contrib.sum(),
+            'residual_return': residual.sum(),
+            'r_squared': 1 - residual.var() / portfolio_returns.var() if portfolio_returns.var() > 0 else 0
+        }