Add macro overlay with VIX, DXY, treasury yields, Fed calendar, CPI tracking

macro_overlay.py

@@ -0,0 +1,300 @@
+"""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)")