4、Python实现风险平价模型:数据获取、协方差矩阵计算、风险贡献计算、权重优化
好,咱们直接进入正题。这一章,我会带你手把手把风险平价模型用Python实现出来。说实话,理论讲再多,不如跑一遍代码来得实在。我当年刚接触这个模型时,也是对着论文看了三天,最后发现——嗯,代码一跑,全明白了。
咱们要做的就四步:拿数据 → 算协方差 → 算风险贡献 → 优化权重。听起来简单?做起来也不复杂,但坑不少。我一个个说。
4.1 数据获取:别拿垃圾数据做分析
我个人习惯用 yfinance 拉数据,免费、方便、够用。但注意一点:数据质量决定模型质量。我曾经有一次用某免费接口拉数据,结果某只ETF的收盘价连续三天都是同一个数——你想想看,这种数据算出来的协方差矩阵能靠谱吗?
这里我选了三类资产做演示:股票(SPY)、债券(TLT)、商品(GLD)。为什么选这三个?因为它们之间的相关性低,风险平价的效果最明显。
import yfinance as yf
import pandas as pd
import numpy as np
from scipy.optimize import minimize
# 获取数据
tickers = ['SPY', 'TLT', 'GLD']
data = yf.download(tickers, start='2020-01-01', end='2023-12-31')['Adj Close']
# 计算日收益率
returns = data.pct_change().dropna()
print(returns.head())
4.2 协方差矩阵计算:别用简单协方差
拿到收益率序列后,第一件事就是算协方差矩阵。但这里有个坑:直接用 np.cov 算出来的协方差矩阵,对异常值非常敏感。
我建议用 指数加权协方差(EWMA)。为什么?因为近期的波动比三个月前的波动更有参考价值。说白了,市场是有记忆的,但记忆会衰退。
# 计算指数加权协方差矩阵
def ewma_covariance(returns, lambda_=0.94):
"""
计算指数加权协方差矩阵
lambda_:衰减因子,通常取0.94(RiskMetrics标准)
"""
mean_returns = returns.mean()
centered = returns - mean_returns
n = len(returns)
weights = np.array([(1 - lambda_) * (lambda_ ** (n - 1 - i)) for i in range(n)])
weights = weights / weights.sum() # 归一化
cov_matrix = np.cov(centered.T, aweights=weights)
return cov_matrix
cov_matrix = ewma_covariance(returns)
print("协方差矩阵:\n", cov_matrix)
4.3 风险贡献计算:核心逻辑在这里
风险平价的核心思想,说白了就是:让每类资产对组合总风险的贡献相等。不是等权重,也不是等金额,而是等风险。
怎么算?分三步:
- 计算组合波动率:
sigma_p = sqrt(w^T * Sigma * w) - 计算边际风险贡献:
MRC = (Sigma * w) / sigma_p - 计算总风险贡献:
TRC = w * MRC
嗯,这里要注意:所有资产的风险贡献之和,正好等于组合波动率。这是一个很好的校验点。我每次跑完模型都会检查这个等式,如果对不上,那一定是代码写错了。
def risk_contribution(weights, cov_matrix):
"""
计算每类资产的风险贡献
"""
portfolio_vol = np.sqrt(weights.T @ cov_matrix @ weights)
marginal_contrib = (cov_matrix @ weights) / portfolio_vol
total_contrib = weights * marginal_contrib
return total_contrib, portfolio_vol
# 测试:等权重组合
w_equal = np.array([1/3, 1/3, 1/3])
trc, vol = risk_contribution(w_equal, cov_matrix)
print("等权重风险贡献:", trc)
print("组合波动率:", vol)
print("风险贡献之和:", trc.sum())
4.4 权重优化:用最优化求解
现在到了最核心的一步:找到一组权重,让所有资产的风险贡献相等。
这本质上是一个最优化问题。目标函数是:各资产风险贡献之间的方差最小化。当方差为0时,所有资产的风险贡献完全相等。
我习惯用 scipy.optimize.minimize 来解。约束条件有两个:权重之和为1,且每个权重在0到1之间(不允许做空)。
def risk_parity_objective(weights, cov_matrix):
"""
风险平价的目标函数:风险贡献的方差最小化
"""
trc, _ = risk_contribution(weights, cov_matrix)
target_risk = trc.mean()
return np.sum((trc - target_risk) ** 2)
# 约束条件
constraints = ({'type': 'eq', 'fun': lambda x: np.sum(x) - 1})
bounds = tuple((0, 1) for _ in range(len(tickers)))
# 初始值:等权重
initial_weights = np.array([1/3, 1/3, 1/3])
# 优化求解
result = minimize(
risk_parity_objective,
initial_weights,
args=(cov_matrix,),
method='SLSQP',
bounds=bounds,
constraints=constraints
)
optimal_weights = result.x
print("最优权重:", optimal_weights)
print("各资产风险贡献:", risk_contribution(optimal_weights, cov_matrix)[0])
method='L-BFGS-B' 跑过一次,结果权重收敛到了局部最优解——债券权重几乎为0。后来换成 SLSQP 才得到合理结果。所以,多试几种优化方法,别一棵树上吊死。
4.5 可视化:一张图看懂风险平价
光看数字不过瘾,咱们画个图。下面这张SVG图展示了风险平价的核心逻辑:从数据到权重,每一步都环环相扣。
4.6 完整代码整合
把上面的代码拼起来,就是一个完整的风险平价实现。我建议你把它封装成一个类,方便以后复用。
class RiskParity:
def __init__(self, tickers, start, end, lambda_=0.94):
self.tickers = tickers
self.start = start
self.end = end
self.lambda_ = lambda_
self.returns = None
self.cov_matrix = None
self.optimal_weights = None
def fetch_data(self):
data = yf.download(self.tickers, start=self.start, end=self.end)['Adj Close']
self.returns = data.pct_change().dropna()
def compute_covariance(self):
mean_returns = self.returns.mean()
centered = self.returns - mean_returns
n = len(self.returns)
weights = np.array([(1 - self.lambda_) * (self.lambda_ ** (n - 1 - i)) for i in range(n)])
weights = weights / weights.sum()
self.cov_matrix = np.cov(centered.T, aweights=weights)
def optimize(self):
n = len(self.tickers)
initial_weights = np.array([1/n] * n)
constraints = ({'type': 'eq', 'fun': lambda x: np.sum(x) - 1})
bounds = tuple((0, 1) for _ in range(n))
result = minimize(
lambda w: self._objective(w),
initial_weights,
method='SLSQP',
bounds=bounds,
constraints=constraints
)
self.optimal_weights = result.x
def _objective(self, weights):
trc, _ = self.risk_contribution(weights)
target = trc.mean()
return np.sum((trc - target) ** 2)
def risk_contribution(self, weights):
port_vol = np.sqrt(weights.T @ self.cov_matrix @ weights)
marginal = (self.cov_matrix @ weights) / port_vol
total = weights * marginal
return total, port_vol
# 使用示例
rp = RiskParity(['SPY', 'TLT', 'GLD'], '2020-01-01', '2023-12-31')
rp.fetch_data()
rp.compute_covariance()
rp.optimize()
print("最优权重:", rp.optimal_weights)
嗯,这一章的内容就到这。代码你可以直接复制跑,但建议你换几只资产试试——比如加上A股、黄金、原油。你会发现不同市场环境下,风险平价的表现差异很大。这就是实战的魅力所在。