spb/vquant Public MIT

1#!/usr/bin/env python32# =============================================================================3#  VibeQuant (vquant) — AI-Powered Financial Intelligence Platform4# -----------------------------------------------------------------------------5#  File:      server/services/monteCarloService.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"""17Monte Carlo Simulation Service18Executes Monte Carlo simulations using financial data from FMP API directly19"""20import json21import sys22import numpy as np23from scipy import stats24import pandas as pd25import os2627# Import our FMP client28sys.path.insert(0, os.path.dirname(__file__))29from fmpClient import get_historical_prices3031def run_monte_carlo_simulation(data_json: str) -> dict:32    """33    Run Monte Carlo simulation based on provided financial data3435    Args:36        data_json: JSON string containing EITHER:37            Option 1 (Direct API mode - RECOMMENDED):38                - symbol: Stock ticker symbol (e.g., 'AAPL')39                - num_simulations: Number of Monte Carlo simulations to run40                - time_horizon: Number of days to project41                - initial_investment: Starting investment amount4243            Option 2 (Manual data mode):44                - historical_prices: List of historical prices45                - num_simulations: Number of Monte Carlo simulations to run46                - time_horizon: Number of days to project47                - initial_investment: Starting investment amount4849    Returns:50        Dictionary containing simulation results and statistics51    """52    try:53        data = json.loads(data_json)5455        # Check if symbol is provided (new API mode)56        symbol = data.get('symbol')57        if symbol:58            # Fetch historical prices directly from FMP API59            print(f"Fetching historical prices for {symbol} from FMP API...", file=sys.stderr)60            hist_data = get_historical_prices(symbol)6162            # Extract closing prices (most recent first, so reverse for chronological order)63            historical_prices = np.array([h['close'] for h in reversed(hist_data['historical'])])64            print(f"Fetched {len(historical_prices)} historical prices for {symbol}", file=sys.stderr)65        else:66            # Use provided historical prices (old mode)67            historical_prices = np.array(data.get('historical_prices', []))6869        # Extract parameters70        num_simulations = data.get('num_simulations', 10000)71        time_horizon = data.get('time_horizon', 252)  # Default 1 year trading days72        initial_investment = data.get('initial_investment', 10000)73        74        if len(historical_prices) < 2:75            return {76                "error": "Need at least 2 historical prices for simulation",77                "success": False78            }79        80        # Calculate daily returns81        returns = np.diff(historical_prices) / historical_prices[:-1]82        83        # Calculate mean and standard deviation of returns84        mean_return = np.mean(returns)85        std_return = np.std(returns)86        87        # Run Monte Carlo simulations88        simulation_results = np.zeros((num_simulations, time_horizon))89        final_values = np.zeros(num_simulations)90        91        for i in range(num_simulations):92            # Generate random returns based on historical distribution93            daily_returns = np.random.normal(mean_return, std_return, time_horizon)94            95            # Calculate price path96            price_path = initial_investment * np.cumprod(1 + daily_returns)97            simulation_results[i] = price_path98            final_values[i] = price_path[-1]99        100        # Calculate statistics101        mean_final_value = np.mean(final_values)102        median_final_value = np.median(final_values)103        std_final_value = np.std(final_values)104        min_final_value = np.min(final_values)105        max_final_value = np.max(final_values)106        107        # Calculate percentiles108        percentile_5 = np.percentile(final_values, 5)109        percentile_25 = np.percentile(final_values, 25)110        percentile_75 = np.percentile(final_values, 75)111        percentile_95 = np.percentile(final_values, 95)112        113        # Calculate probability of profit114        prob_profit = np.sum(final_values > initial_investment) / num_simulations * 100115        116        # Get sample paths for visualization (10 random simulations)117        sample_indices = np.random.choice(num_simulations, min(10, num_simulations), replace=False)118        sample_paths = simulation_results[sample_indices].tolist()119        120        # Return results121        results = {122            "success": True,123            "parameters": {124                "num_simulations": num_simulations,125                "time_horizon": time_horizon,126                "initial_investment": initial_investment,127                "mean_return": float(mean_return),128                "std_return": float(std_return)129            },130            "statistics": {131                "mean_final_value": float(mean_final_value),132                "median_final_value": float(median_final_value),133                "std_final_value": float(std_final_value),134                "min_final_value": float(min_final_value),135                "max_final_value": float(max_final_value),136                "percentile_5": float(percentile_5),137                "percentile_25": float(percentile_25),138                "percentile_75": float(percentile_75),139                "percentile_95": float(percentile_95),140                "probability_of_profit": float(prob_profit)141            },142            "sample_paths": sample_paths,143            "final_values_distribution": {144                "bins": np.histogram(final_values, bins=50)[1].tolist(),145                "counts": np.histogram(final_values, bins=50)[0].tolist()146            }147        }148        149        return results150        151    except Exception as e:152        return {153            "error": str(e),154            "success": False155        }156157if __name__ == "__main__":158    # Read input from stdin159    input_data = sys.stdin.read()160    161    # Run simulation162    result = run_monte_carlo_simulation(input_data)163    164    # Output result as JSON165    print(json.dumps(result))166