capm_analysis.py

"""
CAPM (Capital Asset Pricing Model) Analysis Module
Implements beta calculation, alpha analysis, and Security Market Line
"""

import numpy as np
import pandas as pd
from typing import Dict, List, Tuple
from sklearn.linear_model import LinearRegression
import warnings
warnings.filterwarnings('ignore')


class CAPMAnalyzer:
    """Class for CAPM analysis including beta, alpha, and SML"""
    
    def __init__(self, stock_returns: Dict[str, pd.Series], market_returns: pd.Series, risk_free_rate: float = 0.02):
        """
        Initialize CAPM analyzer
        
        Args:
            stock_returns: Dictionary of stock returns
            market_returns: Market index returns (e.g., S&P 500)
            risk_free_rate: Annual risk-free rate (default 2%)
        """
        self.stock_returns = stock_returns
        self.market_returns = market_returns
        self.risk_free_rate = risk_free_rate
        self.stock_symbols = list(stock_returns.keys())
        
    def calculate_beta_alpha(self, stock_symbol: str) -> Dict[str, float]:
        """
        Calculate beta and alpha for a specific stock
        
        Args:
            stock_symbol: Stock symbol to analyze
            
        Returns:
            Dictionary with beta, alpha, and other CAPM metrics
        """
        if stock_symbol not in self.stock_returns:
            raise ValueError(f"Stock {stock_symbol} not found in data")
            
        # Align stock and market returns by date
        stock_ret = self.stock_returns[stock_symbol]
        market_ret = self.market_returns
        
        # Find common dates
        common_dates = stock_ret.index.intersection(market_ret.index)
        stock_aligned = stock_ret.loc[common_dates]
        market_aligned = market_ret.loc[common_dates]
        
        if len(stock_aligned) < 30:  # Need sufficient data points
            return {'error': 'Insufficient data points for CAPM analysis'}
        
        # Calculate excess returns (returns - risk_free_rate/252 for daily data)
        daily_rf_rate = self.risk_free_rate / 252
        stock_excess = stock_aligned - daily_rf_rate
        market_excess = market_aligned - daily_rf_rate
        
        # Perform linear regression: R_stock - Rf = alpha + beta * (R_market - Rf) + epsilon
        X = market_excess.values.reshape(-1, 1)
        y = stock_excess.values
        
        model = LinearRegression()
        model.fit(X, y)
        
        beta = model.coef_[0]
        alpha = model.intercept_
        
        # Calculate R-squared
        y_pred = model.predict(X)
        ss_res = np.sum((y - y_pred) ** 2)
        ss_tot = np.sum((y - np.mean(y)) ** 2)
        r_squared = 1 - (ss_res / ss_tot) if ss_tot != 0 else 0
        
        # Annualize alpha
        alpha_annualized = alpha * 252
        
        # Calculate expected return using CAPM
        market_return_annual = market_aligned.mean() * 252
        expected_return = self.risk_free_rate + beta * (market_return_annual - self.risk_free_rate)
        actual_return = stock_aligned.mean() * 252
        
        return {
            'beta': beta,
            'alpha': alpha_annualized,
            'alpha_daily': alpha,
            'r_squared': r_squared,
            'expected_return': expected_return,
            'actual_return': actual_return,
            'excess_return': actual_return - expected_return,
            'data_points': len(stock_aligned)
        }
    
    def analyze_all_stocks(self) -> pd.DataFrame:
        """
        Perform CAPM analysis for all stocks
        
        Returns:
            DataFrame with CAPM metrics for all stocks
        """
        results = []
        
        for symbol in self.stock_symbols:
            try:
                metrics = self.calculate_beta_alpha(symbol)
                if 'error' not in metrics:
                    metrics['symbol'] = symbol
                    results.append(metrics)
            except Exception as e:
                print(f"Error analyzing {symbol}: {str(e)}")
        
        if not results:
            return pd.DataFrame()
            
        df = pd.DataFrame(results)
        df = df.set_index('symbol')
        
        # Round numerical columns
        numeric_cols = ['beta', 'alpha', 'r_squared', 'expected_return', 'actual_return', 'excess_return']
        for col in numeric_cols:
            if col in df.columns:
                df[col] = df[col].round(4)
        
        return df
    
    def calculate_security_market_line(self, beta_range: Tuple[float, float] = (-1, 3)) -> pd.DataFrame:
        """
        Calculate Security Market Line (SML) points
        
        Args:
            beta_range: Range of beta values for SML
            
        Returns:
            DataFrame with SML data points
        """
        market_return_annual = self.market_returns.mean() * 252
        market_premium = market_return_annual - self.risk_free_rate
        
        # Generate beta values
        betas = np.linspace(beta_range[0], beta_range[1], 100)
        
        # Calculate expected returns using CAPM
        expected_returns = self.risk_free_rate + betas * market_premium
        
        sml_data = pd.DataFrame({
            'beta': betas,
            'expected_return': expected_returns
        })
        
        return sml_data
    
    def get_stock_positions_on_sml(self) -> pd.DataFrame:
        """
        Get actual stock positions relative to Security Market Line
        
        Returns:
            DataFrame with stock positions
        """
        capm_results = self.analyze_all_stocks()
        
        if capm_results.empty:
            return pd.DataFrame()
        
        # Add market premium information
        market_return_annual = self.market_returns.mean() * 252
        market_premium = market_return_annual - self.risk_free_rate
        
        positions = capm_results[['beta', 'actual_return', 'expected_return']].copy()
        positions['market_premium'] = market_premium
        positions['risk_free_rate'] = self.risk_free_rate
        
        return positions
    
    def calculate_portfolio_beta(self, weights: Dict[str, float]) -> float:
        """
        Calculate portfolio beta as weighted average of individual stock betas
        
        Args:
            weights: Dictionary with stock symbols as keys and weights as values
            
        Returns:
            Portfolio beta
        """
        capm_results = self.analyze_all_stocks()
        
        if capm_results.empty:
            return np.nan
        
        portfolio_beta = 0
        total_weight = 0
        
        for symbol, weight in weights.items():
            if symbol in capm_results.index:
                portfolio_beta += weight * capm_results.loc[symbol, 'beta']
                total_weight += weight
        
        return portfolio_beta / total_weight if total_weight > 0 else np.nan
    
    def calculate_portfolio_alpha(self, weights: Dict[str, float]) -> float:
        """
        Calculate portfolio alpha as weighted average of individual stock alphas
        
        Args:
            weights: Dictionary with stock symbols as keys and weights as values
            
        Returns:
            Portfolio alpha
        """
        capm_results = self.analyze_all_stocks()
        
        if capm_results.empty:
            return np.nan
        
        portfolio_alpha = 0
        total_weight = 0
        
        for symbol, weight in weights.items():
            if symbol in capm_results.index:
                portfolio_alpha += weight * capm_results.loc[symbol, 'alpha']
                total_weight += weight
        
        return portfolio_alpha / total_weight if total_weight > 0 else np.nan
    
    def get_risk_metrics(self) -> Dict[str, float]:
        """
        Calculate overall market risk metrics
        
        Returns:
            Dictionary with market risk metrics
        """
        market_return_annual = self.market_returns.mean() * 252
        market_volatility = self.market_returns.std() * np.sqrt(252)
        market_sharpe = (market_return_annual - self.risk_free_rate) / market_volatility
        
        return {
            'market_return': market_return_annual,
            'market_volatility': market_volatility,
            'market_sharpe_ratio': market_sharpe,
            'risk_free_rate': self.risk_free_rate,
            'market_premium': market_return_annual - self.risk_free_rate
        }