add Market Risk Engine

examples/risk_engine_example.py

@@ -0,0 +1,154 @@
+"""
+Market Risk Engine - Usage Examples
+
+Simple examples showing how to use the Market Risk Engine.
+"""
+
+import asyncio
+from src.core.macroeconomics.markets.risk_engine import (
+    MarketRiskEngine,
+    RiskRegime,
+    RiskEngineConfig,
+)
+
+
+async def example_1_basic_usage():
+    """Example 1: Basic risk assessment."""
+    print("\n=== Example 1: Basic Risk Assessment ===\n")
+
+    # Initialize engine
+    engine = MarketRiskEngine()
+
+    # Get current risk score
+    risk_score = await engine.assess_current_risk()
+
+    # Display results
+    print(f"Global Risk Score: {risk_score.risk_score:.1f}/100")
+    print(f"Risk Regime: {risk_score.regime.value.upper()}")
+
+    # Act based on regime
+    if risk_score.regime == RiskRegime.RED:
+        print("⚠️ HIGH RISK: Consider defensive positioning")
+    elif risk_score.regime == RiskRegime.YELLOW:
+        print("⚡ MODERATE RISK: Monitor closely")
+    else:
+        print("✅ LOW RISK: Favorable environment for risk assets")
+
+
+async def example_2_category_breakdown():
+    """Example 2: Detailed category breakdown."""
+    print("\n=== Example 2: Category Breakdown ===\n")
+
+    engine = MarketRiskEngine()
+    risk_score = await engine.assess_current_risk()
+
+    print("Risk by Category:")
+    for category, score in risk_score.category_scores.items():
+        cat_name = category.value.replace('_', ' ').title()
+        print(f"  {cat_name:20s} {score.score:5.1f}/100")
+
+
+async def example_3_custom_config():
+    """Example 3: Custom configuration."""
+    print("\n=== Example 3: Custom Configuration ===\n")
+
+    # Create custom config
+    config = RiskEngineConfig(
+        lookback_days=252  # 1 year instead of default 2 years
+    )
+
+    engine = MarketRiskEngine(config=config)
+    risk_score = await engine.assess_current_risk()
+
+    print(f"Risk Score (1-year lookback): {risk_score.risk_score:.1f}/100")
+
+
+async def example_4_full_report():
+    """Example 4: Generate full report."""
+    print("\n=== Example 4: Full Report ===\n")
+
+    engine = MarketRiskEngine()
+
+    # Generate comprehensive report
+    report = await engine.get_full_report()
+
+    print(report)
+
+
+async def example_5_top_risks():
+    """Example 5: Identify top risk categories."""
+    print("\n=== Example 5: Top Risk Categories ===\n")
+
+    engine = MarketRiskEngine()
+    risk_score = await engine.assess_current_risk()
+
+    # Get top 3 most stressed categories
+    top_risks = engine.aggregator.get_top_risks(risk_score, n=3)
+
+    print("Top 3 Risk Areas:")
+    for i, cat_score in enumerate(top_risks, 1):
+        cat_name = cat_score.category.value.replace('_', ' ').title()
+        print(f"\n{i}. {cat_name}")
+        print(f"   Score: {cat_score.score:.1f}/100")
+        print(f"   {cat_score.description}")
+
+
+async def example_6_programmatic_usage():
+    """Example 6: Programmatic decision making."""
+    print("\n=== Example 6: Programmatic Usage ===\n")
+
+    engine = MarketRiskEngine()
+    risk_score = await engine.assess_current_risk()
+
+    # Make trading decisions based on risk
+    if risk_score.risk_score > 70:
+        print("🔴 Risk Score > 70: Reduce exposure")
+        print("   Action: Move to defensive positions (TLT, GLD, cash)")
+
+    elif risk_score.risk_score > 50:
+        print("⚡ Risk Score 50-70: Cautious positioning")
+        print("   Action: Reduce leverage, tighten stops")
+
+    else:
+        print("✅ Risk Score < 50: Favorable for risk assets")
+        print("   Action: Normal positioning, consider opportunities")
+
+    # Check specific categories
+    credit_score = risk_score.category_scores.get(
+        next(c for c in risk_score.category_scores.keys()
+             if c.value == "credit")
+    )
+
+    if credit_score and credit_score.score > 65:
+        print("\n⚠️ Credit stress detected:")
+        print("   Consider: Reduce HY exposure, focus on quality")
+
+
+async def main():
+    """Run all examples."""
+    examples = [
+        example_1_basic_usage,
+        example_2_category_breakdown,
+        example_3_custom_config,
+        example_4_full_report,
+        example_5_top_risks,
+        example_6_programmatic_usage,
+    ]
+
+    for example in examples:
+        try:
+            await example()
+        except Exception as e:
+            print(f"\n✗ Example failed: {e}")
+            import traceback
+            traceback.print_exc()
+
+        print("\n" + "=" * 70 + "\n")
+
+
+if __name__ == "__main__":
+    # Run single example
+    # asyncio.run(example_1_basic_usage())
+
+    # Or run all examples
+    asyncio.run(main())

requirements.txt

@@ -6,6 +6,7 @@ pandas>=1.5.0,<3.0.0
 openpyxl>=3.0.0
 
 yfinance>=0.2.18
+pandas-datareader>=0.10.0
 finnhub-python>=2.4.0
 
 fastapi>=0.100.0

src/core/macroeconomics/markets/risk_engine/__init__.py

@@ -50,6 +50,9 @@ from src.core.macroeconomics.markets.risk_engine.data_sources import (
     YAHOO_SYMBOLS,
 )
 
+from src.core.macroeconomics.markets.risk_engine.aggregator import RiskAggregator
+from src.core.macroeconomics.markets.risk_engine.engine import MarketRiskEngine
+
 __all__ = [
     # Core Models
     "RiskRegime",
@@ -65,6 +68,9 @@ __all__ = [
     "YahooSource",
     "FRED_SYMBOLS",
     "YAHOO_SYMBOLS",
+    # Engine
+    "MarketRiskEngine",
+    "RiskAggregator",
 ]
 
 __version__ = "0.1.0"

src/core/macroeconomics/markets/risk_engine/aggregator.py

@@ -0,0 +1,349 @@
+"""
+Aggregation and scoring logic for risk engine.
+
+Combines individual indicators into category scores and global risk score.
+"""
+
+from typing import Dict, List, Optional
+from collections import defaultdict
+import numpy as np
+
+from src.core.macroeconomics.markets.risk_engine.models import (
+    IndicatorSignal,
+    IndicatorCategory,
+    CategoryScore,
+    GlobalRiskScore,
+    RiskRegime,
+)
+from src.telegram_bot.logger import main_logger as logger
+
+
+class RiskAggregator:
+    """
+    Aggregates individual indicators into category and global scores.
+
+    Architecture:
+    1. Group indicators by category
+    2. Calculate category-level stress scores
+    3. Combine categories into global risk score
+    4. Map to risk regime (GREEN/YELLOW/RED)
+    """
+
+    # Category weights for global score
+    # Based on systemic importance
+    CATEGORY_WEIGHTS = {
+        IndicatorCategory.VOLATILITY: 0.20,     # 20% - Market fear gauge
+        IndicatorCategory.CREDIT: 0.18,         # 18% - Credit markets critical
+        IndicatorCategory.LIQUIDITY: 0.15,      # 15% - Liquidity crises = systemic
+        IndicatorCategory.MACRO_RATES: 0.15,    # 15% - Macro regime shifts
+        IndicatorCategory.FX_FUNDING: 0.12,     # 12% - Dollar funding stress
+        IndicatorCategory.BREADTH: 0.10,        # 10% - Market internals
+        IndicatorCategory.TAIL_RISK: 0.10,      # 10% - Tail events
+    }
+
+    # Risk regime thresholds
+    REGIME_THRESHOLDS = {
+        "green_yellow": 40,   # Below 40 = GREEN (Risk-On)
+        "yellow_red": 65,     # Above 65 = RED (Deleveraging Risk)
+    }
+
+    def __init__(self):
+        """Initialize aggregator."""
+        # Validate weights sum to 1
+        total_weight = sum(self.CATEGORY_WEIGHTS.values())
+        if not np.isclose(total_weight, 1.0):
+            logger.warning(
+                f"Category weights sum to {total_weight}, not 1.0. "
+                "Normalizing..."
+            )
+            # Normalize
+            factor = 1.0 / total_weight
+            self.CATEGORY_WEIGHTS = {
+                k: v * factor for k, v in self.CATEGORY_WEIGHTS.items()
+            }
+
+    def aggregate(
+        self,
+        signals: List[IndicatorSignal]
+    ) -> GlobalRiskScore:
+        """
+        Aggregate indicators into global risk score.
+
+        Args:
+            signals: List of indicator signals
+
+        Returns:
+            GlobalRiskScore with regime and breakdown
+        """
+        if not signals:
+            logger.warning("No signals provided for aggregation")
+            return self._create_default_score()
+
+        # Group signals by category
+        category_signals = self._group_by_category(signals)
+
+        # Calculate category scores
+        category_scores = {}
+        for category, cat_signals in category_signals.items():
+            category_scores[category] = self._calculate_category_score(
+                category,
+                cat_signals
+            )
+
+        # Calculate global risk score
+        global_score = self._calculate_global_score(category_scores)
+
+        # Determine risk regime
+        regime = self._determine_regime(global_score)
+
+        return GlobalRiskScore(
+            risk_score=global_score,
+            regime=regime,
+            category_scores=category_scores
+        )
+
+    def _group_by_category(
+        self,
+        signals: List[IndicatorSignal]
+    ) -> Dict[IndicatorCategory, List[IndicatorSignal]]:
+        """Group signals by category."""
+        grouped = defaultdict(list)
+        for signal in signals:
+            grouped[signal.category].append(signal)
+        return dict(grouped)
+
+    def _calculate_category_score(
+        self,
+        category: IndicatorCategory,
+        signals: List[IndicatorSignal]
+    ) -> CategoryScore:
+        """
+        Calculate category-level score.
+
+        Uses equal-weighted average of indicator stress levels.
+        Tracks contributing indicators and their signals.
+
+        Args:
+            category: Indicator category
+            signals: Signals in this category
+
+        Returns:
+            CategoryScore
+        """
+        if not signals:
+            return CategoryScore(
+                category=category,
+                score=50.0,  # Neutral
+                active_indicators=0,
+                description="No data available"
+            )
+
+        # Equal-weighted average of stress levels
+        stress_levels = [s.stress_level for s in signals]
+        avg_stress = np.mean(stress_levels)
+
+        # Find most stressed indicator for description
+        max_signal = max(signals, key=lambda s: s.stress_level)
+
+        # Count active (non-neutral) indicators
+        active_count = sum(1 for s in signals if s.stress_level > 55 or s.stress_level < 45)
+
+        # Create description highlighting key signals
+        if avg_stress > 70:
+            status = "🔴 SEVERE"
+        elif avg_stress > 55:
+            status = "⚠️ ELEVATED"
+        elif avg_stress < 40:
+            status = "✅ CALM"
+        else:
+            status = "⚡ WATCH"
+
+        description = (
+            f"{status}: {category.value.replace('_', ' ').title()} "
+            f"({len(signals)} indicators, {active_count} active)"
+        )
+
+        return CategoryScore(
+            category=category,
+            score=avg_stress,
+            active_indicators=len(signals),
+            description=description
+        )
+
+    def _calculate_global_score(
+        self,
+        category_scores: Dict[IndicatorCategory, CategoryScore]
+    ) -> float:
+        """
+        Calculate global risk score from category scores.
+
+        Uses weighted average based on CATEGORY_WEIGHTS.
+        Missing categories are treated as neutral (50.0).
+
+        Args:
+            category_scores: Category-level scores
+
+        Returns:
+            Global risk score (0-100)
+        """
+        weighted_sum = 0.0
+        total_weight = 0.0
+
+        for category, weight in self.CATEGORY_WEIGHTS.items():
+            if category in category_scores:
+                score = category_scores[category].score
+                weighted_sum += score * weight
+                total_weight += weight
+            else:
+                # Missing category: use neutral score
+                weighted_sum += 50.0 * weight
+                total_weight += weight
+
+        if total_weight == 0:
+            return 50.0  # Neutral
+
+        global_score = weighted_sum / total_weight
+        return float(np.clip(global_score, 0.0, 100.0))
+
+    def _determine_regime(self, risk_score: float) -> RiskRegime:
+        """
+        Map risk score to regime.
+
+        Thresholds:
+        - [0, 40): GREEN (Risk-On)
+        - [40, 65): YELLOW (Fragile)
+        - [65, 100]: RED (Deleveraging Risk)
+
+        Args:
+            risk_score: Global risk score (0-100)
+
+        Returns:
+            RiskRegime
+        """
+        if risk_score < self.REGIME_THRESHOLDS["green_yellow"]:
+            return RiskRegime.GREEN
+        elif risk_score < self.REGIME_THRESHOLDS["yellow_red"]:
+            return RiskRegime.YELLOW
+        else:
+            return RiskRegime.RED
+
+    def _create_default_score(self) -> GlobalRiskScore:
+        """Create default score when no data available."""
+        category_scores = {
+            category: CategoryScore(
+                category=category,
+                score=50.0,
+                active_indicators=0,
+                description="No data"
+            )
+            for category in IndicatorCategory
+        }
+
+        return GlobalRiskScore(
+            risk_score=50.0,
+            regime=RiskRegime.YELLOW,
+            category_scores=category_scores
+        )
+
+    def get_top_risks(
+        self,
+        global_score: GlobalRiskScore,
+        n: int = 5
+    ) -> List[CategoryScore]:
+        """
+        Get top N most stressed categories.
+
+        Args:
+            global_score: Global risk score
+            n: Number of top risks to return
+
+        Returns:
+            List of category scores, sorted by stress level
+        """
+        sorted_categories = sorted(
+            global_score.category_scores.values(),
+            key=lambda c: c.score,
+            reverse=True
+        )
+        return sorted_categories[:n]
+
+    def get_regime_interpretation(self, regime: RiskRegime) -> str:
+        """
+        Get human-readable interpretation of regime.
+
+        Args:
+            regime: Risk regime
+
+        Returns:
+            Interpretation string
+        """
+        interpretations = {
+            RiskRegime.GREEN: (
+                "✅ RISK-ON: Markets healthy, favorable environment for risk assets. "
+                "Low volatility, tight credit spreads, strong liquidity."
+            ),
+            RiskRegime.YELLOW: (
+                "⚡ FRAGILE: Markets showing stress signals but not yet critical. "
+                "Elevated volatility, widening spreads, or deteriorating breadth. "
+                "Monitor closely for regime shift."
+            ),
+            RiskRegime.RED: (
+                "🔴 DELEVERAGING RISK: Severe market stress detected. "
+                "Multiple warning signals flashing. High probability of forced selling, "
+                "liquidity issues, or systemic stress. Risk-off positioning recommended."
+            )
+        }
+        return interpretations.get(regime, "Unknown regime")
+
+    def format_summary(self, global_score: GlobalRiskScore) -> str:
+        """
+        Format global score as text summary.
+
+        Args:
+            global_score: Global risk score
+
+        Returns:
+            Formatted summary string
+        """
+        lines = []
+        lines.append("=" * 50)
+        lines.append("MARKET RISK ENGINE - GLOBAL ASSESSMENT")
+        lines.append("=" * 50)
+        lines.append("")
+
+        # Global score
+        regime_emoji = {
+            RiskRegime.GREEN: "✅",
+            RiskRegime.YELLOW: "⚡",
+            RiskRegime.RED: "🔴"
+        }
+        lines.append(
+            f"Global Risk Score: {global_score.risk_score:.1f}/100 "
+            f"{regime_emoji[global_score.regime]} {global_score.regime.value.upper()}"
+        )
+        lines.append("")
+        lines.append(self.get_regime_interpretation(global_score.regime))
+        lines.append("")
+
+        # Category breakdown
+        lines.append("Category Breakdown:")
+        lines.append("-" * 50)
+
+        # Sort categories by score (descending)
+        sorted_cats = sorted(
+            global_score.category_scores.items(),
+            key=lambda x: x[1].score,
+            reverse=True
+        )
+
+        for category, score in sorted_cats:
+            cat_name = category.value.replace('_', ' ').title()
+            weight = self.CATEGORY_WEIGHTS[category] * 100
+            lines.append(
+                f"  {cat_name:20s} {score.score:5.1f}/100 "
+                f"(weight: {weight:4.1f}%) - {score.description}"
+            )
+
+        lines.append("=" * 50)
+
+        return "\n".join(lines)

src/core/macroeconomics/markets/risk_engine/engine.py

@@ -0,0 +1,373 @@
+"""
+Main Market Risk Engine.
+
+Orchestrates data loading, indicator calculation, and risk scoring.
+"""
+
+import asyncio
+from typing import Dict, List, Optional
+from datetime import datetime, timedelta
+
+from src.core.macroeconomics.markets.risk_engine.models import (
+    TimeSeriesData,
+    GlobalRiskScore,
+    IndicatorSignal,
+    RiskEngineConfig,
+)
+from src.core.macroeconomics.markets.risk_engine.data_sources import (
+    FREDSource,
+    YahooSource,
+)
+from src.core.macroeconomics.markets.risk_engine.indicators import *
+from src.core.macroeconomics.markets.risk_engine.aggregator import RiskAggregator
+from src.telegram_bot.logger import main_logger as logger
+
+
+class MarketRiskEngine:
+    """
+    Professional market risk engine for turbulence detection.
+
+    Architecture:
+    1. Data Sources → Fetch macro/market data
+    2. Indicators → Calculate stress signals
+    3. Aggregation → Combine into category scores
+    4. Risk Score → Global regime (GREEN/YELLOW/RED)
+
+    Example:
+        engine = MarketRiskEngine()
+        risk_score = await engine.assess_current_risk()
+        print(f"Risk: {risk_score.risk_score}/100")
+        print(f"Regime: {risk_score.regime.value}")
+    """
+
+    def __init__(self, config: Optional[RiskEngineConfig] = None):
+        """
+        Initialize risk engine.
+
+        Args:
+            config: Engine configuration (optional)
+        """
+        self.config = config or RiskEngineConfig()
+
+        # Initialize data sources
+        self.fred_source = FREDSource()
+        self.yahoo_source = YahooSource()
+
+        # Initialize aggregator
+        self.aggregator = RiskAggregator()
+
+        # Initialize all indicators
+        self.indicators = self._initialize_indicators()
+
+        logger.info(
+            f"MarketRiskEngine initialized with {len(self.indicators)} indicators"
+        )
+
+    def _initialize_indicators(self) -> List:
+        """Initialize all indicator instances."""
+        return [
+            # Volatility (4)
+            VIXStressIndicator(),
+            VIXTermStructureIndicator(),
+            VVIXIndicator(),
+            RealizedVolatilityIndicator(),
+
+            # Credit (4)
+            CreditSpreadWideningIndicator(),
+            IGvsHYDivergenceIndicator(),
+            CreditETFPressureIndicator(),
+            CorporateBondStressIndicator(),
+
+            # Liquidity (4)
+            AmihudIlliquidityIndicator(),
+            BidAskSpreadIndicator(),
+            TreasuryLiquidityIndicator(),
+            MarketDepthIndicator(),
+
+            # Macro/Rates (5)
+            YieldCurveIndicator(),
+            NFCIIndicator(),
+            RepoStressIndicator(),
+            MOVEIndexIndicator(),
+            RealRatesIndicator(),
+
+            # FX/Funding (4)
+            DXYSpikeIndicator(),
+            JPYCarryUnwindIndicator(),
+            CrossCurrencyBasisIndicator(),
+            EmergingMarketStressIndicator(),
+
+            # Breadth (4)
+            SPYQQQDivergenceIndicator(),
+            SectorRotationIndicator(),
+            MarketConcentrationIndicator(),
+            SmallCapWeaknessIndicator(),
+
+            # Tail Risk (5)
+            CorrelationSpikeIndicator(),
+            VolatilityRegimeIndicator(),
+            DrawdownIndicator(),
+            SkewIndicator(),
+            TailEventIndicator(),
+        ]
+
+    async def assess_current_risk(
+        self,
+        lookback_days: Optional[int] = None
+    ) -> GlobalRiskScore:
+        """
+        Assess current market risk.
+
+        Main entry point for risk assessment.
+
+        Args:
+            lookback_days: Days of historical data to fetch (default from config)
+
+        Returns:
+            GlobalRiskScore with regime and breakdown
+        """
+        lookback = lookback_days or self.config.lookback_days
+
+        logger.info(f"Starting risk assessment (lookback: {lookback} days)")
+
+        # Step 1: Load data
+        data = await self._load_data(lookback)
+
+        if not data:
+            logger.error("Failed to load any data")
+            return self.aggregator._create_default_score()
+
+        logger.info(f"Loaded data for {len(data)} symbols")
+
+        # Step 2: Calculate indicators
+        signals = self._calculate_indicators(data)
+
+        if not signals:
+            logger.warning("No indicators calculated successfully")
+            return self.aggregator._create_default_score()
+
+        logger.info(
+            f"Calculated {len(signals)}/{len(self.indicators)} indicators successfully"
+        )
+
+        # Step 3: Aggregate into risk score
+        risk_score = self.aggregator.aggregate(signals)
+
+        logger.info(
+            f"Risk assessment complete: {risk_score.risk_score:.1f}/100 "
+            f"({risk_score.regime.value})"
+        )
+
+        return risk_score
+
+    async def _load_data(
+        self,
+        lookback_days: int
+    ) -> Dict[str, TimeSeriesData]:
+        """
+        Load data from all sources.
+
+        Args:
+            lookback_days: Days of historical data
+
+        Returns:
+            Dictionary of symbol -> TimeSeriesData
+        """
+        # Calculate date range
+        end_date = datetime.now()
+        start_date = end_date - timedelta(days=lookback_days)
+
+        logger.info(
+            f"Loading data from {start_date.date()} to {end_date.date()}"
+        )
+
+        # Load from both sources in parallel
+        fred_task = self.fred_source.fetch_multiple(start_date, end_date)
+        yahoo_task = self.yahoo_source.fetch_multiple(start_date, end_date)
+
+        fred_data, yahoo_data = await asyncio.gather(
+            fred_task,
+            yahoo_task,
+            return_exceptions=True
+        )
+
+        # Combine data
+        combined = {}
+
+        if isinstance(fred_data, dict):
+            combined.update(fred_data)
+            logger.info(f"Loaded {len(fred_data)} FRED symbols")
+        else:
+            logger.error(f"FRED data loading failed: {fred_data}")
+
+        if isinstance(yahoo_data, dict):
+            combined.update(yahoo_data)
+            logger.info(f"Loaded {len(yahoo_data)} Yahoo symbols")
+        else:
+            logger.error(f"Yahoo data loading failed: {yahoo_data}")
+
+        return combined
+
+    def _calculate_indicators(
+        self,
+        data: Dict[str, TimeSeriesData]
+    ) -> List[IndicatorSignal]:
+        """
+        Calculate all indicators.
+
+        Args:
+            data: Market/macro data
+
+        Returns:
+            List of indicator signals
+        """
+        signals = []
+
+        for indicator in self.indicators:
+            try:
+                signal = indicator.compute(data)
+                if signal is not None:
+                    signals.append(signal)
+                else:
+                    logger.debug(
+                        f"{indicator.name}: No signal (likely missing data)"
+                    )
+            except Exception as e:
+                logger.error(
+                    f"{indicator.name}: Calculation failed: {e}",
+                    exc_info=True
+                )
+
+        return signals
+
+    def get_indicator_details(
+        self,
+        signals: List[IndicatorSignal]
+    ) -> str:
+        """
+        Format detailed indicator breakdown.
+
+        Args:
+            signals: List of indicator signals
+
+        Returns:
+            Formatted string with all indicator details
+        """
+        lines = []
+        lines.append("=" * 60)
+        lines.append("INDICATOR DETAILS")
+        lines.append("=" * 60)
+        lines.append("")
+
+        # Group by category
+        from collections import defaultdict
+        by_category = defaultdict(list)
+        for signal in signals:
+            by_category[signal.category].append(signal)
+
+        # Sort categories
+        sorted_cats = sorted(by_category.items(), key=lambda x: x[0].value)
+
+        for category, cat_signals in sorted_cats:
+            lines.append(f"\n### {category.value.replace('_', ' ').upper()} ###")
+            lines.append("-" * 60)
+
+            # Sort signals by stress level (descending)
+            sorted_signals = sorted(
+                cat_signals,
+                key=lambda s: s.stress_level,
+                reverse=True
+            )
+
+            for signal in sorted_signals:
+                stress_bar = self._create_stress_bar(signal.stress_level)
+                lines.append(f"\n{signal.indicator_name}")
+                lines.append(f"  Stress: {signal.stress_level:.1f}/100 {stress_bar}")
+                lines.append(f"  Value: {signal.description}")
+                lines.append(f"  {signal.interpretation}")
+
+        lines.append("\n" + "=" * 60)
+
+        return "\n".join(lines)
+
+    def _create_stress_bar(self, stress_level: float) -> str:
+        """Create visual stress level bar."""
+        # 10 blocks, each = 10%
+        filled = int(stress_level / 10)
+        empty = 10 - filled
+
+        if stress_level >= 70:
+            bar_char = "█"
+            color = "🔴"
+        elif stress_level >= 55:
+            bar_char = "▓"
+            color = "⚠️"
+        else:
+            bar_char = "░"
+            color = "✅"
+
+        bar = bar_char * filled + "░" * empty
+        return f"{color} [{bar}]"
+
+    async def get_full_report(
+        self,
+        lookback_days: Optional[int] = None
+    ) -> str:
+        """
+        Generate complete risk report.
+
+        Args:
+            lookback_days: Days of historical data
+
+        Returns:
+            Full formatted report
+        """
+        # Get risk score
+        risk_score = await self.assess_current_risk(lookback_days)
+
+        # Get data for indicator details
+        lookback = lookback_days or self.config.lookback_days
+        data = await self._load_data(lookback)
+        signals = self._calculate_indicators(data)
+
+        # Build report
+        report_parts = []
+
+        # 1. Global summary
+        report_parts.append(self.aggregator.format_summary(risk_score))
+        report_parts.append("\n")
+
+        # 2. Top risks
+        report_parts.append("=" * 60)
+        report_parts.append("TOP RISKS")
+        report_parts.append("=" * 60)
+        report_parts.append("")
+
+        top_risks = self.aggregator.get_top_risks(risk_score, n=3)
+        for i, cat_score in enumerate(top_risks, 1):
+            cat_name = cat_score.category.value.replace('_', ' ').title()
+            report_parts.append(
+                f"{i}. {cat_name}: {cat_score.score:.1f}/100 - {cat_score.description}"
+            )
+
+        report_parts.append("\n")
+
+        # 3. Indicator details
+        report_parts.append(self.get_indicator_details(signals))
+
+        # 4. Timestamp
+        report_parts.append("\n")
+        report_parts.append(f"Report generated: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}")
+
+        return "\n".join(report_parts)
+
+    def get_config_summary(self) -> str:
+        """Get configuration summary."""
+        lines = []
+        lines.append("Risk Engine Configuration:")
+        lines.append(f"  Lookback days: {self.config.lookback_days}")
+        lines.append(f"  Total indicators: {len(self.indicators)}")
+        lines.append(f"  Data sources: FRED, Yahoo Finance")
+        lines.append(f"  Regime thresholds: GREEN<{self.aggregator.REGIME_THRESHOLDS['green_yellow']}, "
+                    f"YELLOW<{self.aggregator.REGIME_THRESHOLDS['yellow_red']}, RED")
+        return "\n".join(lines)

src/core/macroeconomics/markets/risk_engine/indicators/__init__.py

@@ -0,0 +1,99 @@
+"""
+Market risk indicators.
+
+Organized by category:
+- Volatility: VIX, VVIX, term structure, realized vol
+- Credit: Spreads, default risk
+- Liquidity: Bid-ask, Amihud
+- Macro/Rates: Yield curve, NFCI, repo
+- FX/Funding: DXY, JPY carry, basis
+- Breadth: Advance/decline, MA divergences
+- Tail Risk: Correlations, regime shifts
+"""
+
+from src.core.macroeconomics.markets.risk_engine.indicators.base import BaseIndicator
+from src.core.macroeconomics.markets.risk_engine.indicators.volatility import (
+    VIXStressIndicator,
+    VIXTermStructureIndicator,
+    VVIXIndicator,
+    RealizedVolatilityIndicator,
+)
+from src.core.macroeconomics.markets.risk_engine.indicators.credit import (
+    CreditSpreadWideningIndicator,
+    IGvsHYDivergenceIndicator,
+    CreditETFPressureIndicator,
+    CorporateBondStressIndicator,
+)
+from src.core.macroeconomics.markets.risk_engine.indicators.liquidity import (
+    AmihudIlliquidityIndicator,
+    BidAskSpreadIndicator,
+    TreasuryLiquidityIndicator,
+    MarketDepthIndicator,
+)
+from src.core.macroeconomics.markets.risk_engine.indicators.macro_rates import (
+    YieldCurveIndicator,
+    NFCIIndicator,
+    RepoStressIndicator,
+    MOVEIndexIndicator,
+    RealRatesIndicator,
+)
+from src.core.macroeconomics.markets.risk_engine.indicators.fx_funding import (
+    DXYSpikeIndicator,
+    JPYCarryUnwindIndicator,
+    CrossCurrencyBasisIndicator,
+    EmergingMarketStressIndicator,
+)
+from src.core.macroeconomics.markets.risk_engine.indicators.breadth import (
+    SPYQQQDivergenceIndicator,
+    SectorRotationIndicator,
+    MarketConcentrationIndicator,
+    SmallCapWeaknessIndicator,
+)
+from src.core.macroeconomics.markets.risk_engine.indicators.tail_risk import (
+    CorrelationSpikeIndicator,
+    VolatilityRegimeIndicator,
+    DrawdownIndicator,
+    SkewIndicator,
+    TailEventIndicator,
+)
+
+__all__ = [
+    "BaseIndicator",
+    # Volatility
+    "VIXStressIndicator",
+    "VIXTermStructureIndicator",
+    "VVIXIndicator",
+    "RealizedVolatilityIndicator",
+    # Credit
+    "CreditSpreadWideningIndicator",
+    "IGvsHYDivergenceIndicator",
+    "CreditETFPressureIndicator",
+    "CorporateBondStressIndicator",
+    # Liquidity
+    "AmihudIlliquidityIndicator",
+    "BidAskSpreadIndicator",
+    "TreasuryLiquidityIndicator",
+    "MarketDepthIndicator",
+    # Macro/Rates
+    "YieldCurveIndicator",
+    "NFCIIndicator",
+    "RepoStressIndicator",
+    "MOVEIndexIndicator",
+    "RealRatesIndicator",
+    # FX/Funding
+    "DXYSpikeIndicator",
+    "JPYCarryUnwindIndicator",
+    "CrossCurrencyBasisIndicator",
+    "EmergingMarketStressIndicator",
+    # Breadth
+    "SPYQQQDivergenceIndicator",
+    "SectorRotationIndicator",
+    "MarketConcentrationIndicator",
+    "SmallCapWeaknessIndicator",
+    # Tail Risk
+    "CorrelationSpikeIndicator",
+    "VolatilityRegimeIndicator",
+    "DrawdownIndicator",
+    "SkewIndicator",
+    "TailEventIndicator",
+]

src/core/macroeconomics/markets/risk_engine/indicators/breadth.py

@@ -0,0 +1,449 @@
+"""
+Market breadth indicators for risk detection.
+
+Monitors advance/decline, participation, and market internals.
+"""
+
+from typing import Dict, Optional
+import pandas as pd
+import numpy as np
+
+from src.core.macroeconomics.markets.risk_engine.models import (
+    IndicatorSignal,
+    IndicatorCategory,
+    TimeSeriesData
+)
+from src.core.macroeconomics.markets.risk_engine.indicators.base import BaseIndicator
+from src.telegram_bot.logger import main_logger as logger
+
+
+class SPYQQQDivergenceIndicator(BaseIndicator):
+    """
+    SPY vs QQQ divergence detector.
+
+    Monitors if large-cap tech (QQQ) diverging from broad market (SPY).
+    Narrowing leadership = fragile rally.
+
+    Signal construction:
+    - Calculate SPY and QQQ 20-day returns
+    - Monitor performance spread
+    - Widening divergence = concentration risk
+
+    Required data:
+    - SPY: S&P 500 ETF
+    - QQQ: Nasdaq-100 ETF
+    """
+
+    def __init__(self):
+        super().__init__(
+            name="SPY-QQQ Divergence",
+            category=IndicatorCategory.BREADTH,
+            description="Large-cap tech vs broad market",
+            zscore_window=252
+        )
+
+    def compute(self, data: Dict[str, TimeSeriesData]) -> Optional[IndicatorSignal]:
+        """Compute SPY-QQQ divergence."""
+        required = ["SPY", "QQQ"]
+        if not self._validate_data(data, required):
+            return None
+
+        try:
+            spy = data["SPY"].data
+            qqq = data["QQQ"].data
+
+            # Align dates
+            combined = pd.DataFrame({
+                'spy': spy,
+                'qqq': qqq
+            }).dropna()
+
+            if len(combined) < self.zscore_window:
+                logger.warning(f"{self.name}: Insufficient data")
+                return None
+
+            # Calculate 20-day returns
+            spy_ret = (combined['spy'].iloc[-1] / combined['spy'].iloc[-21] - 1) * 100
+            qqq_ret = (combined['qqq'].iloc[-1] / combined['qqq'].iloc[-21] - 1) * 100
+
+            # Performance spread (QQQ - SPY)
+            # Positive = QQQ outperforming = concentration
+            spread = qqq_ret - spy_ret
+
+            # Calculate rolling spread for z-score
+            spy_rets = combined['spy'].pct_change(20).dropna() * 100
+            qqq_rets = combined['qqq'].pct_change(20).dropna() * 100
+            spread_series = qqq_rets - spy_rets
+
+            if len(spread_series) < self.zscore_window:
+                return None
+
+            # Z-score of spread
+            spread_zscore = self.calculate_zscore(spread_series)
+
+            # Stress logic:
+            # - QQQ massively outperforming = narrow leadership = stress
+            # - QQQ massively underperforming = tech weakness = stress
+            # Use absolute z-score
+            stress_zscore = abs(spread_zscore)
+
+            stress_level = self.normalize_to_level(stress_zscore)
+
+            return IndicatorSignal(
+                indicator_name=self.name,
+                category=self.category,
+                value=spread,
+                zscore=spread_zscore,
+                stress_level=stress_level,
+                description=(
+                    f"SPY 20d: {spy_ret:+.1f}%, "
+                    f"QQQ 20d: {qqq_ret:+.1f}%, "
+                    f"Spread: {spread:+.1f}% (z={spread_zscore:+.1f})"
+                ),
+                interpretation=self._interpret(stress_level, spread, spread_zscore)
+            )
+
+        except Exception as e:
+            logger.error(f"{self.name} calculation failed: {e}")
+            return None
+
+    def _interpret(
+        self,
+        stress_level: float,
+        spread: float,
+        zscore: float
+    ) -> str:
+        """Generate interpretation."""
+        if stress_level > 70:
+            if zscore > 2:
+                return f"🔴 NARROW: Extreme QQQ outperformance (narrow leadership, {spread:+.1f}%)"
+            else:
+                return f"🔴 ROTATION: Extreme QQQ underperformance (tech weakness, {spread:+.1f}%)"
+        elif stress_level > 55:
+            if zscore > 0:
+                return "⚠️ CONCENTRATION: QQQ leading (narrow breadth)"
+            else:
+                return "⚠️ DIVERGENCE: QQQ lagging (sector rotation)"
+        elif stress_level < 40:
+            return "✅ HEALTHY: SPY and QQQ moving together"
+        else:
+            return "⚡ WATCH: Moderate SPY-QQQ divergence"
+
+
+class SectorRotationIndicator(BaseIndicator):
+    """
+    Sector rotation stress indicator.
+
+    Monitor defensive vs cyclical sector performance.
+    Flight to defensives = risk-off signal.
+
+    Use sector ETFs as proxy:
+    - XLU (Utilities) = defensive
+    - XLF (Financials) = cyclical
+
+    Signal construction:
+    - Calculate XLU/XLF relative strength ratio
+    - Rising ratio = defensive leadership = risk-off
+
+    Required data:
+    - XLU: Utilities sector ETF (defensive)
+    - XLF: Financials sector ETF (cyclical)
+    """
+
+    def __init__(self):
+        super().__init__(
+            name="Sector Rotation",
+            category=IndicatorCategory.BREADTH,
+            description="Defensive vs cyclical sector strength",
+            zscore_window=252
+        )
+
+    def compute(self, data: Dict[str, TimeSeriesData]) -> Optional[IndicatorSignal]:
+        """Compute sector rotation signal."""
+        # Check for sector ETFs (may not be in current data sources)
+        has_xlu = "XLU" in data and len(data["XLU"].data) > 0
+        has_xlf = "XLF" in data and len(data["XLF"].data) > 0
+
+        if not (has_xlu and has_xlf):
+            logger.warning(f"{self.name}: Sector ETF data not available")
+            return None
+
+        try:
+            xlu = data["XLU"].data
+            xlf = data["XLF"].data
+
+            # Align dates
+            combined = pd.DataFrame({
+                'xlu': xlu,
+                'xlf': xlf
+            }).dropna()
+
+            if len(combined) < self.zscore_window:
+                logger.warning(f"{self.name}: Insufficient data")
+                return None
+
+            # Calculate relative strength ratio (XLU/XLF)
+            # Rising = defensives outperforming = risk-off
+            ratio = combined['xlu'] / combined['xlf']
+
+            # Z-score of ratio
+            ratio_zscore = self.calculate_zscore(ratio)
+
+            # Calculate ratio acceleration
+            ratio_accel = self.calculate_acceleration(ratio, periods=10)
+
+            # Positive z-score = defensive leadership = stress
+            stress_level = self.normalize_to_level(ratio_zscore)
+
+            current_ratio = float(ratio.iloc[-1])
+            xlu_ret = (combined['xlu'].iloc[-1] / combined['xlu'].iloc[-21] - 1) * 100
+            xlf_ret = (combined['xlf'].iloc[-1] / combined['xlf'].iloc[-21] - 1) * 100
+
+            return IndicatorSignal(
+                indicator_name=self.name,
+                category=self.category,
+                value=current_ratio,
+                zscore=ratio_zscore,
+                stress_level=stress_level,
+                description=(
+                    f"XLU/XLF: {current_ratio:.3f} "
+                    f"(z={ratio_zscore:+.1f}), "
+                    f"XLU: {xlu_ret:+.1f}%, XLF: {xlf_ret:+.1f}%"
+                ),
+                interpretation=self._interpret(stress_level, ratio_zscore, xlu_ret, xlf_ret)
+            )
+
+        except Exception as e:
+            logger.error(f"{self.name} calculation failed: {e}")
+            return None
+
+    def _interpret(
+        self,
+        stress_level: float,
+        zscore: float,
+        xlu_ret: float,
+        xlf_ret: float
+    ) -> str:
+        """Generate interpretation."""
+        if stress_level > 70:
+            return "🔴 DEFENSIVE: Flight to defensive sectors (risk-off)"
+        elif stress_level > 55:
+            return "⚠️ ROTATION: Defensives outperforming (caution)"
+        elif stress_level < 40:
+            return "✅ CYCLICAL: Cyclicals leading (risk-on)"
+        else:
+            return "⚡ WATCH: Mixed sector performance"
+
+
+class MarketConcentrationIndicator(BaseIndicator):
+    """
+    Market concentration indicator.
+
+    Compare SPY (broad S&P 500) vs largest components.
+    Use QQQ as mega-cap proxy.
+
+    Extreme concentration = fragile market structure.
+
+    Signal construction:
+    - Calculate rolling correlation between SPY and QQQ
+    - Very high correlation = concentrated market
+    - Also monitor relative volatility
+
+    Required data:
+    - SPY: S&P 500
+    - QQQ: Nasdaq-100 (mega-cap proxy)
+    """
+
+    def __init__(self):
+        super().__init__(
+            name="Market Concentration",
+            category=IndicatorCategory.BREADTH,
+            description="Mega-cap concentration risk",
+            zscore_window=252
+        )
+
+    def compute(self, data: Dict[str, TimeSeriesData]) -> Optional[IndicatorSignal]:
+        """Compute market concentration."""
+        required = ["SPY", "QQQ"]
+        if not self._validate_data(data, required):
+            return None
+
+        try:
+            spy = data["SPY"].data
+            qqq = data["QQQ"].data
+
+            # Align dates
+            combined = pd.DataFrame({
+                'spy': spy,
+                'qqq': qqq
+            }).dropna()
+
+            if len(combined) < self.zscore_window + 60:
+                logger.warning(f"{self.name}: Insufficient data")
+                return None
+
+            # Calculate returns
+            spy_ret = combined['spy'].pct_change().dropna()
+            qqq_ret = combined['qqq'].pct_change().dropna()
+
+            # Calculate rolling 60-day correlation
+            correlation = spy_ret.rolling(60).corr(qqq_ret).dropna()
+
+            if len(correlation) < self.zscore_window:
+                return None
+
+            # Z-score of correlation
+            # High correlation = concentrated market = stress
+            corr_zscore = self.calculate_zscore(correlation)
+
+            stress_level = self.normalize_to_level(corr_zscore)
+
+            current_corr = float(correlation.iloc[-1])
+
+            # Also calculate relative volatility
+            spy_vol = spy_ret.rolling(20).std().iloc[-1] * np.sqrt(252) * 100
+            qqq_vol = qqq_ret.rolling(20).std().iloc[-1] * np.sqrt(252) * 100
+
+            return IndicatorSignal(
+                indicator_name=self.name,
+                category=self.category,
+                value=current_corr,
+                zscore=corr_zscore,
+                stress_level=stress_level,
+                description=(
+                    f"SPY-QQQ corr: {current_corr:.2f} "
+                    f"(z={corr_zscore:+.1f}), "
+                    f"SPY vol: {spy_vol:.1f}%, QQQ vol: {qqq_vol:.1f}%"
+                ),
+                interpretation=self._interpret(stress_level, current_corr, corr_zscore)
+            )
+
+        except Exception as e:
+            logger.error(f"{self.name} calculation failed: {e}")
+            return None
+
+    def _interpret(
+        self,
+        stress_level: float,
+        corr: float,
+        zscore: float
+    ) -> str:
+        """Generate interpretation."""
+        if stress_level > 70:
+            return f"🔴 CONCENTRATED: Extreme correlation (market = mega-caps, {corr:.2f})"
+        elif stress_level > 55:
+            return "⚠️ NARROWING: Market concentration increasing"
+        elif stress_level < 40:
+            return "✅ BROAD: Healthy market breadth"
+        else:
+            return "⚡ WATCH: Moderate concentration"
+
+
+class SmallCapWeaknessIndicator(BaseIndicator):
+    """
+    Small-cap weakness indicator.
+
+    Monitor IWM (Russell 2000) vs SPY.
+    Small-caps underperforming = risk appetite declining.
+
+    Signal construction:
+    - Calculate IWM/SPY relative strength
+    - Falling ratio = small-cap weakness = stress
+
+    Required data:
+    - IWM: Russell 2000 small-cap ETF
+    - SPY: S&P 500
+    """
+
+    def __init__(self):
+        super().__init__(
+            name="Small-Cap Weakness",
+            category=IndicatorCategory.BREADTH,
+            description="Small-cap underperformance",
+            zscore_window=252
+        )
+
+    def compute(self, data: Dict[str, TimeSeriesData]) -> Optional[IndicatorSignal]:
+        """Compute small-cap weakness."""
+        # Check for IWM
+        has_iwm = "IWM" in data and len(data["IWM"].data) > 0
+        has_spy = "SPY" in data and len(data["SPY"].data) > 0
+
+        if not (has_iwm and has_spy):
+            logger.warning(f"{self.name}: IWM or SPY data not available")
+            return None
+
+        try:
+            iwm = data["IWM"].data
+            spy = data["SPY"].data
+
+            # Align dates
+            combined = pd.DataFrame({
+                'iwm': iwm,
+                'spy': spy
+            }).dropna()
+
+            if len(combined) < self.zscore_window:
+                logger.warning(f"{self.name}: Insufficient data")
+                return None
+
+            # Calculate relative strength ratio (IWM/SPY)
+            # Falling = small-caps lagging = risk-off
+            ratio = combined['iwm'] / combined['spy']
+
+            # Calculate momentum (20-day % change of ratio)
+            ratio_momentum = (ratio.iloc[-1] / ratio.iloc[-21] - 1) * 100
+
+            # Z-score of momentum
+            momentum_series = ratio.pct_change(20).dropna() * 100
+            if len(momentum_series) < self.zscore_window:
+                return None
+
+            momentum_zscore = self.calculate_zscore(momentum_series)
+
+            # Negative momentum = small-caps lagging = stress
+            # Flip sign
+            stress_zscore = -momentum_zscore
+
+            stress_level = self.normalize_to_level(stress_zscore)
+
+            current_ratio = float(ratio.iloc[-1])
+            iwm_ret = (combined['iwm'].iloc[-1] / combined['iwm'].iloc[-21] - 1) * 100
+            spy_ret = (combined['spy'].iloc[-1] / combined['spy'].iloc[-21] - 1) * 100
+
+            return IndicatorSignal(
+                indicator_name=self.name,
+                category=self.category,
+                value=current_ratio,
+                zscore=stress_zscore,
+                stress_level=stress_level,
+                description=(
+                    f"IWM/SPY: {current_ratio:.3f} "
+                    f"(momentum: {ratio_momentum:+.1f}%), "
+                    f"IWM: {iwm_ret:+.1f}%, SPY: {spy_ret:+.1f}%"
+                ),
+                interpretation=self._interpret(stress_level, ratio_momentum, iwm_ret, spy_ret)
+            )
+
+        except Exception as e:
+            logger.error(f"{self.name} calculation failed: {e}")
+            return None
+
+    def _interpret(
+        self,
+        stress_level: float,
+        momentum: float,
+        iwm_ret: float,
+        spy_ret: float
+    ) -> str:
+        """Generate interpretation."""
+        underperformance = spy_ret - iwm_ret
+
+        if stress_level > 70:
+            return f"🔴 WEAK: Small-caps severely lagging ({underperformance:+.1f}% vs SPY)"
+        elif stress_level > 55:
+            return "⚠️ LAGGING: Small-caps underperforming (risk appetite declining)"
+        elif stress_level < 40:
+            return "✅ STRONG: Small-caps leading (healthy risk appetite)"
+        else:
+            return "⚡ WATCH: Small-caps moderately weak"

src/core/macroeconomics/markets/risk_engine/indicators/credit.py

@@ -0,0 +1,414 @@
+"""
+Credit market indicators for risk detection.
+
+Monitors corporate bond spreads, credit stress, and default risk.
+"""
+
+from typing import Dict, Optional
+import pandas as pd
+import numpy as np
+
+from src.core.macroeconomics.markets.risk_engine.models import (
+    IndicatorSignal,
+    IndicatorCategory,
+    TimeSeriesData
+)
+from src.core.macroeconomics.markets.risk_engine.indicators.base import BaseIndicator
+from src.telegram_bot.logger import main_logger as logger
+
+
+class CreditSpreadWideningIndicator(BaseIndicator):
+    """
+    Detect credit spread widening (HY vs IG).
+
+    Credit spreads widen during stress as investors demand higher yields
+    for corporate bonds vs risk-free Treasuries.
+
+    Signal construction:
+    - Monitor High Yield spread (BAMLH0A0HYM2)
+    - Calculate z-score vs historical levels
+    - Rising spreads = credit stress
+
+    Required data:
+    - BAMLH0A0HYM2: ICE BofA US High Yield Option-Adjusted Spread
+    """
+
+    def __init__(self):
+        super().__init__(
+            name="HY Credit Spread",
+            category=IndicatorCategory.CREDIT,
+            description="High yield credit spread widening",
+            zscore_window=252
+        )
+
+    def compute(self, data: Dict[str, TimeSeriesData]) -> Optional[IndicatorSignal]:
+        """Compute credit spread signal."""
+        if not self._validate_data(data, ["BAMLH0A0HYM2"]):
+            return None
+
+        try:
+            spread_data = data["BAMLH0A0HYM2"].data
+
+            if len(spread_data) < self.zscore_window:
+                logger.warning(
+                    f"{self.name}: Insufficient data "
+                    f"({len(spread_data)} < {self.zscore_window})"
+                )
+                return None
+
+            # Calculate z-score of spread level
+            spread_zscore = self.calculate_zscore(spread_data)
+
+            # Calculate spread acceleration (widening speed)
+            spread_accel = self.calculate_acceleration(spread_data, periods=5)
+
+            # Combined signal: level + momentum
+            # If spreads are wide AND widening fast = severe stress
+            if spread_zscore > 0 and spread_accel > 0:
+                # Amplify signal when both level and momentum elevated
+                amplified_zscore = spread_zscore * (1 + min(spread_accel / 50, 0.5))
+            else:
+                amplified_zscore = spread_zscore
+
+            stress_level = self.normalize_to_level(amplified_zscore)
+
+            current_spread = float(spread_data.iloc[-1])
+
+            return IndicatorSignal(
+                indicator_name=self.name,
+                category=self.category,
+                value=current_spread,
+                zscore=spread_zscore,
+                stress_level=stress_level,
+                description=(
+                    f"HY spread: {current_spread:.0f} bps "
+                    f"(z={spread_zscore:+.1f}), "
+                    f"acceleration: {spread_accel:+.1f} bps"
+                ),
+                interpretation=self._interpret(stress_level, spread_zscore, spread_accel)
+            )
+
+        except Exception as e:
+            logger.error(f"{self.name} calculation failed: {e}")
+            return None
+
+    def _interpret(
+        self,
+        stress_level: float,
+        zscore: float,
+        acceleration: float
+    ) -> str:
+        """Generate interpretation."""
+        if stress_level > 75:
+            if acceleration > 0:
+                return "🔴 SEVERE: Credit spreads blowing out (accelerating)"
+            else:
+                return "🔴 ELEVATED: Wide spreads (but stabilizing)"
+        elif stress_level > 60:
+            return "⚠️ WARNING: Credit spreads widening"
+        elif stress_level < 40:
+            return "✅ HEALTHY: Tight credit spreads"
+        else:
+            return "⚡ WATCH: Credit spreads moderately wide"
+
+
+class IGvsHYDivergenceIndicator(BaseIndicator):
+    """
+    Detect Investment Grade vs High Yield divergence.
+
+    During stress, HY spreads widen much faster than IG spreads.
+    Divergence indicates flight-to-quality and credit tiering.
+
+    Signal construction:
+    - Calculate HY/IG spread ratio
+    - Z-score of ratio
+    - Rising ratio = HY stress relative to IG
+
+    Required data:
+    - BAMLH0A0HYM2: High Yield spread
+    - BAMLC0A0CM: Investment Grade spread
+    """
+
+    def __init__(self):
+        super().__init__(
+            name="IG vs HY Divergence",
+            category=IndicatorCategory.CREDIT,
+            description="High yield underperforming investment grade",
+            zscore_window=252
+        )
+
+    def compute(self, data: Dict[str, TimeSeriesData]) -> Optional[IndicatorSignal]:
+        """Compute IG/HY divergence signal."""
+        required = ["BAMLH0A0HYM2", "BAMLC0A0CM"]
+        if not self._validate_data(data, required):
+            return None
+
+        try:
+            hy_spread = data["BAMLH0A0HYM2"].data
+            ig_spread = data["BAMLC0A0CM"].data
+
+            # Align dates
+            combined = pd.DataFrame({
+                'hy': hy_spread,
+                'ig': ig_spread
+            }).dropna()
+
+            if len(combined) < self.zscore_window:
+                logger.warning(f"{self.name}: Insufficient data")
+                return None
+
+            # Calculate HY/IG ratio
+            ratio = combined['hy'] / combined['ig']
+
+            # Z-score of ratio
+            ratio_zscore = self.calculate_zscore(ratio)
+
+            # Calculate ratio acceleration
+            ratio_accel = self.calculate_acceleration(ratio, periods=5)
+
+            stress_level = self.normalize_to_level(ratio_zscore)
+
+            current_hy = float(hy_spread.iloc[-1])
+            current_ig = float(ig_spread.iloc[-1])
+            current_ratio = current_hy / current_ig
+
+            return IndicatorSignal(
+                indicator_name=self.name,
+                category=self.category,
+                value=current_ratio,
+                zscore=ratio_zscore,
+                stress_level=stress_level,
+                description=(
+                    f"HY: {current_hy:.0f} bps, IG: {current_ig:.0f} bps, "
+                    f"Ratio: {current_ratio:.2f} (z={ratio_zscore:+.1f})"
+                ),
+                interpretation=self._interpret(stress_level, ratio_zscore, ratio_accel)
+            )
+
+        except Exception as e:
+            logger.error(f"{self.name} calculation failed: {e}")
+            return None
+
+    def _interpret(
+        self,
+        stress_level: float,
+        zscore: float,
+        acceleration: float
+    ) -> str:
+        """Generate interpretation."""
+        if stress_level > 75:
+            return "🔴 SEVERE: HY severely underperforming IG (flight-to-quality)"
+        elif stress_level > 60:
+            return "⚠️ WARNING: HY/IG divergence increasing"
+        elif stress_level < 40:
+            return "✅ STABLE: HY and IG spreads moving together"
+        else:
+            return "⚡ WATCH: HY starting to underperform IG"
+
+
+class CreditETFPressureIndicator(BaseIndicator):
+    """
+    Detect credit ETF price pressure.
+
+    Monitor HYG (High Yield ETF) and LQD (Investment Grade ETF) for selling.
+    Falling credit ETF prices indicate institutional deleveraging.
+
+    Signal construction:
+    - Calculate HYG and LQD returns
+    - Z-score of returns (negative returns = stress)
+    - Compare vs SPY to detect credit-specific weakness
+
+    Required data:
+    - HYG: iShares High Yield Corporate Bond ETF
+    - LQD: iShares Investment Grade Corporate Bond ETF
+    - SPY: S&P 500 for comparison
+    """
+
+    def __init__(self):
+        super().__init__(
+            name="Credit ETF Pressure",
+            category=IndicatorCategory.CREDIT,
+            description="Credit ETF selling pressure",
+            zscore_window=252
+        )
+
+    def compute(self, data: Dict[str, TimeSeriesData]) -> Optional[IndicatorSignal]:
+        """Compute credit ETF pressure signal."""
+        required = ["HYG", "LQD", "SPY"]
+        if not self._validate_data(data, required):
+            return None
+
+        try:
+            hyg_data = data["HYG"].data
+            lqd_data = data["LQD"].data
+            spy_data = data["SPY"].data
+
+            # Align dates
+            combined = pd.DataFrame({
+                'hyg': hyg_data,
+                'lqd': lqd_data,
+                'spy': spy_data
+            }).dropna()
+
+            if len(combined) < self.zscore_window:
+                logger.warning(f"{self.name}: Insufficient data")
+                return None
+
+            # Calculate 20-day returns
+            hyg_ret = (combined['hyg'].iloc[-1] / combined['hyg'].iloc[-21] - 1) * 100
+            lqd_ret = (combined['lqd'].iloc[-1] / combined['lqd'].iloc[-21] - 1) * 100
+            spy_ret = (combined['spy'].iloc[-1] / combined['spy'].iloc[-21] - 1) * 100
+
+            # Calculate rolling returns for z-score
+            hyg_rets = combined['hyg'].pct_change(20).dropna() * 100
+            lqd_rets = combined['lqd'].pct_change(20).dropna() * 100
+
+            # Combined credit return
+            credit_ret = (hyg_ret + lqd_ret) / 2
+            credit_rets = (hyg_rets + lqd_rets) / 2
+
+            # Z-score of credit returns (negative z = underperformance)
+            credit_zscore = self.calculate_zscore(credit_rets)
+
+            # Flip sign: negative returns = positive stress
+            stress_zscore = -credit_zscore
+
+            # Amplify if credit underperforming equity
+            if credit_ret < spy_ret - 2.0:
+                # Credit lagging equity by >2% = amplify signal
+                underperformance = spy_ret - credit_ret
+                stress_zscore *= (1 + min(underperformance / 10, 0.5))
+
+            stress_level = self.normalize_to_level(stress_zscore)
+
+            return IndicatorSignal(
+                indicator_name=self.name,
+                category=self.category,
+                value=credit_ret,
+                zscore=stress_zscore,
+                stress_level=stress_level,
+                description=(
+                    f"HYG 20d: {hyg_ret:+.1f}%, "
+                    f"LQD 20d: {lqd_ret:+.1f}%, "
+                    f"SPY 20d: {spy_ret:+.1f}%"
+                ),
+                interpretation=self._interpret(stress_level, credit_ret, spy_ret)
+            )
+
+        except Exception as e:
+            logger.error(f"{self.name} calculation failed: {e}")
+            return None
+
+    def _interpret(
+        self,
+        stress_level: float,
+        credit_ret: float,
+        spy_ret: float
+    ) -> str:
+        """Generate interpretation."""
+        underperf = spy_ret - credit_ret
+
+        if stress_level > 75:
+            return f"🔴 SEVERE: Credit ETFs selling hard ({credit_ret:+.1f}%, {underperf:+.1f}% vs SPY)"
+        elif stress_level > 60:
+            return f"⚠️ WARNING: Credit ETF weakness ({credit_ret:+.1f}%)"
+        elif stress_level < 40:
+            return "✅ STABLE: Credit ETFs performing well"
+        else:
+            return "⚡ WATCH: Credit ETFs under mild pressure"
+
+
+class CorporateBondStressIndicator(BaseIndicator):
+    """
+    Composite corporate bond stress indicator.
+
+    Combines spread levels, spread velocity, and price action
+    into single corporate credit stress measure.
+
+    Signal construction:
+    - Average of: HY spread z-score, IG spread z-score, HYG return z-score
+    - Equal-weighted composite
+
+    Required data:
+    - BAMLH0A0HYM2: HY spread
+    - BAMLC0A0CM: IG spread
+    - HYG: High Yield ETF
+    """
+
+    def __init__(self):
+        super().__init__(
+            name="Corporate Bond Stress",
+            category=IndicatorCategory.CREDIT,
+            description="Composite corporate credit stress",
+            zscore_window=252
+        )
+
+    def compute(self, data: Dict[str, TimeSeriesData]) -> Optional[IndicatorSignal]:
+        """Compute composite credit stress."""
+        required = ["BAMLH0A0HYM2", "BAMLC0A0CM", "HYG"]
+        if not self._validate_data(data, required):
+            return None
+
+        try:
+            hy_spread = data["BAMLH0A0HYM2"].data
+            ig_spread = data["BAMLC0A0CM"].data
+            hyg_price = data["HYG"].data
+
+            # Align dates
+            combined = pd.DataFrame({
+                'hy': hy_spread,
+                'ig': ig_spread,
+                'hyg': hyg_price
+            }).dropna()
+
+            if len(combined) < self.zscore_window:
+                logger.warning(f"{self.name}: Insufficient data")
+                return None
+
+            # Component 1: HY spread z-score
+            hy_zscore = self.calculate_zscore(combined['hy'])
+
+            # Component 2: IG spread z-score
+            ig_zscore = self.calculate_zscore(combined['ig'])
+
+            # Component 3: HYG return z-score (negative = stress)
+            hyg_ret = combined['hyg'].pct_change(20).dropna() * 100
+            hyg_ret_zscore = -self.calculate_zscore(hyg_ret)  # Flip: negative ret = positive stress
+
+            # Composite: equal-weighted average
+            composite_zscore = (hy_zscore + ig_zscore + hyg_ret_zscore) / 3
+
+            stress_level = self.normalize_to_level(composite_zscore)
+
+            current_hy = float(hy_spread.iloc[-1])
+            current_ig = float(ig_spread.iloc[-1])
+            current_hyg_ret = (combined['hyg'].iloc[-1] / combined['hyg'].iloc[-21] - 1) * 100
+
+            return IndicatorSignal(
+                indicator_name=self.name,
+                category=self.category,
+                value=composite_zscore,
+                zscore=composite_zscore,
+                stress_level=stress_level,
+                description=(
+                    f"HY: {current_hy:.0f}bps (z={hy_zscore:+.1f}), "
+                    f"IG: {current_ig:.0f}bps (z={ig_zscore:+.1f}), "
+                    f"HYG: {current_hyg_ret:+.1f}%"
+                ),
+                interpretation=self._interpret(stress_level, composite_zscore)
+            )
+
+        except Exception as e:
+            logger.error(f"{self.name} calculation failed: {e}")
+            return None
+
+    def _interpret(self, stress_level: float, zscore: float) -> str:
+        """Generate interpretation."""
+        if stress_level > 75:
+            return "🔴 SEVERE: Corporate credit under severe stress"
+        elif stress_level > 60:
+            return "⚠️ WARNING: Elevated corporate credit stress"
+        elif stress_level < 40:
+            return "✅ HEALTHY: Corporate credit markets healthy"
+        else:
+            return "⚡ WATCH: Corporate credit showing early stress signals"

src/core/macroeconomics/markets/risk_engine/indicators/fx_funding.py

@@ -0,0 +1,415 @@
+"""
+FX and funding indicators for market risk detection.
+
+Monitors dollar strength, carry trade unwinding, and cross-currency funding stress.
+"""
+
+from typing import Dict, Optional
+import pandas as pd
+import numpy as np
+
+from src.core.macroeconomics.markets.risk_engine.models import (
+    IndicatorSignal,
+    IndicatorCategory,
+    TimeSeriesData
+)
+from src.core.macroeconomics.markets.risk_engine.indicators.base import BaseIndicator
+from src.telegram_bot.logger import main_logger as logger
+
+
+class DXYSpikeIndicator(BaseIndicator):
+    """
+    Dollar Index (DXY) spike detector.
+
+    Rapid USD strengthening = global funding stress.
+    Flight to safety, EM stress, commodity pressure.
+
+    Signal construction:
+    - Calculate DXY level z-score
+    - Calculate DXY velocity (rate of rise)
+    - Fast rising DXY = stress
+
+    Required data:
+    - DXY: US Dollar Index (from Yahoo or FRED)
+    """
+
+    def __init__(self):
+        super().__init__(
+            name="DXY Spike",
+            category=IndicatorCategory.FX_FUNDING,
+            description="Dollar strength (funding stress)",
+            zscore_window=252
+        )
+
+    def compute(self, data: Dict[str, TimeSeriesData]) -> Optional[IndicatorSignal]:
+        """Compute DXY spike signal."""
+        # Try to get DXY from either source
+        if "DXY" in data and len(data["DXY"].data) > 0:
+            dxy_data = data["DXY"].data
+        elif "DTWEXBGS" in data and len(data["DTWEXBGS"].data) > 0:
+            # Trade-weighted dollar index from FRED
+            dxy_data = data["DTWEXBGS"].data
+        else:
+            logger.warning(f"{self.name}: No DXY data available")
+            return None
+
+        try:
+            if len(dxy_data) < self.zscore_window:
+                logger.warning(f"{self.name}: Insufficient data")
+                return None
+
+            # Calculate z-score of DXY level
+            dxy_zscore = self.calculate_zscore(dxy_data)
+
+            # Calculate DXY velocity (20-day % change)
+            dxy_velocity = (dxy_data.iloc[-1] / dxy_data.iloc[-21] - 1) * 100
+
+            # Calculate acceleration
+            dxy_accel = self.calculate_acceleration(dxy_data, periods=5)
+
+            # Stress logic: Rising DXY = stress
+            # Combine level and momentum
+            if dxy_zscore > 0 and dxy_velocity > 2:
+                # Dollar rising and elevated = amplify signal
+                amplified_zscore = dxy_zscore * (1 + min(dxy_velocity / 10, 0.5))
+            elif dxy_velocity > 5:
+                # Rapid rise even from low levels = stress
+                amplified_zscore = dxy_velocity / 5
+            else:
+                amplified_zscore = dxy_zscore
+
+            stress_level = self.normalize_to_level(amplified_zscore)
+
+            current_dxy = float(dxy_data.iloc[-1])
+
+            return IndicatorSignal(
+                indicator_name=self.name,
+                category=self.category,
+                value=current_dxy,
+                zscore=dxy_zscore,
+                stress_level=stress_level,
+                description=(
+                    f"DXY: {current_dxy:.2f} "
+                    f"(z={dxy_zscore:+.1f}), "
+                    f"20d: {dxy_velocity:+.1f}%"
+                ),
+                interpretation=self._interpret(stress_level, dxy_zscore, dxy_velocity)
+            )
+
+        except Exception as e:
+            logger.error(f"{self.name} calculation failed: {e}")
+            return None
+
+    def _interpret(
+        self,
+        stress_level: float,
+        zscore: float,
+        velocity: float
+    ) -> str:
+        """Generate interpretation."""
+        if stress_level > 75:
+            return f"🔴 SEVERE: Dollar surging (global funding stress)"
+        elif stress_level > 60:
+            return "⚠️ RISING: Dollar strengthening (EM pressure)"
+        elif stress_level < 40:
+            if zscore < -1:
+                return "✅ WEAK: Dollar weakening (risk-on supportive)"
+            else:
+                return "✅ STABLE: Dollar at normal levels"
+        else:
+            return "⚡ WATCH: Dollar moderately strong"
+
+
+class JPYCarryUnwindIndicator(BaseIndicator):
+    """
+    JPY carry trade unwind detector.
+
+    USD/JPY falling = yen strengthening = carry unwind = risk-off.
+    Rapid yen strength often coincides with equity selloffs.
+
+    March 2020: USD/JPY fell 10% in days during panic.
+
+    Signal construction:
+    - Calculate USD/JPY velocity
+    - Detect rapid JPY strengthening (USD/JPY falling)
+    - Fast unwind = stress
+
+    Required data:
+    - DEXJPUS: Yen/Dollar exchange rate (need to invert)
+    """
+
+    def __init__(self):
+        super().__init__(
+            name="JPY Carry Unwind",
+            category=IndicatorCategory.FX_FUNDING,
+            description="Yen strength (carry trade unwinding)",
+            zscore_window=252
+        )
+
+    def compute(self, data: Dict[str, TimeSeriesData]) -> Optional[IndicatorSignal]:
+        """Compute JPY carry unwind signal."""
+        if not self._validate_data(data, ["DEXJPUS"]):
+            return None
+
+        try:
+            jpyusd = data["DEXJPUS"].data  # JPY per USD
+
+            if len(jpyusd) < self.zscore_window:
+                logger.warning(f"{self.name}: Insufficient data")
+                return None
+
+            # Convert to USD/JPY for intuitive interpretation
+            usdjpy = 1 / jpyusd
+
+            # Calculate 20-day velocity of USD/JPY
+            # Negative = JPY strengthening = carry unwind = stress
+            usdjpy_velocity = (usdjpy.iloc[-1] / usdjpy.iloc[-21] - 1) * 100
+
+            # Calculate z-score of velocity
+            velocity_series = usdjpy.pct_change(20).dropna() * 100
+            if len(velocity_series) < self.zscore_window:
+                logger.warning(f"{self.name}: Insufficient velocity data")
+                return None
+
+            velocity_zscore = self.calculate_zscore(velocity_series)
+
+            # Stress logic: Negative velocity (JPY strength) = stress
+            # Flip sign so that JPY strength = positive stress signal
+            stress_zscore = -velocity_zscore
+
+            # Amplify if very rapid unwinding
+            if usdjpy_velocity < -3:  # >3% JPY strength
+                stress_zscore *= (1 + abs(usdjpy_velocity) / 10)
+
+            stress_level = self.normalize_to_level(stress_zscore)
+
+            current_usdjpy = float(usdjpy.iloc[-1])
+
+            return IndicatorSignal(
+                indicator_name=self.name,
+                category=self.category,
+                value=current_usdjpy,
+                zscore=stress_zscore,
+                stress_level=stress_level,
+                description=(
+                    f"USD/JPY: {current_usdjpy:.2f} "
+                    f"(20d: {usdjpy_velocity:+.1f}%), "
+                    f"stress_z: {stress_zscore:+.1f}"
+                ),
+                interpretation=self._interpret(stress_level, usdjpy_velocity, stress_zscore)
+            )
+
+        except Exception as e:
+            logger.error(f"{self.name} calculation failed: {e}")
+            return None
+
+    def _interpret(
+        self,
+        stress_level: float,
+        velocity: float,
+        zscore: float
+    ) -> str:
+        """Generate interpretation."""
+        if stress_level > 75:
+            return f"🔴 SEVERE: JPY surging (carry unwind panic, USD/JPY {velocity:+.1f}%)"
+        elif stress_level > 60:
+            return "⚠️ UNWINDING: JPY strengthening (carry trades closing)"
+        elif stress_level < 40:
+            if velocity > 2:
+                return "✅ CARRY ON: JPY weakening (carry trades profitable)"
+            else:
+                return "✅ STABLE: JPY at normal levels"
+        else:
+            return "⚡ WATCH: JPY moderately strong"
+
+
+class CrossCurrencyBasisIndicator(BaseIndicator):
+    """
+    Cross-currency basis indicator.
+
+    True basis swaps not available in free data.
+    Use EUR/USD volatility and DXY stress as proxy.
+
+    Widening basis = dollar funding shortage.
+
+    Signal construction:
+    - Calculate EUR/USD realized volatility
+    - High FX vol = funding stress proxy
+
+    Required data:
+    - DEXUSEU: Euro/Dollar exchange rate
+    """
+
+    def __init__(self):
+        super().__init__(
+            name="FX Funding Stress",
+            category=IndicatorCategory.FX_FUNDING,
+            description="Cross-currency funding stress (EUR/USD vol)",
+            zscore_window=252
+        )
+
+    def compute(self, data: Dict[str, TimeSeriesData]) -> Optional[IndicatorSignal]:
+        """Compute FX funding stress signal."""
+        if not self._validate_data(data, ["DEXUSEU"]):
+            return None
+
+        try:
+            eurusd = data["DEXUSEU"].data
+
+            if len(eurusd) < self.zscore_window + 20:
+                logger.warning(f"{self.name}: Insufficient data")
+                return None
+
+            # Calculate log returns
+            returns = np.log(eurusd / eurusd.shift(1)).dropna()
+
+            # Calculate 20-day realized volatility (annualized)
+            realized_vol = returns.rolling(20).std() * np.sqrt(252) * 100
+            realized_vol = realized_vol.dropna()
+
+            if len(realized_vol) < self.zscore_window:
+                return None
+
+            # Z-score of FX volatility
+            vol_zscore = self.calculate_zscore(realized_vol)
+
+            # High FX vol = funding stress
+            stress_level = self.normalize_to_level(vol_zscore)
+
+            current_vol = float(realized_vol.iloc[-1])
+
+            return IndicatorSignal(
+                indicator_name=self.name,
+                category=self.category,
+                value=current_vol,
+                zscore=vol_zscore,
+                stress_level=stress_level,
+                description=(
+                    f"EUR/USD vol: {current_vol:.1f}% "
+                    f"(z={vol_zscore:+.1f})"
+                ),
+                interpretation=self._interpret(stress_level, vol_zscore)
+            )
+
+        except Exception as e:
+            logger.error(f"{self.name} calculation failed: {e}")
+            return None
+
+    def _interpret(self, stress_level: float, zscore: float) -> str:
+        """Generate interpretation."""
+        if stress_level > 75:
+            return "🔴 SEVERE: Extreme FX volatility (funding crisis)"
+        elif stress_level > 60:
+            return "⚠️ ELEVATED: FX volatility rising (funding stress)"
+        elif stress_level < 40:
+            return "✅ CALM: Low FX volatility (stable funding)"
+        else:
+            return "⚡ WATCH: FX volatility moderately elevated"
+
+
+class EmergingMarketStressIndicator(BaseIndicator):
+    """
+    Emerging market stress indicator.
+
+    Monitor gold (safe haven) vs dollar strength.
+    Gold rising + Dollar rising = extreme stress/uncertainty.
+
+    Signal construction:
+    - Calculate correlation between GLD and DXY
+    - Typically negative (dollar up, gold down)
+    - Positive correlation = crisis mode
+
+    Required data:
+    - GLD: Gold ETF
+    - DXY: Dollar Index
+    """
+
+    def __init__(self):
+        super().__init__(
+            name="EM Stress (Gold/Dollar)",
+            category=IndicatorCategory.FX_FUNDING,
+            description="Emerging market stress via gold-dollar",
+            zscore_window=252
+        )
+
+    def compute(self, data: Dict[str, TimeSeriesData]) -> Optional[IndicatorSignal]:
+        """Compute EM stress signal."""
+        # Try to get both GLD and DXY
+        has_gld = "GLD" in data and len(data["GLD"].data) > 0
+        has_dxy = "DXY" in data and len(data["DXY"].data) > 0
+
+        if not (has_gld and has_dxy):
+            logger.warning(f"{self.name}: Missing GLD or DXY data")
+            return None
+
+        try:
+            gld = data["GLD"].data
+            dxy = data["DXY"].data
+
+            # Align dates
+            combined = pd.DataFrame({
+                'gld': gld,
+                'dxy': dxy
+            }).dropna()
+
+            if len(combined) < self.zscore_window:
+                logger.warning(f"{self.name}: Insufficient data")
+                return None
+
+            # Calculate returns
+            gld_ret = combined['gld'].pct_change(20).dropna() * 100
+            dxy_ret = combined['dxy'].pct_change(20).dropna() * 100
+
+            # Calculate rolling 60-day correlation
+            correlation = gld_ret.rolling(60).corr(dxy_ret).dropna()
+
+            if len(correlation) < self.zscore_window:
+                return None
+
+            # Z-score of correlation
+            # Positive correlation (both up) = crisis = stress
+            corr_zscore = self.calculate_zscore(correlation)
+
+            stress_level = self.normalize_to_level(corr_zscore)
+
+            current_corr = float(correlation.iloc[-1])
+            current_gld_ret = float(gld_ret.iloc[-1])
+            current_dxy_ret = float(dxy_ret.iloc[-1])
+
+            return IndicatorSignal(
+                indicator_name=self.name,
+                category=self.category,
+                value=current_corr,
+                zscore=corr_zscore,
+                stress_level=stress_level,
+                description=(
+                    f"GLD-DXY corr: {current_corr:+.2f} "
+                    f"(z={corr_zscore:+.1f}), "
+                    f"GLD: {current_gld_ret:+.1f}%, DXY: {current_dxy_ret:+.1f}%"
+                ),
+                interpretation=self._interpret(stress_level, current_corr, current_gld_ret, current_dxy_ret)
+            )
+
+        except Exception as e:
+            logger.error(f"{self.name} calculation failed: {e}")
+            return None
+
+    def _interpret(
+        self,
+        stress_level: float,
+        corr: float,
+        gld_ret: float,
+        dxy_ret: float
+    ) -> str:
+        """Generate interpretation."""
+        if stress_level > 70 and corr > 0.3:
+            if gld_ret > 0 and dxy_ret > 0:
+                return "🔴 CRISIS: Gold AND dollar rising (extreme safe haven demand)"
+            else:
+                return "⚠️ UNUSUAL: Abnormal gold-dollar correlation"
+        elif stress_level > 60:
+            return "⚠️ STRESS: Gold-dollar correlation rising"
+        elif stress_level < 40:
+            return "✅ NORMAL: Healthy gold-dollar dynamics"
+        else:
+            return "⚡ WATCH: Gold-dollar correlation shifting"

src/core/macroeconomics/markets/risk_engine/indicators/liquidity.py

@@ -0,0 +1,429 @@
+"""
+Liquidity indicators for market risk detection.
+
+Monitors market liquidity, trading costs, and depth.
+"""
+
+from typing import Dict, Optional
+import pandas as pd
+import numpy as np
+
+from src.core.macroeconomics.markets.risk_engine.models import (
+    IndicatorSignal,
+    IndicatorCategory,
+    TimeSeriesData
+)
+from src.core.macroeconomics.markets.risk_engine.indicators.base import BaseIndicator
+from src.telegram_bot.logger import main_logger as logger
+
+
+class AmihudIlliquidityIndicator(BaseIndicator):
+    """
+    Amihud illiquidity measure.
+
+    Measures price impact per dollar traded.
+    Higher values = lower liquidity = harder to trade without moving market.
+
+    Formula: Illiquidity = |Return| / Volume
+    Averaged over rolling window, then z-scored.
+
+    Spikes during liquidity crises (2008, 2020 March).
+
+    Required data:
+    - SPY: S&P 500 ETF (has volume data)
+    """
+
+    def __init__(self):
+        super().__init__(
+            name="Amihud Illiquidity",
+            category=IndicatorCategory.LIQUIDITY,
+            description="Price impact per dollar traded",
+            zscore_window=252
+        )
+
+    def compute(self, data: Dict[str, TimeSeriesData]) -> Optional[IndicatorSignal]:
+        """Compute Amihud illiquidity."""
+        if not self._validate_data(data, ["SPY"]):
+            return None
+
+        try:
+            spy_ts = data["SPY"]
+            if spy_ts.volume is None or len(spy_ts.volume) == 0:
+                logger.warning(f"{self.name}: No volume data available")
+                return None
+
+            price_data = spy_ts.data
+            volume_data = spy_ts.volume
+
+            # Align price and volume
+            combined = pd.DataFrame({
+                'price': price_data,
+                'volume': volume_data
+            }).dropna()
+
+            if len(combined) < self.zscore_window + 20:
+                logger.warning(f"{self.name}: Insufficient data")
+                return None
+
+            # Calculate returns
+            returns = combined['price'].pct_change().dropna()
+
+            # Amihud measure: |return| / volume (in millions)
+            # Use volume in millions to get reasonable scale
+            amihud = (returns.abs() / (combined['volume'].iloc[1:] / 1e6))
+            amihud = amihud.replace([np.inf, -np.inf], np.nan).dropna()
+
+            # Calculate 20-day average illiquidity
+            amihud_ma = amihud.rolling(20).mean().dropna()
+
+            if len(amihud_ma) < self.zscore_window:
+                logger.warning(f"{self.name}: Insufficient data for z-score")
+                return None
+
+            # Z-score of illiquidity
+            illiq_zscore = self.calculate_zscore(amihud_ma)
+
+            # Calculate acceleration
+            illiq_accel = self.calculate_acceleration(amihud_ma, periods=5)
+
+            stress_level = self.normalize_to_level(illiq_zscore)
+
+            current_illiq = float(amihud_ma.iloc[-1])
+
+            return IndicatorSignal(
+                indicator_name=self.name,
+                category=self.category,
+                value=current_illiq * 1e6,  # Scale for display
+                zscore=illiq_zscore,
+                stress_level=stress_level,
+                description=(
+                    f"Illiquidity: {current_illiq*1e6:.2f} "
+                    f"(z={illiq_zscore:+.1f}), "
+                    f"acceleration: {illiq_accel*1e6:+.2f}"
+                ),
+                interpretation=self._interpret(stress_level, illiq_zscore, illiq_accel)
+            )
+
+        except Exception as e:
+            logger.error(f"{self.name} calculation failed: {e}")
+            return None
+
+    def _interpret(
+        self,
+        stress_level: float,
+        zscore: float,
+        acceleration: float
+    ) -> str:
+        """Generate interpretation."""
+        if stress_level > 75:
+            if acceleration > 0:
+                return "🔴 SEVERE: Liquidity evaporating (illiquidity spiking)"
+            else:
+                return "🔴 STRESSED: Low liquidity (but stabilizing)"
+        elif stress_level > 60:
+            return "⚠️ WARNING: Liquidity deteriorating"
+        elif stress_level < 40:
+            return "✅ DEEP: High market liquidity"
+        else:
+            return "⚡ WATCH: Liquidity moderately thin"
+
+
+class BidAskSpreadIndicator(BaseIndicator):
+    """
+    Bid-ask spread proxy using high-low range.
+
+    True bid-ask spreads not available in free data.
+    Use daily high-low range as proxy for trading costs.
+
+    Wider ranges = higher transaction costs = liquidity stress.
+
+    Signal construction:
+    - Calculate (High - Low) / Close ratio
+    - 20-day moving average
+    - Z-score vs historical
+
+    Required data:
+    - SPY: S&P 500 ETF
+    """
+
+    def __init__(self):
+        super().__init__(
+            name="Bid-Ask Spread Proxy",
+            category=IndicatorCategory.LIQUIDITY,
+            description="High-low range as liquidity proxy",
+            zscore_window=252
+        )
+
+    def compute(self, data: Dict[str, TimeSeriesData]) -> Optional[IndicatorSignal]:
+        """Compute bid-ask spread proxy."""
+        if not self._validate_data(data, ["SPY"]):
+            return None
+
+        try:
+            spy_ts = data["SPY"]
+
+            # Check if we have OHLC data
+            if (spy_ts.high is None or spy_ts.low is None or
+                len(spy_ts.high) == 0 or len(spy_ts.low) == 0):
+                logger.warning(f"{self.name}: No OHLC data available")
+                return None
+
+            # Combine all OHLC data
+            combined = pd.DataFrame({
+                'high': spy_ts.high,
+                'low': spy_ts.low,
+                'close': spy_ts.data
+            }).dropna()
+
+            if len(combined) < self.zscore_window + 20:
+                logger.warning(f"{self.name}: Insufficient data")
+                return None
+
+            # Calculate high-low spread as % of close
+            hl_spread = ((combined['high'] - combined['low']) / combined['close'] * 100)
+
+            # 20-day moving average to smooth noise
+            hl_spread_ma = hl_spread.rolling(20).mean().dropna()
+
+            if len(hl_spread_ma) < self.zscore_window:
+                logger.warning(f"{self.name}: Insufficient data for z-score")
+                return None
+
+            # Z-score
+            spread_zscore = self.calculate_zscore(hl_spread_ma)
+            stress_level = self.normalize_to_level(spread_zscore)
+
+            current_spread = float(hl_spread_ma.iloc[-1])
+
+            return IndicatorSignal(
+                indicator_name=self.name,
+                category=self.category,
+                value=current_spread,
+                zscore=spread_zscore,
+                stress_level=stress_level,
+                description=(
+                    f"H-L spread: {current_spread:.2f}% "
+                    f"(z={spread_zscore:+.1f})"
+                ),
+                interpretation=self._interpret(stress_level, spread_zscore)
+            )
+
+        except Exception as e:
+            logger.error(f"{self.name} calculation failed: {e}")
+            return None
+
+    def _interpret(self, stress_level: float, zscore: float) -> str:
+        """Generate interpretation."""
+        if stress_level > 75:
+            return "🔴 SEVERE: Very wide spreads (high trading costs)"
+        elif stress_level > 60:
+            return "⚠️ WARNING: Spreads widening"
+        elif stress_level < 40:
+            return "✅ TIGHT: Low trading costs"
+        else:
+            return "⚡ WATCH: Spreads moderately wide"
+
+
+class TreasuryLiquidityIndicator(BaseIndicator):
+    """
+    Treasury market liquidity stress.
+
+    Monitor Treasury ETF (TLT, IEF, SHY) price action.
+    Unusual Treasury volatility or selling suggests liquidity issues.
+
+    During extreme stress (Mar 2020), even Treasuries become illiquid.
+
+    Signal construction:
+    - Calculate Treasury ETF realized volatility
+    - Z-score vs historical
+    - High vol in "safe" Treasuries = liquidity crisis
+
+    Required data:
+    - TLT: 20+ Year Treasury ETF
+    - IEF: 7-10 Year Treasury ETF
+    """
+
+    def __init__(self):
+        super().__init__(
+            name="Treasury Liquidity",
+            category=IndicatorCategory.LIQUIDITY,
+            description="Treasury market liquidity stress",
+            zscore_window=252
+        )
+
+    def compute(self, data: Dict[str, TimeSeriesData]) -> Optional[IndicatorSignal]:
+        """Compute Treasury liquidity stress."""
+        # Try to use both TLT and IEF if available
+        available = []
+        if "TLT" in data and len(data["TLT"].data) > 0:
+            available.append("TLT")
+        if "IEF" in data and len(data["IEF"].data) > 0:
+            available.append("IEF")
+
+        if len(available) == 0:
+            logger.warning(f"{self.name}: No Treasury ETF data available")
+            return None
+
+        try:
+            # Combine available Treasury ETFs
+            tsy_prices = []
+            for ticker in available:
+                tsy_prices.append(data[ticker].data)
+
+            # Average price if multiple ETFs
+            if len(tsy_prices) == 1:
+                tsy_price = tsy_prices[0]
+            else:
+                combined = pd.DataFrame({f'tsy{i}': p for i, p in enumerate(tsy_prices)})
+                combined = combined.dropna()
+                tsy_price = combined.mean(axis=1)
+
+            if len(tsy_price) < self.zscore_window + 20:
+                logger.warning(f"{self.name}: Insufficient data")
+                return None
+
+            # Calculate 20-day realized volatility
+            returns = np.log(tsy_price / tsy_price.shift(1)).dropna()
+            realized_vol = returns.rolling(20).std() * np.sqrt(252) * 100
+            realized_vol = realized_vol.dropna()
+
+            if len(realized_vol) < self.zscore_window:
+                return None
+
+            # Z-score of Treasury volatility
+            vol_zscore = self.calculate_zscore(realized_vol)
+
+            # High Treasury vol = liquidity stress
+            stress_level = self.normalize_to_level(vol_zscore)
+
+            current_vol = float(realized_vol.iloc[-1])
+
+            return IndicatorSignal(
+                indicator_name=self.name,
+                category=self.category,
+                value=current_vol,
+                zscore=vol_zscore,
+                stress_level=stress_level,
+                description=(
+                    f"Treasury vol: {current_vol:.1f}% "
+                    f"(z={vol_zscore:+.1f}) "
+                    f"[{', '.join(available)}]"
+                ),
+                interpretation=self._interpret(stress_level, vol_zscore)
+            )
+
+        except Exception as e:
+            logger.error(f"{self.name} calculation failed: {e}")
+            return None
+
+    def _interpret(self, stress_level: float, zscore: float) -> str:
+        """Generate interpretation."""
+        if stress_level > 75:
+            return "🔴 CRISIS: Treasury market dysfunction (liquidity crisis)"
+        elif stress_level > 60:
+            return "⚠️ WARNING: Elevated Treasury volatility"
+        elif stress_level < 40:
+            return "✅ STABLE: Normal Treasury market function"
+        else:
+            return "⚡ WATCH: Treasury volatility increasing"
+
+
+class MarketDepthIndicator(BaseIndicator):
+    """
+    Market depth indicator using volume analysis.
+
+    During liquidity stress, trading volume can spike (forced selling)
+    or collapse (market freeze). Monitor volume extremes.
+
+    Signal construction:
+    - Calculate volume z-score (both high and low extremes)
+    - Volume acceleration
+    - Extreme volume = liquidity stress
+
+    Required data:
+    - SPY: S&P 500 ETF with volume
+    """
+
+    def __init__(self):
+        super().__init__(
+            name="Market Depth",
+            category=IndicatorCategory.LIQUIDITY,
+            description="Trading volume extremes",
+            zscore_window=252
+        )
+
+    def compute(self, data: Dict[str, TimeSeriesData]) -> Optional[IndicatorSignal]:
+        """Compute market depth signal."""
+        if not self._validate_data(data, ["SPY"]):
+            return None
+
+        try:
+            spy_ts = data["SPY"]
+            if spy_ts.volume is None or len(spy_ts.volume) == 0:
+                logger.warning(f"{self.name}: No volume data available")
+                return None
+
+            volume_data = spy_ts.volume
+
+            if len(volume_data) < self.zscore_window + 20:
+                logger.warning(f"{self.name}: Insufficient data")
+                return None
+
+            # Calculate 20-day average volume
+            volume_ma = volume_data.rolling(20).mean().dropna()
+
+            if len(volume_ma) < self.zscore_window:
+                return None
+
+            # Z-score of volume
+            vol_zscore = self.calculate_zscore(volume_ma)
+
+            # Take absolute value: both very high and very low volume = stress
+            # Very high = forced selling, Very low = market freeze
+            vol_zscore_abs = abs(vol_zscore)
+
+            stress_level = self.normalize_to_level(vol_zscore_abs)
+
+            current_vol = float(volume_data.iloc[-1])
+            avg_vol = float(volume_ma.iloc[-1])
+
+            return IndicatorSignal(
+                indicator_name=self.name,
+                category=self.category,
+                value=current_vol,
+                zscore=vol_zscore,
+                stress_level=stress_level,
+                description=(
+                    f"Volume: {current_vol/1e6:.1f}M "
+                    f"(20d avg: {avg_vol/1e6:.1f}M, "
+                    f"z={vol_zscore:+.1f})"
+                ),
+                interpretation=self._interpret(stress_level, vol_zscore, current_vol, avg_vol)
+            )
+
+        except Exception as e:
+            logger.error(f"{self.name} calculation failed: {e}")
+            return None
+
+    def _interpret(
+        self,
+        stress_level: float,
+        zscore: float,
+        current_vol: float,
+        avg_vol: float
+    ) -> str:
+        """Generate interpretation."""
+        if stress_level > 75:
+            if zscore > 2:
+                return "🔴 EXTREME: Panic selling (volume spike)"
+            else:
+                return "🔴 FROZEN: Market freeze (volume collapse)"
+        elif stress_level > 60:
+            if zscore > 0:
+                return "⚠️ ELEVATED: High volume (potential distribution)"
+            else:
+                return "⚠️ THIN: Low volume (reduced liquidity)"
+        elif stress_level < 40:
+            return "✅ NORMAL: Healthy trading volume"
+        else:
+            return "⚡ WATCH: Volume at moderately unusual levels"

src/core/macroeconomics/markets/risk_engine/indicators/macro_rates.py

@@ -0,0 +1,551 @@
+"""
+Macro and rates indicators for market risk detection.
+
+Monitors yield curve, financial conditions, repo market, and bond volatility.
+"""
+
+from typing import Dict, Optional
+import pandas as pd
+import numpy as np
+
+from src.core.macroeconomics.markets.risk_engine.models import (
+    IndicatorSignal,
+    IndicatorCategory,
+    TimeSeriesData
+)
+from src.core.macroeconomics.markets.risk_engine.indicators.base import BaseIndicator
+from src.telegram_bot.logger import main_logger as logger
+
+
+class YieldCurveIndicator(BaseIndicator):
+    """
+    Yield curve inversion and re-steepening detector.
+
+    Monitors 10Y-2Y Treasury spread:
+    - Inversion (spread < 0) = recession warning
+    - Re-steepening after inversion = recession imminent
+    - Extreme steepening = crisis/Fed intervention
+
+    Signal construction:
+    - Calculate 10Y-2Y spread
+    - Z-score of spread
+    - Detect inversion and re-steepening dynamics
+
+    Required data:
+    - DGS10: 10-Year Treasury Constant Maturity Rate
+    - DGS2: 2-Year Treasury Constant Maturity Rate
+    """
+
+    def __init__(self):
+        super().__init__(
+            name="Yield Curve",
+            category=IndicatorCategory.MACRO_RATES,
+            description="10Y-2Y spread (inversion & re-steepening)",
+            zscore_window=252
+        )
+
+    def compute(self, data: Dict[str, TimeSeriesData]) -> Optional[IndicatorSignal]:
+        """Compute yield curve signal."""
+        required = ["DGS10", "DGS2"]
+        if not self._validate_data(data, required):
+            return None
+
+        try:
+            dgs10 = data["DGS10"].data
+            dgs2 = data["DGS2"].data
+
+            # Align dates
+            combined = pd.DataFrame({
+                'dgs10': dgs10,
+                'dgs2': dgs2
+            }).dropna()
+
+            if len(combined) < self.zscore_window:
+                logger.warning(f"{self.name}: Insufficient data")
+                return None
+
+            # Calculate spread (10Y - 2Y)
+            spread = combined['dgs10'] - combined['dgs2']
+
+            # Check for inversion
+            is_inverted = spread.iloc[-1] < 0
+
+            # Calculate spread z-score
+            spread_zscore = self.calculate_zscore(spread)
+
+            # Calculate spread acceleration (rate of change)
+            spread_accel = self.calculate_acceleration(spread, periods=20)
+
+            # Risk logic:
+            # 1. Inverted curve = warning
+            # 2. Re-steepening after inversion = imminent risk
+            # 3. Extreme steepening = crisis mode
+
+            if is_inverted:
+                # Inverted: moderate risk
+                base_stress = 65.0
+                # If re-steepening (positive acceleration), increase risk
+                if spread_accel > 5:  # Steepening > 5bps
+                    stress_level = min(base_stress + 20, 95.0)
+                    signal_type = "re-steepening"
+                else:
+                    stress_level = base_stress
+                    signal_type = "inverted"
+            else:
+                # Not inverted: use z-score
+                # Negative z-score (flattening) = rising risk
+                # Positive z-score (steepening) = falling risk unless extreme
+                if spread_zscore < -1.5:
+                    # Flattening toward inversion
+                    stress_level = self.normalize_to_level(-spread_zscore)
+                    signal_type = "flattening"
+                elif spread_zscore > 2.0:
+                    # Extreme steepening (crisis intervention)
+                    stress_level = self.normalize_to_level(spread_zscore * 0.5)
+                    signal_type = "extreme_steep"
+                else:
+                    # Normal regime
+                    stress_level = 50.0 - (spread_zscore * 5)  # Flatter = higher stress
+                    stress_level = np.clip(stress_level, 0, 100)
+                    signal_type = "normal"
+
+            current_spread = float(spread.iloc[-1])
+
+            return IndicatorSignal(
+                indicator_name=self.name,
+                category=self.category,
+                value=current_spread,
+                zscore=spread_zscore,
+                stress_level=stress_level,
+                description=(
+                    f"10Y-2Y: {current_spread:+.0f}bps "
+                    f"(z={spread_zscore:+.1f}), "
+                    f"accel: {spread_accel:+.1f}bps [{signal_type}]"
+                ),
+                interpretation=self._interpret(stress_level, signal_type, current_spread, spread_accel)
+            )
+
+        except Exception as e:
+            logger.error(f"{self.name} calculation failed: {e}")
+            return None
+
+    def _interpret(
+        self,
+        stress_level: float,
+        signal_type: str,
+        spread: float,
+        accel: float
+    ) -> str:
+        """Generate interpretation."""
+        if signal_type == "re-steepening":
+            return f"🔴 CRITICAL: Curve re-steepening after inversion (recession signal)"
+        elif signal_type == "inverted":
+            return f"⚠️ INVERTED: Yield curve inverted ({spread:+.0f}bps)"
+        elif signal_type == "extreme_steep":
+            return f"🔴 CRISIS: Extreme steepening (Fed intervention signal)"
+        elif signal_type == "flattening":
+            return f"⚠️ FLATTENING: Curve flattening toward inversion"
+        else:
+            if stress_level < 40:
+                return "✅ NORMAL: Healthy yield curve"
+            else:
+                return "⚡ WATCH: Yield curve moderately flat"
+
+
+class NFCIIndicator(BaseIndicator):
+    """
+    National Financial Conditions Index.
+
+    Chicago Fed's comprehensive financial stress measure.
+    Incorporates credit, leverage, and risk indicators.
+
+    NFCI = 0: Average conditions
+    NFCI > 0: Tighter than average (stress)
+    NFCI < 0: Looser than average (easy conditions)
+
+    Signal construction:
+    - Use NFCI value directly (already standardized)
+    - Calculate acceleration
+    - Positive NFCI = stress
+
+    Required data:
+    - NFCI: Chicago Fed National Financial Conditions Index
+    """
+
+    def __init__(self):
+        super().__init__(
+            name="NFCI",
+            category=IndicatorCategory.MACRO_RATES,
+            description="National Financial Conditions Index",
+            zscore_window=252
+        )
+
+    def compute(self, data: Dict[str, TimeSeriesData]) -> Optional[IndicatorSignal]:
+        """Compute NFCI signal."""
+        if not self._validate_data(data, ["NFCI"]):
+            return None
+
+        try:
+            nfci_data = data["NFCI"].data
+
+            if len(nfci_data) < self.zscore_window:
+                logger.warning(f"{self.name}: Insufficient data")
+                return None
+
+            # NFCI is already standardized (mean=0, std=1)
+            # So we can use it directly as a z-score
+            current_nfci = float(nfci_data.iloc[-1])
+
+            # Calculate acceleration
+            nfci_accel = self.calculate_acceleration(nfci_data, periods=4)  # Weekly data
+
+            # Convert NFCI to stress level
+            # NFCI ranges roughly -1.5 to +3 in normal times
+            # Map to 0-100 scale
+            stress_level = self.normalize_to_level(current_nfci, clip=2.0)
+
+            return IndicatorSignal(
+                indicator_name=self.name,
+                category=self.category,
+                value=current_nfci,
+                zscore=current_nfci,  # NFCI is already z-scored
+                stress_level=stress_level,
+                description=(
+                    f"NFCI: {current_nfci:+.2f} "
+                    f"(accel: {nfci_accel:+.2f})"
+                ),
+                interpretation=self._interpret(stress_level, current_nfci, nfci_accel)
+            )
+
+        except Exception as e:
+            logger.error(f"{self.name} calculation failed: {e}")
+            return None
+
+    def _interpret(
+        self,
+        stress_level: float,
+        nfci: float,
+        acceleration: float
+    ) -> str:
+        """Generate interpretation."""
+        if stress_level > 75:
+            if acceleration > 0.1:
+                return "🔴 SEVERE: Financial conditions tightening rapidly"
+            else:
+                return "🔴 TIGHT: Severe financial stress (stabilizing)"
+        elif stress_level > 60:
+            return "⚠️ ELEVATED: Financial conditions tightening"
+        elif stress_level < 40:
+            if stress_level < 25:
+                return "⚠️ LOOSE: Extremely easy conditions (bubble risk)"
+            else:
+                return "✅ EASY: Accommodative financial conditions"
+        else:
+            return "⚡ WATCH: Financial conditions moderately tight"
+
+
+class RepoStressIndicator(BaseIndicator):
+    """
+    Repo market stress indicator.
+
+    Monitors spread between SOFR and EFFR (Fed Funds).
+    Wide spread = funding stress in repo market.
+
+    September 2019: Repo crisis with SOFR spike.
+    March 2020: Severe repo dysfunction.
+
+    Signal construction:
+    - Calculate SOFR - EFFR spread
+    - Z-score of spread
+    - Positive spread = repo stress
+
+    Required data:
+    - SOFR: Secured Overnight Financing Rate
+    - DFF: Federal Funds Effective Rate (EFFR)
+    """
+
+    def __init__(self):
+        super().__init__(
+            name="Repo Stress",
+            category=IndicatorCategory.MACRO_RATES,
+            description="SOFR-EFFR spread (repo market stress)",
+            zscore_window=252
+        )
+
+    def compute(self, data: Dict[str, TimeSeriesData]) -> Optional[IndicatorSignal]:
+        """Compute repo stress signal."""
+        required = ["SOFR", "DFF"]
+        if not self._validate_data(data, required):
+            return None
+
+        try:
+            sofr = data["SOFR"].data
+            effr = data["DFF"].data
+
+            # Align dates
+            combined = pd.DataFrame({
+                'sofr': sofr,
+                'effr': effr
+            }).dropna()
+
+            if len(combined) < self.zscore_window:
+                logger.warning(f"{self.name}: Insufficient data")
+                return None
+
+            # Calculate spread (SOFR - EFFR)
+            # Positive spread = repo more expensive than Fed Funds = stress
+            spread = combined['sofr'] - combined['effr']
+
+            # Z-score of spread
+            spread_zscore = self.calculate_zscore(spread)
+
+            # Calculate spread acceleration
+            spread_accel = self.calculate_acceleration(spread, periods=5)
+
+            stress_level = self.normalize_to_level(spread_zscore)
+
+            current_spread = float(spread.iloc[-1])
+
+            return IndicatorSignal(
+                indicator_name=self.name,
+                category=self.category,
+                value=current_spread,
+                zscore=spread_zscore,
+                stress_level=stress_level,
+                description=(
+                    f"SOFR-EFFR: {current_spread:+.1f}bps "
+                    f"(z={spread_zscore:+.1f}), "
+                    f"accel: {spread_accel:+.2f}bps"
+                ),
+                interpretation=self._interpret(stress_level, spread_zscore, current_spread)
+            )
+
+        except Exception as e:
+            logger.error(f"{self.name} calculation failed: {e}")
+            return None
+
+    def _interpret(
+        self,
+        stress_level: float,
+        zscore: float,
+        spread: float
+    ) -> str:
+        """Generate interpretation."""
+        if stress_level > 75:
+            return f"🔴 SEVERE: Repo market dysfunction (spread: {spread:+.1f}bps)"
+        elif stress_level > 60:
+            return "⚠️ ELEVATED: Repo market stress building"
+        elif stress_level < 40:
+            return "✅ STABLE: Repo market functioning normally"
+        else:
+            return "⚡ WATCH: Repo market showing mild stress"
+
+
+class MOVEIndexIndicator(BaseIndicator):
+    """
+    MOVE Index - bond market volatility.
+
+    Merrill Lynch Option Volatility Estimate.
+    VIX equivalent for Treasury market.
+
+    Rising MOVE = bond market stress/uncertainty.
+    Often spikes during macro regime changes.
+
+    Signal construction:
+    - Calculate MOVE z-score
+    - Monitor acceleration
+    - High MOVE = rates volatility = macro uncertainty
+
+    Required data:
+    - MOVE: Merrill Option Volatility Estimate Index
+    """
+
+    def __init__(self):
+        super().__init__(
+            name="MOVE Index",
+            category=IndicatorCategory.MACRO_RATES,
+            description="Bond market volatility (rates VIX)",
+            zscore_window=252
+        )
+
+    def compute(self, data: Dict[str, TimeSeriesData]) -> Optional[IndicatorSignal]:
+        """Compute MOVE index signal."""
+        if not self._validate_data(data, ["MOVE"]):
+            return None
+
+        try:
+            move_data = data["MOVE"].data
+
+            if len(move_data) < self.zscore_window:
+                logger.warning(f"{self.name}: Insufficient data")
+                return None
+
+            # Calculate z-score
+            move_zscore = self.calculate_zscore(move_data)
+
+            # Calculate acceleration
+            move_accel = self.calculate_acceleration(move_data, periods=5)
+
+            stress_level = self.normalize_to_level(move_zscore)
+
+            current_move = float(move_data.iloc[-1])
+
+            return IndicatorSignal(
+                indicator_name=self.name,
+                category=self.category,
+                value=current_move,
+                zscore=move_zscore,
+                stress_level=stress_level,
+                description=(
+                    f"MOVE: {current_move:.1f} "
+                    f"(z={move_zscore:+.1f}), "
+                    f"accel: {move_accel:+.1f}"
+                ),
+                interpretation=self._interpret(stress_level, move_zscore, move_accel)
+            )
+
+        except Exception as e:
+            logger.error(f"{self.name} calculation failed: {e}")
+            return None
+
+    def _interpret(
+        self,
+        stress_level: float,
+        zscore: float,
+        acceleration: float
+    ) -> str:
+        """Generate interpretation."""
+        if stress_level > 75:
+            if acceleration > 0:
+                return "🔴 EXTREME: Bond volatility surging (macro regime shift)"
+            else:
+                return "🔴 ELEVATED: High bond volatility (stabilizing)"
+        elif stress_level > 60:
+            return "⚠️ RISING: Bond market volatility increasing"
+        elif stress_level < 40:
+            return "✅ CALM: Low bond market volatility"
+        else:
+            return "⚡ WATCH: Bond volatility moderately elevated"
+
+
+class RealRatesIndicator(BaseIndicator):
+    """
+    Real interest rates indicator.
+
+    Monitors 10Y Treasury yield adjusted for inflation expectations.
+    Negative real rates = loose policy, asset bubbles.
+    Rising real rates = tightening, risk-off.
+
+    Signal construction:
+    - Calculate 10Y - expected inflation (proxy: 5Y avg)
+    - Monitor real rate level and change
+    - Rising real rates = headwind for risk assets
+
+    Required data:
+    - DGS10: 10-Year Treasury yield
+    - T10YIE: 10-Year Breakeven Inflation Rate (if available)
+    """
+
+    def __init__(self):
+        super().__init__(
+            name="Real Rates",
+            category=IndicatorCategory.MACRO_RATES,
+            description="Real interest rates (nominal - inflation)",
+            zscore_window=252
+        )
+
+    def compute(self, data: Dict[str, TimeSeriesData]) -> Optional[IndicatorSignal]:
+        """Compute real rates signal."""
+        # Try to get 10Y yield
+        if not self._validate_data(data, ["DGS10"]):
+            return None
+
+        try:
+            dgs10 = data["DGS10"].data
+
+            # Check if we have breakeven inflation
+            if "T10YIE" in data and len(data["T10YIE"].data) > 0:
+                inflation = data["T10YIE"].data
+                combined = pd.DataFrame({
+                    'nominal': dgs10,
+                    'inflation': inflation
+                }).dropna()
+
+                if len(combined) < self.zscore_window:
+                    return None
+
+                # Real rate = Nominal - Inflation
+                real_rate = combined['nominal'] - combined['inflation']
+            else:
+                # Fallback: use simple moving average of 10Y as inflation proxy
+                # This is crude but gives us something
+                logger.warning(f"{self.name}: No breakeven inflation data, using approximation")
+                if len(dgs10) < self.zscore_window + 252:
+                    return None
+
+                inflation_proxy = dgs10.rolling(252).mean()  # 1-year average as "expected"
+                real_rate = dgs10 - inflation_proxy
+                real_rate = real_rate.dropna()
+
+            if len(real_rate) < self.zscore_window:
+                return None
+
+            # Calculate z-score of real rate level
+            real_rate_zscore = self.calculate_zscore(real_rate)
+
+            # Calculate acceleration (how fast are real rates rising?)
+            real_rate_accel = self.calculate_acceleration(real_rate, periods=20)
+
+            # Stress logic:
+            # - Rapidly rising real rates (tightening) = stress
+            # - Very negative real rates (bubble territory) = moderate stress
+            if real_rate_accel > 0.2:  # Rising
+                # Tightening = risk-off stress
+                stress_level = self.normalize_to_level(real_rate_accel * 10)
+                signal_type = "tightening"
+            elif real_rate.iloc[-1] < -1.0:  # Very negative
+                # Bubble territory
+                stress_level = 55.0
+                signal_type = "bubble"
+            else:
+                # Normal: falling real rates = lower stress
+                stress_level = self.normalize_to_level(-real_rate_accel * 5)
+                signal_type = "normal"
+
+            current_real = float(real_rate.iloc[-1])
+
+            return IndicatorSignal(
+                indicator_name=self.name,
+                category=self.category,
+                value=current_real,
+                zscore=real_rate_zscore,
+                stress_level=stress_level,
+                description=(
+                    f"Real rate: {current_real:+.1f}% "
+                    f"(z={real_rate_zscore:+.1f}), "
+                    f"accel: {real_rate_accel:+.2f}% [{signal_type}]"
+                ),
+                interpretation=self._interpret(stress_level, signal_type, current_real, real_rate_accel)
+            )
+
+        except Exception as e:
+            logger.error(f"{self.name} calculation failed: {e}")
+            return None
+
+    def _interpret(
+        self,
+        stress_level: float,
+        signal_type: str,
+        real_rate: float,
+        accel: float
+    ) -> str:
+        """Generate interpretation."""
+        if signal_type == "tightening":
+            return f"🔴 TIGHTENING: Real rates rising fast (headwind for risk assets)"
+        elif signal_type == "bubble":
+            return f"⚠️ BUBBLE: Deeply negative real rates ({real_rate:+.1f}%)"
+        else:
+            if stress_level < 40:
+                return "✅ SUPPORTIVE: Falling/stable real rates"
+            else:
+                return "⚡ WATCH: Real rates moderately elevated"

src/core/macroeconomics/markets/risk_engine/indicators/tail_risk.py

@@ -0,0 +1,538 @@
+"""
+Tail risk indicators for market risk detection.
+
+Monitors correlation spikes, regime shifts, and extreme events.
+"""
+
+from typing import Dict, Optional
+import pandas as pd
+import numpy as np
+
+from src.core.macroeconomics.markets.risk_engine.models import (
+    IndicatorSignal,
+    IndicatorCategory,
+    TimeSeriesData
+)
+from src.core.macroeconomics.markets.risk_engine.indicators.base import BaseIndicator
+from src.telegram_bot.logger import main_logger as logger
+
+
+class CorrelationSpikeIndicator(BaseIndicator):
+    """
+    Cross-asset correlation spike detector.
+
+    During crises, correlations spike to 1 (everything falls together).
+    Diversification breaks down = systemic stress.
+
+    Signal construction:
+    - Calculate rolling correlation between SPY, TLT, GLD
+    - Monitor average pairwise correlation
+    - Spike to 1 or -1 = stress
+
+    Required data:
+    - SPY: Equities
+    - TLT: Bonds
+    - GLD: Gold
+    """
+
+    def __init__(self):
+        super().__init__(
+            name="Correlation Spike",
+            category=IndicatorCategory.TAIL_RISK,
+            description="Cross-asset correlation breakdown",
+            zscore_window=252
+        )
+
+    def compute(self, data: Dict[str, TimeSeriesData]) -> Optional[IndicatorSignal]:
+        """Compute correlation spike signal."""
+        # Require at least 2 of the 3 assets
+        available = []
+        for ticker in ["SPY", "TLT", "GLD"]:
+            if ticker in data and len(data[ticker].data) > 0:
+                available.append(ticker)
+
+        if len(available) < 2:
+            logger.warning(f"{self.name}: Insufficient asset data")
+            return None
+
+        try:
+            # Combine available assets
+            price_dict = {ticker: data[ticker].data for ticker in available}
+            prices = pd.DataFrame(price_dict).dropna()
+
+            if len(prices) < self.zscore_window + 60:
+                logger.warning(f"{self.name}: Insufficient data")
+                return None
+
+            # Calculate returns
+            returns = prices.pct_change().dropna()
+
+            # Calculate rolling 60-day correlation matrix
+            # Get average absolute correlation (excluding diagonal)
+            window = 60
+            correlations = []
+
+            for i in range(window, len(returns)):
+                window_rets = returns.iloc[i-window:i]
+                corr_matrix = window_rets.corr()
+
+                # Get off-diagonal correlations
+                n_assets = len(available)
+                off_diag = []
+                for j in range(n_assets):
+                    for k in range(j+1, n_assets):
+                        off_diag.append(abs(corr_matrix.iloc[j, k]))
+
+                avg_corr = np.mean(off_diag) if off_diag else 0.0
+                correlations.append(avg_corr)
+
+            corr_series = pd.Series(
+                correlations,
+                index=returns.index[window:]
+            )
+
+            if len(corr_series) < self.zscore_window:
+                logger.warning(f"{self.name}: Insufficient correlation data")
+                return None
+
+            # Z-score of average correlation
+            # High correlation = stress
+            corr_zscore = self.calculate_zscore(corr_series)
+
+            stress_level = self.normalize_to_level(corr_zscore)
+
+            current_corr = float(corr_series.iloc[-1])
+
+            return IndicatorSignal(
+                indicator_name=self.name,
+                category=self.category,
+                value=current_corr,
+                zscore=corr_zscore,
+                stress_level=stress_level,
+                description=(
+                    f"Avg correlation: {current_corr:.2f} "
+                    f"(z={corr_zscore:+.1f}) "
+                    f"[{', '.join(available)}]"
+                ),
+                interpretation=self._interpret(stress_level, current_corr, corr_zscore)
+            )
+
+        except Exception as e:
+            logger.error(f"{self.name} calculation failed: {e}")
+            return None
+
+    def _interpret(
+        self,
+        stress_level: float,
+        corr: float,
+        zscore: float
+    ) -> str:
+        """Generate interpretation."""
+        if stress_level > 75:
+            return f"🔴 BREAKDOWN: Correlations spiking (diversification failing, {corr:.2f})"
+        elif stress_level > 60:
+            return "⚠️ RISING: Correlations increasing (contagion risk)"
+        elif stress_level < 40:
+            return "✅ DIVERSIFIED: Normal correlation structure"
+        else:
+            return "⚡ WATCH: Correlations moderately elevated"
+
+
+class VolatilityRegimeIndicator(BaseIndicator):
+    """
+    Volatility regime shift detector.
+
+    Detect transitions between low-vol and high-vol regimes.
+    Regime changes = structural shift = risk.
+
+    Signal construction:
+    - Calculate 20-day realized volatility
+    - Compare to 252-day average
+    - Rapid regime shift = stress
+
+    Required data:
+    - SPY: S&P 500 for volatility calculation
+    """
+
+    def __init__(self):
+        super().__init__(
+            name="Volatility Regime",
+            category=IndicatorCategory.TAIL_RISK,
+            description="Volatility regime shift detection",
+            zscore_window=252
+        )
+
+    def compute(self, data: Dict[str, TimeSeriesData]) -> Optional[IndicatorSignal]:
+        """Compute volatility regime signal."""
+        if not self._validate_data(data, ["SPY"]):
+            return None
+
+        try:
+            spy = data["SPY"].data
+
+            if len(spy) < self.zscore_window + 20:
+                logger.warning(f"{self.name}: Insufficient data")
+                return None
+
+            # Calculate returns
+            returns = np.log(spy / spy.shift(1)).dropna()
+
+            # Calculate short-term (20-day) and long-term (252-day) volatility
+            short_vol = returns.rolling(20).std() * np.sqrt(252) * 100
+            long_vol = returns.rolling(252).std() * np.sqrt(252) * 100
+
+            # Align
+            vol_data = pd.DataFrame({
+                'short': short_vol,
+                'long': long_vol
+            }).dropna()
+
+            if len(vol_data) < self.zscore_window:
+                return None
+
+            # Calculate volatility ratio (short / long)
+            # Ratio > 1 = elevated regime
+            # Ratio >> 1 = regime shift in progress
+            vol_ratio = vol_data['short'] / vol_data['long']
+
+            # Z-score of ratio
+            ratio_zscore = self.calculate_zscore(vol_ratio)
+
+            # Also monitor ratio acceleration (regime shift speed)
+            ratio_accel = self.calculate_acceleration(vol_ratio, periods=5)
+
+            stress_level = self.normalize_to_level(ratio_zscore)
+
+            current_ratio = float(vol_ratio.iloc[-1])
+            current_short = float(vol_data['short'].iloc[-1])
+            current_long = float(vol_data['long'].iloc[-1])
+
+            return IndicatorSignal(
+                indicator_name=self.name,
+                category=self.category,
+                value=current_ratio,
+                zscore=ratio_zscore,
+                stress_level=stress_level,
+                description=(
+                    f"Vol ratio: {current_ratio:.2f} "
+                    f"(20d: {current_short:.1f}%, 252d: {current_long:.1f}%), "
+                    f"z={ratio_zscore:+.1f}"
+                ),
+                interpretation=self._interpret(stress_level, current_ratio, ratio_zscore, ratio_accel)
+            )
+
+        except Exception as e:
+            logger.error(f"{self.name} calculation failed: {e}")
+            return None
+
+    def _interpret(
+        self,
+        stress_level: float,
+        ratio: float,
+        zscore: float,
+        accel: float
+    ) -> str:
+        """Generate interpretation."""
+        if stress_level > 75:
+            if accel > 0.05:
+                return f"🔴 SHIFT: Regime shifting to high-vol (ratio: {ratio:.2f})"
+            else:
+                return f"🔴 HIGH-VOL: High volatility regime (ratio: {ratio:.2f})"
+        elif stress_level > 60:
+            return "⚠️ ELEVATED: Volatility regime changing"
+        elif stress_level < 40:
+            return "✅ STABLE: Low volatility regime"
+        else:
+            return "⚡ WATCH: Volatility moderately elevated"
+
+
+class DrawdownIndicator(BaseIndicator):
+    """
+    Maximum drawdown monitor.
+
+    Track current drawdown from recent peak.
+    Large drawdowns = tail risk materialized.
+
+    Signal construction:
+    - Calculate running maximum
+    - Current drawdown from peak
+    - Drawdown magnitude and duration
+
+    Required data:
+    - SPY: S&P 500
+    """
+
+    def __init__(self):
+        super().__init__(
+            name="Market Drawdown",
+            category=IndicatorCategory.TAIL_RISK,
+            description="Drawdown from recent peak",
+            zscore_window=252
+        )
+
+    def compute(self, data: Dict[str, TimeSeriesData]) -> Optional[IndicatorSignal]:
+        """Compute drawdown signal."""
+        if not self._validate_data(data, ["SPY"]):
+            return None
+
+        try:
+            spy = data["SPY"].data
+
+            if len(spy) < self.zscore_window:
+                logger.warning(f"{self.name}: Insufficient data")
+                return None
+
+            # Calculate running maximum
+            running_max = spy.expanding().max()
+
+            # Calculate drawdown
+            drawdown = (spy / running_max - 1) * 100
+
+            # Current drawdown
+            current_dd = float(drawdown.iloc[-1])
+
+            # Days in drawdown
+            if current_dd < 0:
+                # Find last peak
+                peak_idx = running_max.iloc[-1:].index[0]
+                days_in_dd = (spy.index[-1] - peak_idx).days
+            else:
+                days_in_dd = 0
+
+            # Calculate z-score of drawdown magnitude
+            dd_zscore = self.calculate_zscore(drawdown)
+
+            # Flip sign: deeper drawdown = more negative = higher stress
+            stress_zscore = -dd_zscore
+
+            stress_level = self.normalize_to_level(stress_zscore)
+
+            return IndicatorSignal(
+                indicator_name=self.name,
+                category=self.category,
+                value=current_dd,
+                zscore=stress_zscore,
+                stress_level=stress_level,
+                description=(
+                    f"Drawdown: {current_dd:.1f}% "
+                    f"(z={stress_zscore:+.1f}), "
+                    f"Duration: {days_in_dd} days"
+                ),
+                interpretation=self._interpret(stress_level, current_dd, days_in_dd)
+            )
+
+        except Exception as e:
+            logger.error(f"{self.name} calculation failed: {e}")
+            return None
+
+    def _interpret(
+        self,
+        stress_level: float,
+        drawdown: float,
+        duration: int
+    ) -> str:
+        """Generate interpretation."""
+        if stress_level > 80:
+            return f"🔴 SEVERE: Deep drawdown ({drawdown:.1f}%, {duration}d)"
+        elif stress_level > 65:
+            return f"⚠️ ELEVATED: Significant drawdown ({drawdown:.1f}%)"
+        elif stress_level < 35:
+            if drawdown >= -2:
+                return "✅ AT PEAK: Near all-time highs"
+            else:
+                return "✅ SHALLOW: Minor pullback"
+        else:
+            return f"⚡ WATCH: Moderate drawdown ({drawdown:.1f}%)"
+
+
+class SkewIndicator(BaseIndicator):
+    """
+    Return skewness monitor.
+
+    Measure return distribution skewness.
+    Negative skew = left-tail risk (crash risk).
+
+    Signal construction:
+    - Calculate rolling 60-day return skewness
+    - Negative skew = tail risk
+    - Monitor for increasingly negative skew
+
+    Required data:
+    - SPY: S&P 500
+    """
+
+    def __init__(self):
+        super().__init__(
+            name="Return Skewness",
+            category=IndicatorCategory.TAIL_RISK,
+            description="Return distribution skew (crash risk)",
+            zscore_window=252
+        )
+
+    def compute(self, data: Dict[str, TimeSeriesData]) -> Optional[IndicatorSignal]:
+        """Compute skewness signal."""
+        if not self._validate_data(data, ["SPY"]):
+            return None
+
+        try:
+            spy = data["SPY"].data
+
+            if len(spy) < self.zscore_window + 60:
+                logger.warning(f"{self.name}: Insufficient data")
+                return None
+
+            # Calculate returns
+            returns = spy.pct_change().dropna() * 100
+
+            # Calculate rolling 60-day skewness
+            skew = returns.rolling(60).skew().dropna()
+
+            if len(skew) < self.zscore_window:
+                return None
+
+            # Z-score of skewness
+            skew_zscore = self.calculate_zscore(skew)
+
+            # Flip sign: negative skew = positive stress
+            # (because negative skew means crash risk)
+            stress_zscore = -skew_zscore
+
+            stress_level = self.normalize_to_level(stress_zscore)
+
+            current_skew = float(skew.iloc[-1])
+
+            return IndicatorSignal(
+                indicator_name=self.name,
+                category=self.category,
+                value=current_skew,
+                zscore=stress_zscore,
+                stress_level=stress_level,
+                description=(
+                    f"60d skew: {current_skew:+.2f} "
+                    f"(stress_z={stress_zscore:+.1f})"
+                ),
+                interpretation=self._interpret(stress_level, current_skew, stress_zscore)
+            )
+
+        except Exception as e:
+            logger.error(f"{self.name} calculation failed: {e}")
+            return None
+
+    def _interpret(
+        self,
+        stress_level: float,
+        skew: float,
+        zscore: float
+    ) -> str:
+        """Generate interpretation."""
+        if stress_level > 70:
+            if skew < -1:
+                return f"🔴 SEVERE: Highly negative skew (crash risk, {skew:+.2f})"
+            else:
+                return "🔴 ELEVATED: Distribution tail risk increasing"
+        elif stress_level > 55:
+            return f"⚠️ WARNING: Negative skew building ({skew:+.2f})"
+        elif stress_level < 40:
+            if skew > 0:
+                return f"✅ POSITIVE: Upside skew ({skew:+.2f})"
+            else:
+                return "✅ NORMAL: Moderate skew"
+        else:
+            return "⚡ WATCH: Skew moderately negative"
+
+
+class TailEventIndicator(BaseIndicator):
+    """
+    Tail event detector.
+
+    Monitor for extreme daily moves (>2 std dev).
+    Frequency of tail events = market fragility.
+
+    Signal construction:
+    - Count tail events in rolling window
+    - Compare to historical frequency
+    - Rising frequency = fragility
+
+    Required data:
+    - SPY: S&P 500
+    """
+
+    def __init__(self):
+        super().__init__(
+            name="Tail Event Frequency",
+            category=IndicatorCategory.TAIL_RISK,
+            description="Extreme move frequency",
+            zscore_window=252
+        )
+
+    def compute(self, data: Dict[str, TimeSeriesData]) -> Optional[IndicatorSignal]:
+        """Compute tail event frequency."""
+        if not self._validate_data(data, ["SPY"]):
+            return None
+
+        try:
+            spy = data["SPY"].data
+
+            if len(spy) < self.zscore_window + 60:
+                logger.warning(f"{self.name}: Insufficient data")
+                return None
+
+            # Calculate returns
+            returns = spy.pct_change().dropna() * 100
+
+            # Calculate rolling std
+            rolling_std = returns.rolling(60).std()
+
+            # Identify tail events (|return| > 2 std)
+            tail_events = (returns.abs() > 2 * rolling_std).astype(int)
+
+            # Count tail events in rolling 60-day window
+            tail_count = tail_events.rolling(60).sum().dropna()
+
+            if len(tail_count) < self.zscore_window:
+                return None
+
+            # Z-score of tail event frequency
+            tail_zscore = self.calculate_zscore(tail_count)
+
+            stress_level = self.normalize_to_level(tail_zscore)
+
+            current_count = int(tail_count.iloc[-1])
+
+            # Recent tail events
+            recent_tails = tail_events.iloc[-20:].sum()
+
+            return IndicatorSignal(
+                indicator_name=self.name,
+                category=self.category,
+                value=float(current_count),
+                zscore=tail_zscore,
+                stress_level=stress_level,
+                description=(
+                    f"Tail events (60d): {current_count} "
+                    f"(z={tail_zscore:+.1f}), "
+                    f"Recent 20d: {recent_tails}"
+                ),
+                interpretation=self._interpret(stress_level, current_count, recent_tails, tail_zscore)
+            )
+
+        except Exception as e:
+            logger.error(f"{self.name} calculation failed: {e}")
+            return None
+
+    def _interpret(
+        self,
+        stress_level: float,
+        count: int,
+        recent: int,
+        zscore: float
+    ) -> str:
+        """Generate interpretation."""
+        if stress_level > 75:
+            return f"🔴 FRAGILE: High tail event frequency ({count} in 60d, {recent} recent)"
+        elif stress_level > 60:
+            return f"⚠️ ELEVATED: Increasing tail events ({count} in 60d)"
+        elif stress_level < 40:
+            return "✅ STABLE: Low tail event frequency"
+        else:
+            return f"⚡ WATCH: Moderate tail events ({count} in 60d)"

src/core/macroeconomics/markets/risk_engine/indicators/volatility.py

@@ -0,0 +1,405 @@
+"""
+Volatility indicators for market risk detection.
+
+Monitors fear gauges, volatility term structure, and volatility of volatility.
+"""
+
+from typing import Dict, Optional
+import pandas as pd
+import numpy as np
+
+from src.core.macroeconomics.markets.risk_engine.models import (
+    IndicatorSignal,
+    IndicatorCategory,
+    TimeSeriesData
+)
+from src.core.macroeconomics.markets.risk_engine.indicators.base import BaseIndicator
+from src.telegram_bot.logger import main_logger as logger
+
+
+class VIXStressIndicator(BaseIndicator):
+    """
+    Detect VIX rising without SPY falling (hidden stress).
+
+    Classic stress signal: VIX spikes while equity markets are stable.
+    Suggests fear building before visible market decline.
+
+    Signal construction:
+    - Calculate VIX z-score over 252-day window
+    - Calculate SPY return over 20 days
+    - If VIX elevated but SPY not falling → stress building
+
+    Required data:
+    - ^VIX: CBOE Volatility Index
+    - SPY: S&P 500 ETF
+    """
+
+    def __init__(self):
+        super().__init__(
+            name="VIX Stress",
+            category=IndicatorCategory.VOLATILITY,
+            description="VIX rising without equity decline (hidden stress)",
+            zscore_window=252
+        )
+
+    def compute(self, data: Dict[str, TimeSeriesData]) -> Optional[IndicatorSignal]:
+        """Compute VIX stress signal."""
+        required = ["^VIX", "SPY"]
+        if not self._validate_data(data, required):
+            return None
+
+        try:
+            vix_data = data["^VIX"].data
+            spy_data = data["SPY"].data
+
+            # Align dates
+            combined = pd.DataFrame({
+                'vix': vix_data,
+                'spy': spy_data
+            }).dropna()
+
+            if len(combined) < self.zscore_window:
+                logger.warning(
+                    f"{self.name}: Insufficient data "
+                    f"({len(combined)} < {self.zscore_window})"
+                )
+                return None
+
+            # VIX z-score (how elevated is VIX?)
+            vix_zscore = self.calculate_zscore(combined['vix'])
+
+            # SPY 20-day return (is equity actually falling?)
+            spy_return_20d = (combined['spy'].iloc[-1] / combined['spy'].iloc[-21] - 1) * 100
+
+            # Stress signal: VIX elevated but SPY not falling
+            # If VIX > +1 std and SPY > -5% = hidden stress
+            if vix_zscore > 1.0 and spy_return_20d > -5.0:
+                # Amplify signal based on disconnect
+                disconnect_factor = (spy_return_20d + 5.0) / 10.0  # 0-1 scale
+                stress_level = self.normalize_to_level(vix_zscore) * (1 + disconnect_factor * 0.5)
+                stress_level = min(stress_level, 100.0)
+            else:
+                # Normal regime: just use VIX z-score
+                stress_level = self.normalize_to_level(vix_zscore)
+
+            return IndicatorSignal(
+                indicator_name=self.name,
+                category=self.category,
+                value=float(vix_data.iloc[-1]),
+                zscore=vix_zscore,
+                stress_level=stress_level,
+                description=(
+                    f"VIX at {vix_data.iloc[-1]:.1f} "
+                    f"(z={vix_zscore:+.1f}), "
+                    f"SPY 20d: {spy_return_20d:+.1f}%"
+                ),
+                interpretation=self._interpret(stress_level, vix_zscore, spy_return_20d)
+            )
+
+        except Exception as e:
+            logger.error(f"{self.name} calculation failed: {e}")
+            return None
+
+    def _interpret(self, stress_level: float, vix_zscore: float, spy_return: float) -> str:
+        """Generate human-readable interpretation."""
+        if stress_level > 75:
+            if spy_return > 0:
+                return "⚠️ SEVERE: VIX spiking while equities rally (major disconnect)"
+            else:
+                return "🔴 EXTREME: VIX at extreme levels with equity weakness"
+        elif stress_level > 60:
+            return "⚠️ ELEVATED: Fear gauge rising, monitor for deterioration"
+        elif stress_level < 40:
+            return "✅ CALM: VIX at normal levels"
+        else:
+            return "⚡ WATCH: VIX moderately elevated"
+
+
+class VIXTermStructureIndicator(BaseIndicator):
+    """
+    Detect VIX term structure inversion (backwardation).
+
+    Normal market: VIX term structure in contango (VIX < VXV)
+    Stress market: Backwardation (VIX > VXV) suggests imminent fear
+
+    Signal construction:
+    - Calculate VIX/VXV ratio
+    - Ratio > 1.0 = backwardation (stress)
+    - Z-score of ratio over time
+
+    Required data:
+    - VIXCLS: VIX from FRED (1-month implied vol)
+    - VXV or 3-month VIX proxy
+
+    Note: If VXV unavailable, we use VIX acceleration as proxy
+    """
+
+    def __init__(self):
+        super().__init__(
+            name="VIX Term Structure",
+            category=IndicatorCategory.VOLATILITY,
+            description="VIX term structure inversion (backwardation)",
+            zscore_window=252
+        )
+
+    def compute(self, data: Dict[str, TimeSeriesData]) -> Optional[IndicatorSignal]:
+        """Compute term structure signal."""
+        # Try VIXCLS from FRED first
+        if "VIXCLS" not in data or len(data["VIXCLS"].data) == 0:
+            logger.warning(f"{self.name}: VIXCLS not available")
+            return None
+
+        try:
+            vix_data = data["VIXCLS"].data
+
+            if len(vix_data) < self.zscore_window:
+                logger.warning(
+                    f"{self.name}: Insufficient data "
+                    f"({len(vix_data)} < {self.zscore_window})"
+                )
+                return None
+
+            # Calculate VIX acceleration as term structure proxy
+            # Rising VIX suggests near-term fear > long-term fear
+            vix_acceleration = self.calculate_acceleration(vix_data, periods=5)
+
+            # Z-score of acceleration
+            accel_series = pd.Series([
+                self.calculate_acceleration(vix_data.iloc[:i], periods=5)
+                for i in range(self.zscore_window, len(vix_data) + 1)
+            ], index=vix_data.iloc[self.zscore_window-1:].index)
+
+            if len(accel_series) < 20:
+                return None
+
+            accel_zscore = self.calculate_zscore(accel_series)
+
+            # Positive acceleration = backwardation signal
+            stress_level = self.normalize_to_level(accel_zscore)
+
+            # Current VIX level
+            current_vix = float(vix_data.iloc[-1])
+
+            return IndicatorSignal(
+                indicator_name=self.name,
+                category=self.category,
+                value=vix_acceleration,
+                zscore=accel_zscore,
+                stress_level=stress_level,
+                description=(
+                    f"VIX at {current_vix:.1f}, "
+                    f"acceleration: {vix_acceleration:+.2f} "
+                    f"(z={accel_zscore:+.1f})"
+                ),
+                interpretation=self._interpret(stress_level, accel_zscore, current_vix)
+            )
+
+        except Exception as e:
+            logger.error(f"{self.name} calculation failed: {e}")
+            return None
+
+    def _interpret(self, stress_level: float, accel_zscore: float, vix: float) -> str:
+        """Generate interpretation."""
+        if stress_level > 75:
+            return "🔴 BACKWARDATION: VIX surging (near-term fear spike)"
+        elif stress_level > 60:
+            return "⚠️ FLATTENING: Term structure compressing"
+        elif stress_level < 40:
+            return "✅ CONTANGO: Normal term structure (stable)"
+        else:
+            return "⚡ WATCH: Term structure shifting"
+
+
+class VVIXIndicator(BaseIndicator):
+    """
+    Volatility of volatility indicator.
+
+    VVIX measures expected volatility of VIX itself.
+    Rising VVIX = uncertainty about uncertainty = regime shift risk
+
+    Often spikes before major market dislocations.
+
+    Signal construction:
+    - Calculate VVIX z-score
+    - Calculate VVIX/VIX ratio (normalized vol-of-vol)
+
+    Required data:
+    - ^VVIX: CBOE VVIX Index
+    - ^VIX: VIX for normalization
+    """
+
+    def __init__(self):
+        super().__init__(
+            name="VVIX (Vol-of-Vol)",
+            category=IndicatorCategory.VOLATILITY,
+            description="Volatility of volatility (uncertainty squared)",
+            zscore_window=252
+        )
+
+    def compute(self, data: Dict[str, TimeSeriesData]) -> Optional[IndicatorSignal]:
+        """Compute VVIX signal."""
+        required = ["^VVIX", "^VIX"]
+        if not self._validate_data(data, required):
+            return None
+
+        try:
+            vvix_data = data["^VVIX"].data
+            vix_data = data["^VIX"].data
+
+            # Align dates
+            combined = pd.DataFrame({
+                'vvix': vvix_data,
+                'vix': vix_data
+            }).dropna()
+
+            if len(combined) < self.zscore_window:
+                logger.warning(
+                    f"{self.name}: Insufficient data "
+                    f"({len(combined)} < {self.zscore_window})"
+                )
+                return None
+
+            # VVIX z-score
+            vvix_zscore = self.calculate_zscore(combined['vvix'])
+
+            # VVIX/VIX ratio (normalized vol-of-vol)
+            ratio = combined['vvix'] / combined['vix']
+            ratio_zscore = self.calculate_zscore(ratio)
+
+            # Combined signal: average of VVIX level and ratio
+            combined_zscore = (vvix_zscore + ratio_zscore) / 2
+            stress_level = self.normalize_to_level(combined_zscore)
+
+            current_vvix = float(vvix_data.iloc[-1])
+            current_vix = float(vix_data.iloc[-1])
+            current_ratio = current_vvix / current_vix
+
+            return IndicatorSignal(
+                indicator_name=self.name,
+                category=self.category,
+                value=current_vvix,
+                zscore=combined_zscore,
+                stress_level=stress_level,
+                description=(
+                    f"VVIX at {current_vvix:.1f} "
+                    f"(z={vvix_zscore:+.1f}), "
+                    f"VVIX/VIX ratio: {current_ratio:.2f} "
+                    f"(z={ratio_zscore:+.1f})"
+                ),
+                interpretation=self._interpret(stress_level, vvix_zscore, ratio_zscore)
+            )
+
+        except Exception as e:
+            logger.error(f"{self.name} calculation failed: {e}")
+            return None
+
+    def _interpret(
+        self,
+        stress_level: float,
+        vvix_zscore: float,
+        ratio_zscore: float
+    ) -> str:
+        """Generate interpretation."""
+        if stress_level > 75:
+            return "🔴 EXTREME: Uncertainty about uncertainty (regime shift risk)"
+        elif stress_level > 60:
+            if ratio_zscore > vvix_zscore:
+                return "⚠️ ELEVATED: Vol-of-vol rising faster than vol (instability)"
+            else:
+                return "⚠️ ELEVATED: High volatility uncertainty"
+        elif stress_level < 40:
+            return "✅ STABLE: Low volatility uncertainty"
+        else:
+            return "⚡ WATCH: Volatility uncertainty increasing"
+
+
+class RealizedVolatilityIndicator(BaseIndicator):
+    """
+    Realized volatility spike detector.
+
+    Measures actual market volatility (not implied like VIX).
+    Compares current realized vol to historical levels.
+
+    Signal construction:
+    - Calculate 20-day realized volatility from SPY returns
+    - Z-score vs historical realized vol
+
+    Required data:
+    - SPY: S&P 500 ETF for realized vol calculation
+    """
+
+    def __init__(self):
+        super().__init__(
+            name="Realized Volatility",
+            category=IndicatorCategory.VOLATILITY,
+            description="Actual market volatility (20-day realized)",
+            zscore_window=252
+        )
+
+    def compute(self, data: Dict[str, TimeSeriesData]) -> Optional[IndicatorSignal]:
+        """Compute realized volatility signal."""
+        if not self._validate_data(data, ["SPY"]):
+            return None
+
+        try:
+            spy_data = data["SPY"].data
+
+            if len(spy_data) < self.zscore_window + 20:
+                logger.warning(f"{self.name}: Insufficient data")
+                return None
+
+            # Calculate log returns
+            returns = np.log(spy_data / spy_data.shift(1)).dropna()
+
+            # Calculate 20-day realized volatility (annualized)
+            realized_vol = returns.rolling(20).std() * np.sqrt(252) * 100
+            realized_vol = realized_vol.dropna()
+
+            if len(realized_vol) < self.zscore_window:
+                return None
+
+            # Z-score of current realized vol
+            rvol_zscore = self.calculate_zscore(realized_vol)
+            stress_level = self.normalize_to_level(rvol_zscore)
+
+            current_rvol = float(realized_vol.iloc[-1])
+
+            # Also calculate acceleration (is vol accelerating?)
+            rvol_accel = self.calculate_acceleration(realized_vol, periods=5)
+
+            return IndicatorSignal(
+                indicator_name=self.name,
+                category=self.category,
+                value=current_rvol,
+                zscore=rvol_zscore,
+                stress_level=stress_level,
+                description=(
+                    f"Realized vol: {current_rvol:.1f}% "
+                    f"(z={rvol_zscore:+.1f}), "
+                    f"acceleration: {rvol_accel:+.2f}"
+                ),
+                interpretation=self._interpret(stress_level, rvol_zscore, rvol_accel)
+            )
+
+        except Exception as e:
+            logger.error(f"{self.name} calculation failed: {e}")
+            return None
+
+    def _interpret(
+        self,
+        stress_level: float,
+        rvol_zscore: float,
+        acceleration: float
+    ) -> str:
+        """Generate interpretation."""
+        if stress_level > 75:
+            if acceleration > 0:
+                return "🔴 EXTREME: Volatility surging (accelerating)"
+            else:
+                return "🔴 ELEVATED: High volatility (but decelerating)"
+        elif stress_level > 60:
+            return "⚠️ ELEVATED: Above-normal market volatility"
+        elif stress_level < 40:
+            return "✅ CALM: Low realized volatility"
+        else:
+            return "⚡ WATCH: Volatility moderately elevated"