create_portfolio

Instructions

# Create crypto portfolio from CoinGecko
result = create_portfolio(
    name="crypto_portfolio",
    symbols=["BTC", "ETH", "SOL"],
    source="crypto"
)

# Create portfolio with synthetic data (for testing)
result = create_portfolio(
    name="test_portfolio",
    symbols=["A", "B", "C"],
    source="synthetic",
    seed=42
)
```

Implementation Reference

• app/tools/portfolio.py:210-298 (handler)

  The create_portfolio MCP tool handler function decorated with @mcp.tool. It delegates to the internal _create_portfolio_impl function which handles synthetic, yahoo, and crypto data sources, validates inputs, builds a FinQuant portfolio, and stores it via PortfolioStore.

  @mcp.tool
  def create_portfolio(
      name: str,
      symbols: list[str],
      weights: dict[str, float] | None = None,
      prices: dict[str, list[float]] | None = None,
      dates: list[str] | None = None,
      days: int = 252,
      risk_free_rate: float = 0.02,
      seed: int | None = None,
      source: str = "synthetic",
      period: str = "1y",
  ) -> dict[str, Any]:
      """Create a new portfolio and store it in RefCache.
  
      Creates a portfolio from real market data, provided data, or synthetic
      data. The portfolio is stored persistently and can be retrieved,
      analyzed, and optimized.
  
      Args:
          name: Unique name for the portfolio (e.g., 'stocks', 'crypto').
          symbols: List of asset symbols.
              - For stocks/ETFs: ['AAPL', 'GOOG', 'MSFT', 'SPY']
              - For crypto via Yahoo: ['BTC-USD', 'ETH-USD']
              - For crypto via CoinGecko: ['BTC', 'ETH', 'SOL']
          weights: Optional allocation weights per symbol.
              Must sum to 1.0. If None, equal weights are used.
          prices: Optional price data per symbol as dict of lists.
              If provided, overrides source parameter.
          dates: Optional list of date strings (ISO format) for price data.
              Required if prices is provided.
          days: Number of trading days for synthetic data (default: 252).
          risk_free_rate: Risk-free rate for calculations (default: 0.02).
          seed: Random seed for synthetic data generation.
          source: Data source for prices (default: "synthetic"):
              - "synthetic": Generate GBM simulated data
              - "yahoo": Fetch from Yahoo Finance (stocks, ETFs, crypto)
              - "crypto": Fetch from CoinGecko API (crypto only)
          period: Period for market data (default: "1y").
              Options: 1d, 5d, 1mo, 3mo, 6mo, 1y, 2y, 5y, 10y, ytd, max
  
      Returns:
          Dictionary containing:
          - name: Portfolio name
          - ref_id: RefCache reference ID for retrieval
          - symbols: List of symbols in the portfolio
          - weights: Allocation weights
          - metrics: Initial portfolio metrics (return, volatility, sharpe)
          - source: Data source used
          - created_at: ISO timestamp
  
      Example:
          ```
          # Create portfolio with real stock data
          result = create_portfolio(
              name="tech_stocks",
              symbols=["AAPL", "GOOG", "MSFT"],
              source="yahoo",
              period="1y"
          )
  
          # Create crypto portfolio from CoinGecko
          result = create_portfolio(
              name="crypto_portfolio",
              symbols=["BTC", "ETH", "SOL"],
              source="crypto"
          )
  
          # Create portfolio with synthetic data (for testing)
          result = create_portfolio(
              name="test_portfolio",
              symbols=["A", "B", "C"],
              source="synthetic",
              seed=42
          )
          ```
      """
      return _create_portfolio_impl(
          name=name,
          symbols=symbols,
          weights=weights,
          prices=prices,
          dates=dates,
          days=days,
          risk_free_rate=risk_free_rate,
          seed=seed,
          source=source,
          period=period,
      )

• app/tools/portfolio.py:35-208 (helper)

  The _create_portfolio_impl internal implementation function that handles all the portfolio creation logic: checking for duplicates, fetching data from Yahoo/crypto/synthetic sources, validating weights, building the FinQuant portfolio via build_portfolio, and storing in RefCache via PortfolioStore.

  def _create_portfolio_impl(
      name: str,
      symbols: list[str],
      weights: dict[str, float] | None = None,
      prices: dict[str, list[float]] | None = None,
      dates: list[str] | None = None,
      days: int = 252,
      risk_free_rate: float = 0.02,
      seed: int | None = None,
      source: str = "synthetic",
      period: str = "1y",
  ) -> dict[str, Any]:
      """Internal implementation for portfolio creation.
  
      This is extracted so it can be called by both create_portfolio tool
      and clone_portfolio tool without going through the MCP tool wrapper.
  
      Args:
          name: Unique name for the portfolio.
          symbols: List of asset symbols.
          weights: Optional allocation weights per symbol. Must sum to 1.0.
          prices: Optional price data per symbol as dict of lists.
          dates: Optional list of date strings (ISO format) for price data.
          days: Number of trading days for synthetic data.
          risk_free_rate: Risk-free rate for calculations.
          seed: Random seed for synthetic data generation.
          source: Data source for prices:
              - "synthetic": Generate GBM data (default)
              - "yahoo": Fetch from Yahoo Finance (stocks, ETFs, crypto with -USD suffix)
              - "crypto": Fetch from CoinGecko (crypto symbols like BTC, ETH)
          period: Period for Yahoo Finance data (e.g., '1y', '6mo', '3mo').
  
      Returns:
          Dictionary containing portfolio info or error.
      """
      # Check if portfolio already exists
      if store.exists(name):
          return {
              "error": f"Portfolio '{name}' already exists. Use update_portfolio or delete it first.",
              "suggestion": f"Try: delete_portfolio(name='{name}') first, or use a different name.",
          }
  
      # Determine data source
      data_source = source.lower() if source else "synthetic"
  
      if prices is not None:
          # User provided prices directly
          if dates is None:
              return {
                  "error": "dates parameter is required when providing price data",
              }
          if len(dates) == 0:
              return {"error": "dates list cannot be empty"}
  
          # Validate all symbols have price data
          for symbol in symbols:
              if symbol not in prices:
                  return {
                      "error": f"Missing price data for symbol '{symbol}'",
                  }
              if len(prices[symbol]) != len(dates):
                  return {
                      "error": f"Price data length mismatch for '{symbol}': "
                      f"got {len(prices[symbol])}, expected {len(dates)}",
                  }
  
      elif data_source in ("yahoo", "crypto"):
          # Fetch real market data
          try:
              fetched = fetch_prices(
                  symbols=symbols,
                  source=data_source,
                  period=period,
                  days=days,
              )
              prices = fetched["prices"]
              dates = fetched["dates"]
              symbols = fetched["symbols"]  # May be filtered if some failed
          except ValueError as error:
              return {
                  "error": f"Failed to fetch market data: {error}",
                  "suggestion": "Check symbol names or try source='synthetic' for testing",
              }
  
      else:
          # Generate synthetic prices using GBM
          if seed is not None:
              np.random.seed(seed)
  
          # Default parameters for synthetic data
          initial_prices = dict.fromkeys(symbols, 100.0)
          annual_drift = 0.08
          annual_volatility = 0.20
  
          daily_drift = annual_drift / 252
          daily_volatility = annual_volatility / np.sqrt(252)
  
          # Generate date index
          end_date = pd.Timestamp.now().normalize()
          date_index = pd.bdate_range(end=end_date, periods=days)
  
          # Generate prices
          prices_dict = {}
          for symbol in symbols:
              daily_returns = daily_drift + daily_volatility * np.random.randn(days)
              log_returns = np.log(1 + daily_returns)
              cumulative = np.cumsum(log_returns)
              prices_dict[symbol] = (
                  initial_prices[symbol] * np.exp(cumulative)
              ).tolist()
  
          prices = prices_dict
          dates = [d.strftime("%Y-%m-%d") for d in date_index]
  
      # Build DataFrame from prices with explicit float64 dtype
      date_index = pd.to_datetime(dates)
      prices_df = pd.DataFrame(prices, index=date_index).astype(np.float64)
  
      # Build allocation DataFrame
      if weights is None:
          # Equal weights
          equal_weight = 1.0 / len(symbols)
          weights = dict.fromkeys(symbols, equal_weight)
      else:
          # Validate weights
          weight_sum = sum(weights.values())
          if not np.isclose(weight_sum, 1.0, atol=0.01):
              return {
                  "error": f"Weights must sum to 1.0, got {weight_sum:.4f}",
                  "suggestion": "Normalize your weights or check for missing symbols",
              }
  
          # Ensure all symbols have weights
          for symbol in symbols:
              if symbol not in weights:
                  return {
                      "error": f"Missing weight for symbol '{symbol}'",
                  }
  
      # Create allocation DataFrame for FinQuant with explicit float64 dtype
      allocation_data = [
          {"Allocation": np.float64(weights[s] * 100), "Name": s} for s in symbols
      ]
      allocation_df = pd.DataFrame(allocation_data)
      allocation_df["Allocation"] = allocation_df["Allocation"].astype(np.float64)
  
      # Build FinQuant portfolio
      portfolio = build_portfolio(data=prices_df, pf_allocation=allocation_df)
      portfolio.risk_free_rate = risk_free_rate
  
      # Store in RefCache
      ref_id = store.store(portfolio, name)
  
      # Return portfolio info
      return {
          "name": name,
          "ref_id": ref_id,
          "symbols": symbols,
          "weights": weights,
          "num_days": len(dates),
          "date_range": {"start": dates[0], "end": dates[-1]},
          "metrics": {
              "expected_return": float(portfolio.expected_return),
              "volatility": float(portfolio.volatility),
              "sharpe_ratio": float(portfolio.sharpe),
              "sortino_ratio": float(portfolio.sortino),
              "value_at_risk": float(portfolio.var),
          },
          "settings": {
              "risk_free_rate": risk_free_rate,
          },
          "source": data_source,
          "created_at": datetime.now().isoformat(),
      }

• app/tools/portfolio.py:27-34 (registration)

  The register_portfolio_tools function that registers all portfolio CRUD tools (including create_portfolio) with the FastMCP server instance.

  def register_portfolio_tools(mcp: FastMCP, store: PortfolioStore) -> None:
      """Register portfolio CRUD tools with the FastMCP server.
  
      Args:
          mcp: The FastMCP server instance.
          store: The portfolio store for persistence.
      """

• app/server.py:137-141 (registration)

  The call site in server.py where register_portfolio_tools is invoked with the mcp instance and store, which registers the create_portfolio tool at server startup.

  register_portfolio_tools(mcp, store)
  register_analysis_tools(mcp, store, cache)
  register_optimization_tools(mcp, store, cache)
  register_data_tools(mcp, store, cache)

• tests/test_portfolio_tools.py:11-141 (helper)

  Test class TestCreatePortfolio with tests for creating portfolios with synthetic data, provided data, duplicate detection, and invalid weights.

  class TestCreatePortfolio:
      """Tests for create_portfolio tool."""
  
      def test_create_with_synthetic_data(self, store: PortfolioStore) -> None:
          """Should create a portfolio with generated synthetic data."""
          from fastmcp import FastMCP
  
          from app.tools.portfolio import register_portfolio_tools
  
          mcp = FastMCP(name="test")
          register_portfolio_tools(mcp, store)
  
          # Get the tool function
          create_portfolio = None
          for tool in mcp._tool_manager._tools.values():
              if tool.name == "create_portfolio":
                  create_portfolio = tool.fn
                  break
  
          assert create_portfolio is not None
  
          result = create_portfolio(
              name="test_synthetic",
              symbols=["AAPL", "GOOG", "AMZN"],
              days=100,
              seed=42,
          )
  
          assert "error" not in result
          assert result["name"] == "test_synthetic"
          # ref_id format is "cache_name:hash", not "namespace:key"
          assert "ref_id" in result
          assert ":" in result["ref_id"]
          assert result["symbols"] == ["AAPL", "GOOG", "AMZN"]
          assert "metrics" in result
          assert "expected_return" in result["metrics"]
          assert "volatility" in result["metrics"]
          assert "sharpe_ratio" in result["metrics"]
  
      def test_create_with_provided_data(
          self,
          store: PortfolioStore,
          sample_prices: dict[str, list[float]],
          sample_dates: list[str],
          sample_weights: dict[str, float],
      ) -> None:
          """Should create a portfolio with provided price data."""
          from fastmcp import FastMCP
  
          from app.tools.portfolio import register_portfolio_tools
  
          mcp = FastMCP(name="test")
          register_portfolio_tools(mcp, store)
  
          create_portfolio = None
          for tool in mcp._tool_manager._tools.values():
              if tool.name == "create_portfolio":
                  create_portfolio = tool.fn
                  break
  
          result = create_portfolio(
              name="test_provided",
              symbols=list(sample_prices.keys()),
              prices=sample_prices,
              dates=sample_dates,
              weights=sample_weights,
          )
  
          assert "error" not in result
          assert result["name"] == "test_provided"
          assert result["weights"] == sample_weights
          assert result["num_days"] == len(sample_dates)
  
      def test_create_duplicate_fails(self, store: PortfolioStore) -> None:
          """Should fail when creating a portfolio with existing name."""
          from fastmcp import FastMCP
  
          from app.tools.portfolio import register_portfolio_tools
  
          mcp = FastMCP(name="test")
          register_portfolio_tools(mcp, store)
  
          create_portfolio = None
          for tool in mcp._tool_manager._tools.values():
              if tool.name == "create_portfolio":
                  create_portfolio = tool.fn
                  break
  
          # Create first
          create_portfolio(
              name="duplicate_test",
              symbols=["AAPL"],
              days=50,
              seed=42,
          )
  
          # Try to create duplicate
          result = create_portfolio(
              name="duplicate_test",
              symbols=["AAPL"],
              days=50,
              seed=42,
          )
  
          assert "error" in result
          assert "already exists" in result["error"]
  
      def test_create_with_invalid_weights(self, store: PortfolioStore) -> None:
          """Should fail when weights don't sum to 1.0."""
          from fastmcp import FastMCP
  
          from app.tools.portfolio import register_portfolio_tools
  
          mcp = FastMCP(name="test")
          register_portfolio_tools(mcp, store)
  
          create_portfolio = None
          for tool in mcp._tool_manager._tools.values():
              if tool.name == "create_portfolio":
                  create_portfolio = tool.fn
                  break
  
          result = create_portfolio(
              name="invalid_weights",
              symbols=["AAPL", "GOOG"],
              weights={"AAPL": 0.3, "GOOG": 0.3},  # Sum = 0.6
              days=50,
          )
  
          assert "error" in result
          assert "sum to 1.0" in result["error"]