earnings volatility filter - implementation and early results

import asyncio
from dataclasses import dataclass
from datetime import datetime, timedelta
from typing import Optional
import numpy as np

@dataclass
class EarningsEvent:
    symbol: str
    earnings_date: datetime
    expected_move: float  # implied from options
    historical_move: float  # avg of last 4 quarters
    iv_rank: float  # current IV percentile (0-100)
    sector: str

class EarningsVolatilityFilter:
    def __init__(self,
                 iv_rank_threshold: float = 35.0,
                 min_premium_yield: float = 0.02,
                 max_days_to_earnings: int = 5,
                 min_days_to_earnings: int = 2):
        self.iv_rank_threshold = iv_rank_threshold
        self.min_premium_yield = min_premium_yield
        self.max_days_to_earnings = max_days_to_earnings
        self.min_days_to_earnings = min_days_to_earnings

        # sector correlations for hedging
        self.sector_correlations = {
            'Technology': 0.85,
            'Financials': 0.72,
            'Healthcare': 0.65,
            'Consumer': 0.78,
            'Energy': 0.68
        }

    def calculate_expected_iv_crush(self, event: EarningsEvent) -> float:
        """
        Estimate post-earnings IV contraction
        Higher IV rank = more crush expected
        """
        base_crush = 0.25  # 25% baseline

        # higher IV rank = more room to fall
        rank_adjustment = (event.iv_rank - 50) / 100 * 0.15

        # historical consistency factor
        move_ratio = event.expected_move / event.historical_move
        consistency_adj = 0.05 if move_ratio > 1.2 else -0.03

        return base_crush + rank_adjustment + consistency_adj

    def score_opportunity(self, event: EarningsEvent,
                         current_date: datetime) -> dict:
        """
        Score earnings premium selling opportunity
        Returns dict with score (0-100) and reasoning
        """
        days_to_earnings = (event.earnings_date - current_date).days

        # timing filter
        if days_to_earnings < self.min_days_to_earnings:
            return {'score': 0, 'reason': 'too close to earnings'}
        if days_to_earnings > self.max_days_to_earnings:
            return {'score': 0, 'reason': 'too far from earnings'}

        # IV rank filter
        if event.iv_rank < self.iv_rank_threshold:
            return {'score': 0, 'reason': f'IV rank {event.iv_rank:.1f} below threshold'}

        # calculate component scores
        iv_score = min((event.iv_rank - self.iv_rank_threshold) * 2, 40)
        timing_score = 25 - abs(days_to_earnings - 3) * 5
        crush_estimate = self.calculate_expected_iv_crush(event)
        crush_score = crush_estimate * 100

        # sector diversification bonus
        correlation = self.sector_correlations.get(event.sector, 0.75)
        diversification_score = (1 - correlation) * 20

        total_score = iv_score + timing_score + crush_score + diversification_score

        return {
            'score': min(total_score, 100),
            'iv_component': iv_score,
            'timing_component': timing_score,
            'crush_estimate': crush_estimate,
            'diversification_bonus': diversification_score,
            'reason': 'opportunity detected' if total_score > 50 else 'below threshold'
        }

    async def scan_earnings_week(self, events: list[EarningsEvent],
                                  current_date: datetime) -> list[dict]:
        """
        Scan upcoming earnings for opportunities
        Returns sorted list of scored opportunities
        """
        opportunities = []

        for event in events:
            score_result = self.score_opportunity(event, current_date)
            if score_result['score'] > 50:
                opportunities.append({
                    'symbol': event.symbol,
                    'earnings_date': event.earnings_date,
                    'iv_rank': event.iv_rank,
                    'expected_move': event.expected_move,
                    **score_result
                })

        # sort by score descending
        opportunities.sort(key=lambda x: x['score'], reverse=True)

        return opportunities[:10]  # top 10 only

    def calculate_position_size(self, opportunity: dict,
                                 account_size: float,
                                 max_risk_per_trade: float = 0.02) -> dict:
        """
        Calculate appropriate position size for earnings trade
        Conservative sizing for binary events
        """
        base_size = account_size * max_risk_per_trade

        # reduce size based on expected move magnitude
        if opportunity['expected_move'] > 8.0:
            size_multiplier = 0.5  # halve size for large movers
        elif opportunity['expected_move'] > 5.0:
            size_multiplier = 0.75
        else:
            size_multiplier = 1.0

        # confidence adjustment based on score
        confidence_mult = opportunity['score'] / 100

        final_size = base_size * size_multiplier * confidence_mult

        return {
            'max_risk': final_size,
            'size_multiplier': size_multiplier,
            'confidence_factor': confidence_mult,
            'reasoning': f"Base ${base_size:.0f} * {size_multiplier:.2f} (move adj) * {confidence_mult:.2f} (confidence)"
        }