Chan/ChanMacro/scoring/volatility_regime.py

"""
scoring/volatility_regime.py — Volatility Regime Classification.

4 regimes from OHLCV data:
  LOW_VOL:       ATR/Close < 2% → compression, breakout imminent
  NORMAL_VOL:    ATR/Close 2-5% → normal trading
  HIGH_VOL:      ATR/Close 5-10% → trend acceleration, wider stops
  EXPLOSIVE_VOL: ATR/Close > 10% → extreme, reduce or wait

Uses: ATR(14)/Close, HV(20)/HV(60) ratio, BB width ratio.
OHLCV-only — never goes offline.
"""

from datetime import date as Date
import sqlite3
import numpy as np
import pandas as pd

from .base import BaseScorer
from .constants import (
    VOL_LOW, VOL_HIGH, VOL_REGIME_SCORES, HV_RATIO_LOW, HV_RATIO_HIGH,
)
from models import FactorScore, VolatilityRegimeScore, VolRegime, MacroDirection
from config import config


class VolatilityRegimeScorer(BaseScorer):
    """Classifies volatility regime from OHLCV data."""

    def compute(self, target_date: Date) -> VolatilityRegimeScore:
        conn = self.get_connection()
        try:
            df = pd.read_sql_query(
                "SELECT * FROM ohlcv_daily WHERE date <= ? ORDER BY date DESC LIMIT 120",
                conn, params=(str(target_date),)
            )
            if df.empty:
                return VolatilityRegimeScore(
                    name="Volatility Regime",
                    score=50.0,
                    label="No Data",
                )

            df = df.sort_values("date").reset_index(drop=True)

            # 1. ATR/Close %
            latest = df.iloc[-1]
            atr = latest.get("atr_14")
            close = float(latest["close"])
            atr_pct = (atr / close * 100) if atr and not pd.isna(atr) and close > 0 else 3.0

            # 2. HV(20) / HV(60) ratio
            hv_ratio = self._compute_hv_ratio(df)

            # 3. BB width ratio
            bb_ratio = self._compute_bb_ratio(df)

            # Classify regime
            regime = self._classify(atr_pct, hv_ratio, bb_ratio)

            # Score
            score = VOL_REGIME_SCORES.get(regime.value, 50)

            # Narrative
            narrative = self._build_narrative(regime, atr_pct, hv_ratio, bb_ratio)

            return VolatilityRegimeScore(
                name="Volatility Regime",
                score=float(score),
                label=regime.value,
                direction=MacroDirection.NEUTRAL,
                vol_regime=regime,
                atr_pct=round(atr_pct, 2),
                hv_ratio=round(hv_ratio, 2),
                bb_width_ratio=round(bb_ratio, 2),
                sub_scores={
                    "atr_pct": round(atr_pct, 2),
                    "hv_ratio": round(hv_ratio, 2),
                    "bb_width_ratio": round(bb_ratio, 2),
                },
                narrative=narrative,
            )
        finally:
            conn.close()

    def _compute_hv_ratio(self, df: pd.DataFrame) -> float:
        """Compute HV(20) / HV(60) ratio."""
        closes = df["close"].astype(float)
        returns = closes.pct_change().dropna()

        if len(returns) < 60:
            return 1.0

        hv20 = returns.tail(20).std() * np.sqrt(365) * 100
        hv60 = returns.tail(60).std() * np.sqrt(365) * 100

        if hv60 == 0:
            return 1.0

        return hv20 / hv60

    def _compute_bb_ratio(self, df: pd.DataFrame) -> float:
        """Compute current BB width / 20d average BB width."""
        bb_widths = df["bb_width"].dropna().tail(40)
        if len(bb_widths) < 20:
            return 1.0

        current = bb_widths.iloc[-1]
        avg = bb_widths.tail(20).mean()
        if avg == 0:
            return 1.0

        return current / avg

    @staticmethod
    def _classify(atr_pct: float, hv_ratio: float, bb_ratio: float) -> VolRegime:
        """Classify volatility regime from multiple indicators."""
        # Primary: ATR/Close %
        if atr_pct > 10.0:
            return VolRegime.EXPLOSIVE_VOL
        elif atr_pct > VOL_HIGH:
            return VolRegime.HIGH_VOL
        elif atr_pct < VOL_LOW:
            return VolRegime.LOW_VOL

        # Secondary: HV ratio and BB ratio for edge cases
        if hv_ratio > HV_RATIO_HIGH and bb_ratio > 1.3:
            return VolRegime.HIGH_VOL
        elif hv_ratio < HV_RATIO_LOW and bb_ratio < 0.8:
            return VolRegime.LOW_VOL

        return VolRegime.NORMAL_VOL

    @staticmethod
    def _build_narrative(regime: VolRegime, atr_pct: float,
                         hv_ratio: float, bb_ratio: float) -> str:
        mapping = {
            VolRegime.LOW_VOL: f"低波动(ATR={atr_pct:.1f}%), 布林带收窄, 突破前兆",
            VolRegime.NORMAL_VOL: f"正常波动(ATR={atr_pct:.1f}%), 正常交易环境",
            VolRegime.HIGH_VOL: f"高波动(ATR={atr_pct:.1f}%), 趋势加速, 放宽止损",
            VolRegime.EXPLOSIVE_VOL: f"极端波动(ATR={atr_pct:.1f}%), 减仓或等待",
        }
        return mapping.get(regime, "Unknown")