| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110 |
- """
- 反脆弱仓位管理器
- 基于改进凯利公式计算最优仓位:
- f* = (p×b - q) / b + γ×Convexity + δ×Regime_Adapt
- """
- from dataclasses import dataclass
- from typing import Dict, Optional, Any
- import numpy as np
- from core.ecosystem import MacroRegime, UnifiedEcosystem
- @dataclass
- class PositionSize:
- """仓位计算结果"""
- base_size: float # 基础仓位
- adjusted_size: float # 调整后的仓位
- kelly_fraction: float # 凯利比例
- convexity_adjustment: float
- regime_adjustment: float
- max_position: float
- reasoning: str
- class AntifragilePositionSizer:
- """
- 反脆弱仓位管理器
- 核心算法:
- 1. 基础凯利公式计算
- 2. 凸性调节(正凸性加分,负凸性减分)
- 3. 生态适应系数调节
- """
- def __init__(
- self,
- kelly_fraction: float = 0.7, # 激进凯利: 70%
- gamma_convexity: float = 0.15,
- max_position: float = 1.0, # 允许满仓
- min_position: float = 0.0
- ):
- self.kelly_fraction = kelly_fraction
- self.gamma_convexity = gamma_convexity
- self.max_position = max_position
- self.min_position = min_position
- # 生态适应系数 - 激进配置
- self.regime_multipliers = {
- MacroRegime.SPRING: 0.9,
- MacroRegime.SUMMER: 1.2, # 夏季超配120%
- MacroRegime.AUTUMN: 0.3,
- MacroRegime.WINTER: 0.0,
- MacroRegime.UNKNOWN: 0.0
- }
- def calculate_position(
- self,
- win_probability: float,
- win_loss_ratio: float,
- ecosystem: Optional[UnifiedEcosystem] = None,
- convexity: float = 0.0
- ) -> PositionSize:
- """
- 计算建议仓位
- Args:
- win_probability: 胜率 (0-1)
- win_loss_ratio: 盈亏比
- ecosystem: 市场生态
- convexity: 凸性特征(正凸性策略为正,负凸性为负)
- Returns:
- PositionSize: 仓位计算结果
- """
- # 基础凯利公式
- q = 1 - win_probability
- if win_loss_ratio <= 0:
- kelly_f = 0.0
- else:
- kelly_f = (win_probability * win_loss_ratio - q) / win_loss_ratio
- kelly_f = max(0, min(1, kelly_f))
- # 应用半凯利
- base_size = kelly_f * self.kelly_fraction
- # 凸性调节
- convexity_adj = self.gamma_convexity * convexity
- # 生态适应调节
- regime_adj = 0.0
- if ecosystem and hasattr(ecosystem, 'macro'):
- regime = ecosystem.macro.regime
- regime_multiplier = self.regime_multipliers.get(regime, 1.0)
- regime_adj = (regime_multiplier - 1.0) * base_size
- # 计算最终仓位
- adjusted_size = base_size + convexity_adj + regime_adj
- adjusted_size = max(self.min_position, min(self.max_position, adjusted_size))
- return PositionSize(
- base_size=base_size,
- adjusted_size=adjusted_size,
- kelly_fraction=kelly_f,
- convexity_adjustment=convexity_adj,
- regime_adjustment=regime_adj,
- max_position=self.max_position,
- reasoning=f"Kelly={kelly_f:.2f}, Convexity={convexity_adj:+.2f}, Regime={regime_adj:+.2f}"
- )
|