myquant/finance/agents/risk_agent.py

"""
RiskAgent — 仓位控制与风险预警。

根据市场波动率、回撤、相关性输出仓位建议和止损线。
"""

import numpy as np
import pandas as pd

from agents.base import BaseAgent


class RiskAgent(BaseAgent):
    """仓位控制 Agent。"""

    name = "Risk"
    description = "仓位控制与风险预警"

    # 风险等级阈值
    THRESHOLDS = {
        "high": {"vol": 35, "dd": -15},
        "medium": {"vol": 25, "dd": -8},
    }

    def execute(
        self,
        holdings: dict[str, float] | None = None,
        market_index: str = "000001.SH",  # 上证指数，非个股
        ts_codes: list[str] | None = None,
    ) -> dict:
        """
        评估市场风险并输出仓位建议。

        参数:
            holdings: {ts_code: 持仓比例}
            market_index: 市场参考标的
            ts_codes: 持仓股票列表

        返回:
            {"risk_level": str, "target_exposure": float, "indicators": dict, "alerts": list}
        """
        holdings = holdings or {}
        price = self.dm.get_daily(market_index)
        if price is None or price.empty:
            return self._default_result()

        price = price.set_index("trade_date").sort_index()
        close = price["close"]
        daily_ret = close.pct_change().dropna()

        # 市场波动率（年化）
        market_vol = float(daily_ret.tail(252).std() * np.sqrt(252) * 100) if len(daily_ret) >= 20 else 30

        # 当前回撤
        peak = close.expanding().max()
        current_dd = float((close.iloc[-1] / peak.iloc[-1] - 1) * 100)

        # 风险等级
        if market_vol > self.THRESHOLDS["high"]["vol"] or current_dd < self.THRESHOLDS["high"]["dd"]:
            risk_level = "high"
            target_exposure = 0.30
        elif market_vol > self.THRESHOLDS["medium"]["vol"] or current_dd < self.THRESHOLDS["medium"]["dd"]:
            risk_level = "medium"
            target_exposure = 0.60
        else:
            risk_level = "low"
            target_exposure = 0.85

        # 最近 N 日涨跌
        ret_5d = float(close.pct_change(5).iloc[-1] * 100) if len(close) >= 6 else 0
        ret_20d = float(close.pct_change(20).iloc[-1] * 100) if len(close) >= 21 else 0

        # 单票上限（风险越高越集中）
        max_single = 0.15 if risk_level == "low" else (0.10 if risk_level == "medium" else 0.05)
        stop_loss = -0.05 if risk_level == "low" else (-0.08 if risk_level == "medium" else -0.12)

        # 持仓预警
        alerts = []
        if current_dd < -10:
            alerts.append(f"市场回撤 {current_dd:.1f}%，考虑减仓")
        if market_vol > 30:
            alerts.append(f"市场波动率 {market_vol:.1f}%，处于高位")
        for code, pct in holdings.items():
            if pct > max_single:
                alerts.append(f"{code} 仓位 {pct:.0%} 超过上限 {max_single:.0%}")

        self.log(f"风险={risk_level} 波动={market_vol:.1f}% 回撤={current_dd:.1f}% 仓位→{target_exposure:.0%}")

        return {
            "risk_level": risk_level,
            "target_exposure": round(target_exposure, 2),
            "max_single_position": round(max_single, 2),
            "stop_loss": round(stop_loss, 2),
            "indicators": {
                "market_volatility": round(market_vol, 1),
                "current_drawdown": round(current_dd, 1),
                "return_5d": round(ret_5d, 1),
                "return_20d": round(ret_20d, 1),
                "close": round(float(close.iloc[-1]), 2),
            },
            "alerts": alerts,
        }

    @staticmethod
    def _default_result() -> dict:
        return {
            "risk_level": "medium",
            "target_exposure": 0.60,
            "max_single_position": 0.10,
            "stop_loss": -0.08,
            "indicators": {},
            "alerts": ["数据不足，使用默认风险参数"],
        }