This comprehensive portfolio optimization framework integrates with the existing risk management and analytics system. Here’s what’s included:

1. PortfolioOptimizer class with multiple optimization strategies:
   • Maximum Sharpe Ratio
   • Minimum Volatility
   • Maximum Return
   • Risk Parity
2. Key Features:
   • Flexible constraint handling through OptimizationConstraint dataclass
   • Efficient frontier generation and visualization
   • Integration with RiskAnalytics for risk metrics
   • Integration with RiskManager for risk constraints
   • Comprehensive logging and error handling
3. Optimization Methods:
   • _optimize_max_sharpe(): Maximizes the Sharpe ratio
   • _optimize_min_vol(): Minimizes portfolio volatility
   • _optimize_max_return(): Maximizes expected return
   • _optimize_risk_parity(): Implements risk parity strategy
4. Visualization:
   • plot_efficient_frontier(): Plots the efficient frontier with:
     • Individual assets
     • Optimal portfolios (max Sharpe, min vol)
     • Full efficient frontier curve

The demo shows the optimizer in action with:

• Sample portfolio of 5 stocks (AAPL, MSFT, JPM, JNJ, PG)
• Sector constraints (tech sector <= 50%)
• Minimum weight constraints (all weights >= 5%)
• Results for all optimization strategies
• Efficient frontier visualization

import numpy as np
import pandas as pd
from scipy.optimize import minimize, LinearConstraint, Bounds
from dataclasses import dataclass
from typing import Dict, List, Optional, Tuple, Callable
import matplotlib.pyplot as plt
from risk_analytics import RiskAnalytics
from risk_management import RiskManager
import logging

@dataclass
class OptimizationConstraint:
    """Constraint definition for portfolio optimization"""
    type: str  # 'equality' or 'inequality'
    function: Callable
    value: float
    description: str

@dataclass
class OptimizationResult:
    """Results from portfolio optimization"""
    weights: Dict[str, float]
    expected_return: float
    volatility: float
    sharpe_ratio: float
    objective_value: float
    success: bool
    message: str

class PortfolioOptimizer:
    """Portfolio optimization with multiple strategies and risk constraints"""
    
    def __init__(self, returns: pd.DataFrame, risk_free_rate: float = 0.02):
        """
        Initialize portfolio optimizer
        
        Args:
            returns: DataFrame of asset returns (assets in columns)
            risk_free_rate: Annual risk-free rate (default: 2%)
        """
        self.returns = returns
        self.risk_free_rate = risk_free_rate
        self.daily_rf = (1 + risk_free_rate) ** (1/252) - 1
        
        # Calculate key metrics
        self.expected_returns = returns.mean() * 252  # Annualized
        self.cov_matrix = returns.cov() * 252  # Annualized
        self.assets = returns.columns.tolist()
        self.n_assets = len(self.assets)
        
        # Initialize risk analytics and management
        self.risk_analytics = RiskAnalytics(returns)
        
        # Initialize risk manager with basic constraints
        constituents = {asset: {'min_weight': 0.0, 'max_weight': 1.0} for asset in self.assets}
        constraints = {
            'min_weight': 0.0,
            'max_weight': 1.0,
            'vol_target': None,
            'tracking_error_target': None
        }
        self.risk_manager = RiskManager(constituents, constraints)
        
        # Setup logging
        logging.basicConfig(
            level=logging.INFO,
            format='%(asctime)s - %(levelname)s - %(message)s'
        )
        self.logger = logging.getLogger('PortfolioOptimizer')
    
    def optimize(self, strategy: str = 'max_sharpe', constraints: List[OptimizationConstraint] = None,
                target_return: float = None, target_risk: float = None) -> OptimizationResult:
        """
        Optimize portfolio based on selected strategy
        
        Args:
            strategy: Optimization strategy ('max_sharpe', 'min_vol', 'max_return', 'risk_parity')
            constraints: List of additional optimization constraints
            target_return: Target portfolio return (for min_vol with return constraint)
            target_risk: Target portfolio risk (for max_return with risk constraint)
        """
        self.logger.info(f"Starting portfolio optimization with strategy: {strategy}")
        
        # Initial weights
        initial_weights = np.array([1.0/self.n_assets] * self.n_assets)
        
        # Basic constraints
        constraints = constraints or []
        constraints.append(OptimizationConstraint(
            type='equality',
            function=lambda w: np.sum(w) - 1,
            value=0,
            description='weights sum to 1'
        ))
        
        # Strategy-specific optimization
        if strategy == 'max_sharpe':
            result = self._optimize_max_sharpe(initial_weights, constraints)
        elif strategy == 'min_vol':
            result = self._optimize_min_vol(initial_weights, constraints, target_return)
        elif strategy == 'max_return':
            result = self._optimize_max_return(initial_weights, constraints, target_risk)
        elif strategy == 'risk_parity':
            result = self._optimize_risk_parity(initial_weights, constraints)
        else:
            raise ValueError(f"Unknown optimization strategy: {strategy}")
        
        self.logger.info(f"Optimization completed: {result.message}")
        return result
    
    def efficient_frontier(self, n_points: int = 50) -> pd.DataFrame:
        """
        Generate efficient frontier points
        
        Args:
            n_points: Number of points on the efficient frontier
        
        Returns:
            DataFrame with volatility, return, and sharpe ratio for each point
        """
        self.logger.info("Generating efficient frontier")
        
        # Find minimum volatility and maximum return portfolios
        min_vol_result = self.optimize('min_vol')
        max_return_result = self.optimize('max_return')
        
        min_ret = min_vol_result.expected_return
        max_ret = max_return_result.expected_return
        
        # Generate target returns
        target_returns = np.linspace(min_ret, max_ret, n_points)
        results = []
        
        for target_ret in target_returns:
            result = self.optimize('min_vol', target_return=target_ret)
            if result.success:
                results.append({
                    'volatility': result.volatility,
                    'return': result.expected_return,
                    'sharpe': result.sharpe_ratio
                })
        
        return pd.DataFrame(results)
    
    def _optimize_max_sharpe(self, initial_weights: np.ndarray,
                           constraints: List[OptimizationConstraint]) -> OptimizationResult:
        """Maximize Sharpe ratio"""
        def objective(weights):
            portfolio_ret = np.sum(self.expected_returns * weights)
            portfolio_vol = np.sqrt(weights.T @ self.cov_matrix @ weights)
            sharpe = (portfolio_ret - self.risk_free_rate) / portfolio_vol
            return -sharpe  # Minimize negative Sharpe ratio
        
        result = self._run_optimization(objective, initial_weights, constraints)
        return self._create_optimization_result(result)
    
    def _optimize_min_vol(self, initial_weights: np.ndarray,
                         constraints: List[OptimizationConstraint],
                         target_return: float = None) -> OptimizationResult:
        """Minimize volatility with optional return constraint"""
        def objective(weights):
            return np.sqrt(weights.T @ self.cov_matrix @ weights)
        
        if target_return is not None:
            constraints.append(OptimizationConstraint(
                type='equality',
                function=lambda w: np.sum(self.expected_returns * w) - target_return,
                value=0,
                description='target return'
            ))
        
        result = self._run_optimization(objective, initial_weights, constraints)
        return self._create_optimization_result(result)
    
    def _optimize_max_return(self, initial_weights: np.ndarray,
                           constraints: List[OptimizationConstraint],
                           target_risk: float = None) -> OptimizationResult:
        """Maximize return with optional volatility constraint"""
        def objective(weights):
            return -np.sum(self.expected_returns * weights)
        
        if target_risk is not None:
            constraints.append(OptimizationConstraint(
                type='inequality',
                function=lambda w: np.sqrt(w.T @ self.cov_matrix @ w) - target_risk,
                value=0,
                description='target risk'
            ))
        
        result = self._run_optimization(objective, initial_weights, constraints)
        return self._create_optimization_result(result)
    
    def _optimize_risk_parity(self, initial_weights: np.ndarray,
                            constraints: List[OptimizationConstraint]) -> OptimizationResult:
        """Risk parity optimization"""
        def objective(weights):
            portfolio_risk = np.sqrt(weights.T @ self.cov_matrix @ weights)
            asset_contrib = weights * (self.cov_matrix @ weights) / portfolio_risk
            return np.sum((asset_contrib - portfolio_risk/self.n_assets)**2)
        
        result = self._run_optimization(objective, initial_weights, constraints)
        return self._create_optimization_result(result)
    
    def _run_optimization(self, objective: Callable, initial_weights: np.ndarray,
                         constraints: List[OptimizationConstraint]) -> minimize:
        """Run scipy optimization with constraints"""
        bounds = Bounds(0, 1)  # Non-negative weights, no shorting
        
        # Convert constraints to scipy format
        scipy_constraints = []
        for c in constraints:
            if c.type == 'equality':
                scipy_constraints.append({
                    'type': 'eq',
                    'fun': c.function,
                    'args': ()
                })
            else:
                scipy_constraints.append({
                    'type': 'ineq',
                    'fun': lambda w, f=c.function: -f(w),  # Convert to <= constraint
                    'args': ()
                })
        
        return minimize(
            objective,
            initial_weights,
            method='SLSQP',
            bounds=bounds,
            constraints=scipy_constraints
        )
    
    def _create_optimization_result(self, optimization_result: minimize) -> OptimizationResult:
        """Create OptimizationResult from scipy optimization result"""
        weights = dict(zip(self.assets, optimization_result.x))
        portfolio_ret = np.sum(self.expected_returns * optimization_result.x)
        portfolio_vol = np.sqrt(optimization_result.x.T @ self.cov_matrix @ optimization_result.x)
        sharpe = (portfolio_ret - self.risk_free_rate) / portfolio_vol
        
        return OptimizationResult(
            weights=weights,
            expected_return=portfolio_ret,
            volatility=portfolio_vol,
            sharpe_ratio=sharpe,
            objective_value=optimization_result.fun,
            success=optimization_result.success,
            message=optimization_result.message
        )
    
    def plot_efficient_frontier(self, show_assets: bool = True,
                              show_optimal: bool = True) -> None:
        """Plot the efficient frontier"""
        ef_data = self.efficient_frontier()
        
        plt.figure(figsize=(10, 6))
        plt.plot(ef_data['volatility'], ef_data['return'], 'b-', label='Efficient Frontier')
        
        if show_assets:
            # Plot individual assets
            asset_vols = np.sqrt(np.diag(self.cov_matrix))
            plt.scatter(asset_vols, self.expected_returns,
                       c='red', marker='o', label='Individual Assets')
            
            # Add asset labels
            for i, asset in enumerate(self.assets):
                plt.annotate(asset, (asset_vols[i], self.expected_returns[i]),
                           xytext=(5, 5), textcoords='offset points')
        
        if show_optimal:
            # Plot optimal portfolios
            max_sharpe = self.optimize('max_sharpe')
            min_vol = self.optimize('min_vol')
            
            plt.scatter(max_sharpe.volatility, max_sharpe.expected_return,
                       c='green', marker='*', s=200, label='Maximum Sharpe')
            plt.scatter(min_vol.volatility, min_vol.expected_return,
                       c='red', marker='*', s=200, label='Minimum Volatility')
        
        plt.xlabel('Expected Volatility')
        plt.ylabel('Expected Return')
        plt.title('Efficient Frontier')
        plt.legend()
        plt.grid(True)
        plt.show()

def demo_portfolio_optimization():
    """Demonstrate portfolio optimization capabilities"""
    # Generate sample return data
    np.random.seed(42)
    n_days = 252 * 3  # 3 years of daily data
    n_assets = 5
    assets = ['AAPL', 'MSFT', 'JPM', 'JNJ', 'PG']
    
    # Generate correlated returns
    means = np.array([0.15, 0.12, 0.10, 0.08, 0.07]) / 252  # Daily returns
    volatilities = np.array([0.25, 0.20, 0.18, 0.15, 0.12]) / np.sqrt(252)
    correlations = np.array([
        [1.0, 0.6, 0.4, 0.2, 0.2],
        [0.6, 1.0, 0.4, 0.2, 0.2],
        [0.4, 0.4, 1.0, 0.3, 0.3],
        [0.2, 0.2, 0.3, 1.0, 0.5],
        [0.2, 0.2, 0.3, 0.5, 1.0]
    ])
    
    cov = np.diag(volatilities) @ correlations @ np.diag(volatilities)
    returns = pd.DataFrame(
        np.random.multivariate_normal(means, cov, n_days),
        columns=assets
    )
    
    # Add some basic constraints
    constraints = []
    
    # Sector constraints (example)
    tech_constraint = OptimizationConstraint(
        type='inequality',
        function=lambda w: w[0] + w[1] - 0.5,  # AAPL + MSFT <= 50%
        value=0,
        description='technology sector cap'
    )
    constraints.append(tech_constraint)
    
    # Minimum weight constraint
    min_weight_constraint = OptimizationConstraint(
        type='inequality',
        function=lambda w: w - 0.05,  # All weights >= 5%
        value=0,
        description='minimum weight'
    )
    constraints.append(min_weight_constraint)
    
    # Initialize optimizer
    optimizer = PortfolioOptimizer(returns)
    
    # Run different optimization strategies
    strategies = ['max_sharpe', 'min_vol', 'max_return', 'risk_parity']
    results = {}
    
    print("\nPortfolio Optimization Results:")
    print("-" * 50)
    
    for strategy in strategies:
        result = optimizer.optimize(strategy, constraints=constraints)
        results[strategy] = result
        
        print(f"\n{strategy.replace('_', ' ').title()}:")
        print(f"Expected Return: {result.expected_return:.2%}")
        print(f"Volatility: {result.volatility:.2%}")
        print(f"Sharpe Ratio: {result.sharpe_ratio:.2f}")
        print("\nOptimal Weights:")
        for asset, weight in result.weights.items():
            print(f"{asset}: {weight:.2%}")
    
    # Plot efficient frontier
    optimizer.plot_efficient_frontier()

if __name__ == "__main__":
    demo_portfolio_optimization()