optimize_portfolio

Instructions

    Finds optimal portfolio weights based on the specified optimization
    method. Uses numerical optimization (scipy) to find the solution.

    Args:
        name: The portfolio name.
        method: Optimization method:
            - "max_sharpe": Maximize Sharpe ratio (default)
            - "min_volatility": Minimize portfolio volatility
            - "efficient_return": Minimize volatility for target return
            - "efficient_volatility": Maximize return for target volatility
        target_return: Required for "efficient_return" method.
            The target annualized return to achieve.
        target_volatility: Required for "efficient_volatility" method.
            The target annualized volatility.

    Returns:
        Dictionary containing:
        - method: Optimization method used
        - optimal_weights: Dict of optimal weights per symbol
        - expected_return: Expected return of optimal portfolio
        - volatility: Volatility of optimal portfolio
        - sharpe_ratio: Sharpe ratio of optimal portfolio
        - original: Original portfolio metrics for comparison
        - improvement: Improvement over original portfolio

    Example:
        ```
        # Maximize Sharpe ratio
        result = optimize_portfolio(name="tech_stocks", method="max_sharpe")

        # Minimize volatility
        result = optimize_portfolio(name="tech_stocks", method="min_volatility")

        # Target 15% return with minimum volatility
        result = optimize_portfolio(
            name="tech_stocks",
            method="efficient_return",
            target_return=0.15
        )
        ```
    

Implementation Reference

• app/tools/optimization.py:37-178 (handler)

  Internal implementation function `_optimize_portfolio_impl` that executes the portfolio optimization logic. It retrieves portfolio data from store, validates the method, rebuilds the portfolio using FinQuant, creates an EfficientFrontier, performs the optimization (max_sharpe, min_volatility, efficient_return, efficient_volatility), calculates optimal metrics and improvement over the original portfolio, and returns the results.

  def _optimize_portfolio_impl(
      name: str,
      method: str = "max_sharpe",
      target_return: float | None = None,
      target_volatility: float | None = None,
  ) -> dict[str, Any]:
      """Internal implementation for portfolio optimization.
  
      This is extracted so it can be called by both optimize_portfolio tool
      and apply_optimization tool without going through the MCP tool wrapper.
      """
      data = store.get(name)
      if data is None:
          return {
              "error": f"Portfolio '{name}' not found",
          }
  
      # Validate method
      valid_methods = [
          "max_sharpe",
          "min_volatility",
          "efficient_return",
          "efficient_volatility",
      ]
      if method not in valid_methods:
          return {
              "error": f"Invalid method: {method}",
              "valid_methods": valid_methods,
          }
  
      # Validate target parameters
      if method == "efficient_return" and target_return is None:
          return {
              "error": "target_return is required for 'efficient_return' method",
          }
      if method == "efficient_volatility" and target_volatility is None:
          return {
              "error": "target_volatility is required for 'efficient_volatility' method",
          }
  
      # Rebuild portfolio
      prices_df = pd.DataFrame(
          data=data["prices"]["values"],
          index=pd.to_datetime(data["prices"]["index"]),
          columns=data["prices"]["columns"],
      )
      allocation_df = pd.DataFrame(
          data=data["allocation"]["values"],
          columns=data["allocation"]["columns"],
      )
      portfolio = build_portfolio(data=prices_df, pf_allocation=allocation_df)
      portfolio.risk_free_rate = data["settings"]["risk_free_rate"]
  
      # Store original metrics
      original_metrics = {
          "expected_return": float(portfolio.expected_return),
          "volatility": float(portfolio.volatility),
          "sharpe_ratio": float(portfolio.sharpe),
      }
  
      # Get weights
      original_weights = {}
      for row in data["allocation"]["values"]:
          original_weights[row[1]] = row[0] / 100.0
  
      # Calculate returns for EfficientFrontier
      returns_df = daily_returns(prices_df).dropna()
      mean_returns = returns_df.mean()
      cov_matrix = returns_df.cov()
  
      # Create EfficientFrontier
      ef = EfficientFrontier(
          mean_returns=mean_returns,
          cov_matrix=cov_matrix,
          risk_free_rate=portfolio.risk_free_rate,
          freq=portfolio.freq,
      )
  
      # Perform optimization based on method
      try:
          if method == "max_sharpe":
              opt_weights = ef.maximum_sharpe_ratio()
          elif method == "min_volatility":
              opt_weights = ef.minimum_volatility()
          elif method == "efficient_return":
              opt_weights = ef.efficient_return(target_return)
          elif method == "efficient_volatility":
              opt_weights = ef.efficient_volatility(target_volatility)
          else:
              return {"error": f"Unknown method: {method}"}
      except Exception as e:
          return {
              "error": f"Optimization failed: {e!s}",
              "suggestion": "Try adjusting target values or using a different method",
          }
  
      # Calculate optimal portfolio metrics
      # opt_weights is a DataFrame with symbols as index and 'Allocation' column
      opt_weights_array = np.array(
          [opt_weights.loc[col, "Allocation"] for col in prices_df.columns]
      )
      opt_return = float(np.sum(mean_returns * opt_weights_array) * 252)
      opt_vol = float(
          np.sqrt(
              np.dot(opt_weights_array.T, np.dot(cov_matrix * 252, opt_weights_array))
          )
      )
      opt_sharpe = (opt_return - portfolio.risk_free_rate) / opt_vol
  
      # Build optimal weights dict
      optimal_weights = {
          symbol: float(opt_weights.loc[symbol, "Allocation"])
          for symbol in prices_df.columns
      }
  
      # Calculate improvement
      improvement = {
          "return_change": opt_return - original_metrics["expected_return"],
          "volatility_change": opt_vol - original_metrics["volatility"],
          "sharpe_ratio_change": opt_sharpe - original_metrics["sharpe_ratio"],
      }
  
      return {
          "portfolio_name": name,
          "method": method,
          "optimal_weights": optimal_weights,
          "expected_return": opt_return,
          "volatility": opt_vol,
          "sharpe_ratio": opt_sharpe,
          "original": {
              "weights": original_weights,
              "expected_return": original_metrics["expected_return"],
              "volatility": original_metrics["volatility"],
              "sharpe_ratio": original_metrics["sharpe_ratio"],
          },
          "improvement": improvement,
          "target": {
              "return": target_return,
              "volatility": target_volatility,
          },
          "optimized_at": datetime.now().isoformat(),
      }

• app/tools/optimization.py:185-239 (handler)

  The `optimize_portfolio` function decorated with `@mcp.tool` and `@cache.cached`. This is the public MCP tool handler that delegates to `_optimize_portfolio_impl`. It accepts name, method, target_return, target_volatility parameters and returns the optimization dictionary.

  def optimize_portfolio(
      name: str,
      method: str = "max_sharpe",
      target_return: float | None = None,
      target_volatility: float | None = None,
  ) -> dict[str, Any]:
      """Optimize portfolio weights using Efficient Frontier.
  
      Finds optimal portfolio weights based on the specified optimization
      method. Uses numerical optimization (scipy) to find the solution.
  
      Args:
          name: The portfolio name.
          method: Optimization method:
              - "max_sharpe": Maximize Sharpe ratio (default)
              - "min_volatility": Minimize portfolio volatility
              - "efficient_return": Minimize volatility for target return
              - "efficient_volatility": Maximize return for target volatility
          target_return: Required for "efficient_return" method.
              The target annualized return to achieve.
          target_volatility: Required for "efficient_volatility" method.
              The target annualized volatility.
  
      Returns:
          Dictionary containing:
          - method: Optimization method used
          - optimal_weights: Dict of optimal weights per symbol
          - expected_return: Expected return of optimal portfolio
          - volatility: Volatility of optimal portfolio
          - sharpe_ratio: Sharpe ratio of optimal portfolio
          - original: Original portfolio metrics for comparison
          - improvement: Improvement over original portfolio
  
      Example:
          ```
          # Maximize Sharpe ratio
          result = optimize_portfolio(name="tech_stocks", method="max_sharpe")
  
          # Minimize volatility
          result = optimize_portfolio(name="tech_stocks", method="min_volatility")
  
          # Target 15% return with minimum volatility
          result = optimize_portfolio(
              name="tech_stocks",
              method="efficient_return",
              target_return=0.15
          )
          ```
      """
      return _optimize_portfolio_impl(
          name=name,
          method=method,
          target_return=target_return,
          target_volatility=target_volatility,
      )

• app/tools/optimization.py:26-35 (registration)

  The `register_optimization_tools` function registers all optimization tools (including optimize_portfolio) with the FastMCP server. It is called from app/server.py line 139 with the mcp, store, and cache instances.

  def register_optimization_tools(
      mcp: FastMCP, store: PortfolioStore, cache: RefCache
  ) -> None:
      """Register optimization tools with the FastMCP server.
  
      Args:
          mcp: The FastMCP server instance.
          store: The portfolio store for persistence.
          cache: The RefCache instance for caching large results.
      """

• app/server.py:139-139 (registration)

  Registration call site: `register_optimization_tools(mcp, store, cache)` inside server.py, which wires the optimization tools (including optimize_portfolio) into the MCP server.

  register_optimization_tools(mcp, store, cache)

• app/server.py:69-73 (schema)

  Server instructions/docs listing optimize_portfolio as an Optimization Tool with description 'Optimize weights using Efficient Frontier' and example usage.

  ## Optimization Tools
  - optimize_portfolio: Optimize weights using Efficient Frontier
  - run_monte_carlo: Run Monte Carlo simulation for optimization
  - get_efficient_frontier: Get frontier data points for visualization
  - apply_optimization: Apply optimization and update portfolio weights