engine/volatility_engine.py

import math
import logging
import pandas as pd
import numpy as np
from typing import Dict, Any, Optional

logger = logging.getLogger("bqe.engine.volatility_engine")

def std_normal_cdf(x: float) -> float:
    """
    Standard normal cumulative distribution function (CDF) approximation using math.erf.
    """
    return 0.5 * (1.0 + math.erf(x / math.sqrt(2.0)))

def black_scholes_option_price(
    S: float, 
    K: float, 
    T: float, 
    r: float, 
    sigma: float, 
    is_call: bool
) -> float:
    """
    Prices an option using the standard Black-Scholes-Merton formula.
    
    Args:
        S: Spot price of the underlying asset.
        K: Strike price of the option contract.
        T: Time to expiration in years.
        r: Annual risk-free interest rate (e.g. 0.07 for 7%).
        sigma: Implied Volatility (annualized, e.g. 0.20 for 20%).
        is_call: True for Call option ('CE'), False for Put option ('PE').
        
    Returns:
        float: Theoretical option premium price.
    """
    if T <= 0.0 or sigma <= 0.0:
        # Return intrinsic option value at expiration boundary
        if is_call:
            return max(0.0, S - K)
        else:
            return max(0.0, K - S)
            
    d1 = (math.log(S / K) + (r + 0.5 * sigma ** 2) * T) / (sigma * math.sqrt(T))
    d2 = d1 - sigma * math.sqrt(T)
    
    if is_call:
        return S * std_normal_cdf(d1) - K * math.exp(-r * T) * std_normal_cdf(d2)
    else:
        return K * math.exp(-r * T) * std_normal_cdf(-d2) - S * std_normal_cdf(-d1)


class DerivativeAnalyticsEngine:
    """
    Unified Derivative Analytics Engine for index option contracts.
    Calculates Put-Call Ratios and extracts Implied Volatility (IV).
    """

    def calculate_pcr_vectors(self, instruments_df: pd.DataFrame) -> Dict[str, float]:
        """
        Groups option instruments by expiry date and calculates the Put-Call Ratio (PCR).
        Utilizes 'volume' (or 'open_interest' / counts as fallback).
        
        Args:
            instruments_df: Options instrument contracts DataFrame.
            
        Returns:
            Dict[str, float]: Expiry date keys mapped to Put-Call Ratio values.
        """
        pcr_dict = {}
        if instruments_df.empty:
            logger.warning("Empty instruments DataFrame passed to PCR calculator.")
            return pcr_dict

        # Group by expiry date
        grouped = instruments_df.groupby('expiry')
        
        for expiry, group in grouped:
            # Filter puts and calls
            puts = group[group['instrument_type'] == 'PE']
            calls = group[group['instrument_type'] == 'CE']
            
            # Select appropriate weighting metric: volume -> open_interest -> row count fallback
            if 'volume' in group.columns and group['volume'].sum() > 0:
                put_vol = puts['volume'].sum()
                call_vol = calls['volume'].sum()
            elif 'open_interest' in group.columns and group['open_interest'].sum() > 0:
                put_vol = puts['open_interest'].sum()
                call_vol = calls['open_interest'].sum()
            else:
                put_vol = len(puts)
                call_vol = len(calls)
                
            pcr = float(put_vol / call_vol) if call_vol > 0 else 0.0
            pcr_dict[str(expiry)] = pcr
            
        logger.info(f"Calculated PCR vectors across {len(pcr_dict)} unique expiration dates.")
        return pcr_dict

    def calculate_atm_iv(
        self, 
        spot_price: float, 
        option_premium: float, 
        strike: float, 
        time_to_expiry: float, 
        is_call: bool, 
        risk_free_rate: float = 0.07
    ) -> float:
        """
        Backs out the annualized Implied Volatility (IV) from standard market option premium
        using an iterative numerical Bisection solver.
        
        Args:
            spot_price: Underlying price.
            option_premium: Market option price.
            strike: Strike price.
            time_to_expiry: Time to expiration in years.
            is_call: True if Call ('CE'), False if Put ('PE').
            risk_free_rate: Risk free rate (default 0.07).
            
        Returns:
            float: Annualized Implied Volatility (e.g. 0.165 for 16.5% IV).
        """
        # Lower boundary check: Option price must be at least its intrinsic value
        intrinsic_val = (spot_price - strike) if is_call else (strike - spot_price)
        intrinsic_val = max(0.0, intrinsic_val)
        
        if option_premium <= intrinsic_val:
            logger.debug("Option premium is below or equal to intrinsic value. Returning 0 IV.")
            return 0.0
            
        # Set up bisection bounds
        low = 1e-6
        high = 5.0  # 500% IV bounds ceiling
        max_iter = 100
        tolerance = 1e-5
        
        for i in range(max_iter):
            mid = 0.5 * (low + high)
            theoretical_price = black_scholes_option_price(
                S=spot_price,
                K=strike,
                T=time_to_expiry,
                r=risk_free_rate,
                sigma=mid,
                is_call=is_call
            )
            
            # Check convergence
            if abs(theoretical_price - option_premium) < tolerance:
                return mid
                
            if theoretical_price < option_premium:
                low = mid
            else:
                high = mid
                
        # Return mid-point fallback if max iterations hit
        return 0.5 * (low + high)