Add event risk guard — pre-trade checks for earnings/FOMC/macro with dynamic size reduction

event_risk_guard.py

@@ -0,0 +1,225 @@
+"""Event Risk Guard v1.0 — Pre-Trade Risk Overlay
+Checks for earnings, FOMC, macro releases, and other event risk before execution.
+Reduces position size dynamically based on event proximity and severity.
+Based on: Lo & MacKinlay (1999) event study methodology
+"""
+import re
+from datetime import datetime, timedelta
+from typing import Dict, Optional, List, Tuple
+import pandas as pd
+
+# Event severity multipliers (0.0 = full halt, 1.0 = no impact)
+EVENT_IMPACT = {
+    'earnings':          {'same_day': 0.00, 'd1': 0.30, 'd3': 0.60, 'd7': 0.80, 'd14': 0.90},
+    'fed_meeting':       {'same_day': 0.00, 'd1': 0.25, 'd3': 0.55, 'd7': 0.75, 'd14': 0.90},
+    'cpi_release':       {'same_day': 0.15, 'd1': 0.40, 'd3': 0.65, 'd7': 0.85, 'd14': 1.00},
+    'jobs_report':       {'same_day': 0.10, 'd1': 0.35, 'd3': 0.60, 'd7': 0.80, 'd14': 0.95},
+    'gdp':              {'same_day': 0.20, 'd1': 0.45, 'd3': 0.70, 'd7': 0.90, 'd14': 1.00},
+    'options_expiry':    {'same_day': 0.20, 'd1': 0.50, 'd3': 0.80, 'd7': 1.00, 'd14': 1.00},
+    'dividend':         {'same_day': 0.50, 'd1': 0.70, 'd3': 0.85, 'd7': 1.00, 'd14': 1.00},
+    'analyst_day':       {'same_day': 0.30, 'd1': 0.55, 'd3': 0.75, 'd7': 0.90, 'd14': 1.00},
+    'product_launch':    {'same_day': 0.20, 'd1': 0.40, 'd3': 0.65, 'd7': 0.85, 'd14': 1.00},
+    'conference':        {'same_day': 0.30, 'd1': 0.60, 'd3': 0.80, 'd7': 1.00, 'd14': 1.00},
+    'lawsuit_hearing':   {'same_day': 0.15, 'd1': 0.35, 'd3': 0.55, 'd7': 0.75, 'd14': 0.90},
+    'merger_vote':       {'same_day': 0.05, 'd1': 0.20, 'd3': 0.45, 'd7': 0.70, 'd14': 0.90},
+    'macro_general':     {'same_day': 0.25, 'd1': 0.50, 'd3': 0.70, 'd7': 0.85, 'd14': 1.00},
+}
+
+# Minimum days between events to consider them distinct (avoid double-counting)
+EVENT_CLUSTER_WINDOW = 3
+
+
+class EventRiskGuard:
+    """Pre-trade event risk overlay with dynamic size reduction."""
+
+    def __init__(self, impact_table: Optional[Dict] = None):
+        self.impact = impact_table or dict(EVENT_IMPACT)
+        self._earnings_cache = {}
+        self._macro_cache = {}
+
+    def classify_event(self, headline_or_title: str) -> Tuple[str, float]:
+        """Classify text into event type and confidence (0-1)."""
+        text = headline_or_title.lower()
+
+        patterns = {
+            'earnings':          ['earnings', 'quarterly', 'revenue', 'eps', 'profit', 'q1', 'q2', 'q3', 'q4', 'fiscal', 'guidance'],
+            'fed_meeting':       ['fomc', 'federal reserve', 'fed meeting', 'fed rate', 'interest rate decision', 'powell'],
+            'cpi_release':       ['cpi', 'consumer price', 'inflation data', 'core pce', 'inflation report'],
+            'jobs_report':       ['jobs report', 'unemployment', 'nonfarm payroll', 'nfp', 'labor market'],
+            'gdp':              ['gdp', 'economic growth', 'recession', 'economic output'],
+            'options_expiry':    ['options expiry', 'options expiration', 'op-ex', 'triple witching', 'quadruple witching'],
+            'dividend':         ['dividend', 'ex-dividend', 'dividend date', 'buyback'],
+            'analyst_day':       ['analyst day', 'investor day', 'management presentation'],
+            'product_launch':    ['product launch', 'new product', 'iphone', 'ai model', 'release date', 'unveil'],
+            'conference':        ['conference call', 'shareholder meeting', 'annual meeting'],
+            'lawsuit_hearing':   ['court hearing', 'trial date', 'lawsuit', 'sec hearing', 'doj'],
+            'merger_vote':       ['merger vote', 'shareholder vote', 'acquisition vote'],
+            'macro_general':     ['macro', 'economic data', 'pmi', 'retail sales', 'housing data'],
+        }
+
+        for etype, keywords in patterns.items():
+            count = sum(1 for kw in keywords if kw in text)
+            if count > 0:
+                confidence = min(1.0, count * 0.3 + 0.1)
+                return etype, confidence
+        return 'unknown', 0.1
+
+    def get_event_multiplier(self, event_type: str, days_until: int) -> float:
+        """Get size multiplier (0.0-1.0) based on event type and proximity."""
+        table = self.impact.get(event_type, self.impact.get('macro_general'))
+        if table is None:
+            return 1.0
+
+        if days_until <= 0:
+            return table.get('same_day', 0.25)
+        elif days_until <= 1:
+            return table.get('d1', 0.35)
+        elif days_until <= 3:
+            # Linear interpolation between d1 and d3
+            d1_mult = table.get('d1', 0.35)
+            d3_mult = table.get('d3', 0.60)
+            return d1_mult + (d3_mult - d1_mult) * (days_until - 1) / 2
+        elif days_until <= 7:
+            d3_mult = table.get('d3', 0.60)
+            d7_mult = table.get('d7', 0.80)
+            return d3_mult + (d7_mult - d3_mult) * (days_until - 3) / 4
+        elif days_until <= 14:
+            d7_mult = table.get('d7', 0.80)
+            d14_mult = table.get('d14', 0.95)
+            return d7_mult + (d14_mult - d7_mult) * (days_until - 7) / 7
+        else:
+            return 1.0
+
+    def check_events(self, events: List[Dict], base_exposure: float) -> Dict:
+        """Check all upcoming events and compute adjusted exposure.
+        
+        events: List of dicts with keys: 'type', 'date' (ISO), 'severity' (1-5, optional)
+        base_exposure: 0.0-1.0 proposed position size
+        
+        Returns adjusted exposure + reasoning.
+        """
+        today = datetime.now().date()
+        multipliers = []
+        reasons = []
+
+        for ev in events:
+            event_type = ev.get('type', 'macro_general')
+            event_date = ev.get('date')
+            if not event_date:
+                continue
+            try:
+                ev_date = datetime.strptime(event_date, '%Y-%m-%d').date()
+            except:
+                continue
+
+            days_until = (ev_date - today).days
+            if days_until < -1:  # Already happened yesterday, minimal impact
+                continue
+            if days_until > 30:  # Too far, ignore
+                continue
+
+            mult = self.get_event_multiplier(event_type, max(0, days_until))
+            # Severity override
+            severity = ev.get('severity', 3)
+            severity_adj = 1.0 - (severity - 3) * 0.1  # Adjust ±20%
+            mult *= severity_adj
+            mult = max(0.0, min(1.0, mult))
+
+            multipliers.append(mult)
+            reasons.append({
+                'event': event_type,
+                'date': event_date,
+                'days_until': days_until,
+                'severity': severity,
+                'raw_multiplier': mult,
+                'severity_adj': severity_adj,
+            })
+
+        if not multipliers:
+            return {
+                'can_trade': True,
+                'adjusted_exposure': base_exposure,
+                'reduction': 0.0,
+                'reason': 'No significant events in next 30 days',
+                'events': [],
+                'is_halted': False,
+            }
+
+        # Use minimum multiplier (most restrictive event dominates)
+        min_mult = min(multipliers)
+        adjusted = base_exposure * min_mult
+        reduction = 1 - min_mult
+        is_halted = min_mult < 0.05
+
+        # Find the binding event
+        binding = reasons[multipliers.index(min_mult)]
+
+        return {
+            'can_trade': not is_halted,
+            'adjusted_exposure': round(adjusted, 4),
+            'reduction': round(reduction, 2),
+            'reason': (f"{binding['event'].upper()} on {binding['date']} "
+                      f"({binding['days_until']} days). "
+                      f"Size reduced by {reduction*100:.0f}%"),
+            'binding_event': binding,
+            'events': reasons,
+            'is_halted': is_halted,
+            'all_multipliers': multipliers,
+        }
+
+    def check_ticker(self, ticker: str, base_exposure: float,
+                     custom_events: Optional[List[Dict]] = None) -> Dict:
+        """Full event risk check for a specific ticker.
+        
+        Uses built-in event calendar + any custom events provided.
+        """
+        # Build default events based on ticker
+        default_events = []
+
+        # Check if it's a known stock with known earnings window
+        # For demo, we use a simple heuristic: estimate next quarterly window
+        # In production, this connects to earnings API
+
+        # Quarterly earnings: next ~30-90 days from now
+        today = datetime.now()
+        # Simplified: assume earnings within next quarter
+        next_q_end = today + timedelta(days=30)
+        default_events.append({
+            'type': 'earnings',
+            'date': next_q_end.strftime('%Y-%m-%d'),
+            'severity': 4,
+        })
+
+        # Fed meetings
+        from macro_overlay import FED_MEETINGS
+        for m in FED_MEETINGS:
+            m_date = datetime.strptime(m, '%Y-%m-%d').date()
+            delta = (m_date - today.date()).days
+            if 0 <= delta <= 14:
+                default_events.append({
+                    'type': 'fed_meeting',
+                    'date': m,
+                    'severity': 5,
+                })
+                break  # Only the next one
+
+        all_events = default_events + (custom_events or [])
+        return self.check_events(all_events, base_exposure)
+
+
+if __name__ == '__main__':
+    guard = EventRiskGuard()
+
+    # Example: MSFT with earnings in 5 days + Fed meeting in 2 days
+    events = [
+        {'type': 'earnings', 'date': '2025-05-15', 'severity': 4},
+        {'type': 'fed_meeting', 'date': '2025-05-12', 'severity': 5},
+    ]
+
+    result = guard.check_events(events, base_exposure=1.0)
+    print(f"Can Trade: {result['can_trade']}")
+    print(f"Adjusted Exposure: {result['adjusted_exposure']*100:.1f}%")
+    print(f"Reduction: {result['reduction']*100:.0f}%")
+    print(f"Reason: {result['reason']}")
+    if result['is_halted']:
+        print("⚠️ TRADING HALTED — Event risk too high")