event_risk_guard.py

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