mcp-optimizer

Instructions

Implementation Reference

• src/mcp_optimizer/tools/financial.py:476-512 (handler)

  The primary handler function for the optimize_portfolio_tool MCP tool. It receives input parameters and delegates to the optimize_portfolio helper function to perform the actual computation.
  def optimize_portfolio_tool( assets: list[dict[str, Any]], objective: str = "maximize_return", budget: float = 1.0, risk_tolerance: float | None = None, sector_constraints: dict[str, float] | None = None, min_allocation: float = 0.0, max_allocation: float = 1.0, solver_name: str = "CBC", time_limit_seconds: float = 30.0, ) -> dict[str, Any]: """Optimize portfolio allocation to maximize return or minimize risk. Args: assets: List of asset dictionaries with expected return, risk, and sector objective: Optimization objective ("maximize_return", "minimize_risk", "maximize_sharpe", "risk_parity") budget: Total budget to allocate (default: 1.0) risk_tolerance: Maximum acceptable portfolio risk (optional) sector_constraints: Maximum allocation per sector (optional) min_allocation: Minimum allocation per asset (default: 0.0) max_allocation: Maximum allocation per asset (default: 1.0) solver_name: Solver to use ("CBC", "GLPK", "GUROBI", "CPLEX") time_limit_seconds: Maximum solving time in seconds (default: 30.0) Returns: Optimization result with optimal portfolio allocation """ return optimize_portfolio( assets, objective, budget, risk_tolerance, sector_constraints, min_allocation, max_allocation, )
• src/mcp_optimizer/tools/financial.py:472-512 (registration)

  Function that registers the financial tools, including optimize_portfolio_tool, by defining it with the @mcp.tool() decorator.
  def register_financial_tools(mcp: FastMCP[Any]) -> None: """Register financial optimization tools with MCP server.""" @mcp.tool() def optimize_portfolio_tool( assets: list[dict[str, Any]], objective: str = "maximize_return", budget: float = 1.0, risk_tolerance: float | None = None, sector_constraints: dict[str, float] | None = None, min_allocation: float = 0.0, max_allocation: float = 1.0, solver_name: str = "CBC", time_limit_seconds: float = 30.0, ) -> dict[str, Any]: """Optimize portfolio allocation to maximize return or minimize risk. Args: assets: List of asset dictionaries with expected return, risk, and sector objective: Optimization objective ("maximize_return", "minimize_risk", "maximize_sharpe", "risk_parity") budget: Total budget to allocate (default: 1.0) risk_tolerance: Maximum acceptable portfolio risk (optional) sector_constraints: Maximum allocation per sector (optional) min_allocation: Minimum allocation per asset (default: 0.0) max_allocation: Maximum allocation per asset (default: 1.0) solver_name: Solver to use ("CBC", "GLPK", "GUROBI", "CPLEX") time_limit_seconds: Maximum solving time in seconds (default: 30.0) Returns: Optimization result with optimal portfolio allocation """ return optimize_portfolio( assets, objective, budget, risk_tolerance, sector_constraints, min_allocation, max_allocation, )
• src/mcp_optimizer/mcp_server.py:143-143 (registration)

  The call to register_financial_tools during MCP server creation, which triggers the registration of the optimize_portfolio_tool.
  register_financial_tools(mcp)
• src/mcp_optimizer/tools/financial.py:21-89 (schema)

  Pydantic BaseModel classes (Asset and PortfolioInput) that define and validate the structure of input data used in the portfolio optimization.
  class Asset(BaseModel): """Asset definition with return and risk characteristics.""" name: str expected_return: float risk: float = Field(ge=0) sector: str | None = None current_price: float | None = Field(default=None, ge=0) min_allocation: float = Field(default=0.0, ge=0, le=1) max_allocation: float = Field(default=1.0, ge=0, le=1) @field_validator("max_allocation") @classmethod def validate_max_allocation(cls, v: float, info: ValidationInfo) -> float: if info.data and "min_allocation" in info.data and v < info.data["min_allocation"]: raise ValueError("max_allocation must be >= min_allocation") return v class PortfolioInput(BaseModel): """Input schema for Portfolio Optimization.""" assets: list[Asset] budget: float = Field(gt=0) risk_tolerance: float = Field(ge=0) min_allocation: float = Field(default=0.0, ge=0, le=1) max_allocation: float = Field(default=1.0, ge=0, le=1) sector_limits: dict[str, float] = Field(default_factory=dict) objective: str = Field( default="maximize_return", pattern="^(maximize_return|minimize_risk|sharpe_ratio)$", ) risk_free_rate: float = Field(default=0.02, ge=0) correlation_matrix: list[list[float]] | None = None @field_validator("assets") @classmethod def validate_assets(cls, v: list[Asset]) -> list[Asset]: if not v: raise ValueError("At least one asset required") return v @field_validator("sector_limits") @classmethod def validate_sector_limits(cls, v: dict[str, float]) -> dict[str, float]: for sector, limit in v.items(): if not (0 <= limit <= 1): raise ValueError(f"Sector limit for {sector} must be between 0 and 1") return v @field_validator("correlation_matrix") @classmethod def validate_correlation_matrix( cls, v: list[list[float]] | None, info: ValidationInfo ) -> list[list[float]] | None: if v is not None and info.data and "assets" in info.data: n = len(info.data["assets"]) if len(v) != n or any(len(row) != n for row in v): raise ValueError("Correlation matrix dimensions must match number of assets") # Check if matrix is symmetric and diagonal elements are 1 for i in range(n): if abs(v[i][i] - 1.0) > 1e-6: raise ValueError("Diagonal elements of correlation matrix must be 1") for j in range(i): if abs(v[i][j] - v[j][i]) > 1e-6: raise ValueError("Correlation matrix must be symmetric") return v
• src/mcp_optimizer/tools/financial.py:91-311 (helper)

  Core helper function implementing the PuLP-based linear programming solver for portfolio optimization, handling objectives like maximize_return, minimize_risk, etc.
  def solve_portfolio_optimization(input_data: dict[str, Any]) -> OptimizationResult: """Solve Portfolio Optimization Problem using PuLP. Args: input_data: Portfolio optimization problem specification Returns: OptimizationResult with optimal portfolio allocation """ start_time = time.time() try: # Parse and validate input portfolio_input = PortfolioInput(**input_data) assets = portfolio_input.assets budget = portfolio_input.budget # Create optimization problem if portfolio_input.objective == "maximize_return": prob = pulp.LpProblem("Portfolio_Optimization", pulp.LpMaximize) else: prob = pulp.LpProblem("Portfolio_Optimization", pulp.LpMinimize) # Decision variables: allocation amounts for each asset allocations = {} for asset in assets: allocations[asset.name] = pulp.LpVariable( f"allocation_{asset.name}", lowBound=asset.min_allocation * budget, upBound=asset.max_allocation * budget, cat="Continuous", ) # Budget constraint prob += pulp.lpSum(allocations.values()) == budget, "Budget_Constraint" # Global allocation constraints for asset in assets: prob += ( allocations[asset.name] >= portfolio_input.min_allocation * budget, f"Min_Allocation_{asset.name}", ) prob += ( allocations[asset.name] <= portfolio_input.max_allocation * budget, f"Max_Allocation_{asset.name}", ) # Sector constraints sectors: dict[str, list[Any]] = {} for asset in assets: if asset.sector: if asset.sector not in sectors: sectors[asset.sector] = [] sectors[asset.sector].append(allocations[asset.name]) for sector, limit in portfolio_input.sector_limits.items(): if sector in sectors: prob += ( pulp.lpSum(sectors[sector]) <= limit * budget, f"Sector_Limit_{sector}", ) # Objective function if portfolio_input.objective == "maximize_return": # Maximize expected return expected_return = pulp.lpSum( allocations[asset.name] * asset.expected_return / budget for asset in assets ) prob += expected_return, "Expected_Return" elif portfolio_input.objective == "minimize_risk": # Minimize portfolio risk (simplified as weighted average of individual risks) # Note: This is a simplification. True portfolio risk requires covariance matrix if portfolio_input.correlation_matrix: # Use correlation matrix to calculate portfolio variance portfolio_variance = 0 for i, asset_i in enumerate(assets): for j, asset_j in enumerate(assets): weight_i = allocations[asset_i.name] / budget weight_j = allocations[asset_j.name] / budget correlation = portfolio_input.correlation_matrix[i][j] portfolio_variance += ( weight_i * weight_j * asset_i.risk * asset_j.risk * correlation ) # Since PuLP doesn't handle quadratic objectives directly, we'll use a linear approximation # This is a limitation - for true portfolio optimization, a QP solver would be better portfolio_risk = pulp.lpSum( allocations[asset.name] * asset.risk / budget for asset in assets ) else: portfolio_risk = pulp.lpSum( allocations[asset.name] * asset.risk / budget for asset in assets ) prob += portfolio_risk, "Portfolio_Risk" elif portfolio_input.objective == "sharpe_ratio": # Maximize Sharpe ratio (simplified) # This is complex to implement directly in linear programming # We'll approximate by maximizing return - risk_penalty * risk risk_penalty = ( 1.0 / portfolio_input.risk_tolerance if portfolio_input.risk_tolerance > 0 else 1.0 ) expected_return = pulp.lpSum( allocations[asset.name] * asset.expected_return / budget for asset in assets ) portfolio_risk = pulp.lpSum( allocations[asset.name] * asset.risk / budget for asset in assets ) sharpe_approximation = expected_return - risk_penalty * portfolio_risk prob += sharpe_approximation, "Sharpe_Approximation" # Risk tolerance constraint if portfolio_input.risk_tolerance > 0: portfolio_risk = pulp.lpSum( allocations[asset.name] * asset.risk / budget for asset in assets ) prob += portfolio_risk <= portfolio_input.risk_tolerance, "Risk_Tolerance" # Solve prob.solve(pulp.PULP_CBC_CMD(msg=0)) # Process results status = pulp.LpStatus[prob.status] execution_time = time.time() - start_time if prob.status == pulp.LpStatusOptimal: # Extract solution portfolio_allocation = {} total_allocation = 0 portfolio_return = 0 portfolio_risk = 0 for asset in assets: allocation_amount = allocations[asset.name].varValue allocation_weight = allocation_amount / budget portfolio_allocation[asset.name] = { "amount": allocation_amount, "weight": allocation_weight, "expected_return": asset.expected_return, "risk": asset.risk, "sector": asset.sector, } total_allocation += allocation_amount portfolio_return += allocation_weight * asset.expected_return portfolio_risk += allocation_weight * asset.risk # Calculate portfolio metrics portfolio_variance = portfolio_risk**2 # Simplified portfolio_std = math.sqrt(portfolio_variance) if portfolio_variance > 0 else 0 sharpe_ratio = ( (portfolio_return - portfolio_input.risk_free_rate) / portfolio_std if portfolio_std > 0 else 0 ) # Sector allocation summary sector_allocation = {} for asset in assets: if asset.sector: if asset.sector not in sector_allocation: sector_allocation[asset.sector] = 0 sector_allocation[asset.sector] += portfolio_allocation[asset.name]["weight"] return OptimizationResult( status=OptimizationStatus.OPTIMAL, objective_value=pulp.value(prob.objective), variables={ "portfolio_allocation": portfolio_allocation, "portfolio_metrics": { "total_allocation": total_allocation, "expected_return": portfolio_return, "portfolio_risk": portfolio_risk, "portfolio_std": portfolio_std, "sharpe_ratio": sharpe_ratio, "risk_free_rate": portfolio_input.risk_free_rate, }, "sector_allocation": sector_allocation, "budget_utilization": total_allocation / budget, }, execution_time=execution_time, solver_info={ "solver_name": "PuLP CBC", "objective": portfolio_input.objective, "num_assets": len(assets), "num_sectors": len(sector_allocation), }, ) elif prob.status == pulp.LpStatusInfeasible: return OptimizationResult( status=OptimizationStatus.INFEASIBLE, error_message="Portfolio optimization problem is infeasible. Check constraints.", execution_time=execution_time, ) elif prob.status == pulp.LpStatusUnbounded: return OptimizationResult( status=OptimizationStatus.UNBOUNDED, error_message="Portfolio optimization problem is unbounded.", execution_time=execution_time, ) else: return OptimizationResult( status=OptimizationStatus.ERROR, error_message=f"Solver failed with status: {status}", execution_time=execution_time, ) except Exception as e: return OptimizationResult( status=OptimizationStatus.ERROR, error_message=f"Portfolio optimization error: {str(e)}", execution_time=time.time() - start_time, )