量化风险管理方法:应用 VaR 模型优化多货币投资组合(使用 Python 和 MetaTrader 5)·综合运用
(3/3)·把 VaR 方程变成 MT5 里能跑的代码,并用它管住多货币网格的尾部风险
把 VaR 跑成能盯盘的图
做风控不能只盯一个汇总数字,得把 VaR 和实际回报叠在一起看。一张图里蓝线是每日实际收益,红线是负值形式的 VaR 边界,红色填充区就是实际亏损穿透预测风险的极端段,哪天漏出了心里马上就有数。 回撤图看的是累计收益曲线相对历史峰值的下沉深度和持续时间,热力图则把多个货币对的风险贡献铺成颜色矩阵,暖色越深代表该品种在组合里的尾部风险权重越大。 实盘统计样本里,这套监控下的收益率约 11%,最大浮动回撤压在 1% 以内,外汇和贵金属本身带高杠杆高风险,该数字仅代表特定时段表现,换环境可能明显漂移。 下面这段 Python 是画图与取数核心,接 MT5 跑通后你就能自己复现三张图。
class="kw">import matplotlib.pyplot as<span class="keyword"> as</span> plt class="kw">import seaborn <span class="keyword">as</span> sns def plot_var_vs_returns(returns, var_predictions): fig, ax = plt.subplots(figsize=(<span class="number">class="num">12</span>, <span class="number">class="num">6</span>)) ax.plot(returns, label=&class="macro">#x27;Actual Returns&class="macro">#x27;) ax.plot(-var_predictions, label=&class="macro">#x27;VaR&class="macro">#x27;, <span class="keyword">class="type">class="kw">color</span>=&class="macro">#x27;red&class="macro">#x27;) ax.fill_between(returns.index, -var_predictions, returns, where=returns < -var_predictions, <span class="keyword">class="type">class="kw">color</span>=&class="macro">#x27;red&class="macro">#x27;, alpha=<span class="number">class="num">0.3</span>) ax.legend() ax.set_title(&class="macro">#x27;VaR vs Actual Returns&class="macro">#x27;) plt.show() def plot_drawdown(returns): drawdown = (returns.cumsum() - returns.cumsum().cummax()) plt.figure(figsize=(<span class="number">class="num">12</span>, <span class="number">class="num">6</span>)) plt.plot(drawdown) plt.title(&class="macro">#x27;Portfolio Drawdown&class="macro">#x27;) plt.show() def plot_var_heatmap(var_matrix): plt.figure(figsize=(<span class="number">class="num">12</span>, <span class="number">class="num">8</span>)) sns.heatmap(var_matrix, annot=True, cmap=&class="macro">#x27;YlOrRd&class="macro">#x27;) plt.title(&class="macro">#x27;VaR Heatmap across Currency Pairs&class="macro">#x27;) plt.show() class="kw">import MetaTrader5 <span class="keyword">as</span> mt5 class="kw">import pandas <span class="keyword">as</span> pd class="kw">import numpy <span class="keyword">as</span> np class="kw">import matplotlib.pyplot <span class="keyword">as</span> plt <span class="keyword">from</span> scipy.stats class="kw">import norm <span class="keyword">from</span> scipy.optimize class="kw">import minimize # Initialize connection to MetaTrader <span class="number">class="num">5</span> <span class="keyword">if</span> not mt5.initialize(): print(<span class="class="type">class="kw">string">"Error initializing MetaTrader class="num">5"</span>) mt5.shutdown() # Parameters symbols = [<span class="class="type">class="kw">string">"EURUSD"</span>, <span class="class="type">class="kw">string">"GBPUSD"</span>, <span class="class="type">class="kw">string">"USDJPY"</span>, <span class="class="type">class="kw">string">"AUDUSD"</span>, <span class="class="type">class="kw">string">"USDCAD"</span>, <span class="class="type">class="kw">string">"NZDUSD"</span>, <span class="class="type">class="kw">string">"EURCHF"</span>, <span class="class="type">class="kw">string">"EURGBP"</span>, <span class="class="type">class="kw">string">"AUDCAD"</span>] timeframe = mt5.TIMEFRAME_D1 start_date = pd.Timestamp(<span class="class="type">class="kw">string">&class="macro">#x27;class="num">2023-class="num">01-class="num">01&class="macro">#x27;</span>) end_date = pd.Timestamp.now() # Function to <span class="keyword">get</span> data def get_data(symbol, timeframe, start_date, end_date): rates = mt5.copy_rates_range(symbol, timeframe, start_date, end_date) df = pd.DataFrame(rates) df[<span class="class="type">class="kw">string">&class="macro">#x27;time&class="macro">#x27;</span>] = pd.to_datetime(df[<span class="class="type">class="kw">string">&class="macro">#x27;time&class="macro">#x27;</span>], unit=<span class="class="type">class="kw">string">&class="macro">#x27;s&class="macro">#x27;</span>) df.set_index(<span class="class="type">class="kw">string">&class="macro">#x27;time&class="macro">#x27;</span>, inplace=True) df[<span class="class="type">class="kw">string">&class="macro">#x27;returns&class="macro">#x27;</span>] = df[<span class="class="type">class="kw">string">&class="macro">#x27;close&class="macro">#x27;</span>].pct_change() <span class="keyword">class="kw">return</span> df # Get data <span class="keyword">for</span> all symbols data = {symbol: get_data(symbol, timeframe, start_date, end_date) <span class="keyword">for</span> symbol <span class="keyword">in</span> symbols} # Function to calculate VaR def calculate_var(returns, confidence_level=<span class="number">class="num">0.95</span>, holding_period=<span class="number">class="num">1</span>): <span class="keyword">class="kw">return</span> np.percentile(returns, (<span class="number">class="num">1</span> - confidence_level) * <span class="number">class="num">100</span>) * np.sqrt(holding_period) # Function to calculate CVaR def calculate_cvar(returns, confidence_level=<span class="number">class="num">0.95</span>, holding_period=<span class="number">class="num">1</span>): <span class="keyword">var</span> = calculate_var(returns, confidence_level, holding_period) <span class="keyword">class="kw">return</span> -returns[returns <= -<span class="keyword">var</span>].mean() * np.sqrt(holding_period) # Function to optimize portfolio def optimize_portfolio(returns, target_return, confidence_level=<span class="number">class="num">0.95</span>): n = len(returns.columns) def portfolio_var(weights): portfolio_returns = returns.dot(weights) <span class="keyword">class="kw">return</span> calculate_var(portfolio_returns, confidence_level) def portfolio_return(weights): <span class="keyword">class="kw">return</span> np.sum(returns.mean() * weights) constraints = ({<span class="class="type">class="kw">string">&class="macro">#x27;type&class="macro">#x27;</span>: <span class="class="type">class="kw">string">&class="macro">#x27;eq&class="macro">#x27;</span>, <span class="class="type">class="kw">string">&class="macro">#x27;fun&class="macro">#x27;</span>: lambda x: np.sum(x) - <span class="number">class="num">1</span>}, {<span class="class="type">class="kw">string">&class="macro">#x27;type&class="macro">#x27;</span>: <span class="class="type">class="kw">string">&class="macro">#x27;eq&class="macro">#x27;</span>, <span class="class="type">class="kw">string">&class="macro">#x27;fun&class="macro">#x27;</span>: lambda x: portfolio_return(x) - target_return}) bounds = tuple((<span class="number">class="num">0</span>, <span class="number">class="num">1</span>) <span class="keyword">for</span> _ <span class="keyword">in</span> range(n)) result = minimize(portfolio_var, n * [<span class="number">class="num">1</span>./n], method=<span class="class="type">class="kw">string">&class="macro">#x27;SLSQP&class="macro">#x27;</span>, bounds=bounds, constraints=constraints)
「组合层面的风险与绩效可视化落地」
上面这段 Python 把多标的收益矩阵收口成组合层指标:先按 optimize_portfolio 给的权重做 dot 乘积得到 portfolio_returns,再算 VaR / CVaR。VaR 效率函数用 95% 置信做校验,violations 与 expected_violations 的偏离度绝对值除以期望次数,数值越接近 0 说明 VaR 模型在样本内越贴合。 plot_var_vs_returns 用红色虚线标 -VaR,并把实际收益跌破该线的区域填红(alpha=0.3),一眼能看出尾部穿透频率。drawdown 直接拿 cumsum 减 cummax,没走第三方库,回撤拐点靠肉眼也能抓。 绩效端给了三个硬指标:profit_factor 用正收益和除以负收益绝对值;sharpe_ratio 默认无风险利率 0.02、按 252 交易日年化。跑完 print 出权重、VaR(保留 4 位)、CVaR、VaR 效率、Profit Factor、Sharpe,接 mt5.shutdown() 断连接。 外汇与贵金属组合用这套,杠杆会放大 VaR 穿透概率,实盘前建议把 confidence_level 调到 0.99 重跑一次看权重漂移。
class="kw">return result.x # Create portfolio returns = pd.DataFrame({symbol: data[symbol][&class="macro">#x27;returns&class="macro">#x27;] for symbol in symbols}).dropna() target_return = returns.mean().mean() weights = optimize_portfolio(returns, target_return) # Calculate VaR and CVaR for the portfolio portfolio_returns = returns.dot(weights) portfolio_var = calculate_var(portfolio_returns) portfolio_cvar = calculate_cvar(portfolio_returns) # Functions for visualization def plot_var_vs_returns(returns, var): fig, ax = plt.subplots(figsize=(class="num">12, class="num">6)) ax.plot(returns, label=&class="macro">#x27;Actual Returns&class="macro">#x27;) ax.axhline(-var, class="type">class="kw">color=&class="macro">#x27;red&class="macro">#x27;, linestyle=&class="macro">#x27;--&class="macro">#x27;, label=&class="macro">#x27;VaR&class="macro">#x27;) ax.fill_between(returns.index, -var, returns, where=returns < -var, class="type">class="kw">color=&class="macro">#x27;red&class="macro">#x27;, alpha=class="num">0.3) ax.legend() ax.set_title(&class="macro">#x27;VaR vs Actual Returns&class="macro">#x27;) plt.show() def plot_drawdown(returns): drawdown = (returns.cumsum() - returns.cumsum().cummax()) plt.figure(figsize=(class="num">12, class="num">6)) plt.plot(drawdown) plt.title(&class="macro">#x27;Portfolio Drawdown&class="macro">#x27;) plt.show() def plot_cumulative_returns(returns): cumulative_returns = (class="num">1 + returns).cumprod() plt.figure(figsize=(class="num">12, class="num">6)) plt.plot(cumulative_returns) plt.title(&class="macro">#x27;Cumulative Portfolio Returns&class="macro">#x27;) plt.ylabel(&class="macro">#x27;Cumulative Returns&class="macro">#x27;) plt.show() # Performance analysis def var_efficiency(returns, var, confidence_level=class="num">0.95): violations = (returns < -var).sum() expected_violations = len(returns) * (class="num">1 - confidence_level) class="kw">return abs(violations - expected_violations) / expected_violations def profit_factor(returns): positive_returns = returns[returns > class="num">0].sum() negative_returns = abs(returns[returns < class="num">0].sum()) class="kw">return positive_returns / negative_returns def sharpe_ratio(returns, risk_free_rate=class="num">0.02): class="kw">return (returns.mean() - risk_free_rate) / returns.std() * np.sqrt(class="num">252) # Output results print(f"Optimal portfolio weights: {dict(zip(symbols, weights))}") print(f"Portfolio VaR: {portfolio_var:.4f}") print(f"Portfolio CVaR: {portfolio_cvar:.4f}") print(f"VaR Efficiency: {var_efficiency(portfolio_returns, portfolio_var):.4f}") print(f"Profit Factor: {profit_factor(portfolio_returns):.4f}") print(f"Sharpe Ratio: {sharpe_ratio(portfolio_returns):.4f}") # Visualization plot_var_vs_returns(portfolio_returns, portfolio_var) plot_drawdown(portfolio_returns) plot_cumulative_returns(portfolio_returns) mt5.shutdown()
◍ 把风险价值塞进多币网格的资金与间距
把 VaR 直接接进多货币网格,核心不是预测方向,而是让仓位和网格步长跟着风险走。动态资金分配按各货币对的 VaR 占比切分总资金:某对 VaR 越低,分到的钱越多,组合层面倾向于压住高相关敞口。 货币对之间不是孤立的。用收益率序列算相关系数再外乘 VaR 向量,得到 VaR 相关矩阵,能暴露 EURUSD 与 GBPUSD 这类过度联动——它们一起跳时,组合实际风险会比单对相加更大。 网格参数也得按对调。以基准步长和平均 VaR 做锚,波动因子 = 当前 VaR / 平均 VaR:因子大于 1 就放大步长、砍层级(下限 3 层),因子小于 1 则缩步长、加层级(上限 10 层)。回测中这种自适应让网格在瑞郎黑天鹅类跳空下爆仓概率倾向更低。 外汇与贵金属属高杠杆品种,多币网格叠加 VaR 模型仍可能连亏,实盘前请用 MT5 策略测试器跑至少一年 tick 数据。
def allocate_capital(total_capital, var_values): total_var = sum(var_values.values()) allocations = {pair: (var / total_var) * total_capital for pair, var in var_values.items()} class="kw">return allocations def calculate_var_correlation_matrix(returns_dict): returns_df = pd.DataFrame(returns_dict) var_values = returns_df.apply(calculate_var) correlation_matrix = returns_df.corr() class="kw">return correlation_matrix * np.outer(var_values, var_values) def adjust_grid_params_multi(var_dict, base_params): adjusted_params = {} for pair, var in var_dict.items(): volatility_factor = var / base_params[pair][&class="macro">#x27;average_var&class="macro">#x27;] step = base_params[pair][&class="macro">#x27;base_step&class="macro">#x27;] * volatility_factor levels = max(class="num">3, min(class="num">10, class="type">int(base_params[pair][&class="macro">#x27;base_levels&class="macro">#x27;] / volatility_factor))) adjusted_params[pair] = {&class="macro">#x27;step&class="macro">#x27;: step, &class="macro">#x27;levels&class="macro">#x27;: levels} class="kw">return adjusted_params
别急着下结论
VaR 从一张基础方程走到多货币动态组合,过程里最实在的一课是:它再强也只是工具,数字本身不会替你盯住突发流动性断裂。把 VaR 塞进交易系统,本质是把资金管理和风险认知重写一遍,交易动作会变得更克制、更有章法。 多货币维度的相关性、相互依赖和动态分配,构成一组难但有趣的 puzzle,VaR 是解这组的钥匙而非终点。外汇与贵金属属高风险市场,模型给出的只是概率边界,黑天鹅仍在肥尾里等着。 下一步把机器学习接进 VaR 预测、用非线性模型兜住肥尾,是未完的活。市场不停,模型也得跟着进化,现在盖棺定论还太早。