Security Portfolio Risk — 安全投资组合风险
v1.0.0为基金经理提供针对中国A股的AI驱动的投资组合风险分析,包括VaR、压力测试、尾部风险、因子暴露和风险归因。
运行时依赖
安装命令
点击复制技能文档
组合风险分析专家——覆盖VaR计算、压力测试、尾部风险度量、因子敞口分析、风险归因。适用:基金经理、风险分析师、机构投资者。
行业痛点 / 行业痛点 痛点 / 痛点 影响 / 影响 解决方案 / 本Skill解决方案 系统风险难预测 黑天鹅事件导致大幅回撤 极端情景压力测试+尾部风险分析 因子敞口不清晰 不知道组合暴露在哪些风险上 因子归因模型+敞口分解 回撤控制困难 持有人体验差,资金赎回压力 动态回撤监控+预警机制 相关性突变 平时低相关的资产大跌时齐跌 相关性压力测试+分散化效果评估 合规要求高 资管新规净值化要求 标准风险指标+监管报告
触发关键词 / 触发关键词 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
核心能力 / 核心能力
- VaR & Risk Metrics / VaR与风险指标
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 - 3z) kurt / 24 - (2z*3 - 5z) skew2 / 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"在{confidence100}%置信度下,最大损失为{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) }
- Stress Testing / 压力测试