"""event_study.py — Event-Driven Alpha & Event Study Analysis Detects and analyzes market events (earnings, macro releases, M&A, FDA decisions) to generate abnormal returns and event-driven trading signals. References: - MacKinlay 1997: "Event Studies in Economics and Finance" - Engelberg et al. 2012: "Market Madness? The Case of Mad Money" - Savor & Wilson 2014: "Asset Pricing: A Tale of Two Days" """ import numpy as np, pandas as pd from scipy import stats class EventStudyAnalyzer: """Event study analysis for earnings, macro, and corporate events.""" def __init__(self, estimation_window=252, event_window=(-5, 20)): self.est_window = estimation_window self.ev_window = event_window def abnormal_returns(self, prices, market_prices, event_dates): """Compute abnormal returns around event dates.""" ret = prices.pct_change().dropna() mkt_ret = market_prices.pct_change().dropna() # Market model: R_i = alpha + beta * R_m + epsilon common = ret.index.intersection(mkt_ret.index) y = ret.loc[common].values X = np.column_stack([np.ones(len(common)), mkt_ret.loc[common].values]) beta = np.linalg.lstsq(X, y, rcond=None)[0] expected = beta[0] + beta[1] * mkt_ret abnormal = ret - expected.reindex(ret.index, fill_value=0) results = [] for date in event_dates: try: idx = ret.index.get_loc(pd.Timestamp(date)) start = max(0, idx + self.ev_window[0]) end = min(len(ret), idx + self.ev_window[1] + 1) ar = abnormal.iloc[start:end] car = ar.cumsum() results.append({ 'event_date': date, 'car_day0': float(ar.iloc[self.ev_window[0]*-1] if len(ar) > self.ev_window[0]*-1 else np.nan), 'car_window': float(car.iloc[-1] - car.iloc[0]), 'max_car': float(car.max()), 'min_car': float(car.min()), 'volatility_event': float(ar.std()), 't_stat': float(ar.mean() / (ar.std() + 1e-10) * np.sqrt(len(ar))) }) except Exception: continue return pd.DataFrame(results) def earnings_event_signal(self, prices, earnings_dates, surprise_threshold=0.05): """Generate post-earnings drift (PED) signal. Positive earnings surprise → buy for 1-60 days. """ ret = prices.pct_change().dropna() signals = pd.Series(0.0, index=prices.index) for ed in earnings_dates: try: idx = ret.index.get_loc(pd.Timestamp(ed)) if idx + 1 < len(ret): # 1-day reaction as proxy for surprise day1_ret = ret.iloc[idx + 1] if abs(day1_ret) > surprise_threshold: # Signal for next 5 days for d in range(1, 6): if idx + 1 + d < len(signals): signals.iloc[idx + 1 + d] = np.sign(day1_ret) * 0.2 except Exception: continue return signals def macro_event_signal(self, prices, macro_dates, volatility_threshold=0.015): """Signal around macro events (FOMC, NFP, CPI). Pre-event volatility compression → post-event volatility expansion. """ ret = prices.pct_change().dropna() vol = ret.rolling(20).std() signals = pd.Series(0.0, index=prices.index) for md in macro_dates: try: idx = ret.index.get_loc(pd.Timestamp(md)) if idx >= 20 and idx + 5 < len(ret): pre_vol = vol.iloc[idx - 5] post_vol = vol.iloc[idx:idx+5].mean() if pre_vol < volatility_threshold and post_vol > pre_vol * 1.5: signals.iloc[idx:idx+5] = np.sign(ret.iloc[idx]) * 0.3 except Exception: continue return signals def merger_arbitrage_signal(self, acquirer_prices, target_prices, spread_threshold=0.02): """Merger arbitrage: long target, short acquirer when deal spread exists.""" target_ret = target_prices.pct_change().dropna() acquirer_ret = acquirer_prices.pct_change().dropna() spread = (target_prices / acquirer_prices - 1).dropna() signal = pd.Series(0.0, index=target_prices.index) signal[spread > spread_threshold] = 1.0 # Long target signal[spread < -spread_threshold] = -0.5 # Short acquirer return signal def event_calendar(self, ticker): """Generate synthetic event calendar for a ticker.""" # In production, this would call an API (EarningsWhispers, etc.) today = pd.Timestamp.now() return pd.DataFrame({ 'date': [today + pd.DateOffset(days=30), today + pd.DateOffset(days=60), today + pd.DateOffset(days=90)], 'event': ['Q3 Earnings', 'FOMC Meeting', 'Q4 GDP Release'], 'type': ['earnings', 'macro', 'macro'], 'impact': ['high', 'high', 'medium'] }) def report(self, prices, market_prices, event_dates, event_type='earnings'): """Full event study report.""" ar = self.abnormal_returns(prices, market_prices, event_dates) if ar.empty: return "## Event Study\n\nNo valid events found in data range." mean_car = ar['car_window'].mean() t_stat = ar['car_window'].mean() / (ar['car_window'].std() + 1e-10) * np.sqrt(len(ar)) p_val = 2 * (1 - stats.t.cdf(abs(t_stat), len(ar)-1)) return f"""## Event Study: {event_type.upper()} | Statistic | Value | |-----------|-------| | Events analyzed | {len(ar)} | | Mean CAR (window) | {mean_car*100:.2f}% | | Median CAR | {ar['car_window'].median()*100:.2f}% | | Day-0 abnormal return | {ar['car_day0'].mean()*100:.2f}% | | t-statistic | {t_stat:.2f} | | p-value | {p_val:.4f} | | Significant? | {'✅ Yes' if p_val < 0.05 else '❌ No'} | **Interpretation:** {mean_car > 0 and 'Positive' or 'Negative'} post-event drift of {abs(mean_car)*100:.2f}% on average. """ if __name__ == '__main__': np.random.seed(42) prices = pd.Series(np.cumprod(1 + np.random.normal(0.0005, 0.015, 500)), index=pd.date_range('2022-01-01', periods=500, freq='B')) market = pd.Series(np.cumprod(1 + np.random.normal(0.0003, 0.012, 500)), index=pd.date_range('2022-01-01', periods=500, freq='B')) events = ['2022-03-15', '2022-06-15', '2022-09-15', '2022-12-15'] esa = EventStudyAnalyzer() print(esa.report(prices, market, events))