cyb50-quant/cyb50-pro/agents/volatility_seller/agent.py

"""
波动率卖家智能体

基于IV Rank分析和卖方策略生成信号
最佳生态：夏季繁荣、春季复苏（避开秋季）
注意：需要期权数据支持，当前为预留接口
"""

from datetime import datetime, timedelta
from typing import Dict, Any, Optional

import numpy as np
import pandas as pd

from agents.base import AgentBase, AgentSignal, SignalDirection, SignalStrength
from core.ecosystem import MacroRegime, UnifiedEcosystem


class VolatilitySellerAgent(AgentBase):
    """
    波动率卖家智能体

    策略逻辑：
    1. 当隐含波动率（IV Rank）> 80%时，做空波动率
    2. 使用卖方策略：卖出宽跨式（Strangle）或铁鹰（Iron Condor）
    3. 避开秋季（高波动不确定期）
    4. 收取权利金，利用波动率均值回归

    注意：此策略需要期权数据支持
    """

    def __init__(
        self,
        config: Optional[Dict[str, Any]] = None,
        iv_rank_high: float = 80,
        iv_rank_low: float = 20,
        strategy_type: str = "strangle",  # "straddle", "strangle", "iron_condor"
        delta_target: float = 0.30,
        dte_target: int = 30  # 到期日目标（天）
    ):
        super().__init__(
            name="volatility_seller",
            config=config,
            max_position=0.4,
            min_confidence=0.70
        )

        self.iv_rank_high = iv_rank_high
        self.iv_rank_low = iv_rank_low
        self.strategy_type = strategy_type
        self.delta_target = delta_target
        self.dte_target = dte_target

        self.preferred_regimes = [MacroRegime.SUMMER, MacroRegime.SPRING]
        self.avoid_regimes = [MacroRegime.AUTUMN]

        # 检查是否有期权数据
        self.has_option_data = False

    def generate_signal(
        self,
        price_data: pd.DataFrame,
        ecosystem: Optional[UnifiedEcosystem] = None,
        option_data: Optional[Dict] = None
    ) -> Optional[AgentSignal]:
        """生成卖方信号"""
        if len(price_data) < 60:
            return None

        # 检查是否在避免的生态
        if ecosystem and hasattr(ecosystem, 'macro'):
            if ecosystem.macro.regime in self.avoid_regimes:
                return None

        # 计算IV Rank（使用历史波动率作为代理）
        iv_rank = self._calculate_iv_rank(price_data)

        # 检查是否满足做空波动率条件
        if iv_rank < self.iv_rank_high:
            return None

        # 检查趋势（避免在强趋势市场做卖方）
        trend_strength = self._calculate_trend_strength(price_data)
        if trend_strength > 0.7:  # 强趋势市场风险高
            return None

        direction = SignalDirection.NEUTRAL  # 卖方策略方向中性
        confidence = min(1.0, (iv_rank - self.iv_rank_high) / 20 + 0.5)

        if not self._validate_signal(direction, confidence, ecosystem):
            return None

        position_size = self._calculate_position_size(ecosystem, confidence, iv_rank)

        strength = (
            SignalStrength.STRONG if confidence > 0.85
            else SignalStrength.MEDIUM if confidence > 0.75
            else SignalStrength.WEAK
        )

        signal = AgentSignal(
            agent_name=self.name,
            direction=direction,
            strength=strength,
            confidence=confidence,
            suggested_position=position_size,
            expected_return=self.get_expected_return(price_data, ecosystem),
            win_probability=self.get_win_probability(price_data, ecosystem),
            timestamp=datetime.now(),
            valid_until=datetime.now() + timedelta(days=self.dte_target),
            metadata={
                "iv_rank": iv_rank,
                "strategy_type": self.strategy_type,
                "delta_target": self.delta_target,
                "dte_target": self.dte_target,
                "trend_strength": trend_strength,
                "note": "Option strategy - requires options data"
            }
        )

        self.current_signal = signal
        self.signal_history.append(signal)

        return signal

    def get_expected_return(
        self,
        price_data: pd.DataFrame,
        ecosystem: Optional[UnifiedEcosystem] = None
    ) -> float:
        """计算预期收益（卖方策略收益稳定但有限）"""
        # 卖方策略通常每月收取1-3%权利金
        monthly_premium = 0.02
        annual_return = (1 + monthly_premium) ** 12 - 1

        return annual_return

    def get_win_probability(
        self,
        price_data: pd.DataFrame,
        ecosystem: Optional[UnifiedEcosystem] = None
    ) -> float:
        """计算胜率（卖方策略胜率通常较高，70-80%）"""
        base_prob = 0.70

        iv_rank = self._calculate_iv_rank(price_data)
        if iv_rank > 85:
            base_prob += 0.10
        elif iv_rank > 80:
            base_prob += 0.05

        # 避开秋季降低胜率预期
        if ecosystem and ecosystem.macro.regime == MacroRegime.AUTUMN:
            base_prob -= 0.15

        return min(0.85, base_prob)

    def _calculate_iv_rank(self, data: pd.DataFrame) -> float:
        """
        计算IV Rank

        使用历史波动率作为隐含波动率的代理
        IV Rank = (当前IV - 52周最低IV) / (52周最高IV - 52周最低IV)
        """
        if len(data) < 252:
            lookback = len(data) // 2
        else:
            lookback = 252

        # 计算历史波动率
        returns = data['close'].pct_change().dropna()
        current_vol = returns.iloc[-20:].std() * np.sqrt(252)

        # 滚动计算历史波动率
        rolling_vol = returns.rolling(20).std() * np.sqrt(252)
        historical_vol = rolling_vol.iloc[-lookback:]

        vol_min = historical_vol.min()
        vol_max = historical_vol.max()

        if vol_max == vol_min:
            return 50.0

        iv_rank = (current_vol - vol_min) / (vol_max - vol_min) * 100
        return iv_rank

    def _calculate_trend_strength(self, data: pd.DataFrame) -> float:
        """计算趋势强度（用于避开关强趋势市场）"""
        if len(data) < 20:
            return 0.0

        # 使用ADX作为趋势强度代理
        high = data['high']
        low = data['low']
        close = data['close']

        plus_dm = high.diff().clip(lower=0)
        minus_dm = (-low.diff()).clip(lower=0)

        tr = pd.concat([high - low, (high - close.shift(1)).abs(), (low - close.shift(1)).abs()], axis=1).max(axis=1)
        atr = tr.rolling(14).mean()

        plus_di = 100 * (plus_dm.rolling(14).mean() / atr)
        minus_di = 100 * (minus_dm.rolling(14).mean() / atr)

        dx = 100 * (plus_di - minus_di).abs() / (plus_di + minus_di)
        adx = dx.rolling(14).mean()

        current_adx = adx.iloc[-1] if not pd.isna(adx.iloc[-1]) else 20

        return min(1.0, current_adx / 50)

    def _calculate_position_size(
        self,
        ecosystem: Optional[UnifiedEcosystem],
        confidence: float,
        iv_rank: float
    ) -> float:
        """计算建议仓位（卖方策略仓位保守）"""
        base_size = self.max_position * confidence * 0.5

        # IV越高，仓位越小（风险控制）
        if iv_rank > 90:
            base_size *= 0.7
        elif iv_rank > 85:
            base_size *= 0.85

        if ecosystem:
            if ecosystem.macro.regime in self.preferred_regimes:
                base_size *= 1.0
            else:
                base_size *= 0.6

        return min(self.max_position, base_size)

    def _check_regime_match(self, ecosystem: Any) -> float:
        """检查生态适配度"""
        if not ecosystem or not hasattr(ecosystem, 'macro'):
            return 0.7

        if ecosystem.macro.regime in self.avoid_regimes:
            return 0.3
        elif ecosystem.macro.regime in self.preferred_regimes:
            return 1.1
        else:
            return 0.8