earnings volatility - how my algos adapt to quarterly chaos

import pandas as pd
import numpy as np
from datetime import datetime, timedelta
from dataclasses import dataclass
from typing import List, Optional
import yfinance as yf

@dataclass
class EarningsEvent:
    symbol: str
    report_date: datetime
    report_time: str  # 'BMO' (before market open) or 'AMC' (after market close)
    expected_move: float  # implied by options
    historical_move_avg: float
    market_cap: float  # for weighting impact

class EarningsVolatilityAdapter:
    """
    Adapts trading parameters around earnings announcements
    """

    def __init__(self):
        self.earnings_calendar: List[EarningsEvent] = []
        self.major_symbols = ['AAPL', 'MSFT', 'GOOGL', 'AMZN', 'META',
                             'TSLA', 'NVDA', 'JPM', 'BAC', 'WMT']
        self.index_impact_threshold = 0.5  # % of index weight

    def load_earnings_calendar(self, start_date: datetime, end_date: datetime):
        """
        Load earnings calendar for major movers
        """
        for symbol in self.major_symbols:
            try:
                ticker = yf.Ticker(symbol)
                earnings_dates = ticker.earnings_dates

                if earnings_dates is not None:
                    for date, row in earnings_dates.iterrows():
                        if start_date <= date.to_pydatetime() <= end_date:
                            self.earnings_calendar.append(EarningsEvent(
                                symbol=symbol,
                                report_date=date.to_pydatetime(),
                                report_time='AMC',  # would need additional data
                                expected_move=self._get_expected_move(symbol),
                                historical_move_avg=self._get_historical_move(symbol),
                                market_cap=ticker.info.get('marketCap', 0)
                            ))
            except Exception as e:
                print(f"Error loading {symbol}: {e}")

    def _get_expected_move(self, symbol: str) -> float:
        """Calculate expected move from ATM straddle price"""
        # Simplified - would use real options data
        return np.random.uniform(3, 8)  # typical 3-8% expected moves

    def _get_historical_move(self, symbol: str) -> float:
        """Get average historical earnings move"""
        # Would query historical data
        return np.random.uniform(4, 10)

    def get_earnings_impact(self, trade_date: datetime) -> dict:
        """
        Calculate earnings impact on trading for given date
        """
        # Check for same-day earnings
        same_day = [e for e in self.earnings_calendar
                   if e.report_date.date() == trade_date.date()]

        # Check for next-day earnings (AMC reports affect next day)
        next_day = [e for e in self.earnings_calendar
                   if e.report_date.date() == (trade_date - timedelta(days=1)).date()
                   and e.report_time == 'AMC']

        # Check for upcoming (within 2 days)
        upcoming = [e for e in self.earnings_calendar
                   if 0 < (e.report_date.date() - trade_date.date()).days <= 2]

        # Calculate impact score
        impact_score = 0

        for event in same_day + next_day:
            if event.symbol in ['AAPL', 'MSFT', 'GOOGL', 'AMZN']:
                impact_score += 3  # mega cap = huge impact
            elif event.symbol in ['META', 'TSLA', 'NVDA']:
                impact_score += 2  # large cap = significant impact
            else:
                impact_score += 1

        # Upcoming events add uncertainty
        impact_score += len(upcoming) * 0.5

        return {
            'date': trade_date,
            'same_day_earnings': [e.symbol for e in same_day],
            'post_earnings': [e.symbol for e in next_day],
            'upcoming_earnings': [e.symbol for e in upcoming],
            'impact_score': impact_score,
            'recommendation': self._get_recommendation(impact_score)
        }

    def _get_recommendation(self, impact_score: float) -> dict:
        """Trading recommendations based on earnings impact"""
        if impact_score >= 4:
            return {
                'action': 'REDUCE_EXPOSURE',
                'position_size_mult': 0.3,
                'avoid_symbols': True,
                'reason': 'High earnings impact - major names reporting'
            }
        elif impact_score >= 2:
            return {
                'action': 'CAUTIOUS',
                'position_size_mult': 0.6,
                'avoid_symbols': True,
                'reason': 'Moderate earnings impact'
            }
        elif impact_score >= 1:
            return {
                'action': 'AWARE',
                'position_size_mult': 0.8,
                'avoid_symbols': False,
                'reason': 'Minor earnings impact - stay alert'
            }
        else:
            return {
                'action': 'NORMAL',
                'position_size_mult': 1.0,
                'avoid_symbols': False,
                'reason': 'No significant earnings impact'
            }

    def adjust_strategy_params(
        self,
        base_params: dict,
        impact: dict
    ) -> dict:
        """
        Adjust strategy parameters for earnings environment
        """
        adjusted = base_params.copy()
        mult = impact['recommendation']['position_size_mult']

        # Reduce position size
        adjusted['risk_per_trade'] = base_params['risk_per_trade'] * mult

        # Widen stops (more volatility expected)
        if mult < 1.0:
            adjusted['stop_loss_mult'] = base_params.get('stop_loss_mult', 1.0) * 1.3

        # Increase entry threshold (require stronger signals)
        if mult < 0.7:
            adjusted['signal_threshold'] = base_params.get('signal_threshold', 2.0) * 1.4

        # Reduce max positions
        if mult < 0.5:
            adjusted['max_positions'] = max(1, base_params.get('max_positions', 5) // 2)

        return adjusted


class EarningsPlayFilter:
    """
    Filter for avoiding direct earnings plays
    (I don't trade earnings announcements directly)
    """

    def __init__(self):
        self.blackout_hours_before = 4
        self.blackout_hours_after = 24

    def is_in_blackout(
        self,
        symbol: str,
        current_time: datetime,
        earnings_events: List[EarningsEvent]
    ) -> bool:
        """
        Check if symbol is in earnings blackout window
        """
        for event in earnings_events:
            if event.symbol != symbol:
                continue

            # Calculate blackout windows
            if event.report_time == 'AMC':
                blackout_start = event.report_date.replace(hour=12)  # afternoon before
                blackout_end = event.report_date + timedelta(hours=24)
            else:  # BMO
                blackout_start = event.report_date - timedelta(hours=16)  # day before close
                blackout_end = event.report_date.replace(hour=12)  # morning after

            if blackout_start <= current_time <= blackout_end:
                return True

        return False

    def filter_signals(
        self,
        signals: List[dict],
        current_time: datetime,
        earnings_events: List[EarningsEvent]
    ) -> List[dict]:
        """
        Remove signals for symbols in earnings blackout
        """
        filtered = []

        for signal in signals:
            symbol = signal.get('symbol', '')

            if self.is_in_blackout(symbol, current_time, earnings_events):
                print(f"Filtered {symbol} - earnings blackout")
                continue

            filtered.append(signal)

        return filtered


# Usage this week
adapter = EarningsVolatilityAdapter()
adapter.load_earnings_calendar(
    datetime(2025, 7, 21),
    datetime(2025, 7, 27)
)

# Check impact for each day
for day_offset in range(7):
    check_date = datetime(2025, 7, 21) + timedelta(days=day_offset)
    impact = adapter.get_earnings_impact(check_date)
    print(f"\n{check_date.strftime('%A %m/%d')}:")
    print(f"  Same day: {impact['same_day_earnings']}")
    print(f"  Impact score: {impact['impact_score']}")
    print(f"  Recommendation: {impact['recommendation']['action']}")