"""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")