macro_overlay.py

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