File size: 5,592 Bytes
f7baaf4 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 | 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)
|