在市场中获得优势·进阶篇
（2/3）·传统指标同质化严重，把美联储公开时序接进 MT5 才可能看到别人漏掉的利率信号
把多周期均线一次性导成 CSV
做价格行为复盘时，MT5 图表上的均线只能看不能算。上面这段逻辑在 EA 里先建了一条 150 周期 EMA 句柄：iMA(_Symbol, PERIOD_CURRENT, 150, 0, MODE_EMA, PRICE_CLOSE)，周期取当前图表周期，平滑用指数均线，价格源是收盘价。 随后用 CopyBuffer 把 5 / 15 / 30 / 150 四条均线的缓冲区拉进数组，每条都接一句 ArraySetAsSeries(..., true)，把索引方向翻成时间序（i=0 是最新一根）。不翻的话，后面写文件时 i 的对应关系会整反。 写文件部分用 FileOpen 开一个 CSV（逗号分隔、ANSI 编码），循环从 i=-1 到 size：首行 i==-1 写表头 Time/Open/High/Low/Close/MA 5/MA 15/MA 30/MA 150，其余行用 iTime/iOpen/iHigh/iLow/iClose 配四个均线读数落盘。 导出后切到 Python 侧，四行 import 把 pandas / numpy / matplotlib / seaborn 载入，再用 pd.read_csv 读回 MT5 的 Files 目录下那个 GBPUSD Market Data As Series.csv。路径里那串 NVUSDVJSNDU3483408FVKDL 是你本机 Terminal 实例 ID，得换成自己的。外汇与贵金属波动剧烈，导出的历史数据仅用于统计观察，实盘决策仍属高风险。
ma_handle_150 = iMA(_Symbol,PERIOD_CURRENT,class="num">150,class="num">0,MODE_EMA,PRICE_CLOSE); class=class="str">"cmt">//---Copy indicator values CopyBuffer(ma_handle_5,class="num">0,class="num">0,size_fetch,ma_reading_5); ArraySetAsSeries(ma_reading_5,true); CopyBuffer(ma_handle_15,class="num">0,class="num">0,size_fetch,ma_reading_15); ArraySetAsSeries(ma_reading_15,true); CopyBuffer(ma_handle_30,class="num">0,class="num">0,size_fetch,ma_reading_30); ArraySetAsSeries(ma_reading_30,true); CopyBuffer(ma_handle_150,class="num">0,class="num">0,size_fetch,ma_reading_150); ArraySetAsSeries(ma_reading_150,true); class=class="str">"cmt">//---Write to file class="type">int file_handle=FileOpen(file_name,FILE_WRITE|FILE_ANSI|FILE_CSV,","); for(class="type">int i=-class="num">1;i<=size;i++){ if(i == -class="num">1){ FileWrite(file_handle,"Time","Open","High","Low","Close","MA class="num">5","MA class="num">15","MA class="num">30","MA class="num">150"); } else{ FileWrite(file_handle,iTime(_Symbol,PERIOD_CURRENT,i), iOpen(_Symbol,PERIOD_CURRENT,i), iHigh(_Symbol,PERIOD_CURRENT,i), iLow(_Symbol,PERIOD_CURRENT,i), iClose(_Symbol,PERIOD_CURRENT,i), ma_reading_5[i], ma_reading_15[i], ma_reading_30[i], ma_reading_150[i] ); } } } class=class="str">"cmt">//+------------------------------------------------------------------+ class="kw">import pandas as pd class="kw">import numpy as np class="kw">import matplotlib.pyplot as plt class="kw">import seaborn as sns GBPUSD = pd.read_csv("C:\\Enter\\Your\\Path\\Here\\MetaQuotes\\Terminal\\NVUSDVJSNDU3483408FVKDL\\MQL5\\Files\\GBPUSD Market Data As Series.csv")
◍ 把英镑隔夜利率塞进回归特征里
这段处理把 GBPUSD 的收盘序列做了 30 根 K 线前向平移，生成监督学习的 Target 列：shift(-30) 意味着用当前 OHLC 与均线去猜 30 根之后的收盘价，外汇与贵金属杠杆高，这类多步预测仅作概率参考。 SOIA 是英镑隔夜指数均值，原始 CSV 里缺失值用点号占位，代码先把它替换成 0，再用非 0 值的均值回填，避免线性回归被脏数据带偏。SOFR 同步读取后按日期索引与价格表 merge，最终特征池 = 常规 4 价 + MA5/15/30/150 + IUDSOIA + SOFR。 用 TimeSeriesSplit 做时序交叉验证，gap 设成 look_ahead+10 = 40，防止训练集泄漏未来信息；30 次切分下分别跑普通特征、替代数据特征、全特征三组线性回归，MSE 写进 accuracy 表。 回测出口落在 'Alternative Data Target Look Ahead 30.csv'，开盘 MT5 导出自己的 GBPUSD 日线，按同样 30 根前瞻窗口重跑，就能比对加不加利率因子时 MSE 谁更小。
GBPUSD = GBPUSD[::-class="num">1] GBPUSD.reset_index(inplace=True) look_ahead = class="num">30 splits = class="num">30 GBPUSD["Target"] = GBPUSD["Close"].shift(-look_ahead) GBPUSD.dropna(inplace=True) GBPUSD["Time"] = pd.to_datetime(GBPUSD["Time"]) GBPUSD.set_index("Time",inplace=True) SOIA = pd.read_csv("C:\\Enter\\Your\\Path\\Here\\Downloads\\FED Data\\Daily Sterling Overnight Index Average\\IUDSOIA.csv") SOIA["DATE"] = pd.to_datetime(SOIA["DATE"]) SOIA["IUDSOIA"] = SOIA["IUDSOIA"].replace(".","class="num">0") SOIA["IUDSOIA"] = pd.to_numeric(SOIA["IUDSOIA"]) non_zero_mean = SOIA.loc[SOIA[&class="macro">#x27;IUDSOIA&class="macro">#x27;] != class="num">0, &class="macro">#x27;IUDSOIA&class="macro">#x27;].mean() SOIA[&class="macro">#x27;IUDSOIA&class="macro">#x27;] = SOIA[&class="macro">#x27;IUDSOIA&class="macro">#x27;].replace(class="num">0, non_zero_mean) SOIA.set_index("DATE",inplace=True) SOFR = pd.read_csv("C:\\Enter\\Your\\Path\\Here\\Downloads\\FED Data\\Secured Overnight Financing Rate\\SOFR.csv") merged_df = SOIA.merge(SOFR,left_index=True,right_index=True) merged_df = merged_df.merge(GBPUSD,left_index=True,right_index=True) merged_df.reset_index(inplace=True) merged_df.drop(columns=["index"],inplace=True) normal_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;MA class="num">5&class="macro">#x27;, &class="macro">#x27;MA class="num">15&class="macro">#x27;,&class="macro">#x27;MA class="num">30&class="macro">#x27;, &class="macro">#x27;MA class="num">150&class="macro">#x27;] alternative_predictors = [&class="macro">#x27;IUDSOIA&class="macro">#x27;, &class="macro">#x27;SOFR&class="macro">#x27;] all_predictors = normal_predictors + alternative_predictors accuracy = pd.DataFrame(columns=["Normal","Alternative","All"],index=np.arange(class="num">0,splits)) merged_df merged_df.to_csv(&class="macro">#x27;Alternative Data Target Look Ahead class="num">30.csv&class="macro">#x27;) from sklearn.preprocessing class="kw">import StandardScaler scaled_data = merged_df.loc[:,all_predictors] scaler = StandardScaler() scaler.fit(scaled_data) scaled_data = pd.DataFrame(scaler.transform(scaled_data),index=merged_df.index,columns=all_predictors) from sklearn.linear_model class="kw">import LinearRegression from sklearn.metrics class="kw">import mean_squared_error from sklearn.model_selection class="kw">import TimeSeriesSplit tscv = TimeSeriesSplit(gap=look_ahead+class="num">10,n_splits=splits) for i,(train,test) in enumerate(tscv.split(merged_df)): model = LinearRegression() model.fit(scaled_data.loc[train[class="num">0]:train[-class="num">1],all_predictors],merged_df.loc[train[class="num">0]:train[-class="num">1],"Target"]) accuracy["All"][i] = mean_squared_error(merged_df.loc[test[class="num">0]:test[-class="num">1],"Target"],model.predict(scaled_data.loc[test[class="num">0]:test[-class="num">1],all_predictors])) fig,axs = plt.subplots(class="num">1,class="num">3,sharex=True,sharey=True,figsize=(class="num">16,class="num">4)) for i,ax in enumerate(axs.flat): ax.boxplot(merged_df.iloc[:,i]) ax.set_title(accuracy.columns[i]) sns.scatterplot(x=merged_df["IUDSOIA"],y=merged_df["Target"])
「按数据脾气挑模型，别迷信深度网络」
模型选型第一件事是看数据体量。行数不到一万、特征列少于 30 的盘面小样本，复杂网络基本学不到真规律，反而容易把噪声背下来；这种规模用线性类方法过度拟合的概率更低。 噪声是另一道坎。缺值多、随机跳动解释不了的行情片段，复杂模型方差高，极易在嘈杂交易日跑偏。实盘里遇到特别脏的会话，宁可空仓也别硬上模型。 维度高就得先降维。特征选择加预处理能帮模型聚焦，不然高列数会把简单算法的边界也拖垮。 若数据量够大、噪声可控且机器算力富余，深度网络才有用武之地——它能啃下非线性结构，但前提是清洗和算力都到位。外汇与贵金属波动杠杆高，模型误判会快速放大亏损，任何信号都只是概率倾向。 我们跑了一轮对照：用 Alternative Data.csv 里的 IUDSOIA、SOFR 及 OHLC 作特征，Target 作标签，TimeSeriesSplit 切 10 段、gap=30 防泄漏。误差摘要里，线性回归均方误差 0.002224、方差 1.34e-06；Sigmoid SVR 直接崩到均方误差 1.669、方差 27.65，明显失稳。 所有模型误差几乎同一水平，简单模型因低方差更可能随时间稳住。本例选线性回归进策略：它的误差不输别人，且过度拟合可能性最小。 下面这段 Python 类伪代码可直接改路径复现箱线图对照，注意它调用的是 sklearn 系，不是 MT5 内置，但特征构造思路能搬去 MQL5 自定义指标。 代码逐行拆解： import statistics as st —— 引入统计模块备用 import pandas as pd —— 读 CSV 做表 import numpy as np —— 数值计算 import seaborn as sns / matplotlib.pyplot as plt —— 画图 from sklearn.linear_model import Lasso,Ridge,LinearRegression —— 三类线性模型 from sklearn.ensemble import RandomForestRegressor —— 树集成 from sklearn.model_selection import TimeSeriesSplit —— 时序切分防未来函数 from sklearn.metrics import mean_squared_error —— 算误差 from sklearn.svm import SVR —— 支持向量回归 from sklearn.preprocessing import StandardScaler —— 标准化 from xgboost import XGBRegressor —— 梯度提升 csv = pd.read_csv("C:\\Enter\\Your\\Path\\Here\\Alternative Data.csv") —— 换你本地路径 predictors = [...] / target = 'Target' —— 特征与目标列 scaled_data = csv.loc[:,predictors] 然后 StandardScaler 拟合变换 —— 去量纲 tscv = TimeSeriesSplit(n_splits=10,gap=30) —— 10 折带间隙 error_df 建空表存各模型误差 循环里只示范了 XGBRegressor 的 fit 与误差记录，其余模型同理填列 最后 plt.subplots(2,5) 把 10 个模型箱线图排出来看分布
class="kw">import statistics <span class="keyword">as</span> st class="kw">import pandas <span class="keyword">as</span> pd class="kw">import numpy <span class="keyword">as</span> np class="kw">import seaborn <span class="keyword">as</span> sns class="kw">import matplotlib.pyplot <span class="keyword">as</span> plt <span class="keyword">from</span> sklearn.linear_model class="kw">import Lasso,Ridge,LinearRegression <span class="keyword">from</span> sklearn.ensemble class="kw">import RandomForestRegressor <span class="keyword">from</span> sklearn.model_selection class="kw">import TimeSeriesSplit <span class="keyword">from</span> sklearn.metrics class="kw">import mean_squared_error <span class="keyword">from</span> sklearn.svm class="kw">import SVR <span class="keyword">from</span> sklearn.preprocessing class="kw">import StandardScaler <span class="keyword">from</span> xgboost class="kw">import XGBRegressor csv = pd.read_csv(<span class="class="type">class="kw">string">"C:\Enter\Your\Path\Here\Alternative Data.csv"</span>) predictors = [&class="macro">#x27;IUDSOIA&class="macro">#x27;, &class="macro">#x27;SOFR&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;] target = &class="macro">#x27;Target&class="macro">#x27; scaled_data = csv.loc[:,predictors] scaler = StandardScaler() scaler.fit(scaled_data) scaled_data = scaler.transform(scaled_data) splits = class="num">10 gap = class="num">30 models = [&class="macro">#x27;Linear&class="macro">#x27;,&class="macro">#x27;Lasso&class="macro">#x27;,&class="macro">#x27;Ridge&class="macro">#x27;,&class="macro">#x27;Random Forest&class="macro">#x27;,&class="macro">#x27;Linear SVR&class="macro">#x27;,&class="macro">#x27;Sigmoid SVR&class="macro">#x27;,&class="macro">#x27;RBF SVR&class="macro">#x27;,&class="macro">#x27;class="num">2 Poly SVR&class="macro">#x27;,&class="macro">#x27;class="num">3 Poly SVR&class="macro">#x27;,&class="macro">#x27;XGB&class="macro">#x27;] tscv = TimeSeriesSplit(n_splits=splits,gap=gap) error_df = pd.DataFrame(index=np.arange(class="num">0,splits),columns=models) <span class="keyword">for</span> i,(train,test) in enumerate(tscv.split(csv)): model=XGBRegressor() model.fit(scaled_data.loc[train[<span class="number">class="num">0</span>]:train[-<span class="number">class="num">1</span>],predictors],csv.loc[train[<span class="number">class="num">0</span>]:train[-<span class="number">class="num">1</span>],target]) error_df.iloc[i,<span class="number">class="num">9</span>] =mean_squared_error(csv.loc[test[<span class="number">class="num">0</span>]:test[-<span class="number">class="num">1</span>],target],model.predict(scaled_data.loc[test[<span class="number">class="num">0</span>]:test[-<span class="number">class="num">1</span>],predictors])) error_df fig , axs = plt.subplots(<span class="number">class="num">2</span>,<span class="number">class="num">5</span>,figsize=(<span class="number">class="num">20</span>,<span class="number">class="num">20</span>),sharex=True) <span class="keyword">for</span> i,ax in enumerate(axs.flat): ax.boxplot(error_df.iloc[:,i]) ax.set_title(error_df.columns[i])
把宏观数据接进MT5的日内循环
前面几节拆开的FRED另类数据和MT5行情接口，到这里直接拼成一条可跑的日线策略。核心是用Python把SOFR、SOIA两个利率序列和GBPUSD的OHLC拼成模型输入，再让算法每天重判一次方向。 登录段先用mt5.initialize带账号密码服务器连终端，连上会打印‘Logged in successfully’；随后遍历symbols_get找GBPUSD，读出volume_min——实测该品种最小交易量就是0.01，VOLUME据此定为0.01*LOT_MULTIPLE。 get_prices()取2024-03-20到当前的D1棒，返回最新一根的OHLC；get_alternative_data()用Fred实例拉SOFR和IUDSOIA的当日值。两者在get_model_inputs()里并成6列数组：Open/High/Low/Close/IUDSOIA/SOFR，正好对应LinearRegression训练时的predictors。 训练只一行model.fit，读本地Alternative Data.csv，目标列是‘Target’。之后进while True：每天取positions总数和ai_forecast()，若现价高于预测则SELL_STATE为真，反之BUY_STATE为真；无持仓就按状态开仓，有持仓且方向相反才平仓，跑完time.sleep到次日。 注意代码里elif(current_price > forecast)和if条件写重了，BUY_STATE分支实际永远进不去，真要跑得把第二个改成current_price < forecast。外汇与贵金属杠杆高，这类基于利率差的日线模型信号失效时回撤可能很快，上MT5前先改这处逻辑漏洞。
from fredapi class="kw">import Fred class="kw">import MetaTrader5 as mt5 class="kw">import pandas as pd class="kw">import numpy as np class="kw">import time from class="type">class="kw">datetime class="kw">import class="type">class="kw">datetime class="kw">import matplotlib.pyplot as plt LOGIN = ENTER_YOUR_LOGIN PASSWORD = &class="macro">#x27;ENTER_YOUR_PASSWORD&class="macro">#x27; SERVER = &class="macro">#x27;ENTER_YOUR_SERVER&class="macro">#x27; SYMBOL = &class="macro">#x27;GBPUSD&class="macro">#x27; TIMEFRAME = mt5.TIMEFRAME_D1 DEVIATION = class="num">1000 VOLUME = class="num">0 LOT_MULTIPLE = class="num">1 FRED = Fred(api_key=&class="macro">#x27;ENTER_YOUR_API_KEY&class="macro">#x27;) if mt5.initialize(login=LOGIN,password=PASSWORD,server=SERVER): print(&class="macro">#x27;Logged in successfully&class="macro">#x27;) else: print(&class="macro">#x27;Failed To Log in&class="macro">#x27;) for index,symbol in enumerate(mt5.symbols_get()): if symbol.name == SYMBOL: print(f"{symbol.name} has minimum volume: {symbol.volume_min}") VOLUME = symbol.volume_min * LOT_MULTIPLE def get_prices(): start = class="type">class="kw">datetime(class="num">2024,class="num">3,class="num">20) end = class="type">class="kw">datetime.now() data = pd.DataFrame(mt5.copy_rates_range(SYMBOL,TIMEFRAME,start,end)) data[&class="macro">#x27;time&class="macro">#x27;] = pd.to_datetime(data[&class="macro">#x27;time&class="macro">#x27;],unit=&class="macro">#x27;s&class="macro">#x27;) data.set_index(&class="macro">#x27;time&class="macro">#x27;,inplace=True) class="kw">return(data.iloc[-class="num">1,:]) def get_alternative_data(): SOFR = FRED.get_series_as_of_date(&class="macro">#x27;SOFR&class="macro">#x27;,class="type">class="kw">datetime.now()) SOFR = SOFR.iloc[-class="num">1,-class="num">1] SOIA = FRED.get_series_as_of_date(&class="macro">#x27;IUDSOIA&class="macro">#x27;,class="type">class="kw">datetime.now()) SOIA = SOIA.iloc[-class="num">1,-class="num">1] class="kw">return(SOFR,SOIA) def get_model_inputs(): LAST_OHLC = get_prices() SOFR , SOIA = get_alternative_data() MODEL_INPUT_DF = pd.DataFrame(index=np.arange(class="num">0,class="num">1),columns=predictors) MODEL_INPUT_DF[&class="macro">#x27;Open&class="macro">#x27;] = LAST_OHLC[&class="macro">#x27;open&class="macro">#x27;] MODEL_INPUT_DF[&class="macro">#x27;High&class="macro">#x27;] = LAST_OHLC[&class="macro">#x27;high&class="macro">#x27;] MODEL_INPUT_DF[&class="macro">#x27;Low&class="macro">#x27;] = LAST_OHLC[&class="macro">#x27;low&class="macro">#x27;] MODEL_INPUT_DF[&class="macro">#x27;Close&class="macro">#x27;] = LAST_OHLC[&class="macro">#x27;close&class="macro">#x27;] MODEL_INPUT_DF[&class="macro">#x27;IUDSOIA&class="macro">#x27;] = SOIA MODEL_INPUT_DF[&class="macro">#x27;SOFR&class="macro">#x27;] = SOFR model_input_array = np.array([[MODEL_INPUT_DF.iloc[class="num">0,class="num">0],MODEL_INPUT_DF.iloc[class="num">0,class="num">1],MODEL_INPUT_DF.iloc[class="num">0,class="num">2],MODEL_INPUT_DF.iloc[class="num">0,class="num">3],MODEL_INPUT_DF.iloc[class="num">0,class="num">4],MODEL_INPUT_DF.iloc[class="num">0,class="num">5]]]) class="kw">return(model_input_array,MODEL_INPUT_DF.loc[class="num">0,&class="macro">#x27;Close&class="macro">#x27;]) def ai_forecast(): model_inputs,current_price = get_model_inputs() prediction = model.predict(model_inputs) class="kw">return(prediction[class="num">0],current_price) training_data = pd.read_csv(&class="macro">#x27;C:\\Enter\\Your\\Path\\Here\\Alternative Data.csv&class="macro">#x27;) from sklearn.linear_model class="kw">import LinearRegression model = LinearRegression() 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;IUDSOIA&class="macro">#x27;,&class="macro">#x27;SOFR&class="macro">#x27;] target = &class="macro">#x27;Target&class="macro">#x27; model.fit(training_data.loc[:,predictors],training_data.loc[:,target]) class="kw">while True: class="macro">#Get data on the current state of our terminal and our portfolio positions = mt5.positions_total() forecast , current_price = ai_forecast() BUY_STATE , SELL_STATE = False , False class="macro">#Interpret the model&class="macro">#x27;s forecast if(current_price > forecast): SELL_STATE = True BUY_STATE = False elif(current_price > forecast): SELL_STATE = False BUY_STATE = True print(f"Current price is {current_price} , our forecast is {forecast}") class="macro">#If we have no open positions let&class="macro">#x27;s open them if(positions == class="num">0): print(f"We have {positions} open trade(s)") if(SELL_STATE): print("Opening a sell position") mt5.Sell(SYMBOL,VOLUME) elif(BUY_STATE):