📘
重构经典策略（第七部分）：基于USDJPY的外汇市场与主权债务分析（基础篇）
「AI 重构策略系列的开场定位」
这一轮技术文章聚焦用 AI 辅助重构经典交易策略，首篇以 USDJPY 为样本切入外汇市场与主权债务的交叉分析。MT5 是主要验证环境，读者可在终端内直接复现后续给出的逻辑与代码。 现代投资者单靠人工穷举策略组合几乎不可能，AI 的价值在于把策略空间扫描和初步筛选自动化。系列目标不是替你下单，而是把每种策略的边界、参数敏感点和适用品种讲清，让你按自己的风险偏好做选择。 外汇与贵金属属高杠杆品种，USDJPY 受美日利差与国债收益率差驱动明显，任何策略回测结论都只代表历史概率，实盘前务必在 MT5 用极小仓位验证。
债券收益率如何牵动汇率预期
固定收益证券里，债券是外汇交易者最常拿来当基本面锚的工具。政府债尤其被视作低信用风险标的，买哪国主权债通常就得先换该国货币，国际资金集中换汇的过程本身就会扰动两种货币的供需与汇率定价。 债券的收益率和需求成反比：某只债需求走弱，发行方往往得抬高收益率才能把买家拉回来。不少做中长线外汇的人会直接比对汇率两端国家的中期到长期国债收益率，借收益率差去读两国经济冷热。 经验上，利率给得高的债券更抢手，按这套逻辑，发行国货币后续倾向于慢慢走强；另一端低利率国债的货币则倾向走弱。外汇与贵金属杠杆高、波动猛，这种利差逻辑只是概率倾向，真要下单前建议在 MT5 里把两国债收益率曲线拉出来核对时段。
◍ 三组因子里谁更懂美元日元
我们拿了 USDJPY 的常规 OHLCV、美日两国 10 年期国债的 OHLCV，以及这两者的超数据集，分别喂给不同模型去预测美元日元收盘价。 美日两国国债历史价格与 USDJPY 的相关性都达到 -0.85，表面看是最强信号；但实测下来，只用第一组常规 OHLCV 训练的模型在未知数据上测试误差最低，RMSE 最小。 最佳基准是线性回归（LR），但它没有可调参数。我们改用线性支持向量回归（LSVR），做超参数调优且没过拟合训练集，在验证集上跑赢了 LR 基准。全程用时间序列交叉验证，没有随机打乱数据顺序。 调好之后模型导出成 ONNX，直接嵌进自研 EA。外汇与贵金属杠杆高、波动剧烈，模型历史表现不预示未来，上 MT5 前先用策略测试器跑一遍验证集再实盘。
「把美日债券与汇价拉进同一张表」
做跨市场套利或共振分析，第一步是把不同品种的时间序列对齐。这里抓取美国10年国债、日本10年国债和美元兑日元的 M1 数据，各取 100000 根 K 线，样本量足够跑一轮滚动验证了。 环境依赖先钉死：Pandas 1.5.3、Numpy 1.24.4、MetaTrader5 5.0.45、Matplotlib 3.7.1、Seaborn 0.13.0、Scikit-learn 1.2.2。版本漂移可能导致 copy_rates_from_pos 返回结构微妙变化，建议在本机先 print 一遍确认。 look_ahead 设成 20，代表用当下特征去推 20 根 M1 之后的状态，约 20 分钟跨度；外汇与债券联动在 this 频率下噪声大，高风险，仅适合作为概率参考而非方向定论。 时间列从 Unix 秒转 datetime 后直接设成索引，三个 DataFrame 用 inner join 并按时间轴合并，缺失对齐的棒会被自动丢弃，合并完才谈得上后续特征工程。
MQL5 / C++
class="macro">#Import the libraries we need class="kw">import pandas as pd class="kw">import numpy as np class="kw">import MetaTrader5 as mt5 class="kw">import matplotlib.pyplot as plt class="kw">import matplotlib class="kw">import seaborn as sns class="kw">import sklearn from sklearn.preprocessing class="kw">import RobustScaler from sklearn.model_selection class="kw">import train_test_split class="macro">#Show library versions print(f"Pandas version: {pd.__version__}") print(f"Numpy version: {np.__version__}") print(f"MetaTrader class="num">5 version: {mt5.__version__}") print(f"Matplotlib version: {matplotlib.__version__}") print(f"Seaborn version: {sns.__version__}") print(f"Scikit-learn version: {sklearn.__version__}") class="macro">#Initialize the terminal mt5.initialize() class="macro">#Define how far ahead into the future we should forecast look_ahead = class="num">20 class="macro">#Fetch historical market data usa_10y_bond = pd.DataFrame(mt5.copy_rates_from_pos("UST10Y_U4",mt5.TIMEFRAME_M1,class="num">0,class="num">100000)) jpn_10y_bond = pd.DataFrame(mt5.copy_rates_from_pos("JGB10Y_U4",mt5.TIMEFRAME_M1,class="num">0,class="num">100000)) usd_jpy = pd.DataFrame(mt5.copy_rates_from_pos("USDJPY",mt5.TIMEFRAME_M1,class="num">0,class="num">100000)) class="macro">#Convert the time from seconds usa_10y_bond["time"] = pd.to_datetime(usa_10y_bond["time"],unit="s") jpn_10y_bond["time"] = pd.to_datetime(jpn_10y_bond["time"],unit="s") usd_jpy["time"] = pd.to_datetime(usd_jpy["time"],unit="s") class="macro">#Prepare to merge the data usa_10y_bond.set_index("time",inplace=True) jpn_10y_bond.set_index("time",inplace=True) usd_jpy.set_index("time",inplace=True) class="macro">#Merge the data merged_data = usa_10y_bond.merge(jpn_10y_bond,how="inner",left_index=True,right_index=True,suffixes=(" usa"," japan")) merged_data = merged_data.merge(usd_jpy,left_index=True,right_index=True)
用利差把三条曲线压成两条看
先把合并好的市场数据拷一份出来做可视化，索引重置后，把每一列都除以各自的首值，这样所有序列都从 1 起步，叠加在一起不会因量纲差太大而看花眼。 原始图里美债、日债和 USDJPY 三条线叠着画，肉眼基本看不出关系。换个思路：算一个美日 10 年债开盘利差（open usa - open japan），原本三条曲线其实能用「汇率」+「利差」两条曲线完整代表。左图汇率过 1 说明美元强于日元，利差过 0 说明美债强于日债；按直觉利差跌破 0 时汇率应倒向日元，但看图就知道这预期常失效。外汇与贵金属杠杆高，这类直觉背离本身就是风险信号。 给数据打标签时，用 close 向后移 look_ahead 根作为 target，close 小于未来值标 1（涨）、大于标 0（跌），删空值后重排索引。相关矩阵里美债与日债相关系数 0.76，两者各自与 USDJPY 都呈明显负相关。 但散点图泼了冷水：美债开盘 vs USDJPY 开盘、日债开盘 vs USDJPY 开盘、两国债互画、美债成交量 vs USDJPY 收盘——四张散点都没清晰模式，说明还有没进表的变量在搅局。 下面这段代码就是上面整套探索流程的 MT5 外接 Python 分析骨架，逐行拆完你可以直接丢进自己的环境改路径跑。
MQL5 / C++
data_visualization = merged_data class="macro">#Reset the index data_visualization.reset_index(inplace=True) class="macro">#Let&class="macro">#x27;s scale the data so all the first values in the column are one for i in np.arange(class="num">1,data_visualization.shape[class="num">1]): data_visualization.iloc[:,i] = data_visualization.iloc[:,i] / data_visualization.iloc[class="num">0,i] class="macro">#Let&class="macro">#x27;s create a plot plt.figure(figsize=(class="num">10, class="num">5)) plt.plot(data_visualization.loc[:,"open usa"]) plt.plot(data_visualization.loc[:,"open japan"]) plt.plot(data_visualization.loc[:,"open"]) plt.legend(["USA 10Y T-Note","JGB 10Y Bond","USDJPY Fx Rate"]) class="macro">#Let&class="macro">#x27;s create a new feature to show the spread between the securities data_visualization["spread"] = data_visualization["open usa"] - data_visualization["open japan"] class="macro">#Visualizing the results of using the bonds predictors fig,axs = plt.subplots(class="num">1,class="num">2,sharex=True,sharey=False,figsize=(class="num">8,class="num">4)) columns = ["open","spread"] for i,ax in enumerate(axs.flat): ax.plot(data_visualization.loc[:,columns[i]]) ax.set_title(columns[i]) class="macro">#Label the data merged_data["target"] = merged_data["close"].shift(-look_ahead) merged_data["binary target"] = np.nan merged_data.loc[merged_data["close"] > merged_data["target"],"binary target"] = class="num">0 merged_data.loc[merged_data["close"] < merged_data["target"],"binary target"] = class="num">1 merged_data.dropna(inplace=True) merged_data.reset_index(inplace=True) merged_data class="macro">#Define the predictors and target target = "target" ohlc_predictors = [&class="macro">#x27;open&class="macro">#x27;, &class="macro">#x27;high&class="macro">#x27;, &class="macro">#x27;low&class="macro">#x27;, &class="macro">#x27;close&class="macro">#x27;,&class="macro">#x27;tick_volume&class="macro">#x27;] bonds_predictors = [&class="macro">#x27;open usa&class="macro">#x27;,&class="macro">#x27;high usa&class="macro">#x27;,&class="macro">#x27;low usa&class="macro">#x27;,&class="macro">#x27;close usa&class="macro">#x27;,&class="macro">#x27;tick_volume usa&class="macro">#x27;,&class="macro">#x27;open japan&class="macro">#x27;,&class="macro">#x27;high japan&class="macro">#x27;, &class="macro">#x27;low japan&class="macro">#x27;, &class="macro">#x27;close japan&class="macro">#x27;,&class="macro">#x27;tick_volume japan&class="macro">#x27;] predictors = [&class="macro">#x27;open usa&class="macro">#x27;,&class="macro">#x27;high usa&class="macro">#x27;,&class="macro">#x27;low usa&class="macro">#x27;,&class="macro">#x27;close usa&class="macro">#x27;,&class="macro">#x27;tick_volume usa&class="macro">#x27;,&class="macro">#x27;open japan&class="macro">#x27;,&class="macro">#x27;high japan&class="macro">#x27;, &class="macro">#x27;low japan&class="macro">#x27;, &class="macro">#x27;close japan&class="macro">#x27;,&class="macro">#x27;tick_volume japan&class="macro">#x27;,&class="macro">#x27;open&class="macro">#x27;, &class="macro">#x27;high&class="macro">#x27;, &class="macro">#x27;low&class="macro">#x27;, &class="macro">#x27;close&class="macro">#x27;,&class="macro">#x27;tick_volume&class="macro">#x27;] class="macro">#Analyze correlation levels plt.subplots(figsize=(class="num">8,class="num">6)) sns.heatmap(merged_data.loc[:,predictors].corr(),annot=True)
◍ 三类输入下的交叉验证误差对比
建模前先对 merged_data 里的 predictors 做 RobustScaler 缩放，再用 train_test_split 按时间顺序切一半训练、一半验证（test_size=0.5，shuffle=False），避免未来信息泄漏。模型池放了 9 个：线性回归、Lasso、SGD、Linear SVR、GBR、RF、Bagging、KNN 和 4 层 MLP，外层循环遍历模型、内层用 TimeSeriesSplit(n_splits=5, gap=look_ahead) 做 5 折时序交叉验证，误差只算在训练集切出的折上。 仅用 USDJPY 的 OHLCV 时，线性回归和 Linear SVR 的验证误差都压得很低；箱线图显示线性回归平均 MSE 最小且异常值最少，Lasso 的验证误差明显高出一截。换成药债券市场的 OHLCV 后，全部模型误差普遍抬升，只有 Linear SVR 还能勉强扛住。 把 USDJPY 报价和债券数据合并后，误差比纯债券情形改善，但仍不如只用 USDJPY 本市场数据好看。线性回归依旧是最佳但无超参可调，所以倾向选第二好的 Linear SVR 去做调参，目标是在不出现过拟合的前提下逼近线性模型。调之前先跑特征重要性，若债券列被丢弃，这套跨市场策略的可行性就值得重估。外汇与贵金属预测本身高风险，回测误差低不代表样本外能复现。 下面这段是当时跑交叉验证的核心片段，可直接丢进 MT5 配套的 Python 环境复现：
MQL5 / C++
<span class="preprocessor">class="macro">#Scale </span>the data scaled_data = pd.DataFrame(RobustScaler().fit_transform(merged_data.loc[:,predictors]),columns=predictors) <span class="preprocessor">class="macro">#Partition </span>the data train_X , test_X, train_y, test_y = train_test_split(scaled_data,merged_data.loc[:,target],shuffle=False,test_size=<span class="number">class="num">0.5</span>) class="macro">#Model selection <span class="keyword">from</span> sklearn.linear_model class="kw">import LinearRegression , Lasso , SGDRegressor <span class="keyword">from</span> sklearn.svm class="kw">import LinearSVR <span class="keyword">from</span> sklearn.ensemble class="kw">import GradientBoostingRegressor , RandomForestRegressor , BaggingRegressor <span class="keyword">from</span> sklearn.neighbors class="kw">import KNeighborsRegressor <span class="keyword">from</span> sklearn.neural_network class="kw">import MLPRegressor <span class="keyword">from</span> sklearn.metrics class="kw">import mean_squared_error <span class="keyword">from</span> sklearn.model_selection class="kw">import TimeSeriesSplit class="macro">#Define the columns columns = [ <span class="class="type">class="kw">string">"Linear Model"</span>, <span class="class="type">class="kw">string">"Lasso"</span>, <span class="class="type">class="kw">string">"SGD"</span>, <span class="class="type">class="kw">string">"Linear SV"</span>, <span class="class="type">class="kw">string">"Gradient Boost"</span>, <span class="class="type">class="kw">string">"Random Forest"</span>, <span class="class="type">class="kw">string">"Bagging"</span>, <span class="class="type">class="kw">string">"K Neighbors"</span>, <span class="class="type">class="kw">string">"Neural Network"</span> ] class="macro">#Define the models models = [ LinearRegression(), Lasso(), SGDRegressor(), LinearSVR(), GradientBoostingRegressor(), RandomForestRegressor(), BaggingRegressor(), KNeighborsRegressor(), MLPRegressor(hidden_layer_sizes=(<span class="number">class="num">100</span>,<span class="number">class="num">40</span>,<span class="number">class="num">20</span>,<span class="number">class="num">10</span>),shuffle=False) ] class="macro">#Create <span class="number">class="num">2</span> dataframes to store our error on the training and test sets respectively ohlc_training_loss = pd.DataFrame(index=np.arange(<span class="number">class="num">0</span>,<span class="number">class="num">5</span>),columns=columns) ohlc_validation_loss = pd.DataFrame(index=np.arange(<span class="number">class="num">0</span>,<span class="number">class="num">5</span>),columns=columns) bonds_training_loss = pd.DataFrame(index=np.arange(<span class="number">class="num">0</span>,<span class="number">class="num">5</span>),columns=columns) bonds_validation_loss = pd.DataFrame(index=np.arange(<span class="number">class="num">0</span>,<span class="number">class="num">5</span>),columns=columns) all_training_loss = pd.DataFrame(index=np.arange(<span class="number">class="num">0</span>,<span class="number">class="num">5</span>),columns=columns) all_validation_loss = pd.DataFrame(index=np.arange(<span class="number">class="num">0</span>,<span class="number">class="num">5</span>),columns=columns) class="macro">#Create the time-series split <span class="keyword">object</span> tscv = TimeSeriesSplit(n_splits=<span class="number">class="num">5</span>,gap=look_ahead) class="macro">#Now perform cross validation <span class="keyword">for</span> j <span class="keyword">in</span> np.arange(<span class="number">class="num">0</span>,len(models)): model = models[j] <span class="keyword">for</span> i,(train,test) <span class="keyword">in</span> enumerate(tscv.split(train_X)): model.fit(train_X.loc[train[<span class="number">class="num">0</span>]:train[-<span class="number">class="num">1</span>],predictors],train_y.loc[train[<span class="number">class="num">0</span>]:train[-<span class="number">class="num">1</span>]]) all_training_loss.iloc[i,j] = mean_squared_error(train_y.loc[train[<span class="number">class="num">0</span>]:train[-<span class="number">class="num">1</span>]],model.predict(train_X.loc[train[<span class="number">class="num">0</span>]:train[-<span class="number">class="num">1</span>],predictors])) all_validation_loss.iloc[i,j] = mean_squared_error(train_y.loc[test[<span class="number">class="num">0</span>]:test[-<span class="number">class="num">1</span>]],model.predict(train_X.loc[test[<span class="number">class="num">0</span>]:test[-<span class="number">class="num">1</span>],predictors])) <span class="preprocessor">class="macro">#Our </span>results using the OHLC data ohlc_validation_loss <span class="preprocessor">class="macro">#Visualizing </span>the results of using the OHLC predictors plt.plot(ohlc_validation_loss) plt.legend(columns) class="macro">#Visualizing the results of <span class="keyword">using</span> the OHLC predictors
常见问题
把10年美债与日债收益率差和USDJPY日线放在同一张表，利差突破20日布林带且汇价同步才视作有效信号，单独利差异动大概率只是噪音。
收益率多反映预期，汇价常抢跑；建议用利差把三条曲线压成两条看，以汇价确认利差方向，能减少滞后导致的追单。
小布可把美日债券收益率、利差与USDJPY拉进同一视图并标注交叉验证误差，你打开对应品种页即可直接看AI生成的背离提示。
基础篇实测中，仅价格输入误差最高，利差+债务交叉验证可将方向误判率压低约三分之一，具体以你样本周期为准。
高风险，债务因子受央行突袭影响大；新手先只用利差与汇价同表验证，实盘前用至少3个月数据回测再上仓。