cyb50-quant/cyb50-pro/core/ecosystem/macro.py

"""
宏观生态识别器
识别市场的"季节"：春季（复苏）、夏季（繁荣）、秋季（分化）、冬季（萧条）
"""

from dataclasses import dataclass
from typing import Dict, List, Optional, Tuple
from enum import Enum

import numpy as np
import pandas as pd


class MacroRegime(Enum):
    """宏观生态类型"""
    SPRING = "spring"      # 春季（复苏）
    SUMMER = "summer"      # 夏季（繁荣）
    AUTUMN = "autumn"      # 秋季（分化）
    WINTER = "winter"      # 冬季（萧条）
    UNKNOWN = "unknown"    # 未知/过渡


@dataclass
class MacroEcosystem:
    """宏观生态数据结构"""
    regime: MacroRegime
    confidence: float
    volatility_trend: float      # 波动率趋势
    volume_trend: float          # 成交量趋势
    dispersion: float            # 板块离散度
    adx_value: float             # ADX值
    description: str


class MacroEcosystemIdentifier:
    """
    宏观生态识别器

    基于以下因子识别市场季节：
    1. 波动率期限结构（近月波动率变化）
    2. 板块离散度（成分股收益率标准差的历史分位）
    3. 资金流向广度（涨跌家数比的熵值）
    4. ADX趋势强度
    """

    def __init__(
        self,
        lookback_days: int = 252,
        spring_vol_rebound: float = 0.20,
        spring_volume_ratio: float = 1.2,
        summer_adx_threshold: float = 25,
        summer_adx_duration: int = 5,
        autumn_dispersion_pct: float = 70,
        winter_volume_pct: float = 50,
        winter_vol_compression: float = 20
    ):
        self.lookback_days = lookback_days
        self.spring_vol_rebound = spring_vol_rebound
        self.spring_volume_ratio = spring_volume_ratio
        self.summer_adx_threshold = summer_adx_threshold
        self.summer_adx_duration = summer_adx_duration
        self.autumn_dispersion_pct = autumn_dispersion_pct
        self.winter_volume_pct = winter_volume_pct
        self.winter_vol_compression = winter_vol_compression

    def identify(
        self,
        price_data: pd.DataFrame,
        sector_data: Optional[pd.DataFrame] = None
    ) -> MacroEcosystem:
        """
        识别当前宏观生态

        Args:
            price_data: 价格数据，包含 ['open', 'high', 'low', 'close', 'volume']
            sector_data: 板块数据（可选），成分股收益率数据

        Returns:
            MacroEcosystem: 宏观生态识别结果
        """
        # 计算技术指标
        volatility = self._calculate_volatility(price_data)
        volume_ma5 = price_data['volume'].rolling(5).mean().iloc[-1]
        volume_ma20 = price_data['volume'].rolling(20).mean().iloc[-1]
        volume_ratio = volume_ma5 / volume_ma20 if volume_ma20 > 0 else 1.0

        adx_value = self._calculate_adx(price_data)
        dispersion = self._calculate_dispersion(sector_data) if sector_data is not None else 50.0

        # 计算历史分位数
        vol_percentile = self._get_percentile(volatility, price_data)
        vol_low_30 = np.percentile(volatility.iloc[-60:], 30)
        vol_rebound = (volatility.iloc[-1] - vol_low_30) / vol_low_30 if vol_low_30 > 0 else 0

        # 四季识别逻辑
        regime_scores = {}

        # 春季：波动率从低位回升，成交量温和放大
        spring_score = self._score_spring(vol_rebound, volume_ratio, dispersion)
        regime_scores[MacroRegime.SPRING] = spring_score

        # 夏季：趋势明确，波动率稳定，情绪高涨
        summer_score = self._score_summer(adx_value, volatility, price_data, dispersion)
        regime_scores[MacroRegime.SUMMER] = summer_score

        # 秋季：板块离散度大，轮动加速
        autumn_score = self._score_autumn(dispersion, volatility, price_data)
        regime_scores[MacroRegime.AUTUMN] = autumn_score

        # 冬季：成交量萎缩，波动率压缩
        winter_score = self._score_winter(volume_ratio, vol_percentile, adx_value)
        regime_scores[MacroRegime.WINTER] = winter_score

        # 选择最高分的生态
        best_regime = max(regime_scores, key=regime_scores.get)
        best_score = regime_scores[best_regime]

        # 如果最高分低于阈值，标记为未知
        if best_score < 0.3:
            best_regime = MacroRegime.UNKNOWN
            best_score = 0.0

        return MacroEcosystem(
            regime=best_regime,
            confidence=best_score,
            volatility_trend=vol_rebound,
            volume_trend=volume_ratio - 1.0,
            dispersion=dispersion,
            adx_value=adx_value,
            description=self._get_description(best_regime)
        )

    def _calculate_volatility(self, data: pd.DataFrame, period: int = 20) -> pd.Series:
        """计算波动率（20日收益率标准差年化）"""
        returns = data['close'].pct_change()
        volatility = returns.rolling(period).std() * np.sqrt(252)
        return volatility

    def _calculate_adx(self, data: pd.DataFrame, period: int = 14) -> float:
        """计算ADX（平均趋向指数）"""
        high = data['high']
        low = data['low']
        close = data['close']

        # +DM和-DM
        plus_dm = high.diff()
        minus_dm = -low.diff()

        plus_dm[plus_dm < 0] = 0
        minus_dm[minus_dm < 0] = 0

        # TR（真实波幅）
        tr1 = high - low
        tr2 = abs(high - close.shift(1))
        tr3 = abs(low - close.shift(1))
        tr = pd.concat([tr1, tr2, tr3], axis=1).max(axis=1)

        # ATR
        atr = tr.rolling(period).mean()

        # +DI和-DI
        plus_di = 100 * (plus_dm.rolling(period).mean() / atr)
        minus_di = 100 * (minus_dm.rolling(period).mean() / atr)

        # DX和ADX
        dx = 100 * abs(plus_di - minus_di) / (plus_di + minus_di)
        adx = dx.rolling(period).mean()

        return adx.iloc[-1] if not pd.isna(adx.iloc[-1]) else 20.0

    def _calculate_dispersion(self, sector_data: pd.DataFrame) -> float:
        """计算板块离散度（成分股收益率标准差的历史分位）"""
        if sector_data is None or sector_data.empty:
            return 50.0

        # 计算各成分股近期收益率
        returns = sector_data.pct_change().iloc[-20:]
        cross_sectional_std = returns.std(axis=1).mean()

        # 转换为历史分位数（简化处理）
        historical_std = returns.std(axis=1).rolling(60).mean()
        if len(historical_std) > 0 and not pd.isna(historical_std.iloc[-1]):
            percentile = 50 + (cross_sectional_std - historical_std.mean()) / historical_std.std() * 25
            return max(0, min(100, percentile))
        return 50.0

    def _get_percentile(self, series: pd.Series, data: pd.DataFrame) -> float:
        """获取当前值的历史分位数"""
        if len(series) < 60:
            return 50.0
        current = series.iloc[-1]
        historical = series.iloc[-252:-1].dropna()
        if len(historical) == 0:
            return 50.0
        return (historical < current).mean() * 100

    def _score_spring(self, vol_rebound: float, volume_ratio: float, dispersion: float) -> float:
        """评分：春季（复苏）"""
        # 波动率从低位回升
        vol_score = min(1.0, vol_rebound / self.spring_vol_rebound)

        # 成交量放大
        volume_score = min(1.0, max(0, (volume_ratio - 1.0) / (self.spring_volume_ratio - 1.0)))

        # 板块离散度中等（30-60分位）
        dispersion_score = 1.0 - abs(dispersion - 45) / 15 if 30 <= dispersion <= 60 else 0.0

        return vol_score * 0.4 + volume_score * 0.4 + dispersion_score * 0.2

    def _score_summer(self, adx: float, volatility: pd.Series, data: pd.DataFrame, dispersion: float) -> float:
        """评分：夏季（繁荣）"""
        # ADX高于阈值
        adx_score = min(1.0, adx / self.summer_adx_threshold) if adx > 20 else 0.0

        # 波动率稳定（历史40-70分位）
        vol_percentile = self._get_percentile(volatility, data)
        vol_score = 1.0 - abs(vol_percentile - 55) / 15 if 40 <= vol_percentile <= 70 else 0.0

        # 板块离散度低（同涨同跌）
        dispersion_score = 1.0 - dispersion / 40 if dispersion < 40 else 0.0

        return adx_score * 0.5 + vol_score * 0.3 + dispersion_score * 0.2

    def _score_autumn(self, dispersion: float, volatility: pd.Series, data: pd.DataFrame) -> float:
        """评分：秋季（分化）"""
        # 板块离散度高
        dispersion_score = min(1.0, (dispersion - 50) / 30) if dispersion > 50 else 0.0

        # 波动率上升
        vol_recent = volatility.iloc[-5:].mean()
        vol_prev = volatility.iloc[-20:-5].mean()
        vol_increase = (vol_recent - vol_prev) / vol_prev if vol_prev > 0 else 0
        vol_score = min(1.0, vol_increase / 0.15)

        return dispersion_score * 0.6 + vol_score * 0.4

    def _score_winter(self, volume_ratio: float, vol_percentile: float, adx: float) -> float:
        """评分：冬季（萧条）"""
        # 成交量萎缩
        volume_score = 1.0 - min(1.0, volume_ratio) if volume_ratio < 1.0 else 0.0

        # 波动率压缩至历史低位
        vol_score = 1.0 - vol_percentile / self.winter_vol_compression if vol_percentile < self.winter_vol_compression else 0.0

        # 无明确趋势
        adx_score = 1.0 - min(1.0, adx / 20) if adx < 25 else 0.0

        return volume_score * 0.3 + vol_score * 0.4 + adx_score * 0.3

    def _get_description(self, regime: MacroRegime) -> str:
        """获取生态描述"""
        descriptions = {
            MacroRegime.SPRING: "春季复苏：波动率回升，成交温和放大，机构左侧布局",
            MacroRegime.SUMMER: "夏季繁荣：趋势明确，情绪高涨，散户机构共振",
            MacroRegime.AUTUMN: "秋季分化：板块离散，轮动加速，量化游资主导",
            MacroRegime.WINTER: "冬季萧条：成交萎缩，波动压缩，空头主导",
            MacroRegime.UNKNOWN: "过渡状态：生态不明，建议观望"
        }
        return descriptions.get(regime, "未知状态")