Security Portfolio Risk

Portfolio Risk Analysis Expert / 组合风险分析专家

English: AI-powered portfolio risk analysis expert — covers VaR calculation, stress testing, tail risk measurement, factor exposure, and risk attribution. Built for fund managers and risk analysts.

中文: 组合风险分析专家——覆盖VaR计算、压力测试、尾部风险度量、因子敞口分析、风险归因。适用：基金经理、风险分析师、机构投资者。

---

证券监管最新动态 [2026-05-25更新]

数据截止: 2026-05-25 | 来源：证监会、NFRA、中证协、安永Q1分析 声明: 以上动态供参考，具体以官方最新发布为准

Industry Pain Points / 行业痛点

---

Trigger Keywords / 触发关键词

English Triggers: portfolio risk, VaR, stress testing, risk decomposition, factor exposure, tail risk, risk attribution, China A-share, fund management, risk management

中文触发词（优先）： 组合风险 / 风险分析 / VaR / 压力测试 / 回撤控制 / 风险归因 / 因子敞口 / 尾部风险 / 风险分解 / 风险预警 / 资产配置 / 分散化 / 相关性分析 / 最大回撤 / 夏普比率 / 波动率 / 风险调整收益 / 风险预算 / VaR计算 / CVaR / ES

---

Core Capabilities / 核心能力

1. VaR & Risk Metrics / VaR与风险指标

import numpy as np
import pandas as pd
from scipy import stats

class PortfolioRiskAnalyzer:
    """组合风险分析引擎"""
    
    def __init__(self, returns: pd.DataFrame, weights: np.ndarray):
        """
        Args:
            returns: 收益率序列（列=资产，行=日期）
            weights: 资产权重向量
        """
        self.returns = returns
        self.weights = weights
        self.n_assets = len(weights)
    
    def calculate_var(self, confidence: float = 0.95, 
                      method: str = "historical") -> dict:
        """计算VaR（Value at Risk）"""
        portfolio_returns = (self.returns * self.weights).sum(axis=1)
        
        if method == "historical":
            var = np.percentile(portfolio_returns, (1 - confidence) * 100)
        elif method == "parametric":
            mu = portfolio_returns.mean()
            sigma = portfolio_returns.std()
            var = stats.norm.ppf(1 - confidence, mu, sigma)
        elif method == "modified":
            # Cornish-Fisher调整
            mu = portfolio_returns.mean()
            sigma = portfolio_returns.std()
            skew = stats.skew(portfolio_returns)
            kurt = stats.kurtosis(portfolio_returns)
            z = stats.norm.ppf(1 - confidence)
            z_cf = (z + (z**2 - 1) * skew / 6 + 
                   (z**3 - 3*z) * kurt / 24 - 
                   (2*z**3 - 5*z) * skew**2 / 36)
            var = mu + sigma * z_cf
        
        return {
            "var": round(var * 100, 2),  # 百分比
            "var_amount": round(var * 1000000, 2),  # 假设100万组合
            "confidence": confidence,
            "method": method,
            "interpretation": f"在{confidence*100}%置信度下，最大损失为{abs(var)*100:.2f}%"
        }
    
    def calculate_cvar(self, confidence: float = 0.95) -> dict:
        """计算CVaR（Conditional VaR / Expected Shortfall）"""
        portfolio_returns = (self.returns * self.weights).sum(axis=1)
        var = np.percentile(portfolio_returns, (1 - confidence) * 100)
        
        cvar = portfolio_returns[portfolio_returns <= var].mean()
        
        return {
            "cvar": round(cvar * 100, 2),
            "cvar_amount": round(cvar * 1000000, 2),
            "interpretation": f"超过VaR时的平均损失为{abs(cvar)*100:.2f}%"
        }
    
    def calculate_max_drawdown(self) -> dict:
        """计算最大回撤"""
        cumulative = (1 + self.returns @ self.weights).cumprod()
        running_max = cumulative.expanding().max()
        drawdown = (cumulative - running_max) / running_max
        
        max_dd = drawdown.min()
        max_dd_end = drawdown.idxmin()
        max_dd_start = cumulative[:max_dd_end].idxmax()
        
        return {
            "max_drawdown": round(max_dd * 100, 2),
            "peak_date": str(max_dd_start.date()),
            "trough_date": str(max_dd_end.date()),
            "recovery_date": None  # 需后续计算
        }
    
    def factor_risk_attribution(self, factor_returns: pd.DataFrame) -> dict:
        """因子风险归因"""
        portfolio_returns = self.returns @ self.weights
        
        # 回归分析
        X = factor_returns.values
        X = np.column_stack([np.ones(len(X)), X])
        y = portfolio_returns.values
        
        coeffs = np.linalg.lstsq(X, y, rcond=None)[0]
        residuals = y - X @ coeffs
        
        # 分解方差
        total_var = np.var(y)
        factor_var = np.var(X[:, 1:] @ coeffs[1:])
        specific_var = np.var(residuals)
        
        return {
            "factor_exposure": {
                "market": round(coeffs[1], 3),
                "factors": {
                    col: round(coef, 3) 
                    for col, coef in zip(factor_returns.columns, coeffs[2:])
                }
            },
            "risk_contribution": {
                "factor_risk": round(factor_var / total_var * 100, 2),
                "specific_risk": round(specific_var / total_var * 100, 2)
            },
            "r_squared": round(1 - specific_var / total_var, 4)
        }

2. Stress Testing / 压力测试

class StressTestScenarios:
    """压力测试情景库"""
    
    SCENARIOS = {
        "2015股灾重演": {
            "description": "假设上证指数单周下跌20%",
            "market_shock": -0.20,
            "sector_impacts": {
                "金融": -0.25,
                "房地产": -0.30,
                "消费": -0.15,
                "科技": -0.20,
                "医药": -0.10
            },
            "liquidity_shock": 0.5  # 流动性降至50%
        },
        
        "利率急升": {
            "description": "假设基准利率上调100bp",
            "rate_shock": 0.01,
            "bond_impact": -0.08,
            "equity_impact": -0.10,
            "bank_impact": -0.05
        },
        
        "人民币急贬": {
            "description": "假设USD/CNY一日升值5%",
            "fx_shock": 0.05,
            "export_related": -0.15,
            "import_related": 0.05,
            "domestic_consumer": -0.08
        },
        
        "黑天鹅-新冠": {
            "description": "类似2020年初疫情冲击",
            "market_shock": -0.12,
            "travel": -0.30,
            "retail": -0.20,
            "healthcare": 0.10,
            "online": 0.05
        }
    }
    
    def run_stress_test(self, portfolio: dict, scenario: str) -> dict:
        """执行压力测试"""
        if scenario not in self.SCENARIOS:
            raise ValueError(f"Unknown scenario: {scenario}")
        
        s = self.SCENARIOS[scenario]
        positions = portfolio["positions"]
        
        stressed_pnl = 0
        stressed_values = []
        
        for pos in positions:
            sector = pos.get("sector", "general")
            weight = pos["weight"]
            
            # 根据情景调整
            if "sector_impacts" in s and sector in s["sector_impacts"]:
                shock = s["sector_impacts"][sector]
            else:
                shock = s.get("market_shock", -0.10)
            
            pos_stressed = weight * (1 + shock)
            stressed_values.append(pos_stressed)
            stressed_pnl += weight * shock
        
        total_value = sum(stressed_values)
        portfolio_stress_loss = total_value - 1  # 假设初始为1
        
        return {
            "scenario": scenario,
            "description": s["description"],
            "portfolio_loss": round(portfolio_stress_loss * 100, 2),
            "portfolio_value_after": round(total_value * 100, 2),
            "position_impacts": [
                {"name": pos["name"], "weight": pos["weight"], 
                 "shock": round(shock * 100, 2), "impact": "loss" if shock < 0 else "gain"}
                for pos, shock in zip(positions, 
                    [s.get("sector_impacts", {}).get(pos.get("sector", ""), 
                     s.get("market_shock", -0.10)) for pos in positions])
            ]
        }

3. Risk Contribution Analysis / 风险贡献分析

    def risk_contribution_by_asset(self) -> dict:
        """计算各资产风险贡献"""
        cov_matrix = self.returns.cov()
        portfolio_vol = np.sqrt(self.weights @ cov_matrix.values @ self.weights)
        
        # 边际风险贡献 (MCTR)
        mctr = (cov_matrix.values @ self.weights) / portfolio_vol
        
        # 风险贡献
        risk_contrib = self.weights * mctr
        
        return {
            "portfolio_volatility": round(portfolio_vol * 100, 2),
            "asset_risk_contribution": {
                self.returns.columns[i]: round(rc * 100, 2)
                for i, rc in enumerate(risk_contrib)
            },
            "concentration_risk": {
                "max_concentration": round(max(risk_contrib) * 100, 2),
                "diversification_benefit": round(
                    (sum([self.returns[col].std() * w 
                         for col, w in zip(self.returns.columns, self.weights)]) - 
                     portfolio_vol) * 100, 2)
            }
        }

---

Quick Command Templates / 快速指令模板

组合风险评估：

分析以下组合的风险：
- 总规模：1000万
- 持仓：[股票A 30%, 股票B 20%, 债券B 50%]
- 置信度：95%

压力测试：

执行"2015股灾重演"情景压力测试

---

Disclaimer

This skill provides risk analysis tools for educational purposes. Risk metrics are based on historical data and statistical models, which do not guarantee future accuracy. Investment decisions should be made based on comprehensive analysis and professional advice.