spb/vquant Public MIT

1#!/usr/bin/env python32# =============================================================================3#  VibeQuant (vquant) — AI-Powered Financial Intelligence Platform4# -----------------------------------------------------------------------------5#  File:      server/services/plotService.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"""17Plot Service18Generates customizable plots and charts for financial data visualization19"""20import json21import sys22import numpy as np23import pandas as pd24import matplotlib25matplotlib.use('Agg')  # Non-interactive backend for server-side rendering26import matplotlib.pyplot as plt27import matplotlib.ticker as ticker28import matplotlib.dates as mdates29import seaborn as sns30from io import BytesIO31import base6432from datetime import datetime33import os3435# Import our FMP client for data fetching36sys.path.insert(0, os.path.dirname(__file__))37from fmpClient import get_historical_prices3839# Set style40sns.set_style("whitegrid")41plt.rcParams['figure.facecolor'] = 'white'42plt.rcParams['axes.facecolor'] = 'white'434445def create_plot(data_json: str) -> dict:46    """47    Create customizable plots for data visualization4849    Args:50        data_json: JSON string containing:51            - plot_type: Type of plot ('line', 'bar', 'scatter', 'histogram', 'candlestick', 'area', 'pie', 'heatmap')52            - data: Plot data - can be:53                * dict with x and y keys (for simple plots)54                * dict with multiple series (for multi-line plots)55                * symbol (string) to auto-fetch from FMP API56            - title: Plot title (optional)57            - xlabel: X-axis label (optional)58            - ylabel: Y-axis label (optional)59            - color: Color or list of colors (optional, defaults: blue, green, red, etc.)60            - figsize: Figure size as [width, height] (optional, default: [12, 6])61            - grid: Show grid (optional, default: True)62            - legend: Show legend (optional, default: True if multiple series)63            - style: Line style for line plots (optional: '-', '--', '-.', ':')64            - marker: Marker style for scatter/line plots (optional: 'o', 's', '^', etc.)65            - alpha: Transparency (optional, 0-1, default: 0.8)66            - theme: Color theme ('default', 'dark', 'colorful') (optional)6768    Returns:69        Dictionary containing plot as base64-encoded PNG image70    """71    try:72        data = json.loads(data_json)7374        plot_type = data.get('plot_type', 'line').lower()75        plot_data = data.get('data')76        title = data.get('title', 'Financial Chart')77        xlabel = data.get('xlabel', '')78        ylabel = data.get('ylabel', '')79        color = data.get('color', None)80        figsize = data.get('figsize', [12, 6])81        show_grid = data.get('grid', True)82        show_legend = data.get('legend', None)83        style = data.get('style', '-')84        marker = data.get('marker', None)85        alpha = data.get('alpha', 0.8)86        theme = data.get('theme', 'default')8788        # Handle symbol-based data fetching89        if isinstance(plot_data, str):90            symbol = plot_data91            print(f"Fetching historical data for {symbol}...", file=sys.stderr)92            hist_data = get_historical_prices(symbol)93            historical = hist_data['historical']9495            # Convert to format suitable for plotting96            dates = [h['date'] for h in reversed(historical)]97            closes = [h['close'] for h in reversed(historical)]9899            # Convert date strings to datetime objects for better formatting100            dates_parsed = [datetime.strptime(d, '%Y-%m-%d') if isinstance(d, str) else d for d in dates]101102            # Only include close price by default (volume would dwarf price on same scale)103            plot_data = {104                'x': dates_parsed,105                'y': closes106            }107108            if not title or title == 'Financial Chart':109                title = f'{symbol} Stock Price'110            if not ylabel:111                ylabel = 'Price ($)'112            if not xlabel:113                xlabel = 'Date'114115        # Apply theme116        if theme == 'dark':117            plt.style.use('dark_background')118        elif theme == 'colorful':119            sns.set_palette("husl")120121        # Create figure122        fig, ax = plt.subplots(figsize=tuple(figsize))123124        # Parse data format125        if isinstance(plot_data, dict):126            # Check if it's simple x/y format or multi-series127            if 'x' in plot_data and 'y' in plot_data:128                # Simple x/y plot129                x_data = plot_data['x']130                y_data = plot_data['y']131                series_data = {'data': {'x': x_data, 'y': y_data}}132            elif 'x' in plot_data:133                # Multi-series with shared x-axis134                x_data = plot_data['x']135                series_data = {}136                for key, values in plot_data.items():137                    if key != 'x':138                        series_data[key] = {'x': x_data, 'y': values}139            else:140                return {141                    "success": False,142                    "error": "Invalid data format. Must include 'x' key."143                }144        elif isinstance(plot_data, list):145            # List of values, create auto x-axis146            x_data = list(range(len(plot_data)))147            y_data = plot_data148            series_data = {'data': {'x': x_data, 'y': y_data}}149        else:150            return {151                "success": False,152                "error": "Data must be a dict or list"153            }154155        # Color palette156        if color is None:157            colors = ['#2563eb', '#16a34a', '#dc2626', '#9333ea', '#ea580c', '#0891b2']158        elif isinstance(color, str):159            colors = [color]160        else:161            colors = color162163        # Generate plot based on type164        if plot_type == 'line':165            for idx, (label, data_dict) in enumerate(series_data.items()):166                x = data_dict['x']167                y = data_dict['y']168                plot_color = colors[idx % len(colors)]169                ax.plot(x, y, label=label, color=plot_color, linewidth=2,170                       linestyle=style, marker=marker, alpha=alpha)171172        elif plot_type == 'bar':173            if len(series_data) == 1:174                # Single series bar chart175                label, data_dict = list(series_data.items())[0]176                x = data_dict['x']177                y = data_dict['y']178                ax.bar(x, y, color=colors[0], alpha=alpha, label=label)179            else:180                # Multiple series bar chart181                x = list(series_data.values())[0]['x']182                width = 0.8 / len(series_data)183                for idx, (label, data_dict) in enumerate(series_data.items()):184                    y = data_dict['y']185                    offset = width * idx - (width * len(series_data) / 2)186                    x_pos = np.arange(len(x)) + offset187                    ax.bar(x_pos, y, width, label=label, color=colors[idx % len(colors)], alpha=alpha)188                ax.set_xticks(np.arange(len(x)))189                ax.set_xticklabels(x, rotation=45, ha='right')190191        elif plot_type == 'scatter':192            for idx, (label, data_dict) in enumerate(series_data.items()):193                x = data_dict['x']194                y = data_dict['y']195                plot_color = colors[idx % len(colors)]196                marker_style = marker if marker else 'o'197                ax.scatter(x, y, label=label, color=plot_color, alpha=alpha,198                          s=100, marker=marker_style)199200        elif plot_type == 'area':201            for idx, (label, data_dict) in enumerate(series_data.items()):202                x = data_dict['x']203                y = data_dict['y']204                plot_color = colors[idx % len(colors)]205                ax.fill_between(x, y, alpha=alpha*0.6, color=plot_color, label=label)206                ax.plot(x, y, color=plot_color, linewidth=2, alpha=alpha)207208        elif plot_type == 'histogram':209            # Histogram only uses y values210            for idx, (label, data_dict) in enumerate(series_data.items()):211                y = data_dict['y']212                plot_color = colors[idx % len(colors)]213                ax.hist(y, bins=30, alpha=alpha*0.7, color=plot_color,214                       edgecolor='black', label=label)215216        elif plot_type == 'pie':217            # Pie chart uses first series only218            label, data_dict = list(series_data.items())[0]219            labels = data_dict['x']220            values = data_dict['y']221            ax.pie(values, labels=labels, autopct='%1.1f%%', startangle=90,222                  colors=colors[:len(values)], textprops={'fontsize': 10})223            ax.axis('equal')224225        elif plot_type == 'candlestick':226            # Candlestick requires OHLC data227            if 'open' in plot_data and 'high' in plot_data and 'low' in plot_data and 'close' in plot_data:228                dates = plot_data.get('x', list(range(len(plot_data['open']))))229                opens = plot_data['open']230                highs = plot_data['high']231                lows = plot_data['low']232                closes = plot_data['close']233234                # Convert dates to numbers if they're strings235                if isinstance(dates[0], str):236                    x_pos = np.arange(len(dates))237                else:238                    x_pos = dates239240                # Plot candlesticks241                for i, (o, h, l, c, x) in enumerate(zip(opens, highs, lows, closes, x_pos)):242                    color_candle = colors[0] if c >= o else colors[2]  # green if up, red if down243                    # High-low line244                    ax.plot([x, x], [l, h], color=color_candle, linewidth=1, alpha=alpha)245                    # Open-close rectangle246                    height = abs(c - o)247                    bottom = min(o, c)248                    ax.add_patch(plt.Rectangle((x - 0.3, bottom), 0.6, height,249                                              facecolor=color_candle, alpha=alpha))250251                # Set x-axis labels if dates are strings252                if isinstance(dates[0], str):253                    # Show every nth label to avoid crowding254                    step = max(1, len(dates) // 10)255                    ax.set_xticks(x_pos[::step])256                    ax.set_xticklabels([dates[i] for i in range(0, len(dates), step)],257                                      rotation=45, ha='right')258            else:259                return {260                    "success": False,261                    "error": "Candlestick plot requires 'open', 'high', 'low', 'close' data"262                }263264        elif plot_type == 'heatmap':265            # Heatmap requires 2D data266            if isinstance(plot_data, dict) and 'values' in plot_data:267                values = np.array(plot_data['values'])268                xlabels = plot_data.get('xlabels', None)269                ylabels = plot_data.get('ylabels', None)270271                im = ax.imshow(values, cmap='RdYlGn', aspect='auto', alpha=alpha)272273                if xlabels:274                    ax.set_xticks(np.arange(len(xlabels)))275                    ax.set_xticklabels(xlabels, rotation=45, ha='right')276                if ylabels:277                    ax.set_yticks(np.arange(len(ylabels)))278                    ax.set_yticklabels(ylabels)279280                plt.colorbar(im, ax=ax)281            else:282                return {283                    "success": False,284                    "error": "Heatmap requires 'values' as 2D array in data"285                }286287        else:288            return {289                "success": False,290                "error": f"Unknown plot type: {plot_type}. Supported types: line, bar, scatter, histogram, area, pie, candlestick, heatmap"291            }292293        # Customize plot294        ax.set_title(title, fontsize=16, fontweight='bold', pad=20)295        if xlabel:296            ax.set_xlabel(xlabel, fontsize=12)297        if ylabel:298            ax.set_ylabel(ylabel, fontsize=12)299300        if show_grid and plot_type not in ['pie', 'heatmap']:301            ax.grid(True, alpha=0.3, linestyle='--')302303        # Show legend if multiple series or explicitly requested304        if show_legend is None:305            show_legend = len(series_data) > 1 or plot_type == 'pie'306307        if show_legend and plot_type not in ['pie', 'heatmap', 'candlestick']:308            ax.legend(loc='best', framealpha=0.9, fontsize=10)309310        # Format axes for better readability311        if plot_type in ['line', 'area', 'scatter', 'bar']:312            x_values = list(series_data.values())[0]['x']313314            # Check if x-axis contains datetime objects315            if len(x_values) > 0 and isinstance(x_values[0], datetime):316                # Use matplotlib date formatting317                ax.xaxis.set_major_formatter(mdates.DateFormatter('%b %Y'))318                ax.xaxis.set_major_locator(mdates.AutoDateLocator())319                plt.setp(ax.xaxis.get_majorticklabels(), rotation=45, ha='right')320            elif len(x_values) > 0 and isinstance(x_values[0], str):321                # Show every nth label to avoid crowding for string labels322                step = max(1, len(x_values) // 15)323                ax.set_xticks(range(0, len(x_values), step))324                ax.set_xticklabels([x_values[i] for i in range(0, len(x_values), step)],325                                  rotation=45, ha='right')326327            # Format y-axis with thousand separators for financial data328            if ylabel and ('$' in ylabel or 'Price' in ylabel or 'Revenue' in ylabel or 'Value' in ylabel):329                ax.yaxis.set_major_formatter(ticker.FuncFormatter(lambda x, p: f'${x:,.0f}' if abs(x) >= 1 else f'${x:.2f}'))330            elif ylabel and '%' in ylabel:331                ax.yaxis.set_major_formatter(ticker.FuncFormatter(lambda x, p: f'{x:.1f}%'))332            else:333                # General number formatting with thousand separators334                ax.yaxis.set_major_formatter(ticker.FuncFormatter(lambda x, p: f'{x:,.0f}' if abs(x) >= 1 else f'{x:.2f}'))335336        plt.tight_layout()337338        # Generate unique filename with timestamp and random UUID339        import uuid340        import time341        timestamp = int(time.time() * 1000)342        unique_id = str(uuid.uuid4())[:8]343        filename = f"plot_{timestamp}_{unique_id}.png"344345        # Get the project root directory (2 levels up from this file)346        project_root = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))347        plots_dir = os.path.join(project_root, 'public', 'plots')348349        # Create directory if it doesn't exist350        os.makedirs(plots_dir, exist_ok=True)351352        # Save the plot to file353        filepath = os.path.join(plots_dir, filename)354        plt.savefig(filepath, format='png', dpi=100, bbox_inches='tight',355                   facecolor='white', edgecolor='none')356        plt.close(fig)357358        # Return the relative URL path359        image_url = f"/plots/{filename}"360361        return {362            "success": True,363            "plot_type": plot_type,364            "image_url": image_url,365            "format": "png",366            "title": title367        }368369    except Exception as e:370        import traceback371        error_details = traceback.format_exc()372        print(f"ERROR: {error_details}", file=sys.stderr)373        return {374            "success": False,375            "error": str(e)376        }377378379if __name__ == "__main__":380    # Read input from stdin381    input_data = sys.stdin.read()382383    # Create plot384    result = create_plot(input_data)385386    # Output result as JSON387    print(json.dumps(result))388