spb/vquant Public MIT
VibeQuant — AI-powered institutional-grade financial intelligence platform.
TypeScript 84.3%
Python 11.7%
JavaScript 1.6%
CSS 1.5%
HTML 0.7%
1#!/usr/bin/env python32# =============================================================================3# VibeQuant (vquant) — AI-Powered Financial Intelligence Platform4# -----------------------------------------------------------------------------5# File: server/services/portfolioOptimizer.py6#7# Author: Simon-Pierre Boucher8# Contact: contact@spboucher.ai9# Website: https://www.spboucher.ai10# Demo: https://www.vquant.ai11# License: MIT (see LICENSE)12#13# Copyright © 2026 Simon-Pierre Boucher. All rights reserved.14# =============================================================================1516"""17Portfolio Optimizer Service18Implements Modern Portfolio Theory (Markowitz) for optimal asset allocation19"""20import json21import sys22import numpy as np23import pandas as pd24import os25from scipy.optimize import minimize26from typing import List, Dict, Tuple2728# Import our FMP client29sys.path.insert(0, os.path.dirname(__file__))30from fmpClient import get_historical_prices313233def get_portfolio_returns_and_covariance(symbols: List[str], data_period: int = 504) -> Tuple[np.ndarray, np.ndarray, pd.DataFrame]:34 """35 Fetch historical data and calculate returns and covariance matrix3637 Args:38 symbols: List of stock ticker symbols39 data_period: Number of days of historical data4041 Returns:42 Tuple of (mean_returns, cov_matrix, prices_df)43 """44 prices_dict = {}4546 # Fetch historical prices for each symbol47 for symbol in symbols:48 print(f"Fetching data for {symbol}...", file=sys.stderr)49 hist_data = get_historical_prices(symbol)5051 # Extract closing prices52 prices = [h['close'] for h in reversed(hist_data['historical'])]53 dates = [h['date'] for h in reversed(hist_data['historical'])]5455 # Limit to requested period56 if len(prices) > data_period:57 prices = prices[-data_period:]58 dates = dates[-data_period:]5960 prices_dict[symbol] = pd.Series(prices, index=pd.to_datetime(dates))6162 # Create DataFrame with all prices63 prices_df = pd.DataFrame(prices_dict)6465 # Remove any rows with missing data66 prices_df = prices_df.dropna()6768 if len(prices_df) < 30:69 raise ValueError(f"Not enough data points after alignment. Need at least 30, got {len(prices_df)}")7071 # Calculate daily returns72 returns_df = prices_df.pct_change().dropna()7374 # Calculate mean returns (annualized)75 mean_returns = returns_df.mean() * 2527677 # Calculate covariance matrix (annualized)78 cov_matrix = returns_df.cov() * 2527980 return mean_returns.values, cov_matrix.values, prices_df818283def calculate_portfolio_performance(weights: np.ndarray, mean_returns: np.ndarray, cov_matrix: np.ndarray) -> Tuple[float, float]:84 """85 Calculate portfolio return and volatility8687 Args:88 weights: Portfolio weights89 mean_returns: Expected returns for each asset90 cov_matrix: Covariance matrix9192 Returns:93 Tuple of (portfolio_return, portfolio_volatility)94 """95 portfolio_return = np.sum(weights * mean_returns)96 portfolio_std = np.sqrt(np.dot(weights.T, np.dot(cov_matrix, weights)))9798 return portfolio_return, portfolio_std99100101def negative_sharpe_ratio(weights: np.ndarray, mean_returns: np.ndarray, cov_matrix: np.ndarray, risk_free_rate: float) -> float:102 """103 Calculate negative Sharpe ratio (for minimization)104105 Args:106 weights: Portfolio weights107 mean_returns: Expected returns for each asset108 cov_matrix: Covariance matrix109 risk_free_rate: Risk-free rate110111 Returns:112 Negative Sharpe ratio113 """114 portfolio_return, portfolio_std = calculate_portfolio_performance(weights, mean_returns, cov_matrix)115116 if portfolio_std == 0:117 return 0118119 sharpe_ratio = (portfolio_return - risk_free_rate) / portfolio_std120 return -sharpe_ratio121122123def portfolio_volatility(weights: np.ndarray, mean_returns: np.ndarray, cov_matrix: np.ndarray) -> float:124 """125 Calculate portfolio volatility (for minimization)126127 Args:128 weights: Portfolio weights129 mean_returns: Expected returns for each asset130 cov_matrix: Covariance matrix131132 Returns:133 Portfolio volatility134 """135 return calculate_portfolio_performance(weights, mean_returns, cov_matrix)[1]136137138def optimize_portfolio(mean_returns: np.ndarray, cov_matrix: np.ndarray, risk_free_rate: float,139 target_return: float = None, min_weight: float = 0.0, max_weight: float = 1.0) -> dict:140 """141 Optimize portfolio allocation142143 Args:144 mean_returns: Expected returns for each asset145 cov_matrix: Covariance matrix146 risk_free_rate: Risk-free rate147 target_return: Target return for minimum variance portfolio (optional)148 min_weight: Minimum weight per asset149 max_weight: Maximum weight per asset150151 Returns:152 Dictionary with optimal weights and performance metrics153 """154 num_assets = len(mean_returns)155156 # Constraints: weights sum to 1157 constraints = [{'type': 'eq', 'fun': lambda x: np.sum(x) - 1}]158159 # Add target return constraint if specified160 if target_return is not None:161 constraints.append({162 'type': 'eq',163 'fun': lambda x: calculate_portfolio_performance(x, mean_returns, cov_matrix)[0] - target_return164 })165166 # Bounds for weights167 bounds = tuple((min_weight, max_weight) for _ in range(num_assets))168169 # Initial guess (equal weights)170 init_guess = np.array([1.0 / num_assets] * num_assets)171172 # Optimize for maximum Sharpe ratio173 result_sharpe = minimize(174 negative_sharpe_ratio,175 init_guess,176 args=(mean_returns, cov_matrix, risk_free_rate),177 method='SLSQP',178 bounds=bounds,179 constraints=constraints180 )181182 # Optimize for minimum volatility183 result_min_vol = minimize(184 portfolio_volatility,185 init_guess,186 args=(mean_returns, cov_matrix),187 method='SLSQP',188 bounds=bounds,189 constraints=constraints190 )191192 # Calculate performance for optimal portfolios193 sharpe_weights = result_sharpe.x194 sharpe_return, sharpe_vol = calculate_portfolio_performance(sharpe_weights, mean_returns, cov_matrix)195 sharpe_ratio = (sharpe_return - risk_free_rate) / sharpe_vol if sharpe_vol > 0 else 0196197 min_vol_weights = result_min_vol.x198 min_vol_return, min_vol_vol = calculate_portfolio_performance(min_vol_weights, mean_returns, cov_matrix)199 min_vol_sharpe = (min_vol_return - risk_free_rate) / min_vol_vol if min_vol_vol > 0 else 0200201 return {202 'max_sharpe': {203 'weights': sharpe_weights.tolist(),204 'return': float(sharpe_return),205 'volatility': float(sharpe_vol),206 'sharpe_ratio': float(sharpe_ratio),207 'success': result_sharpe.success208 },209 'min_volatility': {210 'weights': min_vol_weights.tolist(),211 'return': float(min_vol_return),212 'volatility': float(min_vol_vol),213 'sharpe_ratio': float(min_vol_sharpe),214 'success': result_min_vol.success215 }216 }217218219def generate_efficient_frontier(mean_returns: np.ndarray, cov_matrix: np.ndarray, risk_free_rate: float,220 num_points: int = 50, min_weight: float = 0.0, max_weight: float = 1.0) -> dict:221 """222 Generate efficient frontier223224 Args:225 mean_returns: Expected returns for each asset226 cov_matrix: Covariance matrix227 risk_free_rate: Risk-free rate228 num_points: Number of points on the frontier229 min_weight: Minimum weight per asset230 max_weight: Maximum weight per asset231232 Returns:233 Dictionary with frontier points234 """235 num_assets = len(mean_returns)236237 # Find minimum and maximum possible returns238 min_return = np.min(mean_returns)239 max_return = np.max(mean_returns)240241 # Generate target returns242 target_returns = np.linspace(min_return, max_return, num_points)243244 frontier_returns = []245 frontier_volatilities = []246 frontier_sharpe_ratios = []247248 constraints = [{'type': 'eq', 'fun': lambda x: np.sum(x) - 1}]249 bounds = tuple((min_weight, max_weight) for _ in range(num_assets))250 init_guess = np.array([1.0 / num_assets] * num_assets)251252 for target_return in target_returns:253 # Add target return constraint254 target_constraints = constraints + [255 {'type': 'eq', 'fun': lambda x, tr=target_return: calculate_portfolio_performance(x, mean_returns, cov_matrix)[0] - tr}256 ]257258 try:259 result = minimize(260 portfolio_volatility,261 init_guess,262 args=(mean_returns, cov_matrix),263 method='SLSQP',264 bounds=bounds,265 constraints=target_constraints,266 options={'maxiter': 1000}267 )268269 if result.success:270 ret, vol = calculate_portfolio_performance(result.x, mean_returns, cov_matrix)271 sharpe = (ret - risk_free_rate) / vol if vol > 0 else 0272273 frontier_returns.append(float(ret))274 frontier_volatilities.append(float(vol))275 frontier_sharpe_ratios.append(float(sharpe))276 except:277 continue278279 return {280 'returns': frontier_returns,281 'volatilities': frontier_volatilities,282 'sharpe_ratios': frontier_sharpe_ratios283 }284285286def run_portfolio_optimization(data_json: str) -> dict:287 """288 Run portfolio optimization analysis289290 Args:291 data_json: JSON string containing:292 - symbols: List of stock ticker symbols (e.g., ['AAPL', 'GOOGL', 'MSFT'])293 - risk_free_rate: Risk-free rate (default: 0.02 for 2%)294 - min_weight: Minimum weight per asset (default: 0.0)295 - max_weight: Maximum weight per asset (default: 1.0)296 - data_period: Historical data period in days (default: 504 for ~2 years)297 - generate_frontier: Whether to generate efficient frontier (default: true)298 - frontier_points: Number of points on efficient frontier (default: 50)299300 Returns:301 Dictionary containing optimal portfolios and efficient frontier302 """303 try:304 data = json.loads(data_json)305306 # Extract parameters307 symbols = data.get('symbols')308 if not symbols or len(symbols) < 2:309 return {310 "error": "At least 2 symbols are required for portfolio optimization",311 "success": False312 }313314 risk_free_rate = data.get('risk_free_rate', 0.02)315 min_weight = data.get('min_weight', 0.0)316 max_weight = data.get('max_weight', 1.0)317 data_period = data.get('data_period', 504)318 generate_frontier_flag = data.get('generate_frontier', True)319 frontier_points = data.get('frontier_points', 50)320321 print(f"Optimizing portfolio for {len(symbols)} assets: {', '.join(symbols)}", file=sys.stderr)322323 # Get returns and covariance matrix324 mean_returns, cov_matrix, prices_df = get_portfolio_returns_and_covariance(symbols, data_period)325326 print(f"Calculated returns and covariance from {len(prices_df)} data points", file=sys.stderr)327328 # Calculate correlation matrix329 returns_df = prices_df.pct_change().dropna()330 corr_matrix = returns_df.corr()331332 # Optimize portfolio333 print("Optimizing portfolio allocations...", file=sys.stderr)334 optimization_results = optimize_portfolio(335 mean_returns, cov_matrix, risk_free_rate, min_weight=min_weight, max_weight=max_weight336 )337338 # Calculate equal-weight portfolio for comparison339 equal_weights = np.array([1.0 / len(symbols)] * len(symbols))340 equal_return, equal_vol = calculate_portfolio_performance(equal_weights, mean_returns, cov_matrix)341 equal_sharpe = (equal_return - risk_free_rate) / equal_vol if equal_vol > 0 else 0342343 # Generate efficient frontier if requested344 efficient_frontier = None345 if generate_frontier_flag:346 print("Generating efficient frontier...", file=sys.stderr)347 efficient_frontier = generate_efficient_frontier(348 mean_returns, cov_matrix, risk_free_rate, frontier_points, min_weight, max_weight349 )350351 # Prepare results352 results = {353 "success": True,354 "symbols": symbols,355 "parameters": {356 "num_assets": len(symbols),357 "risk_free_rate": risk_free_rate,358 "min_weight": min_weight,359 "max_weight": max_weight,360 "data_points": len(prices_df),361 "data_period_days": data_period362 },363 "asset_statistics": {364 symbol: {365 "expected_return": round(float(mean_returns[i]) * 100, 2),366 "volatility": round(float(np.sqrt(cov_matrix[i, i])) * 100, 2),367 "sharpe_ratio": round(float((mean_returns[i] - risk_free_rate) / np.sqrt(cov_matrix[i, i])), 4) if cov_matrix[i, i] > 0 else 0368 }369 for i, symbol in enumerate(symbols)370 },371 "correlation_matrix": {372 symbols[i]: {373 symbols[j]: round(float(corr_matrix.iloc[i, j]), 4)374 for j in range(len(symbols))375 }376 for i in range(len(symbols))377 },378 "optimal_portfolios": {379 "max_sharpe_ratio": {380 "description": "Portfolio with maximum Sharpe ratio",381 "allocation": {382 symbols[i]: round(optimization_results['max_sharpe']['weights'][i] * 100, 2)383 for i in range(len(symbols))384 },385 "expected_return": round(optimization_results['max_sharpe']['return'] * 100, 2),386 "volatility": round(optimization_results['max_sharpe']['volatility'] * 100, 2),387 "sharpe_ratio": round(optimization_results['max_sharpe']['sharpe_ratio'], 4),388 "optimization_success": optimization_results['max_sharpe']['success']389 },390 "min_volatility": {391 "description": "Portfolio with minimum volatility",392 "allocation": {393 symbols[i]: round(optimization_results['min_volatility']['weights'][i] * 100, 2)394 for i in range(len(symbols))395 },396 "expected_return": round(optimization_results['min_volatility']['return'] * 100, 2),397 "volatility": round(optimization_results['min_volatility']['volatility'] * 100, 2),398 "sharpe_ratio": round(optimization_results['min_volatility']['sharpe_ratio'], 4),399 "optimization_success": optimization_results['min_volatility']['success']400 },401 "equal_weight": {402 "description": "Equal weight portfolio (benchmark)",403 "allocation": {404 symbols[i]: round(equal_weights[i] * 100, 2)405 for i in range(len(symbols))406 },407 "expected_return": round(equal_return * 100, 2),408 "volatility": round(equal_vol * 100, 2),409 "sharpe_ratio": round(equal_sharpe, 4)410 }411 },412 "efficient_frontier": efficient_frontier,413 "interpretation": {414 "diversification_benefit": round(415 (equal_vol - optimization_results['min_volatility']['volatility']) / equal_vol * 100, 2416 ) if equal_vol > 0 else 0,417 "notes": [418 "Max Sharpe portfolio optimizes risk-adjusted returns",419 "Min Volatility portfolio minimizes risk",420 "Equal Weight portfolio is the naive 1/N allocation",421 "Diversification benefit shows risk reduction vs. equal weighting",422 f"Correlations range from {round(float(corr_matrix.min().min()), 2)} to {round(float(corr_matrix.max().max()), 2)}"423 ]424 }425 }426427 return results428429 except Exception as e:430 import traceback431 error_details = traceback.format_exc()432 print(f"ERROR: {error_details}", file=sys.stderr)433 return {434 "error": str(e),435 "success": False436 }437438439if __name__ == "__main__":440 # Read input from stdin441 input_data = sys.stdin.read()442443 # Run portfolio optimization444 result = run_portfolio_optimization(input_data)445446 # Output result as JSON447 print(json.dumps(result))448