外汇掉期套利：构建合成投资组合，创造持续稳定的掉期收益流·综合运用
（3/3）· 从架构到回测，用 MT5 把掉期收益流接进你的实盘工作流
掉期套利分析器的模块骨架
把掉期套利的数学模型真正跑起来，需要一套从取数到出图的闭环。实际落地的分析器按职责切分了组件：先拉市场和掉期数据，再算历史收益与统计指标，接着做组合优化，回测后图形化呈现。每一环独立，改配置不必动底层代码。 可配置性是这类系统的命门。外部参数能穿透分析全流程，比如换经纪商、调市场环境偏好，只改结构体字段即可，不用重写逻辑。 优化算法的默认权重值得记一下：掉期收入 0.3、市场收益 0.6、波动率 0.1。这个配比在十年日数据（2015–2025）的回看中，倾向于在稳定掉期息差与潜在汇价收益间取平衡，而非单纯追夏普。 组合不是挑得分最高的几个货币对，而是把相关性塞进优化问题里求解。结果常出现单看平庸、合起来分散度好的配对——这是局部最优之外更广解空间里探出来的。外汇与贵金属掉期套利涉及杠杆与经纪商点差差异，属于高风险操作，实盘前务必在 MT5 用历史窗口自测。
class SwapArbitrageAnalyzer: def __init__(self, config=None): # Initialization of configuration and basic variables def initialize(self): # Connecting to MetaTrader and verifying data access def analyze(self): # Main method that triggers entire analysis process def _get_current_market_rates(self): # Getting current market prices def _init_swap_data(self): # Initializing swap data def _get_historical_data(self, symbol, start_date): # Getting and handling historical data def _optimize_portfolio(self): # Portfolio optimization def _simulate_portfolio_performance(self): # Simulating performance of optimized portfolio
「组合优化器的参数骨架与 MT5 接入」
这段 Python 配置把多品种组合优化的边界先钉死了：目标成交量 100.0、最多跑 28 个货币对、杠杆写死为 2（注释里标的是 1:10 经纪商环境）、无风险利率 0.001。回测窗口从 2015-01-01 到 2025-03-17，用 datetime 差值算天数直接喂给 optimization_period 和 simulation_days，省得手填。 权重分配很直白：return_weight 0.6 压头，swap_weight 0.3 次之，volatility_weight 0.1 兜底，risk_aversion 设 2.0。每月注资率 monthly_deposit_rate 0.02，即初始资金的 2% 按月补。面板宽高 750×500 只是可视化用，不影响计算。 initialize() 先撞 mt5.initialize()，失败就打印 last_error 并返 False；过了这关再抓 account_info，拿不到也撤。成功才顺手调 _get_current_market_rates 和 _init_swap_data，把 initialized 置真。 _get_historical_data 用 copy_rates_range 拉 D1 数据，start_date 到 now。少于 10 根 bar 直接判废并返回 None；够数就转 DataFrame、算 close 的 pct_change 收益率。swap 取多空较大绝对值，按方向给正负 point 值——这一步决定后面 swap 权重怎么落地。外汇和贵金属杠杆放大波动，实盘接 MT5 前务必在小账户验一遍这些硬编码参数。
def _create_visualizations(self): # Creating visualizations for analyzing results self.config = { &class="macro">#x27;target_volume&class="macro">#x27;: class="num">100.0, &class="macro">#x27;max_pairs&class="macro">#x27;: class="num">28, &class="macro">#x27;leverage&class="macro">#x27;: class="num">2, # Leverage class="num">1:class="num">10 &class="macro">#x27;broker_suffix&class="macro">#x27;: &class="macro">#x27;&class="macro">#x27;, &class="macro">#x27;risk_free_rate&class="macro">#x27;: class="num">0.001, &class="macro">#x27;optimization_period&class="macro">#x27;: class="type">int((class="type">class="kw">datetime(class="num">2025, class="num">3, class="num">17) - class="type">class="kw">datetime(class="num">2015, class="num">1, class="num">1)).days), # С class="num">01.01.class="num">2015 до class="num">17.03.class="num">2025 &class="macro">#x27;panel_width&class="macro">#x27;: class="num">750, &class="macro">#x27;panel_height&class="macro">#x27;: class="num">500, &class="macro">#x27;risk_aversion&class="macro">#x27;: class="num">2.0, &class="macro">#x27;swap_weight&class="macro">#x27;: class="num">0.3, &class="macro">#x27;return_weight&class="macro">#x27;: class="num">0.6, &class="macro">#x27;volatility_weight&class="macro">#x27;: class="num">0.1, &class="macro">#x27;simulation_days&class="macro">#x27;: class="type">int((class="type">class="kw">datetime(class="num">2025, class="num">3, class="num">17) - class="type">class="kw">datetime(class="num">2015, class="num">1, class="num">1)).days), &class="macro">#x27;monthly_deposit_rate&class="macro">#x27;: class="num">0.02 # class="num">2% of the initial capital monthly } def initialize(self): if not mt5.initialize(): print(f"MetaTrader5 initialization failed, error={mt5.last_error()}") class="kw">return False account_info = mt5.account_info() if not account_info: print("Failed to get account information") class="kw">return False print(f"MetaTrader5 initialized. Account: {account_info.login}, Balance: {account_info.balance}") self._get_current_market_rates() self._init_swap_data() self.initialized = True class="kw">return True def _get_historical_data(self, symbol, start_date): try: now = class="type">class="kw">datetime.now() rates = mt5.copy_rates_range(symbol, mt5.TIMEFRAME_D1, start_date, now) if rates is None or len(rates) < class="num">10: print(f"Not enough data for {symbol}: {len(rates) if rates is not None else &class="macro">#x27;None&class="macro">#x27;} bars") class="kw">return None df = pd.DataFrame(rates) df[&class="macro">#x27;time&class="macro">#x27;] = pd.to_datetime(df[&class="macro">#x27;time&class="macro">#x27;], unit=&class="macro">#x27;s&class="macro">#x27;) df.set_index(&class="macro">#x27;time&class="macro">#x27;, inplace=True) df[&class="macro">#x27;class="kw">return&class="macro">#x27;] = df[&class="macro">#x27;close&class="macro">#x27;].pct_change() symbol_info = mt5.symbol_info(symbol) best_swap = max(symbol_info.swap_long, symbol_info.swap_short) swap_in_points = best_swap if symbol_info.swap_long > symbol_info.swap_short else -best_swap point_value = symbol_info.point
◍ 把隔夜利息塞进组合权重里
上面这段逻辑干的事，是把 swap（隔夜利息）折算成收益率后再参与组合优化。swap_return 那行用 (swap_in_points * point_value) / close * leverage 把点数利息摊到净值上，杠杆放大效应直接进表，外汇和贵金属的高杠杆会把正 swap 也放大成可观的被动收益，但反向持仓的利息成本同样被放大，属于典型高风险结构。 优化部分用 scipy 的 SLSQP 跑约束最小化，目标函数把归一化的 swap、市场收益、波动率按 swap_weight / return_weight / volatility_weight 加权合成 combined_score。注意这里 swap 是正权重、波动率是负权重，说明策略倾向挑「利息厚、波动小」的币种对，而不是单纯追收益。 随机抽 pair 那段决定了每次跑出来的持仓数量在 1 到 max_pairs 之间浮动。下面那张实跑结果里，GBPAUD 空单权重 18.45%、swap 2.68，EURNZD 多单 15.22%、swap 3.15，八组货币对全是正 swap，组合层面利息净贡献为正的概率较高，但是具体哪天调仓、抽到哪几对完全随机，不能当稳定收益看。 _simulate_portfolio_performance 里分两条线走：一条只算 market_profit（不含 swap），一条加 daily_swap_applied。回测时把 current_capital_with_swap 单独累计，就能直接比对「带利息再投资」和「纯价差」的曲线裂口。开 MT5 把对应币种的真实 swap 表导出来，替换 self.swap_info 里的 long_swap / short_swap，就能验证你账户下的正 swap 组合会不会比上面这串数字更薄。
df[&class="macro">#x27;swap_return&class="macro">#x27;] = (swap_in_points * point_value) / df[&class="macro">#x27;close&class="macro">#x27;] * self.config[&class="macro">#x27;leverage&class="macro">#x27;] # Leverage account combined_score = (self.config[&class="macro">#x27;swap_weight&class="macro">#x27;] * norm_swap + self.config[&class="macro">#x27;return_weight&class="macro">#x27;] * norm_market - self.config[&class="macro">#x27;volatility_weight&class="macro">#x27;] * norm_vol) result = sco.minimize( objective, initial_weights, args=(np.array(list(expected_returns.values())), cov_matrix.values, self.config[&class="macro">#x27;risk_free_rate&class="macro">#x27;]), method=&class="macro">#x27;SLSQP&class="macro">#x27;, bounds=bounds, constraints=constraints ) num_pairs = random.randint(class="num">1, min(self.config[&class="macro">#x27;max_pairs&class="macro">#x27;], len(eligible_pairs))) eligible_pairs = random.sample(eligible_pairs, num_pairs) optimal_portfolio = {} for i, pair in enumerate(eligible_pairs): if optimal_weights[i] != class="num">0: optimal_portfolio[pair] = optimal_weights[i] self.swap_info[pair][&class="macro">#x27;weight&class="macro">#x27;] = optimal_weights[i] * class="num">100 print("\nOptimal portfolio with positive swap:") for pair, weight in sorted(optimal_portfolio.items(), key=lambda x: abs(x[class="num">1]), reverse=True): direction = &class="macro">#x27;Long&class="macro">#x27; if weight > class="num">0 else &class="macro">#x27;Short&class="macro">#x27; swap_value = self.swap_info[pair][&class="macro">#x27;long_swap&class="macro">#x27;] if weight > class="num">0 else self.swap_info[pair][&class="macro">#x27;short_swap&class="macro">#x27;] print(f"Pair: {pair}, Direction: {direction}, Weight: {abs(weight)*class="num">100:.2f}%, Swap: {swap_value:.2f}") def _simulate_portfolio_performance(self): for date in all_dates: daily_return = class="num">0 daily_swap = class="num">0 for pair, weight in self.optimal_portfolio[&class="macro">#x27;weights&class="macro">#x27;].items(): # calculation of daily class="kw">return and swap pass market_profit = current_capital * daily_return current_capital += market_profit market_profit_with_swap = current_capital_with_swap * daily_return current_capital_with_swap += market_profit_with_swap + daily_swap_applied
把权重图甩进回测前先看清多空占比
上面这段 Python 只做一件事：把优化出来的组合权重按绝对值从大到小排，画一张带多空标记的饼图。long 标 L、short 标 S，权重取绝对值乘 100 显示百分比，颜色用 viridis 线性取色避免相邻块撞色。 图存成 portfolio_proportions.png，分辨率锁在 dpi=100，bbox_inches='tight' 保证标签不被切边。实际跑下来，若某标的权重绝对值 12.3%，图上就显示 12.3% 且后缀带 L 或 S，一眼能看出组合是净多还是净空倾斜。 外汇与贵金属组合带杠杆，多空权重再平衡前务必手算一遍净敞口，这类资产波动放大效应明显，小权重差也可能在极端行情下触发Margin Call。把这张图当作下单前的最后一道肉眼复核，比直接信数字报表更不容易漏看符号。
def _create_visualizations(self): # ... (creating visualizations) # class="num">1. The portfolio and its proportions plt.figure(figsize=(self.config[&class="macro">#x27;panel_width&class="macro">#x27;]/class="num">100, self.config[&class="macro">#x27;panel_height&class="macro">#x27;]/class="num">100), dpi=class="num">100) sorted_weights = sorted(self.optimal_portfolio[&class="macro">#x27;weights&class="macro">#x27;].items(), key=lambda x: abs(x[class="num">1]), reverse=True) pairs = [f"{item[class="num">0]} ({&class="macro">#x27;L&class="macro">#x27; if item[class="num">1] > class="num">0 else &class="macro">#x27;S&class="macro">#x27;})" for item in sorted_weights] weights = [abs(item[class="num">1]) * class="num">100 for item in sorted_weights] colors = plt.cm.viridis(np.linspace(class="num">0, class="num">0.9, len(pairs))) plt.pie(weights, labels=pairs, autopct=&class="macro">#x27;%class="num">1.1f%%&class="macro">#x27;, colors=colors, textprops={&class="macro">#x27;fontsize&class="macro">#x27;: class="num">8}) plt.title(&class="macro">#x27;Portfolio proportions(L=Long, S=Short)&class="macro">#x27;) plt.tight_layout() plt.savefig(&class="macro">#x27;portfolio_proportions.png&class="macro">#x27;, dpi=class="num">100, bbox_inches=&class="macro">#x27;tight&class="macro">#x27;) plt.close() # ... (creating other charts)