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/riskMetrics.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"""17Risk Metrics Analyzer Service18Calculates comprehensive risk and performance metrics for investment analysis19"""20import json21import sys22import numpy as np23import pandas as pd24import os25from typing import Tuple2627# Import our FMP client28sys.path.insert(0, os.path.dirname(__file__))29from fmpClient import get_historical_prices303132def calculate_returns(prices: np.ndarray) -> np.ndarray:33 """Calculate daily returns from prices"""34 return np.diff(prices) / prices[:-1]353637def calculate_sharpe_ratio(returns: np.ndarray, risk_free_rate: float = 0.02) -> float:38 """39 Calculate Sharpe Ratio4041 Args:42 returns: Array of returns43 risk_free_rate: Annual risk-free rate4445 Returns:46 Sharpe ratio47 """48 if len(returns) == 0:49 return 0.05051 # Annualize returns and volatility52 mean_return = np.mean(returns) * 25253 std_return = np.std(returns, ddof=1) * np.sqrt(252)5455 if std_return == 0:56 return 0.05758 sharpe = (mean_return - risk_free_rate) / std_return59 return float(sharpe)606162def calculate_sortino_ratio(returns: np.ndarray, risk_free_rate: float = 0.02, target_return: float = 0.0) -> float:63 """64 Calculate Sortino Ratio (uses downside deviation instead of total volatility)6566 Args:67 returns: Array of returns68 risk_free_rate: Annual risk-free rate69 target_return: Target return threshold (daily)7071 Returns:72 Sortino ratio73 """74 if len(returns) == 0:75 return 0.07677 # Annualize returns78 mean_return = np.mean(returns) * 2527980 # Calculate downside deviation (only negative returns)81 downside_returns = returns[returns < target_return]82 if len(downside_returns) == 0:83 return 0.08485 downside_std = np.std(downside_returns, ddof=1) * np.sqrt(252)8687 if downside_std == 0:88 return 0.08990 sortino = (mean_return - risk_free_rate) / downside_std91 return float(sortino)929394def calculate_calmar_ratio(returns: np.ndarray, prices: np.ndarray) -> float:95 """96 Calculate Calmar Ratio (annualized return / maximum drawdown)9798 Args:99 returns: Array of returns100 prices: Array of prices101102 Returns:103 Calmar ratio104 """105 if len(returns) == 0:106 return 0.0107108 mean_return = np.mean(returns) * 252109 max_dd = calculate_max_drawdown(prices)[0]110111 if max_dd == 0:112 return 0.0113114 calmar = mean_return / abs(max_dd)115 return float(calmar)116117118def calculate_max_drawdown(prices: np.ndarray) -> Tuple[float, int, int, int]:119 """120 Calculate maximum drawdown and recovery time121122 Args:123 prices: Array of prices124125 Returns:126 Tuple of (max_drawdown, start_idx, end_idx, recovery_days)127 """128 if len(prices) == 0:129 return 0.0, 0, 0, 0130131 # Calculate cumulative maximum132 cummax = np.maximum.accumulate(prices)133134 # Calculate drawdown135 drawdown = (prices - cummax) / cummax136137 # Find maximum drawdown138 max_dd_idx = np.argmin(drawdown)139 max_dd = drawdown[max_dd_idx]140141 # Find start of drawdown (last peak before max drawdown)142 start_idx = np.argmax(cummax[:max_dd_idx + 1] == cummax[max_dd_idx])143144 # Find recovery point (when price exceeds previous peak)145 recovery_idx = None146 peak_price = prices[start_idx]147 for i in range(max_dd_idx + 1, len(prices)):148 if prices[i] >= peak_price:149 recovery_idx = i150 break151152 # Calculate recovery time153 if recovery_idx is not None:154 recovery_days = recovery_idx - max_dd_idx155 else:156 recovery_days = len(prices) - max_dd_idx # Still in drawdown157158 return float(max_dd), int(start_idx), int(max_dd_idx), int(recovery_days)159160161def calculate_beta_alpha(asset_returns: np.ndarray, benchmark_returns: np.ndarray, risk_free_rate: float = 0.02) -> Tuple[float, float]:162 """163 Calculate Beta and Alpha relative to benchmark164165 Args:166 asset_returns: Asset returns167 benchmark_returns: Benchmark returns168 risk_free_rate: Annual risk-free rate169170 Returns:171 Tuple of (beta, alpha)172 """173 if len(asset_returns) != len(benchmark_returns) or len(asset_returns) == 0:174 return 0.0, 0.0175176 # Calculate beta (covariance / variance)177 covariance = np.cov(asset_returns, benchmark_returns)[0, 1]178 benchmark_variance = np.var(benchmark_returns, ddof=1)179180 if benchmark_variance == 0:181 beta = 0.0182 else:183 beta = covariance / benchmark_variance184185 # Calculate alpha (annualized)186 asset_return = np.mean(asset_returns) * 252187 benchmark_return = np.mean(benchmark_returns) * 252188189 alpha = asset_return - (risk_free_rate + beta * (benchmark_return - risk_free_rate))190191 return float(beta), float(alpha)192193194def calculate_information_ratio(asset_returns: np.ndarray, benchmark_returns: np.ndarray) -> float:195 """196 Calculate Information Ratio (excess return / tracking error)197198 Args:199 asset_returns: Asset returns200 benchmark_returns: Benchmark returns201202 Returns:203 Information ratio204 """205 if len(asset_returns) != len(benchmark_returns) or len(asset_returns) == 0:206 return 0.0207208 # Calculate excess returns209 excess_returns = asset_returns - benchmark_returns210211 # Annualized excess return212 mean_excess = np.mean(excess_returns) * 252213214 # Tracking error (annualized)215 tracking_error = np.std(excess_returns, ddof=1) * np.sqrt(252)216217 if tracking_error == 0:218 return 0.0219220 information_ratio = mean_excess / tracking_error221 return float(information_ratio)222223224def calculate_treynor_ratio(asset_returns: np.ndarray, benchmark_returns: np.ndarray, risk_free_rate: float = 0.02) -> float:225 """226 Calculate Treynor Ratio (excess return / beta)227228 Args:229 asset_returns: Asset returns230 benchmark_returns: Benchmark returns231 risk_free_rate: Annual risk-free rate232233 Returns:234 Treynor ratio235 """236 if len(asset_returns) == 0:237 return 0.0238239 beta, _ = calculate_beta_alpha(asset_returns, benchmark_returns, risk_free_rate)240241 if beta == 0:242 return 0.0243244 asset_return = np.mean(asset_returns) * 252245 treynor = (asset_return - risk_free_rate) / beta246247 return float(treynor)248249250def calculate_tracking_error(asset_returns: np.ndarray, benchmark_returns: np.ndarray) -> float:251 """252 Calculate Tracking Error (annualized standard deviation of excess returns)253254 Args:255 asset_returns: Asset returns256 benchmark_returns: Benchmark returns257258 Returns:259 Tracking error260 """261 if len(asset_returns) != len(benchmark_returns) or len(asset_returns) == 0:262 return 0.0263264 excess_returns = asset_returns - benchmark_returns265 tracking_error = np.std(excess_returns, ddof=1) * np.sqrt(252)266267 return float(tracking_error)268269270def calculate_downside_capture(asset_returns: np.ndarray, benchmark_returns: np.ndarray) -> float:271 """272 Calculate Downside Capture Ratio273274 Args:275 asset_returns: Asset returns276 benchmark_returns: Benchmark returns277278 Returns:279 Downside capture ratio (%)280 """281 if len(asset_returns) != len(benchmark_returns) or len(asset_returns) == 0:282 return 0.0283284 # Filter for periods when benchmark was negative285 down_periods = benchmark_returns < 0286287 if not np.any(down_periods):288 return 0.0289290 asset_down = asset_returns[down_periods]291 benchmark_down = benchmark_returns[down_periods]292293 # Calculate average returns during down periods294 asset_down_avg = np.mean(asset_down)295 benchmark_down_avg = np.mean(benchmark_down)296297 if benchmark_down_avg == 0:298 return 0.0299300 downside_capture = (asset_down_avg / benchmark_down_avg) * 100301302 return float(downside_capture)303304305def calculate_upside_capture(asset_returns: np.ndarray, benchmark_returns: np.ndarray) -> float:306 """307 Calculate Upside Capture Ratio308309 Args:310 asset_returns: Asset returns311 benchmark_returns: Benchmark returns312313 Returns:314 Upside capture ratio (%)315 """316 if len(asset_returns) != len(benchmark_returns) or len(asset_returns) == 0:317 return 0.0318319 # Filter for periods when benchmark was positive320 up_periods = benchmark_returns > 0321322 if not np.any(up_periods):323 return 0.0324325 asset_up = asset_returns[up_periods]326 benchmark_up = benchmark_returns[up_periods]327328 # Calculate average returns during up periods329 asset_up_avg = np.mean(asset_up)330 benchmark_up_avg = np.mean(benchmark_up)331332 if benchmark_up_avg == 0:333 return 0.0334335 upside_capture = (asset_up_avg / benchmark_up_avg) * 100336337 return float(upside_capture)338339340def run_risk_metrics_analysis(data_json: str) -> dict:341 """342 Run comprehensive risk metrics analysis343344 Args:345 data_json: JSON string containing:346 - symbol: Stock ticker symbol (e.g., 'AAPL')347 - benchmark_symbol: Benchmark symbol (e.g., 'SPY') - optional348 - risk_free_rate: Annual risk-free rate (default: 0.02 for 2%)349 - data_period: Historical data period in days (default: 504 for ~2 years)350351 Returns:352 Dictionary containing comprehensive risk metrics353 """354 try:355 data = json.loads(data_json)356357 # Extract parameters358 symbol = data.get('symbol')359 if not symbol:360 return {361 "error": "Symbol is required",362 "success": False363 }364365 benchmark_symbol = data.get('benchmark_symbol', 'SPY')366 risk_free_rate = data.get('risk_free_rate', 0.02)367 data_period = data.get('data_period', 504)368369 # Fetch asset historical prices370 print(f"Fetching historical prices for {symbol} from FMP API...", file=sys.stderr)371 asset_hist = get_historical_prices(symbol)372 asset_prices = np.array([h['close'] for h in reversed(asset_hist['historical'])])373 asset_dates = [h['date'] for h in reversed(asset_hist['historical'])]374375 # Limit to requested period376 if len(asset_prices) > data_period:377 asset_prices = asset_prices[-data_period:]378 asset_dates = asset_dates[-data_period:]379380 print(f"Using {len(asset_prices)} historical prices for {symbol}", file=sys.stderr)381382 if len(asset_prices) < 30:383 return {384 "error": f"Need at least 30 historical prices (got {len(asset_prices)})",385 "success": False386 }387388 # Calculate asset returns389 asset_returns = calculate_returns(asset_prices)390391 # Fetch benchmark historical prices392 print(f"Fetching benchmark data for {benchmark_symbol}...", file=sys.stderr)393 benchmark_hist = get_historical_prices(benchmark_symbol)394 benchmark_prices = np.array([h['close'] for h in reversed(benchmark_hist['historical'])])395 benchmark_dates = [h['date'] for h in reversed(benchmark_hist['historical'])]396397 # Align benchmark with asset dates398 asset_dates_set = set(asset_dates)399 aligned_benchmark = []400 for i, date in enumerate(benchmark_dates):401 if date in asset_dates_set:402 aligned_benchmark.append(benchmark_prices[i])403404 benchmark_prices_aligned = np.array(aligned_benchmark)405406 # Ensure same length407 min_len = min(len(asset_prices), len(benchmark_prices_aligned))408 asset_prices = asset_prices[-min_len:]409 benchmark_prices_aligned = benchmark_prices_aligned[-min_len:]410411 benchmark_returns = calculate_returns(benchmark_prices_aligned)412 asset_returns = asset_returns[-len(benchmark_returns):]413414 print(f"Calculated {len(asset_returns)} aligned returns for analysis", file=sys.stderr)415416 # Calculate all metrics417 print("Calculating risk metrics...", file=sys.stderr)418419 # Return statistics420 mean_return = float(np.mean(asset_returns) * 252)421 volatility = float(np.std(asset_returns, ddof=1) * np.sqrt(252))422423 # Sharpe, Sortino, Calmar424 sharpe = calculate_sharpe_ratio(asset_returns, risk_free_rate)425 sortino = calculate_sortino_ratio(asset_returns, risk_free_rate)426 calmar = calculate_calmar_ratio(asset_returns, asset_prices)427428 # Maximum Drawdown429 max_dd, dd_start, dd_end, recovery_days = calculate_max_drawdown(asset_prices)430431 # Beta and Alpha432 beta, alpha = calculate_beta_alpha(asset_returns, benchmark_returns, risk_free_rate)433434 # Information Ratio and Treynor Ratio435 information_ratio = calculate_information_ratio(asset_returns, benchmark_returns)436 treynor = calculate_treynor_ratio(asset_returns, benchmark_returns, risk_free_rate)437438 # Tracking Error439 tracking_error = calculate_tracking_error(asset_returns, benchmark_returns)440441 # Capture Ratios442 downside_capture = calculate_downside_capture(asset_returns, benchmark_returns)443 upside_capture = calculate_upside_capture(asset_returns, benchmark_returns)444445 # Win rate446 winning_days = np.sum(asset_returns > 0)447 total_days = len(asset_returns)448 win_rate = (winning_days / total_days * 100) if total_days > 0 else 0449450 # Average win/loss451 wins = asset_returns[asset_returns > 0]452 losses = asset_returns[asset_returns < 0]453 avg_win = float(np.mean(wins)) if len(wins) > 0 else 0454 avg_loss = float(np.mean(losses)) if len(losses) > 0 else 0455456 # Profit factor457 total_wins = float(np.sum(wins)) if len(wins) > 0 else 0458 total_losses = float(abs(np.sum(losses))) if len(losses) > 0 else 0459 profit_factor = (total_wins / total_losses) if total_losses > 0 else 0460461 # Benchmark statistics462 benchmark_return = float(np.mean(benchmark_returns) * 252)463 benchmark_volatility = float(np.std(benchmark_returns, ddof=1) * np.sqrt(252))464465 # Correlation466 correlation = float(np.corrcoef(asset_returns, benchmark_returns)[0, 1]) if len(asset_returns) > 1 else 0467468 # Results469 results = {470 "success": True,471 "symbol": symbol,472 "benchmark": benchmark_symbol,473 "parameters": {474 "risk_free_rate": risk_free_rate,475 "data_points": len(asset_returns),476 "data_period_days": data_period,477 "start_date": asset_dates[-len(asset_returns)],478 "end_date": asset_dates[-1]479 },480 "return_metrics": {481 "annualized_return": round(mean_return * 100, 2),482 "annualized_volatility": round(volatility * 100, 2),483 "benchmark_return": round(benchmark_return * 100, 2),484 "benchmark_volatility": round(benchmark_volatility * 100, 2),485 "excess_return": round((mean_return - benchmark_return) * 100, 2)486 },487 "risk_adjusted_metrics": {488 "sharpe_ratio": round(sharpe, 4),489 "sortino_ratio": round(sortino, 4),490 "calmar_ratio": round(calmar, 4),491 "information_ratio": round(information_ratio, 4),492 "treynor_ratio": round(treynor, 4)493 },494 "market_metrics": {495 "beta": round(beta, 4),496 "alpha_annualized": round(alpha * 100, 2),497 "correlation": round(correlation, 4),498 "tracking_error": round(tracking_error * 100, 2),499 "r_squared": round(correlation ** 2, 4)500 },501 "drawdown_metrics": {502 "max_drawdown_pct": round(max_dd * 100, 2),503 "max_drawdown_start_idx": dd_start,504 "max_drawdown_end_idx": dd_end,505 "recovery_days": recovery_days,506 "currently_in_drawdown": dd_end == len(asset_prices) - 1,507 "current_drawdown_pct": round(((asset_prices[-1] - np.max(asset_prices)) / np.max(asset_prices)) * 100, 2)508 },509 "capture_ratios": {510 "upside_capture_pct": round(upside_capture, 2),511 "downside_capture_pct": round(downside_capture, 2),512 "capture_ratio": round(upside_capture / downside_capture, 4) if downside_capture != 0 else 0513 },514 "trading_statistics": {515 "win_rate_pct": round(win_rate, 2),516 "average_win_pct": round(avg_win * 100, 4),517 "average_loss_pct": round(avg_loss * 100, 4),518 "profit_factor": round(profit_factor, 4),519 "win_loss_ratio": round(abs(avg_win / avg_loss), 4) if avg_loss != 0 else 0520 },521 "interpretation": {522 "risk_rating": "low" if volatility < 0.15 else "moderate" if volatility < 0.25 else "high",523 "performance_vs_benchmark": "outperforming" if mean_return > benchmark_return else "underperforming",524 "risk_adjusted_performance": "excellent" if sharpe > 2 else "good" if sharpe > 1 else "moderate" if sharpe > 0 else "poor",525 "market_sensitivity": "defensive" if beta < 0.8 else "neutral" if beta < 1.2 else "aggressive",526 "notes": [527 f"Sharpe ratio of {round(sharpe, 2)} indicates {'excellent' if sharpe > 2 else 'good' if sharpe > 1 else 'moderate' if sharpe > 0 else 'poor'} risk-adjusted returns",528 f"Beta of {round(beta, 2)} means the asset is {'less' if beta < 1 else 'more'} volatile than the market",529 f"Maximum drawdown of {round(abs(max_dd) * 100, 2)}% shows the worst peak-to-trough decline",530 f"Downside capture of {round(downside_capture, 2)}% shows protection in market declines",531 f"Alpha of {round(alpha * 100, 2)}% shows {'outperformance' if alpha > 0 else 'underperformance'} vs. expected return"532 ]533 }534 }535536 return results537538 except Exception as e:539 import traceback540 error_details = traceback.format_exc()541 print(f"ERROR: {error_details}", file=sys.stderr)542 return {543 "error": str(e),544 "success": False545 }546547548if __name__ == "__main__":549 # Read input from stdin550 input_data = sys.stdin.read()551552 # Run risk metrics analysis553 result = run_risk_metrics_analysis(input_data)554555 # Output result as JSON556 print(json.dumps(result))557