"""Macro Overlay v1.0 — Real-Time Macro Regime Detection & Market Conditions Tracks VIX, DXY, Treasury yields, Fed calendar, and CPI data for context. Falls back to yfinance tickers when direct API unavailable. """ import yfinance as yf import numpy as np import pandas as pd from datetime import datetime, timedelta from typing import Dict, Optional, Tuple MACRO_TICKERS = { 'VIX': '^VIX', # CBOE Volatility Index 'DXY': 'DX-Y.NYB', # US Dollar Index (yfinance alternative) 'TNX': '^TNX', # 10-Year Treasury Yield 'FVX': '^FVX', # 5-Year Treasury Yield 'IRX': '^IRX', # 13-Week Treasury Yield 'SPY': 'SPY', # S&P 500 'QQQ': 'QQQ', # NASDAQ 100 'IWM': 'IWM', # Russell 2000 'GLD': 'GLD', # Gold 'USO': 'USO', # Oil 'TLT': 'TLT', # 20+ Year Treasury 'HYG': 'HYG', # High Yield Corporate } # Fed meeting dates (2025-2026). Update as needed. FED_MEETINGS = [ '2025-01-29', '2025-03-19', '2025-05-07', '2025-06-18', '2025-07-30', '2025-09-17', '2025-11-05', '2025-12-10', '2026-01-28', '2026-03-18', '2026-05-06', '2026-06-17', '2026-07-29', '2026-09-16', '2026-11-04', '2026-12-09', ] class MacroOverlay: """Real-time macro regime classification for trading context.""" def __init__(self, tickers: Optional[Dict[str, str]] = None): self.tickers = tickers or dict(MACRO_TICKERS) self._cache = {} # ticker -> (df, timestamp) self._cache_ttl = 300 # 5 min def _fetch(self, ticker: str, period: str = '3mo') -> Optional[pd.DataFrame]: """Fetch with caching.""" cache_key = f"{ticker}_{period}" now = datetime.now() if cache_key in self._cache: df, ts = self._cache[cache_key] if (now - ts).total_seconds() < self._cache_ttl: return df try: df = yf.Ticker(ticker).history(period=period) if df.empty: return None self._cache[cache_key] = (df, now) return df except Exception: return None def vix_context(self) -> Dict: """VIX regime classification.""" df = self._fetch(self.tickers.get('VIX', '^VIX')) if df is None: return {'level': 20.0, 'regime': 'normal', 'score': 50} last = df['Close'].iloc[-1] ma20 = df['Close'].rolling(20).mean().iloc[-1] vol = df['Close'].std() regime = 'normal' if last > 30: regime = 'crisis' elif last > 25: regime = 'elevated' elif last < 15: regime = 'complacent' elif last < ma20 * 0.9 and ma20 > 20: regime = 'declining' elif last > ma20 * 1.2: regime = 'spiking' # VIX score: lower = better for risk assets, but too low = complacency if regime == 'complacent': score = 40 # Quiet before storm elif regime == 'normal': score = 75 elif regime == 'declining': score = 85 # Fear receding elif regime == 'elevated': score = 35 # Elevated risk elif regime == 'spiking': score = 15 # Fear building elif regime == 'crisis': score = 10 # Max fear else: score = 50 return { 'level': round(float(last), 2), 'ma20': round(float(ma20), 2), 'regime': regime, 'score': score, } def treasury_yield_context(self) -> Dict: """Yield curve context.""" tnx = self._fetch(self.tickers.get('TNX', '^TNX')) fvx = self._fetch(self.tickers.get('FVX', '^FVX')) irx = self._fetch(self.tickers.get('IRX', '^IRX')) if tnx is None: return {'yield_10y': 4.2, 'regime': 'normal', 'score': 50} y10 = tnx['Close'].iloc[-1] / 100 # TNX is in basis points * 10 spread = None if fvx is not None: y5 = fvx['Close'].iloc[-1] / 100 spread_5_10 = y10 - y5 else: spread_5_10 = None if irx is not None: y3m = irx['Close'].iloc[-1] / 100 spread_3m_10y = y10 - y3m else: spread_3m_10y = None # Score: rising rates hurt growth stocks, inverted = recession risk score = 50 if y10 > 0.05: score -= 20 # 5%+ rates = restrictive elif y10 > 0.04: score -= 10 elif y10 < 0.03: score += 10 # Low rates = supportive if spread_3m_10y is not None: if spread_3m_10y < -0.5: # Deep inversion score -= 25 elif spread_3m_10y < -0.2: # Mild inversion score -= 15 elif spread_3m_10y > 1.0: # Steep = healthy score += 10 if spread_5_10 is not None: if spread_5_10 < 0: # 5-10 inversion score -= 5 regime = 'normal' if y10 > 0.05: regime = 'high_rates' elif spread_3m_10y is not None and spread_3m_10y < 0: regime = 'inverted' if spread_3m_10y < -0.3 else 'flat' elif y10 < 0.025: regime = 'low_rates' elif spread_3m_10y is not None and spread_3m_10y > 1.5: regime = 'steep' return { 'yield_10y': round(float(y10), 3), 'yield_5y': round(float(y5), 3) if fvx is not None else None, 'yield_3m': round(float(y3m), 3) if irx is not None else None, 'spread_3m_10y': round(float(spread_3m_10y), 3) if spread_3m_10y is not None else None, 'regime': regime, 'score': max(0, min(100, score)), } def dollar_context(self) -> Dict: """Dollar strength context.""" df = self._fetch(self.tickers.get('DXY', 'DX-Y.NYB')) if df is None: return {'level': 105.0, 'regime': 'normal', 'score': 50} last = df['Close'].iloc[-1] ma20 = df['Close'].rolling(20).mean().iloc[-1] score = 50 regime = 'normal' if last > ma20 * 1.02: regime = 'strengthening' score -= 10 # Strong dollar bad for EM and exporters elif last < ma20 * 0.98: regime = 'weakening' score += 10 # Weak dollar good for risk if last > 110: score -= 10 # Very strong elif last < 100: score += 10 # Very weak return { 'level': round(float(last), 2), 'regime': regime, 'score': max(0, min(100, score)), } def equity_context(self) -> Dict: """Broader equity market context.""" spy = self._fetch(self.tickers.get('SPY', 'SPY')) qqq = self._fetch(self.tickers.get('QQQ', 'QQQ')) iwm = self._fetch(self.tickers.get('IWM', 'IWM')) ctx = {} for name, df in [('SPY', spy), ('QQQ', qqq), ('IWM', iwm)]: if df is None: continue ret_20d = df['Close'].pct_change(20).iloc[-1] * 100 ret_5d = df['Close'].pct_change(5).iloc[-1] * 100 above_50d = df['Close'].iloc[-1] > df['Close'].rolling(50).mean().iloc[-1] ctx[name] = { 'return_20d': round(float(ret_20d), 2), 'return_5d': round(float(ret_5d), 2), 'above_50d': bool(above_50d), } # Score based on breadth breadth_score = 50 breadth_signals = [] for name, data in ctx.items(): if data.get('return_20d', 0) > 5: breadth_score += 5 breadth_signals.append(f"{name} +20d") if data.get('return_20d', 0) < -5: breadth_score -= 10 breadth_signals.append(f"{name} -20d") if data.get('above_50d', False): breadth_score += 5 return { 'breadth_score': max(0, min(100, breadth_score)), 'indices': ctx, 'signals': breadth_signals, } def fed_context(self) -> Dict: """Fed meeting proximity and rate regime.""" today = datetime.now().date() upcoming = [] for m in FED_MEETINGS: d = datetime.strptime(m, '%Y-%m-%d').date() delta = (d - today).days if delta >= -1 and delta <= 45: # Within 45 days upcoming.append({'date': m, 'days_until': delta}) next_meeting = min(upcoming, key=lambda x: abs(x['days_until'])) if upcoming else None days_until = next_meeting['days_until'] if next_meeting else 999 # Fed proximity penalty score = 50 if days_until <= 0: score -= 30 elif days_until <= 2: score -= 20 elif days_until <= 7: score -= 15 elif days_until <= 14: score -= 10 elif days_until <= 30: score -= 5 return { 'next_meeting': next_meeting['date'] if next_meeting else 'none', 'days_until': days_until, 'score': max(0, min(100, score)), } def full_macro_snapshot(self) -> Dict: """Complete macro dashboard.""" vix = self.vix_context() yield_ctx = self.treasury_yield_context() dollar = self.dollar_context() equity = self.equity_context() fed = self.fed_context() # Composite macro score: equal weight of components components = { 'vix': vix['score'], 'yield_curve': yield_ctx['score'], 'dollar': dollar['score'], 'equity_breadth': equity['breadth_score'], 'fed': fed['score'], } composite = np.mean(list(components.values())) # Regime classification if composite > 75: macro_regime = 'risk_on' elif composite < 35: macro_regime = 'risk_off' elif vix['regime'] == 'elevated' or vix['regime'] == 'spiking': macro_regime = 'risk_off_building' elif yield_ctx['regime'] == 'inverted': macro_regime = 'late_cycle' else: macro_regime = 'mixed' return { 'timestamp': datetime.now().isoformat(), 'composite_score': round(composite, 1), 'regime': macro_regime, 'components': components, 'vix': vix, 'yield_curve': yield_ctx, 'dollar': dollar, 'equity': equity, 'fed': fed, } if __name__ == '__main__': macro = MacroOverlay() snap = macro.full_macro_snapshot() print(f"Macro Regime: {snap['regime'].upper()}") print(f"Composite Score: {snap['composite_score']}/100") print(f"VIX: {snap['vix']['level']} ({snap['vix']['regime']})") if snap['yield_curve']['yield_10y']: print(f"10Y Yield: {snap['yield_curve']['yield_10y']}%") print(f"DXY Regime: {snap['dollar']['regime']}") print(f"Next Fed: {snap['fed']['next_meeting']} ({snap['fed']['days_until']} days)")