""" 波动率卖家智能体 基于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