spb/vquant Public MIT

1#!/usr/bin/env python32# =============================================================================3#  VibeQuant (vquant) — AI-Powered Financial Intelligence Platform4# -----------------------------------------------------------------------------5#  File:      server/services/volatilitySurfaceService.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"""17Volatility Surface Service18Generates 3D volatility surface visualization for options19"""20import json21import sys22import numpy as np23import pandas as pd24import matplotlib25matplotlib.use('Agg')26import matplotlib.pyplot as plt27from mpl_toolkits.mplot3d import Axes3D28from matplotlib import cm29from datetime import datetime, timedelta30import os31import requests32from scipy.interpolate import griddata3334# API Configuration (EODHD UnicornBay options add-on)35EODHD_API_KEY = os.environ.get('EODHD_API_KEY')36API_BASE_URL = 'https://eodhd.com/api'373839def fetch_option_chain(symbol: str) -> list:40    """41    Fetch option chain data (with implied volatility) from the EODHD42    UnicornBay options API. Paginates up to 5000 contracts in one pass —43    IV is included per contract, so no per-contract follow-up calls needed.44    """45    if not EODHD_API_KEY:46        raise ValueError("EODHD_API_KEY environment variable not set")4748    underlying = symbol.strip().upper().removesuffix('.US')49    url = f"{API_BASE_URL}/mp/unicornbay/options/contracts"50    options = []51    offset = 052    page_limit = 10005354    try:55        for _ in range(5):  # cap at 5 pages / 5000 contracts56            params = {57                'filter[underlying_symbol]': underlying,58                'fields[options-contracts]': 'contract,exp_date,strike,type,volatility',59                'sort': 'exp_date',60                'page[limit]': page_limit,61                'page[offset]': offset,62                'api_token': EODHD_API_KEY,63                'fmt': 'json',64            }65            response = requests.get(url, params=params, timeout=20)66            if response.status_code in (402, 403):67                raise RuntimeError(68                    "EODHD Options add-on (UnicornBay marketplace) is not active on this API key. "69                    "Subscribe at https://eodhd.com/marketplace/unicornbay/options"70                )71            response.raise_for_status()72            payload = response.json()73            page = payload.get('data') or []74            for item in page:75                attrs = item.get('attributes') or {}76                options.append({77                    'identifier': attrs.get('contract'),78                    'strike': attrs.get('strike'),79                    'expiration_date': attrs.get('exp_date'),80                    'option_type': attrs.get('type'),81                    'implied_volatility': attrs.get('volatility'),82                })83            total = (payload.get('meta') or {}).get('total', len(options))84            offset += page_limit85            if len(page) < page_limit or offset >= total:86                break87        print(f"Fetched {len(options)} contracts from EODHD", file=sys.stderr)88        return options89    except Exception as e:90        print(f"Error fetching option chain: {e}", file=sys.stderr)91        raise929394def smart_sample_contracts(options: list, max_contracts: int = 60) -> list:95    """96    Intelligently sample option contracts to get good coverage of strikes and expirations97    while keeping the number of API calls reasonable.98    """99    if len(options) <= max_contracts:100        return options101102    # Group by expiration date103    by_expiration = {}104    for opt in options:105        exp = opt.get('expiration_date') or opt.get('expirationDate') or opt.get('expiration')106        if exp:107            if exp not in by_expiration:108                by_expiration[exp] = []109            by_expiration[exp].append(opt)110111    # Select contracts: take a subset from each expiration112    sampled = []113    contracts_per_exp = max(3, max_contracts // len(by_expiration))114115    for exp_date, contracts in by_expiration.items():116        # Sort by strike (handle both strike and strike_price fields)117        sorted_contracts = sorted(contracts, key=lambda x: float(x.get('strike_price') or x.get('strike', 0)))118119        # Sample evenly across strike range120        if len(sorted_contracts) <= contracts_per_exp:121            sampled.extend(sorted_contracts)122        else:123            # Take contracts at regular intervals124            step = len(sorted_contracts) // contracts_per_exp125            for i in range(0, len(sorted_contracts), step):126                if len(sampled) < max_contracts:127                    sampled.append(sorted_contracts[i])128129    print(f"Sampled {len(sampled)} contracts from {len(options)} total", file=sys.stderr)130    return sampled131132133def calculate_days_to_expiration(expiration_date: str) -> int:134    """135    Calculate days until option expiration136    """137    try:138        exp_date = datetime.strptime(expiration_date, '%Y-%m-%d')139        today = datetime.now()140        delta = exp_date - today141        return max(1, delta.days)  # At least 1 day142    except:143        # If date format is different, try other formats144        try:145            # Try YYMMDD format (from option identifier)146            exp_date = datetime.strptime(expiration_date, '%y%m%d')147            today = datetime.now()148            delta = exp_date - today149            return max(1, delta.days)150        except:151            return 30  # Default to 30 days if parsing fails152153154def create_volatility_surface(data_json: str) -> dict:155    """156    Create 3D volatility surface visualization157158    Args:159        data_json: JSON string containing:160            - symbol: Stock ticker symbol161            - title: Optional custom title162            - figsize: Optional figure size [width, height]163            - color_map: Optional colormap name (default: 'viridis')164165    Returns:166        Dictionary with success status and image URL167    """168    try:169        data = json.loads(data_json)170        symbol = data.get('symbol')171        title = data.get('title', f'{symbol} Options - Volatility Surface')172        figsize = data.get('figsize', [14, 10])173        color_map = data.get('color_map', 'viridis')174175        if not symbol:176            return {177                "success": False,178                "error": "Symbol is required"179            }180181        print(f"Fetching option chain for {symbol}...", file=sys.stderr)182183        # Fetch option chain data184        chain_data = fetch_option_chain(symbol)185186        # Extract option contracts187        options = chain_data if isinstance(chain_data, list) else chain_data.get('options', [])188189        if not options or len(options) == 0:190            return {191                "success": False,192                "error": f"No option data found for {symbol}"193            }194195        print(f"Processing {len(options)} option contracts...", file=sys.stderr)196197        # Sample contracts for good strike/expiration coverage (IV comes with the chain)198        sampled_contracts = smart_sample_contracts(options, max_contracts=300)199        print(f"Fetching IV data for {len(sampled_contracts)} contracts...", file=sys.stderr)200201        # Prepare data for surface plot202        strikes = []203        days_to_exp = []204        implied_vols = []205        option_types = []206207        # IV is already included in the chain data — no per-contract calls needed208        for i, option in enumerate(sampled_contracts):209            identifier = option.get('contract_name') or option.get('identifier') or option.get('contractSymbol')210211            if not identifier:212                continue213214            price_data = option215            iv = price_data.get('implied_volatility') or price_data.get('impliedVolatility') or price_data.get('iv')216217            if not iv or iv <= 0:218                continue219220            strike = option.get('strike_price') or option.get('strike') or price_data.get('strike_price') or price_data.get('strike')221            expiration = option.get('expiration_date') or option.get('expirationDate') or option.get('expiration') or price_data.get('expiration_date')222            option_type = option.get('option_type') or option.get('type') or price_data.get('option_type', 'call')223224            if strike and expiration:225                days = calculate_days_to_expiration(expiration)226227                strikes.append(float(strike))228                days_to_exp.append(days)229                implied_vols.append(float(iv) * 100)  # Convert to percentage230                option_types.append(option_type.lower())231232            # Progress indicator233            if (i + 1) % 10 == 0:234                print(f"Processed {i + 1}/{len(sampled_contracts)} contracts, found {len(strikes)} with IV", file=sys.stderr)235236        if len(strikes) < 3:237            return {238                "success": False,239                "error": f"Insufficient option data with implied volatility. Found only {len(strikes)} valid contracts."240            }241242        print(f"Valid contracts with IV: {len(strikes)}", file=sys.stderr)243244        # Convert to numpy arrays245        strikes = np.array(strikes)246        days_to_exp = np.array(days_to_exp)247        implied_vols = np.array(implied_vols)248249        # Create figure with 3D subplot250        fig = plt.figure(figsize=tuple(figsize))251        ax = fig.add_subplot(111, projection='3d')252253        # Create grid for surface interpolation254        strike_range = np.linspace(strikes.min(), strikes.max(), 30)255        days_range = np.linspace(days_to_exp.min(), days_to_exp.max(), 30)256        strike_grid, days_grid = np.meshgrid(strike_range, days_range)257258        # Interpolate volatility surface259        try:260            iv_grid = griddata(261                (strikes, days_to_exp),262                implied_vols,263                (strike_grid, days_grid),264                method='cubic',265                fill_value=np.nan266            )267268            # Handle NaN values (replace with linear interpolation)269            mask = np.isnan(iv_grid)270            if mask.any():271                iv_grid_linear = griddata(272                    (strikes, days_to_exp),273                    implied_vols,274                    (strike_grid, days_grid),275                    method='linear',276                    fill_value=implied_vols.mean()277                )278                iv_grid[mask] = iv_grid_linear[mask]279280        except Exception as e:281            print(f"Cubic interpolation failed, using linear: {e}", file=sys.stderr)282            iv_grid = griddata(283                (strikes, days_to_exp),284                implied_vols,285                (strike_grid, days_grid),286                method='linear',287                fill_value=implied_vols.mean()288            )289290        # Plot surface291        surf = ax.plot_surface(292            strike_grid,293            days_grid,294            iv_grid,295            cmap=color_map,296            alpha=0.8,297            edgecolor='none',298            antialiased=True299        )300301        # Scatter actual data points302        ax.scatter(303            strikes,304            days_to_exp,305            implied_vols,306            c='red',307            marker='o',308            s=20,309            alpha=0.6,310            label='Market Data'311        )312313        # Labels and title314        ax.set_xlabel('Strike Price ($)', fontsize=11, labelpad=10)315        ax.set_ylabel('Days to Expiration', fontsize=11, labelpad=10)316        ax.set_zlabel('Implied Volatility (%)', fontsize=11, labelpad=10)317        ax.set_title(title, fontsize=14, fontweight='bold', pad=20)318319        # Add colorbar320        fig.colorbar(surf, ax=ax, shrink=0.5, aspect=5, label='Implied Volatility (%)')321322        # Set viewing angle for better perspective323        ax.view_init(elev=25, azim=45)324325        # Add legend326        ax.legend(loc='upper left', fontsize=9)327328        # Grid329        ax.grid(True, alpha=0.3)330331        # Statistics text332        stats_text = f'Contracts: {len(strikes)} | IV Range: {implied_vols.min():.1f}%-{implied_vols.max():.1f}%'333        fig.text(0.5, 0.02, stats_text, ha='center', fontsize=9, style='italic', color='gray')334335        plt.tight_layout()336337        # Generate unique filename338        import uuid339        import time340        timestamp = int(time.time() * 1000)341        unique_id = str(uuid.uuid4())[:8]342        filename = f"volatility_surface_{timestamp}_{unique_id}.png"343344        # Get project root and create plots directory345        project_root = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))346        plots_dir = os.path.join(project_root, 'public', 'plots')347        os.makedirs(plots_dir, exist_ok=True)348349        # Save plot350        filepath = os.path.join(plots_dir, filename)351        plt.savefig(filepath, format='png', dpi=120, bbox_inches='tight',352                   facecolor='white', edgecolor='none')353        plt.close(fig)354355        print(f"Volatility surface saved: {filename}", file=sys.stderr)356357        return {358            "success": True,359            "image_url": f"/plots/{filename}",360            "format": "png",361            "title": title,362            "stats": {363                "total_contracts": len(strikes),364                "min_iv": float(implied_vols.min()),365                "max_iv": float(implied_vols.max()),366                "avg_iv": float(implied_vols.mean()),367                "strike_range": [float(strikes.min()), float(strikes.max())],368                "expiration_range_days": [int(days_to_exp.min()), int(days_to_exp.max())]369            }370        }371372    except Exception as e:373        import traceback374        error_details = traceback.format_exc()375        print(f"ERROR: {error_details}", file=sys.stderr)376        return {377            "success": False,378            "error": str(e)379        }380381382if __name__ == "__main__":383    # Read input from stdin384    input_data = sys.stdin.read()385386    # Create volatility surface387    result = create_volatility_surface(input_data)388389    # Output result as JSON390    print(json.dumps(result))391