agents24 量化交易插件中的风险指标计算:VaR、CVaR、回撤与压力测试的完整 Python 实现指南
2026/9/10 16:11:20 网站建设 项目流程

agents24 量化交易插件中的风险指标计算:VaR、CVaR、回撤与压力测试的完整 Python 实现指南

【免费下载链接】agentsMulti-harness agentic plugin marketplace for Claude Code, Codex, Cursor, OpenCode, GitHub Copilot, and Google Antigravity项目地址: https://gitcode.com/GitHub_Trending/agents24/agents

本篇技术指南围绕 agents24 仓库中quantitative-trading插件的risk-metrics-calculation技能展开,完整讲解单资产风险度量、组合风险分解、滚动风险监控与压力测试四大实现模式的 Python 代码细节。读完本文,你将掌握基于 NumPy / pandas / SciPy 编写一套可用于组合管理、风险限额、风险看板与监管报送的可复用风险计算工具箱,并理解每个指标的年化逻辑、边界条件与适用场景。

本文的全部代码与配置均来自 details.md,并与同插件的 SKILL.md、risk-manager.md 等仓库文档相互印证。

一、技能定位:这个风险指标库解决什么问题

risk-metrics-calculation是 agents24 仓库中quantitative-trading插件下的一个 Skill,其定位是"组合管理的综合性风险度量工具箱",覆盖 Value at Risk(VaR)、Expected Shortfall(CVaR)与回撤分析。对应 Skill 的元信息描述为:

Calculate portfolio risk metrics including VaR, CVaR, Sharpe, Sortino, and drawdown analysis. Use when measuring portfolio risk, implementing risk limits, or building risk monitoring systems.

该技能在 SKILL.md 中明确了典型使用场景:

  • 度量组合风险(Measuring portfolio risk)
  • 实施风险限额(Implementing risk limits)
  • 构建风险看板(Building risk dashboards)
  • 计算风险调整后收益(Calculating risk-adjusted returns)
  • 设定仓位大小(Setting position sizes)
  • 监管报送(Regulatory reporting)

在 agent 层,该插件提供两个角色配合使用:quant-analyst.md(opus,负责金融建模、交易策略与风险指标计算)与 risk-manager.md(sonnet,负责组合风险监控、R-multiple 跟踪与头寸限额)。risk-manager 的职责清单中明确包含"Value at Risk (VaR) calculations""Correlation and beta analysis""Stress testing and scenario analysis""Risk-adjusted performance metrics",与本文要实现的指标一一对应。

安装该插件的方式为(见 docs/plugins.md):

/plugin install quantitative-trading

下文将逐一展开 details.md 中四个实现模式:核心风险指标、组合风险、滚动风险指标与压力测试,最后给出快速使用参考与最佳实践清单。

二、风险指标的分类框架与时间尺度(继承 SKILL.md 核心概念)

在进入代码之前,先建立分类框架。SKILL.md 将风险指标划分为四个类别:

类别(Category)指标(Metrics)用途(Use Case)
波动率(Volatility)标准差(Std Dev)、Beta一般风险度量
尾部风险(Tail Risk)VaR、CVaR极端亏损度量
回撤(Drawdown)最大回撤(Max DD)、Calmar资本保全
风险调整收益(Risk-Adjusted)Sharpe、Sortino绩效评估

此外,指标的时间尺度应与投资决策周期匹配:

Intraday: Minute/hourly VaR for day traders(日内:分钟/小时级 VaR,用于日内交易者) Daily: Standard risk reporting(日频:标准风险报告) Weekly: Rebalancing decisions(周频:再平衡决策) Monthly: Performance attribution(月频:业绩归因) Annual: Strategic allocation(年频:战略配置)

代码中默认的年化因子ann_factor = 252即建立在"日频收益、一年 252 个交易日"的假设之上——这是整套实现最重要的前提假设,涉及任何年化计算时都应显式确认。

三、Pattern 1:单资产核心风险指标(RiskMetrics 类)

RiskMetrics是整套风险库的地基,负责对单一收益序列计算波动率、VaR/CVaR、回撤与风险调整收益。构造函数如下:

import numpy as np import pandas as pd from scipy import stats from typing import Dict, Optional, Tuple class RiskMetrics: """Core risk metric calculations.""" def __init__(self, returns: pd.Series, rf_rate: float = 0.02): """ Args: returns: Series of periodic returns rf_rate: Annual risk-free rate """ self.returns = returns self.rf_rate = rf_rate self.ann_factor = 252 # Trading days per year

参数说明:

  • returns:周期收益序列(pandas Series),默认按日频收益假设处理;若传入周频或月频数据,需同步调整ann_factor
  • rf_rate:年化无风险利率,默认 2%,用于 Sharpe / Sortino 比率的超额收益计算。
  • ann_factor:年化交易日数,默认 252,是全部年化计算的基准。

3.1 波动率类指标:volatility、downside_deviation、beta

# Volatility Metrics def volatility(self, annualized: bool = True) -> float: """Standard deviation of returns.""" vol = self.returns.std() if annualized: vol *= np.sqrt(self.ann_factor) return vol def downside_deviation(self, threshold: float = 0, annualized: bool = True) -> float: """Standard deviation of returns below threshold.""" downside = self.returns[self.returns < threshold] if len(downside) == 0: return 0.0 dd = downside.std() if annualized: dd *= np.sqrt(self.ann_factor) return dd def beta(self, market_returns: pd.Series) -> float: """Beta relative to market.""" aligned = pd.concat([self.returns, market_returns], axis=1).dropna() if len(aligned) < 2: return np.nan cov = np.cov(aligned.iloc[:, 0], aligned.iloc[:, 1]) return cov[0, 1] / cov[1, 1] if cov[1, 1] != 0 else 0

实现要点:

  • volatility使用returns.std()(样本标准差,ddof=1),年化时乘sqrt(252),即日频波动率平方根法则放大。若不年化,可传annualized=False得到周期波动率。
  • downside_deviation只取低于阈值的样本计算标准差,是 Sortino 比率的输入。注意边界处理:当没有任何收益低于阈值时返回0.0,避免 NaN 污染后续比率。
  • beta通过pd.concat对齐两序列并dropna(),用np.cov计算协方差矩阵,beta = Cov(p, m) / Var(m)。边界处理:样本不足 2 个返回np.nan;市场方差为 0 时返回 0。

3.2 尾部风险:三种 VaR 与 CVaR

# Value at Risk def var_historical(self, confidence: float = 0.95) -> float: """Historical VaR at confidence level.""" return -np.percentile(self.returns, (1 - confidence) * 100) def var_parametric(self, confidence: float = 0.95) -> float: """Parametric VaR assuming normal distribution.""" z_score = stats.norm.ppf(confidence) return self.returns.mean() - z_score * self.returns.std() def var_cornish_fisher(self, confidence: float = 0.95) -> float: """VaR with Cornish-Fisher expansion for non-normality.""" z = stats.norm.ppf(confidence) s = stats.skew(self.returns) # Skewness k = stats.kurtosis(self.returns) # Excess kurtosis # Cornish-Fisher expansion z_cf = (z + (z**2 - 1) * s / 6 + (z**3 - 3*z) * k / 24 - (2*z**3 - 5*z) * s**2 / 36) return -(self.returns.mean() + z_cf * self.returns.std()) # Conditional VaR (Expected Shortfall) def cvar(self, confidence: float = 0.95) -> float: """Expected Shortfall / CVaR / Average VaR.""" var = self.var_historical(confidence) return -self.returns[self.returns <= -var].mean()

三种 VaR 的差异与适用场景:

  • 历史模拟法(historical):直接取收益分布的分位数取负,np.percentile(returns, (1-confidence)*100)即左侧尾部下分位点。无分布假设,稳健直观,是默认推荐。
  • 参数法(parametric):假设收益服从正态分布,用stats.norm.ppf(confidence)取标准正态分位数 z,VaR = mean - z * std。计算快,但对肥尾分布会系统性低估风险。
  • Cornish-Fisher 展开:在正态分位数基础上用偏度s和超额峰度k修正,z_cf公式为z + (z²-1)s/6 + (z³-3z)k/24 - (2z³-5z)s²/36。这是文档与代码注释反复强调的关键点——金融收益具有肥尾特征,不要默认假设正态性(见 SKILL.md 的 Don'ts 清单)。
  • CVaR / 期望损失(Expected Shortfall):取所有亏损超过历史 VaR 阈值的样本均值再取负,衡量"最坏 α 尾部内的平均损失"。它是比 VaR 更完备的尾部风险度量——VaR 只回答"阈值在哪",CVaR 回答"一旦突破阈值,平均亏多少"。

3.3 回撤分析:drawdowns、max_drawdown、avg_drawdown、drawdown_duration

# Drawdown Analysis def drawdowns(self) -> pd.Series: """Calculate drawdown series.""" cumulative = (1 + self.returns).cumprod() running_max = cumulative.cummax() return (cumulative - running_max) / running_max def max_drawdown(self) -> float: """Maximum drawdown.""" return self.drawdowns().min() def avg_drawdown(self) -> float: """Average drawdown.""" dd = self.drawdowns() return dd[dd < 0].mean() if (dd < 0).any() else 0 def drawdown_duration(self) -> Dict[str, int]: """Drawdown duration statistics.""" dd = self.drawdowns() in_drawdown = dd < 0 # Find drawdown periods drawdown_starts = in_drawdown & ~in_drawdown.shift(1).fillna(False) drawdown_ends = ~in_drawdown & in_drawdown.shift(1).fillna(False) durations = [] current_duration = 0 for i in range(len(dd)): if in_drawdown.iloc[i]: current_duration += 1 elif current_duration > 0: durations.append(current_duration) current_duration = 0 if current_duration > 0: durations.append(current_duration) return { "max_duration": max(durations) if durations else 0, "avg_duration": np.mean(durations) if durations else 0, "current_duration": current_duration }

实现要点:

  • drawdowns先计算累计净值cumprod,再用cummax得到历史峰值序列,回撤 =(净值 - 峰值) / 峰值。结果为 ≤ 0 的序列。
  • max_drawdown取回撤序列最小值(最深的负回撤)。
  • avg_drawdown只对回撤为负的区间求均值;若从未出现回撤返回 0,避免mean()对空切片告警。
  • drawdown_duration通过逐日扫描统计回撤持续期:返回max_duration(最长回撤天数)、avg_duration(平均回撤天数)与current_duration(当前正处于回撤中的天数)。该统计对资本保全与心理预期管理非常实用——很多策略"回撤深度可控但回撤时间过长"。

3.4 风险调整收益:Sharpe、Sortino、Calmar、Omega

# Risk-Adjusted Returns def sharpe_ratio(self) -> float: """Annualized Sharpe ratio.""" excess_return = self.returns.mean() * self.ann_factor - self.rf_rate vol = self.volatility(annualized=True) return excess_return / vol if vol > 0 else 0 def sortino_ratio(self) -> float: """Sortino ratio using downside deviation.""" excess_return = self.returns.mean() * self.ann_factor - self.rf_rate dd = self.downside_deviation(threshold=0, annualized=True) return excess_return / dd if dd > 0 else 0 def calmar_ratio(self) -> float: """Calmar ratio (return / max drawdown).""" annual_return = (1 + self.returns).prod() ** (self.ann_factor / len(self.returns)) - 1 max_dd = abs(self.max_drawdown()) return annual_return / max_dd if max_dd > 0 else 0 def omega_ratio(self, threshold: float = 0) -> float: """Omega ratio.""" returns_above = self.returns[self.returns > threshold] - threshold returns_below = threshold - self.returns[self.returns <= threshold] if returns_below.sum() == 0: return np.inf return returns_above.sum() / returns_below.sum()

四个比率的口径差异:

  • Sharpe:超额收益(年化平均收益减无风险利率)÷ 年化总波动率。波动率用全样本标准差,对上行波动与下行波动"一视同仁"。
  • Sortino:分子与 Sharpe 相同,但分母换成下行偏差(downside_deviation(threshold=0)),只惩罚亏损波动,更适合收益分布不对称的策略。
  • Calmar:年化收益 ÷ |最大回撤|,衡量"每承受 1 单位最深回撤能换来多少年化收益",是趋势型/CTA 策略常用指标。注意其年化收益使用复利几何平均(1+r).prod() ** (ann_factor/n) - 1,与 Sharpe 中使用的算术年化(mean * ann_factor)口径不同。
  • Omega:阈值之上收益总和 ÷ 阈值之下亏损总和,不依赖收益分布形态。当没有下行收益时返回np.inf(纯赚不亏的理想情形)。

3.5 信息比率与综合摘要

# Information Ratio def information_ratio(self, benchmark_returns: pd.Series) -> float: """Information ratio vs benchmark.""" active_returns = self.returns - benchmark_returns tracking_error = active_returns.std() * np.sqrt(self.ann_factor) active_return = active_returns.mean() * self.ann_factor return active_return / tracking_error if tracking_error > 0 else 0 # Summary def summary(self) -> Dict[str, float]: """Generate comprehensive risk summary.""" dd_stats = self.drawdown_duration() return { # Returns "total_return": (1 + self.returns).prod() - 1, "annual_return": (1 + self.returns).prod() ** (self.ann_factor / len(self.returns)) - 1, # Volatility "annual_volatility": self.volatility(), "downside_deviation": self.downside_deviation(), # VaR & CVaR "var_95_historical": self.var_historical(0.95), "var_99_historical": self.var_historical(0.99), "cvar_95": self.cvar(0.95), # Drawdowns "max_drawdown": self.max_drawdown(), "avg_drawdown": self.avg_drawdown(), "max_drawdown_duration": dd_stats["max_duration"], # Risk-Adjusted "sharpe_ratio": self.sharpe_ratio(), "sortino_ratio": self.sortino_ratio(), "calmar_ratio": self.calmar_ratio(), "omega_ratio": self.omega_ratio(), # Distribution "skewness": stats.skew(self.returns), "kurtosis": stats.kurtosis(self.returns), }

information_ratio计算相对基准的超额收益与跟踪误差之比:主动收益 = 组合收益 − 基准收益,跟踪误差 = 主动收益标准差 × √252。这是衡量"主动管理能力"的标准指标。

summary()一次性输出 14 项指标,涵盖收益、波动、VaR/CVaR、回撤、风险调整收益与分布形态(偏度、超额峰度),可直接作为风险看板的数据源。注意 Cornish-Fisher 修正所需的偏度与峰度也被一并输出,方便交叉核验尾部假设是否成立。

四、Pattern 2:组合层风险分解(PortfolioRisk 类)

单资产指标无法回答"多资产组合整体风险多大、每只资产贡献多少"的问题,PortfolioRisk负责组合层计算。

class PortfolioRisk: """Portfolio-level risk calculations.""" def __init__( self, returns: pd.DataFrame, weights: Optional[pd.Series] = None ): """ Args: returns: DataFrame with asset returns (columns = assets) weights: Portfolio weights (default: equal weight) """ self.returns = returns self.weights = weights if weights is not None else \ pd.Series(1/len(returns.columns), index=returns.columns) self.ann_factor = 252 def portfolio_return(self) -> float: """Weighted portfolio return.""" return (self.returns @ self.weights).mean() * self.ann_factor def portfolio_volatility(self) -> float: """Portfolio volatility.""" cov_matrix = self.returns.cov() * self.ann_factor port_var = self.weights @ cov_matrix @ self.weights return np.sqrt(port_var) def marginal_risk_contribution(self) -> pd.Series: """Marginal contribution to risk by asset.""" cov_matrix = self.returns.cov() * self.ann_factor port_vol = self.portfolio_volatility() # Marginal contribution mrc = (cov_matrix @ self.weights) / port_vol return mrc def component_risk(self) -> pd.Series: """Component contribution to total risk.""" mrc = self.marginal_risk_contribution() return self.weights * mrc def risk_parity_weights(self, target_vol: float = None) -> pd.Series: """Calculate risk parity weights.""" from scipy.optimize import minimize n = len(self.returns.columns) cov_matrix = self.returns.cov() * self.ann_factor def risk_budget_objective(weights): port_vol = np.sqrt(weights @ cov_matrix @ weights) mrc = (cov_matrix @ weights) / port_vol rc = weights * mrc target_rc = port_vol / n # Equal risk contribution return np.sum((rc - target_rc) ** 2) constraints = [ {"type": "eq", "fun": lambda w: np.sum(w) - 1}, # Weights sum to 1 ] bounds = [(0.01, 1.0) for _ in range(n)] # Min 1%, max 100% x0 = np.array([1/n] * n) result = minimize( risk_budget_objective, x0, method="SLSQP", bounds=bounds, constraints=constraints ) return pd.Series(result.x, index=self.returns.columns) def correlation_matrix(self) -> pd.DataFrame: """Asset correlation matrix.""" return self.returns.corr() def diversification_ratio(self) -> float: """Diversification ratio (higher = more diversified).""" asset_vols = self.returns.std() * np.sqrt(self.ann_factor) weighted_vol = (self.weights * asset_vols).sum() port_vol = self.portfolio_volatility() return weighted_vol / port_vol if port_vol > 0 else 1 def tracking_error(self, benchmark_returns: pd.Series) -> float: """Tracking error vs benchmark.""" port_returns = self.returns @ self.weights active_returns = port_returns - benchmark_returns return active_returns.std() * np.sqrt(self.ann_factor) def conditional_correlation( self, threshold_percentile: float = 10 ) -> pd.DataFrame: """Correlation during stress periods.""" port_returns = self.returns @ self.weights threshold = np.percentile(port_returns, threshold_percentile) stress_mask = port_returns <= threshold return self.returns[stress_mask].corr()

实现要点逐项拆解:

  • 组合收益:用矩阵乘法returns @ weights得到组合日收益序列,再取均值年化。权重缺省时等权分配1/n
  • 组合波动率:先算年化协方差矩阵cov * 252,再用二次型wᵀΣw求方差、开方得波动率。这是 Markowitz 组合理论的核心公式。
  • 边际风险贡献(MRC)Σw / σp,表示每增加 1 单位某资产权重引起的组合波动率边际变化。
  • 成分风险(Component Risk)wᵢ × MRCᵢ,各资产成分风险之和恰好等于组合波动率(欧拉分解),可直接回答"组合风险中每只资产占了多少"。
  • 风险平价权重(Risk Parity):通过scipy.optimize.minimize以 SLSQP 求解,目标函数为最小化各资产风险贡献与等风险贡献目标σp/n的平方偏差;约束权重和为 1,边界为每只资产 1%~100%。这是"每个资产贡献相同风险"的经典风险预算方案,与 risk-manager 关注的相关性集中风险(Monitor correlations to avoid concentration)直接相关。
  • 分散化比率(Diversification Ratio):加权单资产波动率之和 ÷ 组合波动率,比值越高说明分散化越有效;当组合波动率为 0 时返回 1 兜底。
  • 条件相关性(Conditional Correlation):取组合收益最低 10% 分位的压力期样本,单独计算相关性矩阵。这对应 SKILL.md 中"Don't ignore correlation - Increases in stress"的警示——正常市况下相关性低,但压力期相关性会飙升,用压力期相关性做风险管理比全样本相关性更保守。

五、Pattern 3:滚动窗口风险监控(RollingRiskMetrics 类)

风险是时变的,静态全样本指标会掩盖 regime(波动率状态)切换。RollingRiskMetrics用滚动窗口把风险指标变成时间序列。

class RollingRiskMetrics: """Rolling window risk calculations.""" def __init__(self, returns: pd.Series, window: int = 63): """ Args: returns: Return series window: Rolling window size (default: 63 = ~3 months) """ self.returns = returns self.window = window def rolling_volatility(self, annualized: bool = True) -> pd.Series: """Rolling volatility.""" vol = self.returns.rolling(self.window).std() if annualized: vol *= np.sqrt(252) return vol def rolling_sharpe(self, rf_rate: float = 0.02) -> pd.Series: """Rolling Sharpe ratio.""" rolling_return = self.returns.rolling(self.window).mean() * 252 rolling_vol = self.rolling_volatility() return (rolling_return - rf_rate) / rolling_vol def rolling_var(self, confidence: float = 0.95) -> pd.Series: """Rolling historical VaR.""" return self.returns.rolling(self.window).apply( lambda x: -np.percentile(x, (1 - confidence) * 100), raw=True ) def rolling_max_drawdown(self) -> pd.Series: """Rolling maximum drawdown.""" def max_dd(returns): cumulative = (1 + returns).cumprod() running_max = cumulative.cummax() drawdowns = (cumulative - running_max) / running_max return drawdowns.min() return self.returns.rolling(self.window).apply(max_dd, raw=False) def rolling_beta(self, market_returns: pd.Series) -> pd.Series: """Rolling beta vs market.""" def calc_beta(window_data): port_ret = window_data.iloc[:, 0] mkt_ret = window_data.iloc[:, 1] cov = np.cov(port_ret, mkt_ret) return cov[0, 1] / cov[1, 1] if cov[1, 1] != 0 else 0 combined = pd.concat([self.returns, market_returns], axis=1) return combined.rolling(self.window).apply( lambda x: calc_beta(x.to_frame()), raw=False ).iloc[:, 0] def volatility_regime( self, low_threshold: float = 0.10, high_threshold: float = 0.20 ) -> pd.Series: """Classify volatility regime.""" vol = self.rolling_volatility() def classify(v): if v < low_threshold: return "low" elif v > high_threshold: return "high" else: return "normal" return vol.apply(classify)

实现要点:

  • 默认窗口 63 天(约 3 个月),window可自行调整;SKILL.md 的 Don'ts 明确提示"Don't use short lookbacks - Miss regime changes",过短的窗口会错过波动率状态切换。
  • rolling_volatility用 pandas 原生rolling(window).std()实现,比循环快几个数量级;年化因子在此硬编码为 252。
  • rolling_sharpe在窗口内用算术年化(mean * 252)计算滚动年化收益。
  • rolling_var/rolling_max_drawdown通过rolling().apply()传入自定义函数;raw=True传 numpy 数组(更快),raw=False传 pandas Series(可访问索引等元数据)。
  • rolling_beta将组合与市场收益 concat 成两列后滚动回归式地计算协方差比。
  • volatility_regime把年化滚动波动率分类为low(<10%)、high(>20%)、normal(介于两者之间),可直接用于驱动风控规则的切换——例如高波动 regime 下自动收紧仓位与止损(对应 risk-manager 的"Implementing risk limits"职责)。

六、Pattern 4:压力测试(StressTester 类)

压力测试回答"极端情景下组合会怎样",分三类:历史危机重演、假设冲击、Monte Carlo 模拟。

class StressTester: """Historical and hypothetical stress testing.""" # Historical crisis periods HISTORICAL_SCENARIOS = { "2008_financial_crisis": ("2008-09-01", "2009-03-31"), "2020_covid_crash": ("2020-02-19", "2020-03-23"), "2022_rate_hikes": ("2022-01-01", "2022-10-31"), "dot_com_bust": ("2000-03-01", "2002-10-01"), "flash_crash_2010": ("2010-05-06", "2010-05-06"), } def __init__(self, returns: pd.Series, weights: pd.Series = None): self.returns = returns self.weights = weights def historical_stress_test( self, scenario_name: str, historical_data: pd.DataFrame ) -> Dict[str, float]: """Test portfolio against historical crisis period.""" if scenario_name not in self.HISTORICAL_SCENARIOS: raise ValueError(f"Unknown scenario: {scenario_name}") start, end = self.HISTORICAL_SCENARIOS[scenario_name] # Get returns during crisis crisis_returns = historical_data.loc[start:end] if self.weights is not None: port_returns = (crisis_returns @ self.weights) else: port_returns = crisis_returns total_return = (1 + port_returns).prod() - 1 max_dd = self._calculate_max_dd(port_returns) worst_day = port_returns.min() return { "scenario": scenario_name, "period": f"{start} to {end}", "total_return": total_return, "max_drawdown": max_dd, "worst_day": worst_day, "volatility": port_returns.std() * np.sqrt(252) } def hypothetical_stress_test( self, shocks: Dict[str, float] ) -> float: """ Test portfolio against hypothetical shocks. Args: shocks: Dict of {asset: shock_return} """ if self.weights is None: raise ValueError("Weights required for hypothetical stress test") total_impact = 0 for asset, shock in shocks.items(): if asset in self.weights.index: total_impact += self.weights[asset] * shock return total_impact def monte_carlo_stress( self, n_simulations: int = 10000, horizon_days: int = 21, vol_multiplier: float = 2.0 ) -> Dict[str, float]: """Monte Carlo stress test with elevated volatility.""" mean = self.returns.mean() vol = self.returns.std() * vol_multiplier simulations = np.random.normal( mean, vol, (n_simulations, horizon_days) ) total_returns = (1 + simulations).prod(axis=1) - 1 return { "expected_loss": -total_returns.mean(), "var_95": -np.percentile(total_returns, 5), "var_99": -np.percentile(total_returns, 1), "worst_case": -total_returns.min(), "prob_10pct_loss": (total_returns < -0.10).mean() } def _calculate_max_dd(self, returns: pd.Series) -> float: cumulative = (1 + returns).cumprod() running_max = cumulative.cummax() drawdowns = (cumulative - running_max) / running_max return drawdowns.min()

实现要点:

  • 历史情景:内置 5 个经典危机区间——2008 金融危机、2020 疫情崩盘、2022 加息周期、2000 互联网泡沫、2010 闪崩。historical_stress_testloc[start:end]截取危机区间收益,输出区间总收益、最大回撤、最差单日、波动率。未知情景名会抛ValueError
  • 假设冲击hypothetical_stress_test接收{资产: 冲击收益}字典,用权重加权求和计算组合冲击影响;未传权重时抛ValueError。适合做"某资产跌 20% 会怎样"的单点分析。
  • Monte Carlo 压力模拟:默认 10000 次模拟、21 天(约一个月)持有期、波动率放大 2 倍,从Normal(mean, vol*2)抽样生成(10000, 21)矩阵,逐行复利累乘得到期末总收益分布,输出期望损失、VaR95/99、最坏情形与亏损超过 10% 的概率。注意该模拟假设正态分布,在极端尾部可能低估风险——建议与历史 VaR / CVaR 交叉验证。

七、快速参考:把整套工具用起来

details.md 提供了开箱即用的日常使用示例:

# Daily usage metrics = RiskMetrics(returns) print(f"Sharpe: {metrics.sharpe_ratio():.2f}") print(f"Max DD: {metrics.max_drawdown():.2%}") print(f"VaR 95%: {metrics.var_historical(0.95):.2%}") # Full summary summary = metrics.summary() for metric, value in summary.items(): print(f"{metric}: {value:.4f}")

接入风险的完整链路可以这样组织(结合同插件 backtesting-frameworks 的向量化回测器):

  1. 用回测器产出组合净值曲线与日收益序列(对应 backtesting-frameworks/references/details.md 中的VectorizedBacktester);
  2. 将日收益喂给RiskMetrics,输出 Sharpe、Sortino、Calmar、VaR、CVaR、最大回撤等单资产指标;
  3. 多资产场景用PortfolioRisk做组合波动率、成分风险分解与风险平价配置;
  4. RollingRiskMetrics监控波动率 regime 变化,动态调整风控阈值;
  5. 上线前用StressTester跑历史危机与 Monte Carlo 压力测试,完成风险预检。

八、最佳实践:Do's 与 Don'ts(继承 SKILL.md)

SKILL.md 给出了整套方法论的纪律性要求:

Do's(应当遵守)

  • Use multiple metrics- 单一指标无法刻画全部风险,多指标交叉印证
  • Consider tail risk- 仅有 VaR 不够,必须配合 CVaR
  • Rolling analysis- 风险随时间变化,静态指标会失效
  • Stress test- 历史情景与假设情景都要测
  • Document assumptions- 明确记录分布假设、回看窗口等关键参数

Don'ts(必须避免)

  • Don't rely on VaR alone- VaR 会低估尾部风险(正态假设下尤其明显)
  • Don't assume normality- 收益是肥尾的,用 Cornish-Fisher 或历史法修正
  • Don't ignore correlation- 压力期相关性会上升(用conditional_correlation验证)
  • Don't use short lookbacks- 过短窗口会错过波动率 regime 切换
  • Don't forget transaction costs- 交易成本会影响实际实现的风险收益

九、小结:从代码到风控体系的落地路径

本文完整继承了 risk-metrics-calculation 技能的全部四个实现模式与快速参考,并逐项注解了参数默认值、年化口径与边界处理。这套代码在 agents24 仓库中的定位是:为 risk-manager 与 quant-analyst 两个 Agent 提供可复用的风险计算基础,支撑其"风险限额实施、R-multiple 跟踪、头寸保护与压力测试"的完整职责链。

需要记住的五个关键设计决策:年化因子默认 252 交易日;尾部风险以"历史法 VaR + CVaR + Cornish-Fisher 修正"三件套互相校验;回撤分析覆盖深度与持续时间两个维度;组合层用欧拉分解回答"谁贡献了风险";压力测试同时覆盖历史危机、假设冲击与 Monte Carlo 三种范式。据此即可搭建从日频监控、动态限额到压力预检的完整风控体系。

【免费下载链接】agentsMulti-harness agentic plugin marketplace for Claude Code, Codex, Cursor, OpenCode, GitHub Copilot, and Google Antigravity项目地址: https://gitcode.com/GitHub_Trending/agents24/agents

创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考

需要专业的网站建设服务?

联系我们获取免费的网站建设咨询和方案报价,让我们帮助您实现业务目标

立即咨询