233 lines
9.1 KiB
Python
233 lines
9.1 KiB
Python
"""
|
|
Intent Classification Service
|
|
Replaces naive direction (BULL/BEAR) with nuanced volatility and hedging intent classification
|
|
"""
|
|
import pandas as pd
|
|
import numpy as np
|
|
from typing import Optional
|
|
from enum import Enum
|
|
from utils.logger import logger
|
|
from utils.error_handler import safe_divide
|
|
|
|
|
|
class VolatilityIntent(Enum):
|
|
"""Volatility and hedging intent classification"""
|
|
LONG_VOL = "LONG_VOL" # Buying volatility (call/put buying, expecting large moves)
|
|
SHORT_VOL = "SHORT_VOL" # Selling volatility (premium collection, expecting low vol)
|
|
DIRECTIONAL = "DIRECTIONAL" # Directional positioning (directional bias)
|
|
HEDGE_UNWIND = "HEDGE_UNWIND" # Hedging or unwinding existing positions
|
|
|
|
|
|
class IntentClassifier:
|
|
"""
|
|
Classifies options flow intent beyond simple BULL/BEAR direction.
|
|
Identifies volatility trading, hedging, and directional positioning.
|
|
"""
|
|
|
|
def __init__(self):
|
|
# Configuration thresholds
|
|
self.otm_threshold_pct = 5.0 # Strikes >5% OTM considered OTM
|
|
self.long_vol_premium_threshold = 100000 # Minimum premium for long vol signal
|
|
self.short_vol_premium_threshold = 200000 # Minimum premium for short vol signal
|
|
|
|
def estimate_delta(self, row: pd.Series) -> float:
|
|
"""
|
|
Estimate option delta from moneyness.
|
|
Returns delta estimate (0-1 for calls, -1-0 for puts, use absolute value).
|
|
"""
|
|
spot_num = row.get('spot_num', 0) or 0
|
|
strike_num = row.get('strike_num', 0) or 0
|
|
cp_norm = row.get('cp_norm', '')
|
|
|
|
if pd.isna(spot_num) or spot_num == 0 or pd.isna(strike_num) or pd.isna(cp_norm):
|
|
return 0.5 # Default to ATM
|
|
|
|
moneyness_ratio = strike_num / spot_num if spot_num > 0 else 1.0
|
|
|
|
if cp_norm == 'CALL':
|
|
if strike_num <= spot_num:
|
|
# ITM call: delta 0.5 to 1.0
|
|
delta = 0.5 + (1.0 - moneyness_ratio) * 0.5
|
|
else:
|
|
# OTM call: delta 0.5 to 0.0
|
|
delta = max(0.0, 0.5 * (1.5 - moneyness_ratio) / 0.5)
|
|
else: # PUT
|
|
if strike_num >= spot_num:
|
|
# ITM put: delta -0.5 to -1.0 (return absolute)
|
|
delta = abs(0.5 + (moneyness_ratio - 1.0) * 0.5)
|
|
else:
|
|
# OTM put: delta -0.5 to 0.0 (return absolute)
|
|
delta = max(0.0, 0.5 * (1.0 - moneyness_ratio) / 0.5)
|
|
|
|
return float(delta)
|
|
|
|
def estimate_gamma(self, row: pd.Series, delta: float) -> float:
|
|
"""
|
|
Estimate option gamma (sensitivity of delta to price changes).
|
|
Higher near ATM, lower ITM/OTM.
|
|
Returns gamma estimate (positive value).
|
|
"""
|
|
spot_num = row.get('spot_num', 0) or 0
|
|
strike_num = row.get('strike_num', 0) or 0
|
|
|
|
if pd.isna(spot_num) or spot_num == 0 or pd.isna(strike_num):
|
|
return 0.0
|
|
|
|
# Gamma is highest at-the-money
|
|
moneyness_ratio = strike_num / spot_num if spot_num > 0 else 1.0
|
|
|
|
# Simple approximation: gamma peaks at 1.0 (ATM) and decays away
|
|
# Use normal distribution approximation
|
|
distance_from_atm = abs(moneyness_ratio - 1.0)
|
|
|
|
# Gamma ≈ exp(-distance^2 / (2*sigma^2)) where sigma ≈ 0.1 (10% moneyness)
|
|
gamma = np.exp(-(distance_from_atm ** 2) / (2 * 0.01))
|
|
|
|
return float(gamma)
|
|
|
|
def calculate_delta_exposure(self, row: pd.Series) -> float:
|
|
"""
|
|
Calculate delta exposure: contracts * delta * 100 * spot_price.
|
|
Positive = long delta (bullish), Negative = short delta (bearish).
|
|
"""
|
|
premium_num = row.get('premium_num', 0) or 0
|
|
spot_num = row.get('spot_num', 0) or 0
|
|
vol_num = row.get('vol_num', 0) or 0
|
|
cp_norm = row.get('cp_norm', '')
|
|
side_norm = row.get('side_norm', '')
|
|
|
|
if pd.isna(spot_num) or spot_num == 0:
|
|
return 0.0
|
|
|
|
# Estimate delta
|
|
delta = self.estimate_delta(row)
|
|
|
|
# Determine sign based on call/put and buy/sell
|
|
if cp_norm == 'CALL':
|
|
if side_norm == 'BUY':
|
|
delta_sign = 1.0 # Long calls = positive delta
|
|
else: # SELL
|
|
delta_sign = -1.0 # Short calls = negative delta
|
|
else: # PUT
|
|
if side_norm == 'BUY':
|
|
delta_sign = -1.0 # Long puts = negative delta
|
|
else: # SELL
|
|
delta_sign = 1.0 # Short puts = positive delta
|
|
|
|
# Delta exposure = contracts * delta * 100 * spot
|
|
# Use volume as proxy for contracts
|
|
contracts = vol_num if not pd.isna(vol_num) else 0
|
|
delta_exposure = contracts * delta * delta_sign * 100 * spot_num
|
|
|
|
return float(delta_exposure)
|
|
|
|
def calculate_gamma_exposure(self, row: pd.Series) -> float:
|
|
"""
|
|
Calculate gamma exposure: contracts * gamma * 100 * spot_price^2.
|
|
Positive = long gamma (volatility long), Negative = short gamma (volatility short).
|
|
"""
|
|
premium_num = row.get('premium_num', 0) or 0
|
|
spot_num = row.get('spot_num', 0) or 0
|
|
vol_num = row.get('vol_num', 0) or 0
|
|
cp_norm = row.get('cp_norm', '')
|
|
side_norm = row.get('side_norm', '')
|
|
|
|
if pd.isna(spot_num) or spot_num == 0:
|
|
return 0.0
|
|
|
|
# Estimate delta and gamma
|
|
delta = self.estimate_delta(row)
|
|
gamma = self.estimate_gamma(row, delta)
|
|
|
|
# Determine sign: buying options = long gamma, selling = short gamma
|
|
if side_norm == 'BUY':
|
|
gamma_sign = 1.0 # Long gamma
|
|
else: # SELL
|
|
gamma_sign = -1.0 # Short gamma
|
|
|
|
# Gamma exposure = contracts * gamma * 100 * spot^2
|
|
contracts = vol_num if not pd.isna(vol_num) else 0
|
|
gamma_exposure = contracts * gamma * gamma_sign * 100 * (spot_num ** 2)
|
|
|
|
return float(gamma_exposure)
|
|
|
|
def classify_volatility_intent(self, row: pd.Series) -> str:
|
|
"""
|
|
Classify volatility intent based on trade characteristics.
|
|
Returns VolatilityIntent enum value as string.
|
|
"""
|
|
premium_num = row.get('premium_num', 0) or 0
|
|
spot_num = row.get('spot_num', 0) or 0
|
|
strike_num = row.get('strike_num', 0) or 0
|
|
cp_norm = row.get('cp_norm', '')
|
|
side_norm = row.get('side_norm', '')
|
|
vol_num = row.get('vol_num', 0) or 0
|
|
oi_num = row.get('oi_num', 0) or 0
|
|
|
|
# Calculate moneyness
|
|
if pd.isna(spot_num) or spot_num == 0 or pd.isna(strike_num):
|
|
mny_pct = 0.0
|
|
else:
|
|
if cp_norm == 'CALL':
|
|
mny_pct = ((strike_num - spot_num) / spot_num * 100.0) if spot_num > 0 else 0
|
|
else: # PUT
|
|
mny_pct = ((spot_num - strike_num) / spot_num * 100.0) if spot_num > 0 else 0
|
|
|
|
is_otm = abs(mny_pct) > self.otm_threshold_pct
|
|
|
|
# Long volatility: buying OTM options (calls or puts)
|
|
# High premium, OTM strikes, buying side
|
|
if (side_norm == 'BUY' and
|
|
is_otm and
|
|
premium_num >= self.long_vol_premium_threshold):
|
|
return VolatilityIntent.LONG_VOL.value
|
|
|
|
# Short volatility: selling options, collecting premium
|
|
# High premium, selling side, vol > OI (opening new short positions)
|
|
if (side_norm == 'SELL' and
|
|
premium_num >= self.short_vol_premium_threshold and
|
|
vol_num > oi_num):
|
|
return VolatilityIntent.SHORT_VOL.value
|
|
|
|
# Directional: ITM options, buying side, strong directional flow
|
|
if (side_norm == 'BUY' and
|
|
not is_otm and
|
|
premium_num >= 50000):
|
|
return VolatilityIntent.DIRECTIONAL.value
|
|
|
|
# Hedge/Unwind: Selling existing positions (vol < OI)
|
|
# Or buying protective puts/calls
|
|
if (side_norm == 'SELL' and vol_num < oi_num) or \
|
|
(side_norm == 'BUY' and is_otm and premium_num < self.long_vol_premium_threshold):
|
|
return VolatilityIntent.HEDGE_UNWIND.value
|
|
|
|
# Default: directional (fallback)
|
|
return VolatilityIntent.DIRECTIONAL.value
|
|
|
|
def enrich_with_intent_classification(self, df: pd.DataFrame) -> pd.DataFrame:
|
|
"""
|
|
Add intent classification metrics to DataFrame.
|
|
Adds: delta_exposure, gamma_exposure, volatility_intent
|
|
"""
|
|
if df.empty:
|
|
return df
|
|
|
|
df = df.copy()
|
|
|
|
# Initialize columns
|
|
df['delta_exposure'] = 0.0
|
|
df['gamma_exposure'] = 0.0
|
|
df['volatility_intent'] = VolatilityIntent.DIRECTIONAL.value
|
|
|
|
# Calculate metrics
|
|
df['delta_exposure'] = df.apply(self.calculate_delta_exposure, axis=1)
|
|
df['gamma_exposure'] = df.apply(self.calculate_gamma_exposure, axis=1)
|
|
df['volatility_intent'] = df.apply(self.classify_volatility_intent, axis=1)
|
|
|
|
# Log distribution
|
|
intent_counts = df['volatility_intent'].value_counts()
|
|
logger.info(f"Intent classification complete. Distribution: {intent_counts.to_dict()}")
|
|
|
|
return df
|
|
|