get_individual_stock_metrics

Instructions

    Calculates return and volatility metrics for each stock
    separately, useful for identifying best/worst performers.

    Args:
        name: The portfolio name.

    Returns:
        Dictionary containing metrics per stock:
        - mean_return: Average daily return (annualized)
        - volatility: Standard deviation (annualized)
        - sharpe_ratio: Individual Sharpe ratio
        - weight: Current allocation weight

    Example:
        ```
        result = get_individual_stock_metrics(name="tech_stocks")
        for symbol, metrics in result['stocks'].items():
            print(f"{symbol}: Return={metrics['mean_return']:.2%}")
        ```
    

Implementation Reference

• app/tools/analysis.py:446-526 (handler)

  The main handler function for get_individual_stock_metrics tool. It retrieves stock metrics (mean_return, volatility, sharpe_ratio, weight) for each stock in a portfolio, sorted by Sharpe ratio.

  def get_individual_stock_metrics(name: str) -> dict[str, Any]:
      """Get metrics for each individual stock in a portfolio.
  
      Calculates return and volatility metrics for each stock
      separately, useful for identifying best/worst performers.
  
      Args:
          name: The portfolio name.
  
      Returns:
          Dictionary containing metrics per stock:
          - mean_return: Average daily return (annualized)
          - volatility: Standard deviation (annualized)
          - sharpe_ratio: Individual Sharpe ratio
          - weight: Current allocation weight
  
      Example:
          ```
          result = get_individual_stock_metrics(name="tech_stocks")
          for symbol, metrics in result['stocks'].items():
              print(f"{symbol}: Return={metrics['mean_return']:.2%}")
          ```
      """
      data = store.get(name)
      if data is None:
          return {
              "error": f"Portfolio '{name}' not found",
          }
  
      # Rebuild price DataFrame
      prices_df = pd.DataFrame(
          data=data["prices"]["values"],
          index=pd.to_datetime(data["prices"]["index"]),
          columns=data["prices"]["columns"],
      )
  
      # Calculate daily returns
      returns_df = daily_returns(prices_df).dropna()
  
      # Get weights
      weights = {}
      for row in data["allocation"]["values"]:
          weights[row[1]] = row[0] / 100.0
  
      risk_free_rate = data["settings"]["risk_free_rate"]
  
      # Calculate metrics per stock
      stocks = {}
      for symbol in returns_df.columns:
          daily_mean = returns_df[symbol].mean()
          daily_std = returns_df[symbol].std()
  
          annual_return = daily_mean * 252
          annual_vol = daily_std * np.sqrt(252)
  
          sharpe = (
              (annual_return - risk_free_rate) / annual_vol if annual_vol > 0 else 0
          )
  
          stocks[symbol] = {
              "mean_return": float(annual_return),
              "volatility": float(annual_vol),
              "sharpe_ratio": float(sharpe),
              "weight": weights.get(symbol, 0),
              "daily_mean": float(daily_mean),
              "daily_std": float(daily_std),
          }
  
      # Sort by Sharpe ratio
      sorted_stocks = sorted(
          stocks.items(), key=lambda x: x[1]["sharpe_ratio"], reverse=True
      )
  
      return {
          "portfolio_name": name,
          "stocks": stocks,
          "ranking_by_sharpe": [s[0] for s in sorted_stocks],
          "best_performer": sorted_stocks[0][0] if sorted_stocks else None,
          "worst_performer": sorted_stocks[-1][0] if sorted_stocks else None,
          "risk_free_rate": risk_free_rate,
      }

• app/tools/analysis.py:441-445 (registration)

  The tool is registered via @mcp.tool decorator and @cache.cached decorator on the handler function inside register_analysis_tools().

  @mcp.tool
  @cache.cached(
      namespace="public",
      ttl=None,  # Deterministic - infinite TTL
  )

• app/tools/analysis.py:22-24 (registration)

  The registration function signature - register_analysis_tools registers this and other tools with the FastMCP server.

  def register_analysis_tools(
      mcp: FastMCP, store: PortfolioStore, cache: RefCache
  ) -> None: