apply_optimization

Instructions

Implementation Reference

• app/tools/optimization.py:533-617 (handler)

  The actual handler function for the apply_optimization MCP tool. Decorated with @mcp.tool, it calls _optimize_portfolio_impl to run optimization, then rebuilds the portfolio with new weights, deletes the old portfolio from the store, stores the updated portfolio, and returns old/new weights, metrics, and improvement data.

  @mcp.tool
  def apply_optimization(
      name: str,
      method: str = "max_sharpe",
      target_return: float | None = None,
      target_volatility: float | None = None,
  ) -> dict[str, Any]:
      """Apply optimization and update portfolio weights.
  
      Optimizes the portfolio using the specified method and updates
      the stored portfolio with the new optimal weights.
  
      Args:
          name: The portfolio name.
          method: Optimization method (same as optimize_portfolio).
          target_return: Target return for "efficient_return" method.
          target_volatility: Target volatility for "efficient_volatility" method.
  
      Returns:
          Updated portfolio information with new weights and metrics.
  
      Example:
          ```
          result = apply_optimization(name="tech_stocks", method="max_sharpe")
          print(f"New Sharpe: {result['new_metrics']['sharpe_ratio']:.2f}")
          ```
      """
      # First, run optimization using internal implementation
      opt_result = _optimize_portfolio_impl(
          name=name,
          method=method,
          target_return=target_return,
          target_volatility=target_volatility,
      )
  
      if "error" in opt_result:
          return opt_result
  
      # Get portfolio data
      data = store.get(name)
      if data is None:
          return {"error": f"Portfolio '{name}' not found"}
  
      # Rebuild with new weights
      prices_df = pd.DataFrame(
          data=data["prices"]["values"],
          index=pd.to_datetime(data["prices"]["index"]),
          columns=data["prices"]["columns"],
      )
  
      # Create new allocation DataFrame
      optimal_weights = opt_result["optimal_weights"]
      symbols = prices_df.columns.tolist()
      allocation_data = [
          {"Allocation": optimal_weights[s] * 100, "Name": s} for s in symbols
      ]
      allocation_df = pd.DataFrame(allocation_data)
  
      # Rebuild portfolio
      portfolio = build_portfolio(data=prices_df, pf_allocation=allocation_df)
      portfolio.risk_free_rate = data["settings"]["risk_free_rate"]
  
      # Update store
      store.delete(name)
      ref_id = store.store(portfolio, name)
  
      return {
          "portfolio_name": name,
          "ref_id": ref_id,
          "method": method,
          "old_weights": opt_result["original"]["weights"],
          "new_weights": optimal_weights,
          "old_metrics": {
              "expected_return": opt_result["original"]["expected_return"],
              "volatility": opt_result["original"]["volatility"],
              "sharpe_ratio": opt_result["original"]["sharpe_ratio"],
          },
          "new_metrics": {
              "expected_return": float(portfolio.expected_return),
              "volatility": float(portfolio.volatility),
              "sharpe_ratio": float(portfolio.sharpe),
          },
          "improvement": opt_result["improvement"],
          "applied_at": datetime.now().isoformat(),
      }

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

  register_optimization_tools function registers all optimization tools with FastMCP, including apply_optimization. The _optimize_portfolio_impl internal function (lines 37-178) is shared by both optimize_portfolio and apply_optimization tools.

  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.
      """
  
      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:37-178 (helper)

  _optimize_portfolio_impl is the internal helper that performs the actual Efficient Frontier optimization (max_sharpe, min_volatility, efficient_return, efficient_volatility) and is called by apply_optimization.

  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/server.py:35-35 (registration)

  Import of register_optimization_tools in the main server module.

  from app.tools.optimization import register_optimization_tools

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

  Actual call to register_optimization_tools(mcp, store, cache) which registers apply_optimization and other optimization tools with the FastMCP server.

  register_optimization_tools(mcp, store, cache)