风险指标计算

用于投资组合管理的全面风险测量工具包，包括风险价值、预期亏损和回撤分析。

何时使用此技能

• - 衡量投资组合风险
• 实施风险限额
• 构建风险仪表盘
• 计算风险调整后收益
• 设定头寸规模
• 监管报告

核心概念

1. 风险指标类别

2. 时间跨度

日内： 日内交易者的分钟/小时级VaR
每日： 标准风险报告
每周： 再平衡决策
每月： 绩效归因
每年： 战略配置

实现

模式1：核心风险指标

python
import numpy as np
import pandas as pd
from scipy import stats
from typing import Dict, Optional, Tuple

class RiskMetrics:
核心风险指标计算。

def init(self, returns: pd.Series, rf_rate: float = 0.02):

参数:
returns: 周期收益率序列
rf_rate: 年化无风险利率

self.returns = returns
self.rfrate = rfrate
self.ann_factor = 252 # 每年交易日数

# 波动率指标
def volatility(self, annualized: bool = True) -> float:
收益率标准差。
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:
低于阈值的收益率标准差。
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:
相对于市场的贝塔值。
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

# 风险价值
def var_historical(self, confidence: float = 0.95) -> float:
置信水平下的历史VaR。
return -np.percentile(self.returns, (1 - confidence) * 100)

def var_parametric(self, confidence: float = 0.95) -> float:
假设正态分布的参数VaR。
z_score = stats.norm.ppf(confidence)
return self.returns.mean() - z_score * self.returns.std()

def varcornishfisher(self, confidence: float = 0.95) -> float:
使用Cornish-Fisher展开的非正态VaR。
z = stats.norm.ppf(confidence)
s = stats.skew(self.returns) # 偏度
k = stats.kurtosis(self.returns) # 超额峰度

# Cornish-Fisher展开
z_cf = (z + (z2 - 1) * s / 6 +
(z3 - 3z) k / 24 -
(2z3 - 5z) * s2 / 36)

return -(self.returns.mean() + z_cf * self.returns.std())

# 条件VaR（预期亏损）
def cvar(self, confidence: float = 0.95) -> float:
预期亏损 / CVaR / 平均VaR。
var = self.var_historical(confidence)
return -self.returns[self.returns <= -var].mean()

# 回撤分析
def drawdowns(self) -> pd.Series:
计算回撤序列。
cumulative = (1 + self.returns).cumprod()
running_max = cumulative.cummax()
return (cumulative - runningmax) / runningmax

def max_drawdown(self) -> float:
最大回撤。
return self.drawdowns().min()

def avg_drawdown(self) -> float:
平均回撤。
dd = self.drawdowns()
return dd[dd < 0].mean() if (dd < 0).any() else 0

def drawdown_duration(self) -> Dict[str, int]:
回撤持续时间统计。
dd = self.drawdowns()
in_drawdown = dd < 0

# 查找回撤期间
drawdownstarts = indrawdown & ~in_drawdown.shift(1).fillna(False)
drawdownends = ~indrawdown & 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,
currentduration: currentduration
}

# 风险调整后收益
def sharpe_ratio(self) -> float:
年化夏普比率。
excessreturn = self.returns.mean() * self.annfactor - self.rf_rate
vol = self.volatility(annualized=True)
return excess_return / vol if vol > 0 else 0

def sortino_ratio(self) -> float:
使用下行偏差的索提诺比率。
excessreturn = self.returns.mean() * self.annfactor - 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:
卡玛比率（收益/最大回撤）。
annualreturn = (1 + self.returns).prod() (self.annfactor / len(self.returns)) - 1
maxdd = abs(self.maxdrawdown())
return annualreturn / maxdd if max_dd > 0 else 0

def omega_ratio(self, threshold: float = 0) -> float:
欧米伽比率。
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 returnsabove.sum() / returnsbelow.sum()

# 信息比率
def informationratio(self, benchmarkreturns: pd.Series) -> float:
相对于基准的信息比率。
activereturns = self.returns - benchmarkreturns
trackingerror = activereturns.std() * np.sqrt(self.ann_factor)
activereturn = activereturns.mean() * self.ann_factor
return activereturn / trackingerror if tracking_error > 0 else 0

# 汇总
def summary(self) -> Dict[str, float]:
生成全面风险汇总。
ddstats = self.drawdownduration()

return {
# 收益
total_return: (1 + self.returns).prod() - 1,
annualreturn: (1 + self.returns).prod() (self.annfactor / len(self.returns)) - 1,

# 波动率
annual_volatility: self.volatility(),
downsidedeviation: self.downsidedeviation(),

# VaR和CVaR
var95historical: self.var_historical(0.95),
var99historical: self.var_historical(0.99),
cvar_95: self.cvar(0.95),

# 回撤
maxdrawdown: self.maxdrawdown(),
avgdrawdown: self.avgdrawdown(),
maxdrawdownduration: ddstats[maxduration],

# 风险调整
sharpe_