cyb50-quant/cyb50-pro/risk/manager.py

"""
风险管理体系

整合仓位管理、风险预算、硬止损等功能
"""

from dataclasses import dataclass
from typing import Dict, List, Optional, Any
from datetime import datetime

from .position_sizing import AntifragilePositionSizer, PositionSize
from .budget import RiskBudgetManager, BudgetState


@dataclass
class RiskCheckResult:
    """风险检查结果"""
    can_trade: bool
    allowed_position: float
    risk_level: str
    warnings: List[str]
    stop_loss_price: Optional[float] = None


class RiskManager:
    """
    风险管理者

    整合所有风险管理组件：
    - 仓位管理
    - 风险预算
    - 组合级硬止损
    """

    def __init__(
        self,
        max_drawdown_stop: float = 0.15,
        max_daily_loss: float = 0.05,
        position_sizer: Optional[AntifragilePositionSizer] = None,
        budget_manager: Optional[RiskBudgetManager] = None
    ):
        self.max_drawdown_stop = max_drawdown_stop
        self.max_daily_loss = max_daily_loss
        self.position_sizer = position_sizer or AntifragilePositionSizer()
        self.budget_manager = budget_manager or RiskBudgetManager()

        # 状态追踪
        self.peak_value: float = 0.0
        self.current_drawdown: float = 0.0
        self.daily_pnl: float = 0.0
        self.last_reset_date: Optional[datetime] = None

    def check_trade_permission(
        self,
        account_value: float,
        proposed_position: float,
        entry_price: float
    ) -> RiskCheckResult:
        """
        检查交易权限

        Returns:
            RiskCheckResult: 检查结果
        """
        warnings = []

        # 1. 检查风险预算
        if not self.budget_manager.check_can_trade():
            return RiskCheckResult(
                can_trade=False,
                allowed_position=0.0,
                risk_level="high",
                warnings=["风险预算已耗尽，处于冷却/停止期"]
            )

        # 2. 检查最大回撤 - 激进模式：允许50%仓位继续交易
        if self.current_drawdown <= -self.max_drawdown_stop:
            warnings.append(f"最大回撤已达到{self.max_drawdown_stop:.1%}，限制仓位")
            proposed_position *= 0.5  # 允许50%仓位
            risk_level = "critical"

        # 3. 检查单日亏损
        if self.daily_pnl <= -account_value * self.max_daily_loss:
            warnings.append(f"单日亏损已达到{self.max_daily_loss:.1%}，建议减仓")
            proposed_position *= 0.5

        # 4. 确定风险等级
        if self.current_drawdown <= -0.10:
            risk_level = "high"
        elif self.current_drawdown <= -0.05:
            risk_level = "medium"
        else:
            risk_level = "low"

        # 计算止损价格
        stop_loss = entry_price * 0.98 if proposed_position > 0 else None

        return RiskCheckResult(
            can_trade=True,
            allowed_position=proposed_position,
            risk_level=risk_level,
            warnings=warnings,
            stop_loss_price=stop_loss
        )

    def update_account_state(
        self,
        account_value: float,
        daily_pnl: Optional[float] = None
    ):
        """更新账户状态"""
        today = datetime.now().date()

        # 重置每日盈亏
        if self.last_reset_date != today:
            self.daily_pnl = 0.0
            self.last_reset_date = today

        if daily_pnl is not None:
            self.daily_pnl += daily_pnl

        # 更新峰值和回撤
        if account_value > self.peak_value:
            self.peak_value = account_value

        if self.peak_value > 0:
            self.current_drawdown = (account_value - self.peak_value) / self.peak_value

    def record_trade_result(self, trade_result: Dict):
        """记录交易结果"""
        # 更新风险预算
        if "realized_loss" in trade_result:
            self.budget_manager.record_trade_result(
                trade_result.get("trade_id", "unknown"),
                trade_result["realized_loss"]
            )

        # 更新账户状态
        if "pnl" in trade_result:
            self.update_account_state(
                trade_result.get("account_value", 0),
                trade_result["pnl"]
            )

    def get_risk_summary(self) -> Dict[str, Any]:
        """获取风险摘要"""
        return {
            "current_drawdown": self.current_drawdown,
            "daily_pnl": self.daily_pnl,
            "max_drawdown_stop": self.max_drawdown_stop,
            "budget_summary": self.budget_manager.get_budget_summary()
        }