Deepvest/ProfilingAI
0
1import numpy as np2import pandas as pd3from typing import List, Dict, Tuple, Optional4from abc import ABC, abstractmethod5from dataclasses import dataclass6from scipy.optimize import minimize7import cvxopt8from cvxopt import matrix, solvers9import statsmodels.api as sm10from sklearn.covariance import LedoitWolf11 12from .market_states import MarketConditions, MarketRegime13 14 15@dataclass16class OptimizationConstraints:17 """Constraints for portfolio optimization"""18 min_weights: np.ndarray19 max_weights: np.ndarray20 asset_classes: Dict[str, List[int]] # Maps asset class to asset indices21 min_class_weights: Dict[str, float]22 max_class_weights: Dict[str, float]23 max_turnover: float = 0.224 min_liquidity_score: float = 0.725 tracking_error_target: Optional[float] = None26 esg_constraints: Optional[Dict[str, float]] = None27 28class PortfolioOptimizer(ABC):29 """Abstract base class defining the interface for portfolio optimization"""30 31 @abstractmethod32 def optimize_portfolio(self, constraints: OptimizationConstraints) -> np.ndarray:33 """34 Optimize portfolio weights given constraints35 36 Args:37 constraints: Portfolio optimization constraints38 39 Returns:40 Array of optimal portfolio weights41 """42 pass43 44 @abstractmethod45 def calculate_portfolio_metrics(self, weights: np.ndarray) -> Dict[str, float]:46 """47 Calculate portfolio metrics for given weights48 49 Args:50 weights: Portfolio weights51 52 Returns:53 Dictionary of portfolio metrics54 """55 pass56 57 @abstractmethod58 def generate_efficient_frontier(self, 59 constraints: OptimizationConstraints,60 n_points: int = 100) -> pd.DataFrame:61 """62 Generate efficient frontier points63 64 Args:65 constraints: Portfolio constraints66 n_points: Number of points on efficient frontier67 68 Returns:69 DataFrame with efficient frontier points70 """71 pass72 73class PortfolioOptimizerImpl(PortfolioOptimizer):74 """Implementation of portfolio optimizer using quadratic programming"""75 76 def __init__(self, 77 returns: pd.DataFrame, 78 risk_free_rate: float,79 market_conditions: Dict, 80 use_robust_covariance: bool = True):81 """82 Initialize the portfolio optimizer with historical returns data.83 84 Args:85 returns: DataFrame of historical returns86 risk_free_rate: Risk-free rate87 market_conditions: Dictionary containing market regime information88 use_robust_covariance: Whether to use robust covariance estimation89 """90 self.returns = returns91 self.risk_free_rate = risk_free_rate92 self.n_assets = returns.shape[1] if returns is not None else 093 self.market_conditions = market_conditions94 95 # Calculate covariance matrix FIRST96 try:97 if returns is not None:98 if use_robust_covariance:99 lw = LedoitWolf()100 self.covariance_matrix = lw.fit(returns).covariance_101 else:102 self.covariance_matrix = returns.cov().values103 except Exception as e:104 print(f"Error calculating covariance matrix: {e}")105 self.covariance_matrix = returns.cov().values if returns is not None else None106 107 # THEN calculate expected returns using Black-Litterman108 try:109 self.expected_returns = self._calculate_black_litterman_returns() if returns is not None else None110 except Exception as e:111 print(f"Error calculating expected returns: {e}")112 self.expected_returns = returns.mean().values if returns is not None else None113 114 def _calculate_black_litterman_returns(self) -> np.ndarray:115 """Calculate expected returns using the Black-Litterman model."""116 try:117 market_weights = self._get_market_weights()118 risk_aversion = 2.5119 120 # Calculate prior returns using the already computed covariance matrix121 pi = risk_aversion * self.covariance_matrix @ market_weights122 123 # Get views124 P, Q, omega = self._generate_views()125 126 # Calculate posterior127 tau = 0.05128 sigma_prior = tau * self.covariance_matrix129 130 # Add small constant to diagonal for numerical stability131 sigma_prior += np.eye(self.n_assets) * 1e-8132 omega += np.eye(omega.shape[0]) * 1e-8133 134 # Calculate posterior distribution135 sigma_post = np.linalg.inv(136 np.linalg.inv(sigma_prior) + 137 P.T @ np.linalg.inv(omega) @ P138 )139 140 mu_post = sigma_post @ (141 np.linalg.inv(sigma_prior) @ pi + 142 P.T @ np.linalg.inv(omega) @ Q143 )144 145 return mu_post146 147 except Exception as e:148 print(f"Error in Black-Litterman calculation: {e}")149 return self.returns.mean().values # Fallback to historical means150 151 def _generate_views(self) -> Tuple[np.ndarray, np.ndarray, np.ndarray]:152 """Generate views based on market conditions."""153 n_views = 3154 P = np.zeros((n_views, self.n_assets))155 Q = np.zeros(n_views)156 157 # View 1: Based on market regime158 regime = self.market_conditions.get('current_regime', 'neutral')159 if regime == "bull_market":160 P[0, :] = [1 if i < self.n_assets // 2 else -1 161 for i in range(self.n_assets)]162 Q[0] = 0.05163 elif regime == "bear_market":164 P[0, :] = [-1 if i < self.n_assets // 2 else 1 165 for i in range(self.n_assets)]166 Q[0] = -0.03167 168 # View 2: Based on volatility level169 volatility = self.market_conditions.get('volatility_level', 'medium')170 if volatility == "high":171 P[1, :] = [1 if i % 2 == 0 else -1 172 for i in range(self.n_assets)]173 Q[1] = -0.02174 175 # View 3: Based on market sentiment176 sentiment = self.market_conditions.get('market_sentiment', 'neutral')177 if sentiment == "positive":178 P[2, :] = [1 if i % 3 == 0 else 0 179 for i in range(self.n_assets)]180 Q[2] = 0.04181 182 # Uncertainty in views183 omega = np.eye(n_views) * 0.05184 185 return P, Q, omega186 187 def _get_market_weights(self) -> np.ndarray:188 """Get market capitalization weights (equal-weighted for simplicity)."""189 return np.ones(self.n_assets) / self.n_assets190 191 def optimize_portfolio(self, constraints: OptimizationConstraints) -> np.ndarray:192 """Optimize portfolio weights using quadratic programming"""193 try:194 if self.returns is None or self.n_assets == 0:195 raise ValueError("No return data available for optimization")196 197 # S'assurer que la matrice de covariance a les bonnes dimensions198 cov_matrix = self.covariance_matrix199 if isinstance(cov_matrix, pd.DataFrame):200 cov_matrix = cov_matrix.values201 202 # S'assurer que les rendements attendus ont les bonnes dimensions 203 exp_returns = self.expected_returns204 if isinstance(exp_returns, pd.Series):205 exp_returns = exp_returns.values206 207 # Reshape pour assurer les bonnes dimensions208 exp_returns = exp_returns.reshape(-1, 1)209 210 # Préparer les matrices pour l'optimisation quadratique211 P = matrix(cov_matrix)212 q = matrix(-exp_returns)213 214 # Contraintes de somme = 1215 A = matrix(1.0, (1, self.n_assets))216 b = matrix(1.0)217 218 # Contraintes de bornes219 min_weights = constraints.min_weights.reshape(-1)220 max_weights = constraints.max_weights.reshape(-1)221 222 G = matrix(np.vstack((-np.eye(self.n_assets), np.eye(self.n_assets))))223 h = matrix(np.concatenate((-min_weights, max_weights)))224 225 # Résoudre l'optimisation226 solvers.options['show_progress'] = False227 solution = solvers.qp(P, q, G, h, A, b)228 229 if solution['status'] != 'optimal':230 raise ValueError(f"Optimization not optimal, status: {solution['status']}")231 232 weights = np.array(solution['x']).flatten()233 weights = np.clip(weights, 0, 1)234 weights = weights / np.sum(weights) # Normaliser235 236 return weights237 238 except Exception as e:239 print(f"Optimization error: {str(e)}")240 return np.ones(self.n_assets) / self.n_assets # Allocation égale en cas d'erreur241 242 def calculate_portfolio_metrics(self, weights: np.ndarray) -> Dict[str, float]:243 """Calculate various portfolio metrics"""244 try:245 if self.returns is None:246 raise ValueError("No return data available for metric calculation")247 248 portfolio_return = weights @ self.expected_returns249 portfolio_vol = np.sqrt(weights @ self.covariance_matrix @ weights)250 251 # Calculate various risk metrics252 returns = self.returns @ weights253 254 metrics = {255 'expected_return': portfolio_return,256 'volatility': portfolio_vol,257 'sharpe_ratio': (portfolio_return - self.risk_free_rate) / portfolio_vol,258 'sortino_ratio': self._calculate_sortino_ratio(returns),259 'max_drawdown': self._calculate_max_drawdown(returns),260 'var_95': self._calculate_var(returns, 0.95),261 'cvar_95': self._calculate_cvar(returns, 0.95),262 'beta': self._calculate_beta(returns),263 'tracking_error': self._calculate_tracking_error(returns)264 }265 266 return metrics267 268 except Exception as e:269 print(f"Error calculating portfolio metrics: {e}")270 return {}271 272 def generate_efficient_frontier(self, 273 constraints: OptimizationConstraints,274 n_points: int = 100) -> pd.DataFrame:275 """Generate efficient frontier points"""276 try:277 if self.returns is None:278 raise ValueError("No return data available for efficient frontier")279 280 min_return = min(self.expected_returns)281 max_return = max(self.expected_returns)282 target_returns = np.linspace(min_return, max_return, n_points)283 284 results = []285 for target_return in target_returns:286 try:287 weights = self.optimize_portfolio(constraints)288 metrics = self.calculate_portfolio_metrics(weights)289 results.append({290 'target_return': target_return,291 'actual_return': metrics['expected_return'],292 'volatility': metrics['volatility'],293 'sharpe_ratio': metrics['sharpe_ratio'],294 'weights': weights295 })296 except ValueError:297 continue298 299 return pd.DataFrame(results)300 301 except Exception as e:302 print(f"Error generating efficient frontier: {e}")303 return pd.DataFrame()304 305 @staticmethod306 def _calculate_sortino_ratio(returns: np.ndarray) -> float:307 """Calculate Sortino ratio."""308 negative_returns = returns[returns < 0]309 downside_std = np.std(negative_returns) if len(negative_returns) > 0 else 1e-6310 return np.mean(returns) / downside_std311 312 @staticmethod313 def _calculate_max_drawdown(returns: np.ndarray) -> float:314 """Calculate maximum drawdown."""315 cumulative = (1 + returns).cumprod()316 running_max = np.maximum.accumulate(cumulative)317 drawdown = (cumulative - running_max) / running_max318 return np.min(drawdown)319 320 @staticmethod321 def _calculate_var(returns: np.ndarray, confidence: float) -> float:322 """Calculate Value at Risk."""323 return np.percentile(returns, (1 - confidence) * 100)324 325 @staticmethod326 def _calculate_cvar(returns: np.ndarray, confidence: float) -> float:327 """Calculate Conditional Value at Risk."""328 var = np.percentile(returns, (1 - confidence) * 100)329 return np.mean(returns[returns <= var])330 331 def _calculate_beta(self, returns: np.ndarray) -> float:332 """Calculate portfolio beta."""333 if self.returns is None:334 return 0.0335 market_returns = self.returns.mean(axis=1) # Using equal-weighted market336 covariance = np.cov(returns, market_returns)[0, 1]337 market_variance = np.var(market_returns)338 return covariance / market_variance339 340 def _calculate_tracking_error(self, returns: np.ndarray) -> float:341 """Calculate tracking error relative to benchmark."""342 if self.returns is None:343 return 0.0344 benchmark_returns = self.returns.mean(axis=1) # Using equal-weighted benchmark345 return np.std(returns - benchmark_returns)346 347 def generate_efficient_frontier(self, 348 constraints: OptimizationConstraints,349 n_points: int = 100) -> pd.DataFrame:350 """351 Generate efficient frontier points.352 353 Args:354 constraints: Portfolio constraints355 n_points: Number of points on the efficient frontier356 357 Returns:358 DataFrame with efficient frontier points359 """360 min_return = min(self.expected_returns)361 max_return = max(self.expected_returns)362 target_returns = np.linspace(min_return, max_return, n_points)363 364 results = []365 for target_return in target_returns:366 try:367 weights = self.optimize_portfolio(constraints, target_return=target_return)368 metrics = self.calculate_portfolio_metrics(weights)369 results.append({370 'target_return': target_return,371 'actual_return': metrics['expected_return'],372 'volatility': metrics['volatility'],373 'sharpe_ratio': metrics['sharpe_ratio'],374 'weights': weights375 })376 except ValueError:377 continue378 379 return pd.DataFrame(results)380