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/customPythonExecutor.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"""17Generic Python Executor with FMP API Access18Allows Claude to write custom Python code with direct access to FMP endpoints19"""2021import sys22import json23import io24import base6425import traceback26import os27from contextlib import redirect_stdout, redirect_stderr28import matplotlib29matplotlib.use('Agg') # Non-interactive backend30import matplotlib.pyplot as plt31import numpy as np32import pandas as pd33from datetime import datetime, timedelta34import warnings35warnings.filterwarnings('ignore') # Suppress warnings for cleaner output3637# Create a comprehensive FMP API client class38class FMPClient:39 """Comprehensive FMP API client for Python custom code"""4041 def __init__(self, api_key):42 self.api_key = api_key43 self.base_url = 'https://financialmodelingprep.com/stable'44 self.base_url_stable = 'https://financialmodelingprep.com/stable'4546 def _request(self, endpoint, params=None):47 """Make API request"""48 import requests49 if params is None:50 params = {}51 params['apikey'] = self.api_key5253 url = f"{self.base_url}{endpoint}"54 response = requests.get(url, params=params, timeout=15)55 response.raise_for_status()56 return response.json()5758 def _request_stable(self, endpoint, params=None):59 """Make API request to stable endpoint"""60 import requests61 if params is None:62 params = {}63 params['apikey'] = self.api_key6465 url = f"{self.base_url_stable}{endpoint}"66 response = requests.get(url, params=params, timeout=15)67 response.raise_for_status()68 return response.json()6970 # ===== MOST USED METHODS =====7172 def get_quote(self, symbol):73 """Get current stock quote"""74 data = self._request('/quote', {'symbol': symbol})75 return data[0] if data else None7677 def get_historical_prices(self, symbol, from_date=None, to_date=None):78 """Get historical prices - returns dict with 'historical' array"""79 params = {'symbol': symbol}80 if from_date:81 params['from'] = from_date82 if to_date:83 params['to'] = to_date84 data = self._request('/historical-price-eod/full', params)85 # Stable API returns flat array; wrap to match legacy format86 if isinstance(data, list):87 return {'symbol': symbol, 'historical': data}88 return data8990 def get_income_statement(self, symbol, period='annual', limit=5):91 """Get income statement"""92 return self._request('/income-statement', {'symbol': symbol, 'period': period, 'limit': limit})9394 def get_balance_sheet(self, symbol, period='annual', limit=5):95 """Get balance sheet"""96 return self._request('/balance-sheet-statement', {'symbol': symbol, 'period': period, 'limit': limit})9798 def get_cash_flow(self, symbol, period='annual', limit=5):99 """Get cash flow statement"""100 return self._request('/cash-flow-statement', {'symbol': symbol, 'period': period, 'limit': limit})101102 def get_key_metrics(self, symbol, period='annual', limit=5):103 """Get key metrics"""104 return self._request('/key-metrics', {'symbol': symbol, 'period': period, 'limit': limit})105106 def get_financial_ratios(self, symbol, period='annual', limit=5):107 """Get financial ratios"""108 return self._request('/ratios', {'symbol': symbol, 'period': period, 'limit': limit})109110 def get_company_profile(self, symbol):111 """Get company profile"""112 return self._request('/profile', {'symbol': symbol})113114 def get_stock_news(self, symbol, limit=20):115 """Get stock news"""116 return self._request('/news/stock', {'symbols': symbol, 'limit': limit})117118 def get_rsi(self, symbol, period=14, time_period='daily'):119 """Get RSI indicator"""120 data = self._request('/technical-indicators/rsi', {'symbol': symbol, 'timeframe': time_period, 'periodLength': period})121 return data[:10] # Limit for performance122123 def get_macd(self, symbol, time_period='daily'):124 """Get MACD indicator"""125 data = self._request('/technical-indicators/macd', {'symbol': symbol, 'timeframe': time_period})126 return data[:10]127128 def search(self, query, limit=10):129 """Search for companies"""130 return self._request('/search-symbol', {'query': query, 'limit': limit})131132# Import additional libraries (optional - fail silently if not available)133AVAILABLE_LIBS = {}134135try:136 import seaborn as sns137 AVAILABLE_LIBS['sns'] = sns138 AVAILABLE_LIBS['seaborn'] = sns139except ImportError:140 pass141142try:143 import plotly.graph_objects as go144 import plotly.express as px145 AVAILABLE_LIBS['go'] = go146 AVAILABLE_LIBS['px'] = px147except ImportError:148 pass149150try:151 from scipy import stats, optimize, signal152 AVAILABLE_LIBS['stats'] = stats153 AVAILABLE_LIBS['optimize'] = optimize154 AVAILABLE_LIBS['signal'] = signal155except ImportError:156 pass157158try:159 from sklearn.linear_model import LinearRegression160 from sklearn.preprocessing import StandardScaler161 from sklearn.decomposition import PCA162 AVAILABLE_LIBS['LinearRegression'] = LinearRegression163 AVAILABLE_LIBS['StandardScaler'] = StandardScaler164 AVAILABLE_LIBS['PCA'] = PCA165except ImportError:166 pass167168try:169 import ta # Technical analysis library170 AVAILABLE_LIBS['ta'] = ta171except ImportError:172 pass173174try:175 import yfinance as yf176 AVAILABLE_LIBS['yf'] = yf177except ImportError:178 pass179180try:181 from statsmodels.tsa.stattools import adfuller, acf, pacf182 from statsmodels.tsa.arima.model import ARIMA183 AVAILABLE_LIBS['adfuller'] = adfuller184 AVAILABLE_LIBS['acf'] = acf185 AVAILABLE_LIBS['pacf'] = pacf186 AVAILABLE_LIBS['ARIMA'] = ARIMA187except ImportError:188 pass189190try:191 import cvxpy as cp192 AVAILABLE_LIBS['cp'] = cp193 AVAILABLE_LIBS['cvxpy'] = cp194except ImportError:195 pass196197# === ARCH / GARCH Models ===198try:199 from arch import arch_model200 from arch.__future__ import reindexing201 AVAILABLE_LIBS['arch_model'] = arch_model202except ImportError:203 pass204205# === Numba (JIT compilation) ===206try:207 import numba208 from numba import jit, njit, prange209 AVAILABLE_LIBS['numba'] = numba210 AVAILABLE_LIBS['jit'] = jit211 AVAILABLE_LIBS['njit'] = njit212 AVAILABLE_LIBS['prange'] = prange213except ImportError:214 pass215216# === NetworkX (Graph analysis) ===217try:218 import networkx as nx219 AVAILABLE_LIBS['nx'] = nx220 AVAILABLE_LIBS['networkx'] = nx221except ImportError:222 pass223224# === SymPy (Symbolic math) ===225try:226 import sympy227 AVAILABLE_LIBS['sympy'] = sympy228except ImportError:229 pass230231# === XGBoost ===232try:233 import xgboost as xgb234 AVAILABLE_LIBS['xgb'] = xgb235 AVAILABLE_LIBS['xgboost'] = xgb236except ImportError:237 pass238239# === LightGBM ===240try:241 import lightgbm as lgb242 AVAILABLE_LIBS['lgb'] = lgb243 AVAILABLE_LIBS['lightgbm'] = lgb244except ImportError:245 pass246247# === PyPortfolioOpt (Portfolio optimization) ===248try:249 from pypfopt import EfficientFrontier, risk_models, expected_returns, HRPOpt, BlackLittermanModel250 from pypfopt.discrete_allocation import DiscreteAllocation251 AVAILABLE_LIBS['EfficientFrontier'] = EfficientFrontier252 AVAILABLE_LIBS['risk_models'] = risk_models253 AVAILABLE_LIBS['expected_returns'] = expected_returns254 AVAILABLE_LIBS['HRPOpt'] = HRPOpt255 AVAILABLE_LIBS['BlackLittermanModel'] = BlackLittermanModel256 AVAILABLE_LIBS['DiscreteAllocation'] = DiscreteAllocation257except ImportError:258 pass259260# === Riskfolio-lib (Advanced portfolio optimization) ===261try:262 import riskfolio as rp263 AVAILABLE_LIBS['rp'] = rp264 AVAILABLE_LIBS['riskfolio'] = rp265except ImportError:266 pass267268# === Prophet (Time series forecasting) ===269try:270 from prophet import Prophet271 AVAILABLE_LIBS['Prophet'] = Prophet272except ImportError:273 pass274275# === pmdarima (Auto ARIMA) ===276try:277 import pmdarima as pm278 from pmdarima import auto_arima279 AVAILABLE_LIBS['pm'] = pm280 AVAILABLE_LIBS['pmdarima'] = pm281 AVAILABLE_LIBS['auto_arima'] = auto_arima282except ImportError:283 pass284285# === Polars (Fast DataFrames) ===286try:287 import polars as pl288 AVAILABLE_LIBS['pl'] = pl289 AVAILABLE_LIBS['polars'] = pl290except ImportError:291 pass292293# === DuckDB (Analytical Database) ===294try:295 import duckdb296 AVAILABLE_LIBS['duckdb'] = duckdb297except ImportError:298 pass299300# === mplfinance (Financial charts) ===301try:302 import mplfinance as mpf303 AVAILABLE_LIBS['mpf'] = mpf304 AVAILABLE_LIBS['mplfinance'] = mpf305except ImportError:306 pass307308# === VectorBT (Backtesting) ===309try:310 import vectorbt as vbt311 AVAILABLE_LIBS['vbt'] = vbt312 AVAILABLE_LIBS['vectorbt'] = vbt313except ImportError:314 pass315316# === HMMLearn (Hidden Markov Models) ===317try:318 from hmmlearn import hmm319 AVAILABLE_LIBS['hmm'] = hmm320 AVAILABLE_LIBS['hmmlearn'] = hmm321except ImportError:322 pass323324# === Additional sklearn modules ===325try:326 from sklearn.ensemble import RandomForestRegressor, RandomForestClassifier, GradientBoostingRegressor327 from sklearn.cluster import KMeans, DBSCAN328 from sklearn.model_selection import train_test_split, cross_val_score, GridSearchCV329 from sklearn.metrics import mean_squared_error, accuracy_score, r2_score330 AVAILABLE_LIBS['RandomForestRegressor'] = RandomForestRegressor331 AVAILABLE_LIBS['RandomForestClassifier'] = RandomForestClassifier332 AVAILABLE_LIBS['GradientBoostingRegressor'] = GradientBoostingRegressor333 AVAILABLE_LIBS['KMeans'] = KMeans334 AVAILABLE_LIBS['DBSCAN'] = DBSCAN335 AVAILABLE_LIBS['train_test_split'] = train_test_split336 AVAILABLE_LIBS['cross_val_score'] = cross_val_score337 AVAILABLE_LIBS['GridSearchCV'] = GridSearchCV338 AVAILABLE_LIBS['mean_squared_error'] = mean_squared_error339 AVAILABLE_LIBS['accuracy_score'] = accuracy_score340 AVAILABLE_LIBS['r2_score'] = r2_score341except ImportError:342 pass343344# === Additional scipy modules ===345try:346 from scipy.interpolate import interp1d, griddata, CubicSpline347 from scipy.integrate import quad, odeint348 from scipy.fft import fft, ifft349 AVAILABLE_LIBS['interp1d'] = interp1d350 AVAILABLE_LIBS['griddata'] = griddata351 AVAILABLE_LIBS['CubicSpline'] = CubicSpline352 AVAILABLE_LIBS['quad'] = quad353 AVAILABLE_LIBS['odeint'] = odeint354 AVAILABLE_LIBS['fft'] = fft355 AVAILABLE_LIBS['ifft'] = ifft356except ImportError:357 pass358359# === Additional statsmodels ===360try:361 from statsmodels.tsa.seasonal import seasonal_decompose362 from statsmodels.tsa.holtwinters import ExponentialSmoothing363 from statsmodels.tsa.vector_ar.var_model import VAR364 from statsmodels.regression.rolling import RollingOLS365 AVAILABLE_LIBS['seasonal_decompose'] = seasonal_decompose366 AVAILABLE_LIBS['ExponentialSmoothing'] = ExponentialSmoothing367 AVAILABLE_LIBS['VAR'] = VAR368 AVAILABLE_LIBS['RollingOLS'] = RollingOLS369except ImportError:370 pass371372373class CustomPythonExecutor:374 """Executor for custom Python code with FMP API access"""375376 def __init__(self, fmp_api_key=None):377 self.fmp_api_key = fmp_api_key or os.getenv('FMP_API_KEY')378 self.fmp = FMPClient(self.fmp_api_key) if FMPClient and self.fmp_api_key else None379 self.output_text = []380 self.figures = []381382 def execute(self, code: str, context: dict = None) -> dict:383 """384 Execute custom Python code with access to FMP API385386 Args:387 code: Python code to execute388 context: Optional context variables to inject389390 Returns:391 dict with:392 - success: bool393 - output: str (printed output)394 - result: any (return value if function defined)395 - figures: list of base64 encoded images396 - error: str (if failed)397 """398 try:399 # Prepare safe execution environment with all available libraries400 exec_globals = {401 '__builtins__': __builtins__,402 # Core libraries403 'np': np,404 'numpy': np,405 'pd': pd,406 'pandas': pd,407 'plt': plt,408 'matplotlib': matplotlib,409 'datetime': datetime,410 'timedelta': timedelta,411 # FMP API client412 'fmp': self.fmp,413 # Custom print414 'print': self._custom_print,415 # Helper functions for common tasks416 'convert_to_datetime': lambda x: pd.to_datetime(x, errors='coerce'),417 'safe_strftime': lambda dt, fmt: dt.strftime(fmt) if pd.notna(dt) and hasattr(dt, 'strftime') else str(dt),418 }419420 # Add all available optional libraries421 exec_globals.update(AVAILABLE_LIBS)422423 # Add context variables if provided424 if context:425 exec_globals.update(context)426427 # Capture stdout/stderr428 stdout_capture = io.StringIO()429 stderr_capture = io.StringIO()430431 result_value = None432433 with redirect_stdout(stdout_capture), redirect_stderr(stderr_capture):434 # Execute the code using a single namespace so variables435 # persist across steps and are visible in nested scopes436 # (functions, comprehensions, lambdas, etc.)437 exec(code, exec_globals)438439 # If a 'main' function is defined, call it440 if 'main' in exec_globals and callable(exec_globals['main']):441 result_value = exec_globals['main']()442443 # Capture any matplotlib figures444 self._capture_figures()445446 # Get captured output447 stdout_text = stdout_capture.getvalue()448 stderr_text = stderr_capture.getvalue()449450 combined_output = '\n'.join(self.output_text)451 if stdout_text:452 combined_output += '\n' + stdout_text453 if stderr_text and not stderr_text.strip().startswith('WARNING'):454 combined_output += '\nStderr: ' + stderr_text455456 return {457 'success': True,458 'output': combined_output.strip(),459 'result': result_value,460 'figures': self.figures,461 'error': None462 }463464 except Exception as e:465 error_msg = f"{type(e).__name__}: {str(e)}\n\n{traceback.format_exc()}"466 return {467 'success': False,468 'output': '\n'.join(self.output_text),469 'result': None,470 'figures': self.figures,471 'error': error_msg472 }473474 def _custom_print(self, *args, **kwargs):475 """Custom print function to capture output"""476 output = ' '.join(str(arg) for arg in args)477 self.output_text.append(output)478479 def _capture_figures(self):480 """Capture all matplotlib figures as base64 images"""481 figs = [plt.figure(i) for i in plt.get_fignums()]482483 for fig in figs:484 # Save figure to bytes buffer485 buf = io.BytesIO()486 fig.savefig(buf, format='png', dpi=150, bbox_inches='tight')487 buf.seek(0)488489 # Encode to base64490 img_base64 = base64.b64encode(buf.read()).decode('utf-8')491 self.figures.append(img_base64)492493 buf.close()494495 # Close all figures to free memory496 plt.close('all')497498499def main():500 """Main entry point for the executor"""501 try:502 # Read input from stdin503 input_data = json.loads(sys.stdin.read())504505 code = input_data.get('code', '')506 context = input_data.get('context', {})507 fmp_api_key = input_data.get('fmp_api_key') or os.getenv('FMP_API_KEY')508509 if not code:510 print(json.dumps({511 'success': False,512 'error': 'No code provided'513 }))514 return515516 # Execute the code517 executor = CustomPythonExecutor(fmp_api_key=fmp_api_key)518 result = executor.execute(code, context)519520 # Return result as JSON521 print(json.dumps(result, default=str)) # default=str to handle datetime, etc.522523 except Exception as e:524 error_result = {525 'success': False,526 'error': f'Executor error: {str(e)}\n{traceback.format_exc()}',527 'output': '',528 'result': None,529 'figures': []530 }531 print(json.dumps(error_result))532533534if __name__ == '__main__':535 main()536