Chan/ChanMacro/regime_detector.py

"""
regime_detector.py — Market regime detection (V1: 3 states).

★ FACTOR-LOCKED: Regime = f(Price Structure, Breadth, Volatility) — forever.
  Fear, Liquidation, ETF, Funding are Context, NOT regime inputs.
  Adding new factors MUST NOT change regime definition.

★ VERSIONED: regime_version = 'v1_price_breadth_vol'.
  Weight changes → new version. Multiple versions coexist.
  Query: WHERE regime_version = 'v1_price_breadth_vol'.

★ CONFIDENCE-BASED: Each regime gets a continuous score. Highest wins.
  No hard thresholds (prevents boundary oscillation).
"""

from datetime import date as Date
from typing import Optional
from collections import deque

from models import MarketRegime, RegimeResult
from config import config


class RegimeDetector:
    """
    Detects market regime from Price + Breadth + Vol.

    V1: 3 regimes (TREND / RANGE / PANIC)
    V2+: Can split TREND→TREND_UP/TREND_DOWN/EUPHORIA when samples > 500/regime.
    """

    def __init__(self, regime_version: Optional[str] = None):
        self.version = regime_version or config.regime_version
        self.w_price = config.regime_w_price
        self.w_breadth = config.regime_w_breadth
        self.w_vol = config.regime_w_vol
        self.panic_w_anti_trend = config.regime_panic_w_anti_trend
        self.panic_w_vol_extreme = config.regime_panic_w_vol_extreme

        # State persistence
        self._current_regime: Optional[MarketRegime] = None
        self._pending_regime: Optional[MarketRegime] = None
        self._confirmation_count: int = 0
        self._consecutive_days: int = 0
        self._regime_history: deque = deque(maxlen=100)

        # Confirmation: 2 days minimum
        self.MIN_CONFIRMATION = 2

    def load_state(self, db_path: str):
        """Restore regime state from the most recent regime_history record."""
        import sqlite3
        try:
            conn = sqlite3.connect(db_path)
            conn.row_factory = sqlite3.Row
            row = conn.execute(
                "SELECT regime, confidence, confirmation_days, maturity_score "
                "FROM regime_history ORDER BY date DESC LIMIT 1"
            ).fetchone()
            conn.close()

            if row:
                regime_str = row["regime"]
                if regime_str in ("TREND", "RANGE", "PANIC"):
                    self._current_regime = MarketRegime(regime_str)
                    self._consecutive_days = row["confirmation_days"] or 1
        except Exception:
            pass  # DB not initialized yet, use defaults

    def detect(self, price_structure_score: float, breadth_score: float,
               volatility_regime: str, date: Date) -> RegimeResult:
        """
        Detect regime from the 3 locked factors.

        Args:
            price_structure_score: 0-100 from PriceStructureScorer
            breadth_score: 0-100 from BreadthScorer
            volatility_regime: 'LOW_VOL'/'NORMAL_VOL'/'HIGH_VOL'/'EXPLOSIVE_VOL'
            date: Target date
        """
        # ── Compute regime scores ────────────────────────
        # TREND: strong price + strong breadth + non-extreme vol
        trend_score = (
            price_structure_score * self.w_price +
            breadth_score * self.w_breadth +
            self._vol_to_trend(volatility_regime) * self.w_vol
        )

        # RANGE: neutral price + neutral breadth + low vol
        # Score how "range-like" each dimension is
        price_neutral = 60 - abs(price_structure_score - 50)
        breadth_neutral = 60 - abs(breadth_score - 50)
        vol_neutral = 80 if volatility_regime in ("LOW_VOL", "NORMAL_VOL") else 30
        range_score = (
            price_neutral * 0.40 +
            breadth_neutral * 0.40 +
            vol_neutral * 0.20
        )

        # PANIC: very weak trend + extreme vol (NO Fear/Liquidation!)
        anti_trend = 100 - trend_score
        vol_extreme = 100 if volatility_regime == "EXPLOSIVE_VOL" else (
            60 if volatility_regime == "HIGH_VOL" else 20
        )
        panic_score = (
            anti_trend * self.panic_w_anti_trend +
            vol_extreme * self.panic_w_vol_extreme
        )

        scores = {
            MarketRegime.TREND: round(trend_score, 1),
            MarketRegime.RANGE: round(range_score, 1),
            MarketRegime.PANIC: round(panic_score, 1),
        }

        best_regime = max(scores, key=scores.get)

        # ── Persistence check ────────────────────────────
        prior_regime = self._current_regime

        if best_regime == self._current_regime:
            self._consecutive_days += 1
            self._pending_regime = None
            self._confirmation_count = 0
        elif best_regime == self._pending_regime:
            self._confirmation_count += 1
            if self._confirmation_count >= self.MIN_CONFIRMATION:
                # Transition confirmed
                prior_regime = self._current_regime
                self._current_regime = best_regime
                self._consecutive_days = self.MIN_CONFIRMATION
                self._pending_regime = None
                self._confirmation_count = 0
        else:
            self._pending_regime = best_regime
            self._confirmation_count = 1

        # Fallback: if no current regime yet (first run)
        if self._current_regime is None:
            self._current_regime = best_regime
            self._consecutive_days = 1

        # ── Confidence: for the CONFIRMED regime, not the raw best ──
        confirmed_regime = self._current_regime
        confidence = scores[confirmed_regime] / 100.0

        # ── Maturity ─────────────────────────────────────
        maturity = self._compute_maturity(
            trend_score, breadth_score, volatility_regime
        )

        # Track history
        self._regime_history.append({
            "date": date,
            "regime": confirmed_regime.value,
            "confidence": round(confidence, 3),
        })

        return RegimeResult(
            date=date,
            regime=confirmed_regime,
            confidence=round(confidence, 3),
            prior_regime=prior_regime,
            regime_version=self.version,
            maturity_score=round(maturity, 1),
            all_scores={k.value: v for k, v in scores.items()},
            confirmation_days=self._consecutive_days,
        )

    @property
    def current_regime(self) -> Optional[MarketRegime]:
        return self._current_regime

    @property
    def pending_regime(self) -> Optional[MarketRegime]:
        return self._pending_regime

    @property
    def confirmation_progress(self) -> tuple[int, int]:
        """(confirmed_days, required_days) for pending transition."""
        return (self._confirmation_count, self.MIN_CONFIRMATION)

    @staticmethod
    def _vol_to_trend(vol_regime: str) -> float:
        """Convert volatility regime to trend-contributing score."""
        mapping = {
            "LOW_VOL": 50,       # Low vol: neutral for trend
            "NORMAL_VOL": 70,    # Normal vol: good for trend
            "HIGH_VOL": 60,      # High vol: trending but risky
            "EXPLOSIVE_VOL": 30, # Explosive: anti-trend
        }
        return mapping.get(vol_regime, 50)

    @staticmethod
    def _compute_maturity(trend_score: float, breadth_score: float,
                          vol_regime: str) -> float:
        """
        Compute regime maturity: 0-100 continuous.
        0-30: EMERGING (trend accelerating, breadth expanding)
        30-70: CONFIRMED (stable)
        70-100: EXHAUSTING (decelerating, vol abnormal)
        """
        # Trend strength contribution
        trend_contrib = trend_score * 0.50

        # Breadth contribution
        breadth_contrib = breadth_score * 0.30

        # Vol contribution (inverted: low vol = early, explosive = late)
        vol_contrib = {"LOW_VOL": 20, "NORMAL_VOL": 40, "HIGH_VOL": 60, "EXPLOSIVE_VOL": 85}
        vol_val = vol_contrib.get(vol_regime, 50) * 0.20

        return trend_contrib + breadth_contrib + vol_val