Add earnings model with calendar detection, implied move, post-earnings drift analysis

earnings_model.py

@@ -0,0 +1,290 @@
+"""Earnings Model v1.0 — Earnings Event Intelligence
+Detects earnings proximity, estimates implied move from options, 
+analyzes post-earnings drift patterns, and adjusts position sizing.
+Based on: Ball & Brown (1968), Frazzini & Lamont (2007) PEAD
+"""
+import yfinance as yf
+import numpy as np
+import pandas as pd
+from datetime import datetime, timedelta
+from typing import Dict, Optional, Tuple
+
+# Typical quarterly earnings windows by month (approximate)
+EARNINGS_CALENDAR = {
+    # Q1 (Jan-Mar): reports Apr-May
+    'Q1':  {'months': [4, 5], 'label': 'Q1'},
+    # Q2 (Apr-Jun): reports Jul-Aug
+    'Q2':  {'months': [7, 8], 'label': 'Q2'},
+    # Q3 (Jul-Sep): reports Oct-Nov
+    'Q3':  {'months': [10, 11], 'label': 'Q3'},
+    # Q4 (Oct-Dec): reports Jan-Feb
+    'Q4':  {'months': [1, 2], 'label': 'Q4'},
+}
+
+# Sector-specific typical implied moves (based on historical options data)
+SECTOR_IMPLIED_MOVES = {
+    'Technology':        0.045,
+    'Healthcare':        0.040,
+    'Financials':        0.030,
+    'Energy':            0.055,
+    'Consumer Discretionary': 0.050,
+    'Consumer Staples': 0.025,
+    'Industrials':       0.035,
+    'Communication':     0.045,
+    'Utilities':         0.020,
+    'Materials':         0.045,
+    'Real Estate':       0.030,
+    'default':           0.040,
+}
+
+
+class EarningsModel:
+    """Earnings event intelligence for quant trading."""
+
+    def __init__(self):
+        self._cache = {}
+
+    def estimate_earnings_date(self, ticker: str) -> Optional[datetime.date]:
+        """Estimate next earnings date from yfinance calendar."""
+        try:
+            t = yf.Ticker(ticker)
+            cal = t.calendar
+            if cal is not None and not cal.empty:
+                # yfinance calendar returns next earnings date
+                if hasattr(cal, 'index') and 'Earnings Date' in cal.index:
+                    date_str = cal.loc['Earnings Date'].values[0]
+                    if isinstance(date_str, str):
+                        return datetime.strptime(date_str, '%Y-%m-%d').date()
+                elif 'Earnings Date' in cal.columns:
+                    date_str = cal['Earnings Date'].iloc[0]
+                    if isinstance(date_str, str):
+                        return datetime.strptime(date_str, '%Y-%m-%d').date()
+            # Fallback: estimate from current month
+            today = datetime.now()
+            for q, data in EARNINGS_CALENDAR.items():
+                if today.month in data['months']:
+                    # Next likely report window: mid-month
+                    return datetime(today.year, today.month, 15).date()
+            return None
+        except Exception:
+            return None
+
+    def days_to_earnings(self, ticker: str) -> Tuple[Optional[int], Optional[datetime.date]]:
+        """Return (days_until, date)."""
+        ed = self.estimate_earnings_date(ticker)
+        if ed is None:
+            return None, None
+        today = datetime.now().date()
+        delta = (ed - today).days
+        return delta, ed
+
+    def implied_move(self, ticker: str, default_move: float = None) -> Dict:
+        """Estimate implied earnings move from options chain if available.
+        Fallback to sector average or default.
+        """
+        move = default_move or 0.04  # 4% default
+        source = 'default'
+
+        try:
+            t = yf.Ticker(ticker)
+            # Try to get implied move from near-the-money straddle
+            # Find the closest expiration before/after estimated earnings
+            expiry = t.options
+            if expiry and len(expiry) > 0:
+                # Get first expiration
+                opt = t.option_chain(expiry[0])
+                calls = opt.calls
+                puts = opt.puts
+
+                if len(calls) > 0 and len(puts) > 0:
+                    # ATM strike
+                    spot = calls['strike'].iloc[len(calls)//2]
+                    atm_call = calls[calls['strike'].sub(spot).abs().idxmin()]
+                    atm_put = puts[puts['strike'].sub(spot).abs().idxmin()]
+
+                    # Straddle price
+                    call_p = atm_call['lastPrice'] if 'lastPrice' in atm_call else atm_call['ask']
+                    put_p = atm_put['lastPrice'] if 'lastPrice' in atm_put else atm_put['ask']
+                    straddle = float(call_p) + float(put_p)
+
+                    # Implied move as % of stock price
+                    current_price = yf.Ticker(ticker).history(period='1d')['Close'].iloc[-1]
+                    move = straddle / current_price
+                    source = 'options_chain'
+
+                    # Annualize: if 1 month until expiry
+                    days = 30  # approximate
+                    move_annual = move * np.sqrt(365 / max(1, days))
+
+                    return {
+                        'implied_move_pct': round(move * 100, 2),
+                        'annualized_pct': round(move_annual * 100, 2),
+                        'straddle_price': round(float(straddle), 2),
+                        'source': source,
+                        'expiry': expiry[0],
+                        'atm_strike': float(spot),
+                    }
+        except Exception as e:
+            pass
+
+        # Try sector-based
+        try:
+            info = yf.Ticker(ticker).info
+            sector = info.get('sector', '')
+            for sector_name, sector_move in SECTOR_IMPLIED_MOVES.items():
+                if sector_name.lower() in sector.lower():
+                    move = sector_move
+                    source = f'sector_{sector_name}'
+                    break
+        except:
+            pass
+
+        return {
+            'implied_move_pct': round(move * 100, 2),
+            'annualized_pct': round(move * np.sqrt(12) * 100, 2),
+            'source': source,
+        }
+
+    def historical_pead(self, ticker: str, lookback_years: int = 3) -> Dict:
+        """Post-Earnings Announcement Drift analysis.
+        Measures how much stock drifts in direction of surprise after earnings.
+        Returns drift score: positive = bullish drift tendency, negative = bearish.
+        """
+        try:
+            # Fetch enough history
+            df = yf.Ticker(ticker).history(period=f"{lookback_years}y")
+            if len(df) < 100:
+                return {'drift_score': 0, 'confidence': 0, 'n_events': 0}
+
+            # Detect earnings dates (volume spikes on quarterly frequency)
+            df['volume_z'] = (df['Volume'] - df['Volume'].rolling(20).mean()) / df['Volume'].rolling(20).std()
+            # Find volume spike days
+            spike_days = df[df['volume_z'] > 2.5].index
+
+            if len(spike_days) < 4:
+                return {'drift_score': 0, 'confidence': 0, 'n_events': 0}
+
+            # Analyze returns post-spike
+            drifts = []
+            for spike in spike_days:
+                try:
+                    idx = df.index.get_loc(spike)
+                    if idx + 5 >= len(df): continue
+                    # Day +1 to +5 return
+                    post_ret = (df['Close'].iloc[idx+5] / df['Close'].iloc[idx]) - 1
+                    # Day 0 overnight surprise (gap)
+                    overnight_gap = (df['Open'].iloc[idx] / df['Close'].iloc[idx-1]) - 1 if idx > 0 else 0
+                    # PEAD: continuation of surprise
+                    drift = np.sign(overnight_gap) * post_ret
+                    drifts.append(drift)
+                except:
+                    continue
+
+            if len(drifts) < 3:
+                return {'drift_score': 0, 'confidence': 0, 'n_events': len(spike_days)}
+
+            drift_score = np.mean(drifts)
+            confidence = min(1.0, len(drifts) / 12)  # More events = more confidence
+
+            return {
+                'drift_score': round(float(drift_score), 4),
+                'confidence': round(float(confidence), 2),
+                'n_events': len(drifts),
+                'avg_overnight_gap': round(float(np.mean([d for d in drifts])), 4) if drifts else 0,
+                'interpretation': (
+                    'Positive PEAD: earnings surprises tend to continue' if drift_score > 0.02 else
+                    'Negative PEAD: post-earnings reversals typical' if drift_score < -0.02 else
+                    'Weak PEAD pattern'
+                ),
+            }
+
+        except Exception:
+            return {'drift_score': 0, 'confidence': 0, 'n_events': 0}
+
+    def earnings_position_size(self, base_size: float, days_to_earnings: int,
+                              implied_move: float = 0.04) -> Dict:
+        """Adjust position size for earnings proximity.
+        
+        Strategy:
+        - D-30 to D-7: Full size (can position for earnings)
+        - D-7 to D-3: Reduce to 50% (avoid theta decay, lock profits)
+        - D-3 to D-1: Reduce to 25% (extreme event risk)
+        - D-Day: 0% (do not hold into earnings)
+        - D+1 to D+5: Can re-enter at 50% (capture PEAD)
+        """
+        if days_to_earnings is None:
+            return {
+                'adjusted_size': base_size,
+                'reduction': 0.0,
+                'strategy': 'No earnings date detected — normal sizing',
+            }
+
+        if days_to_earnings > 7:
+            adj = base_size
+            strategy = 'Pre-earnings positioning window — full size'
+        elif days_to_earnings > 3:
+            adj = base_size * 0.5
+            strategy = 'Reduce to 50% — earnings week risk building'
+        elif days_to_earnings > 0:
+            adj = base_size * 0.25
+            strategy = 'Reduce to 25% — imminent earnings, extreme theta'
+        elif days_to_earnings == 0:
+            adj = 0.0
+            strategy = 'DO NOT HOLD INTO EARNINGS — day-of closure'
+        elif days_to_earnings >= -1:
+            adj = base_size * 0.5
+            strategy = 'Post-earnings PEAD window — 50% re-entry'
+        elif days_to_earnings >= -5:
+            adj = base_size * 0.5
+            strategy = 'PEAD continuation — 50% size'
+        else:
+            adj = base_size
+            strategy = 'Post-earnings quiet period — normal sizing'
+
+        # Adjust for implied move magnitude
+        if implied_move > 0.08:  # > 8% expected move
+            adj *= 0.7
+            strategy += ' | High implied move (>8%), further reduced'
+        elif implied_move > 0.06:
+            adj *= 0.85
+            strategy += ' | Elevated implied move, slight reduction'
+
+        return {
+            'days_to_earnings': days_to_earnings,
+            'implied_move_pct': round(implied_move * 100, 2),
+            'base_size': round(base_size, 4),
+            'adjusted_size': round(adj, 4),
+            'reduction': round(1 - (adj / (base_size + 1e-10)), 2),
+            'strategy': strategy,
+        }
+
+    def full_analysis(self, ticker: str, base_size: float = 1.0) -> Dict:
+        """Complete earnings intelligence for a ticker."""
+        days, ed = self.days_to_earnings(ticker)
+        implied = self.implied_move(ticker)
+        pead = self.historical_pead(ticker)
+
+        sizing = self.earnings_position_size(
+            base_size, days,
+            implied.get('implied_move_pct', 4) / 100
+        )
+
+        return {
+            'ticker': ticker,
+            'estimated_earnings_date': ed.strftime('%Y-%m-%d') if ed else 'unknown',
+            'days_to_earnings': days,
+            'implied_move': implied,
+            'pead_analysis': pead,
+            'position_sizing': sizing,
+            'recommendation': sizing['strategy'],
+        }
+
+
+if __name__ == '__main__':
+    model = EarningsModel()
+    result = model.full_analysis('AAPL', base_size=1.0)
+    print(f"Estimated Earnings: {result['estimated_earnings_date']}")
+    print(f"Days Until: {result['days_to_earnings']}")
+    print(f"Implied Move: {result['implied_move'].get('implied_move_pct', 'N/A')}%")
+    print(f"PEAD Score: {result['pead_analysis'].get('drift_score', 0):.4f}")
+    print(f"Position Sizing: {result['position_sizing']['adjusted_size']*100:.0f}% ({result['position_sizing']['strategy']})")