get_covariance_matrix

Instructions

    Calculates pairwise covariances between all assets in the
    portfolio based on daily returns.

    Args:
        name: The portfolio name.
        annualized: If True, annualize the covariance (multiply by 252).

    Returns:
        Dictionary containing:
        - symbols: List of symbols
        - covariance_matrix: 2D covariance matrix
        - variances: Individual asset variances (diagonal)

    Example:
        ```
        result = get_covariance_matrix(name="tech_stocks")
        print(f"GOOG variance: {result['variances']['GOOG']}")
        ```
    

Implementation Reference

• app/tools/analysis.py:289-351 (handler)

  The actual handler function for get_covariance_matrix tool. Calculates covariance matrix from portfolio price data, with optional annualization (multiply by 252). Returns symbols, covariance_matrix, and variances.

  @mcp.tool
  @cache.cached(
      namespace="public",
      ttl=None,  # Deterministic - infinite TTL
  )
  def get_covariance_matrix(name: str, annualized: bool = True) -> dict[str, Any]:
      """Get the covariance matrix for portfolio assets.
  
      Calculates pairwise covariances between all assets in the
      portfolio based on daily returns.
  
      Args:
          name: The portfolio name.
          annualized: If True, annualize the covariance (multiply by 252).
  
      Returns:
          Dictionary containing:
          - symbols: List of symbols
          - covariance_matrix: 2D covariance matrix
          - variances: Individual asset variances (diagonal)
  
      Example:
          ```
          result = get_covariance_matrix(name="tech_stocks")
          print(f"GOOG variance: {result['variances']['GOOG']}")
          ```
      """
      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()
  
      # Calculate covariance matrix
      cov_matrix = returns_df.cov()
  
      # Annualize if requested
      if annualized:
          cov_matrix = cov_matrix * 252
  
      # Extract variances
      symbols = cov_matrix.columns.tolist()
      variances = {
          symbol: float(cov_matrix.loc[symbol, symbol]) for symbol in symbols
      }
  
      return {
          "portfolio_name": name,
          "symbols": symbols,
          "annualized": annualized,
          "covariance_matrix": cov_matrix.values.tolist(),
          "variances": variances,
      }

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

  The registration function register_analysis_tools which registers get_covariance_matrix (and other tools) via the @mcp.tool decorator on the inner function.

  def register_analysis_tools(
      mcp: FastMCP, store: PortfolioStore, cache: RefCache
  ) -> None:
      """Register analysis 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/tools/analysis.py:13-14 (helper)

  The daily_returns function imported from finquant.returns, used inside get_covariance_matrix to compute daily returns from price data.

  from finquant.returns import cumulative_returns, daily_log_returns, daily_returns