量化风险管理方法:应用 VaR 模型优化多货币投资组合(使用 Python 和 MetaTrader 5)·综合运用
📉

量化风险管理方法:应用 VaR 模型优化多货币投资组合(使用 Python 和 MetaTrader 5)·综合运用

(3/3)·把 VaR 方程变成 MT5 里能跑的代码,并用它管住多货币网格的尾部风险

含代码示例偏理论 第 3/3 篇
很多人算完 VaR 就扔进 Excel 再也不看,实盘一遇到跳空就被尾部损失打穿。把模型嵌进 MT5 却不同步时区,回测和实盘的时间戳错位会让整组风险数字失真。

把 VaR 跑成能盯盘的图

做风控不能只盯一个汇总数字,得把 VaR 和实际回报叠在一起看。一张图里蓝线是每日实际收益,红线是负值形式的 VaR 边界,红色填充区就是实际亏损穿透预测风险的极端段,哪天漏出了心里马上就有数。 回撤图看的是累计收益曲线相对历史峰值的下沉深度和持续时间,热力图则把多个货币对的风险贡献铺成颜色矩阵,暖色越深代表该品种在组合里的尾部风险权重越大。 实盘统计样本里,这套监控下的收益率约 11%,最大浮动回撤压在 1% 以内,外汇和贵金属本身带高杠杆高风险,该数字仅代表特定时段表现,换环境可能明显漂移。 下面这段 Python 是画图与取数核心,接 MT5 跑通后你就能自己复现三张图。

MQL5 / C++
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):
&nbsp;&nbsp;&nbsp;&nbsp;fig, ax = plt.subplots(figsize=(<span class="number">class="num">12</span>, <span class="number">class="num">6</span>))
&nbsp;&nbsp;&nbsp;&nbsp;ax.plot(returns, label=&class="macro">#x27;Actual Returns&class="macro">#x27;)
&nbsp;&nbsp;&nbsp;&nbsp;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;)
&nbsp;&nbsp;&nbsp;&nbsp;ax.fill_between(returns.index, -var_predictions, returns, where=returns &lt; -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>)
&nbsp;&nbsp;&nbsp;&nbsp;ax.legend()
&nbsp;&nbsp;&nbsp;&nbsp;ax.set_title(&class="macro">#x27;VaR vs Actual Returns&class="macro">#x27;)
&nbsp;&nbsp;&nbsp;&nbsp;plt.show()
def plot_drawdown(returns):
&nbsp;&nbsp;&nbsp;&nbsp;drawdown = (returns.cumsum() - returns.cumsum().cummax())
&nbsp;&nbsp;&nbsp;&nbsp;plt.figure(figsize=(<span class="number">class="num">12</span>, <span class="number">class="num">6</span>))
&nbsp;&nbsp;&nbsp;&nbsp;plt.plot(drawdown)
&nbsp;&nbsp;&nbsp;&nbsp;plt.title(&class="macro">#x27;Portfolio Drawdown&class="macro">#x27;)
&nbsp;&nbsp;&nbsp;&nbsp;plt.show()
def plot_var_heatmap(var_matrix):
&nbsp;&nbsp;&nbsp;&nbsp;plt.figure(figsize=(<span class="number">class="num">12</span>, <span class="number">class="num">8</span>))
&nbsp;&nbsp;&nbsp;&nbsp;sns.heatmap(var_matrix, annot=True, cmap=&class="macro">#x27;YlOrRd&class="macro">#x27;)
&nbsp;&nbsp;&nbsp;&nbsp;plt.title(&class="macro">#x27;VaR Heatmap across Currency Pairs&class="macro">#x27;)
&nbsp;&nbsp;&nbsp;&nbsp;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():
&nbsp;&nbsp;&nbsp;&nbsp;print(<span class="class="type">class="kw">string">"Error initializing MetaTrader class="num">5"</span>)
&nbsp;&nbsp;&nbsp;&nbsp;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):
&nbsp;&nbsp;&nbsp;&nbsp;rates = mt5.copy_rates_range(symbol, timeframe, start_date, end_date)
&nbsp;&nbsp;&nbsp;&nbsp;df = pd.DataFrame(rates)
&nbsp;&nbsp;&nbsp;&nbsp;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>)
&nbsp;&nbsp;&nbsp;&nbsp;df.set_index(<span class="class="type">class="kw">string">&class="macro">#x27;time&class="macro">#x27;</span>, inplace=True)
&nbsp;&nbsp;&nbsp;&nbsp;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()
&nbsp;&nbsp;&nbsp;&nbsp;<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>):
&nbsp;&nbsp;&nbsp;&nbsp;<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>):
&nbsp;&nbsp;&nbsp;&nbsp;<span class="keyword">var</span> = calculate_var(returns, confidence_level, holding_period)
&nbsp;&nbsp;&nbsp;&nbsp;<span class="keyword">class="kw">return</span> -returns[returns &lt;= -<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>):
&nbsp;&nbsp;&nbsp;&nbsp;n = len(returns.columns)
&nbsp;&nbsp;&nbsp;&nbsp;
&nbsp;&nbsp;&nbsp;&nbsp;def portfolio_var(weights):
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;portfolio_returns = returns.dot(weights)
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;<span class="keyword">class="kw">return</span> calculate_var(portfolio_returns, confidence_level)
&nbsp;&nbsp;&nbsp;&nbsp;
&nbsp;&nbsp;&nbsp;&nbsp;def portfolio_return(weights):
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;<span class="keyword">class="kw">return</span> np.sum(returns.mean() * weights)
&nbsp;&nbsp;&nbsp;&nbsp;
&nbsp;&nbsp;&nbsp;&nbsp;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>},
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp; {<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})
&nbsp;&nbsp;&nbsp;&nbsp;
&nbsp;&nbsp;&nbsp;&nbsp;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))
&nbsp;&nbsp;&nbsp;&nbsp;
&nbsp;&nbsp;&nbsp;&nbsp;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 重跑一次看权重漂移。

MQL5 / C++
    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 数据。

MQL5 / C++
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 预测、用非线性模型兜住肥尾,是未完的活。市场不停,模型也得跟着进化,现在盖棺定论还太早。

让小布替你跑这套 VaR 巡检
这些诊断小布盯盘的 AIGC 已内置,打开对应品种页即可看到多货币组合的波动率与尾部暴露,把重复劳动交给小布,你专注决策。

常见问题

历史模拟依赖真实行情样本,对跳跃行情更敏感;蒙特卡洛可补全稀薄尾部,但吃算力。实盘多货币组合常用历史模拟做日频、蒙特卡洛做压力测试。
VaR 只给某个分位的可能损失上限,CVaR 进一步求超过该上限后的平均损失,更贴近黑天鹅下的真实回撤。
网格在横盘吃肉、单边崩盘吃土,VaR 主要用来掐掉极端暴露,可能牺牲部分频率但倾向保住本金。
目前小布盯盘内置的是基于报价流的近似风险面板,若你用 Python 导出 VaR 快照到本地,可对照品种页的波动带做交叉验证。
以经纪商服务器时间与本机 UTC 的固差为准,多数情况下是常数,遇夏令时切换需手动校准一次。