重构经典策略(第十部分):人工智能(AI)能否为MACD提供动力?·进阶篇
🤖

重构经典策略(第十部分):人工智能(AI)能否为MACD提供动力?·进阶篇

(2/3)· 当滞后指标遇上机器学习,欧元兑美元上的实证拆解揭示特征权重与调参路径

案例拆解 第 2/3 篇
把 MACD 当成纯滞后信号直接跟单,常在趋势耗尽时才进场。本篇用 EURUSD 真实数据做建模,看 AI 能否把它的预测力往前推一步。

MACD 读数里藏着的 nonlinearity 陷阱

把未来价格水平对当前 MACD 读数画成散点图,能看出两者确实有关联,但关系不是一条直线——图里呈现的是非线性且结构杂乱的形态,单看 MACD 变化方向,没法直接判定后续是偏多还是偏空的价格表现。 三维散点把 MACD 主线、信号线同欧元兑美元收盘价叠在一起,边界模糊到没法清晰切分样本,分类预期数据会很吃力。唯一能合理推断的是:当 MACD 冲到极端值后,市场倾向于快速向中心区回归。外汇与贵金属杠杆高,这种回归只是概率倾向,不是确定性规律。 小提琴图对比了 MACD 上行与下行两组后续价格分布,蓝色轮廓几乎重合,内嵌箱线图的均值也基本一样。也就是说,指标处在看涨或看跌状态,对未来价格水平的统计分布影响微弱,别把方向信号当因果。 下面这段 Python(seaborn + matplotlib)就是上面三张图的生成逻辑,直接在 MT5 导出的 CSV 上跑就能复现:

MQL5 / C++
sns.scatterplot(data=data,x="MACD Main",y="MACD Signal",hue="Price Binary Target")
class="macro">#Define the 3D Plot
fig = plt.figure(figsize=(class="num">7,class="num">7))
ax = plt.axes(projection="3d")
ax.scatter(data["MACD Main"],data["MACD Signal"],data["Close"],c=data["Price Binary Target"])
ax.set_xlabel("MACD Main")
ax.set_ylabel("MACD Signal")
ax.set_zlabel("EURUSD Close")
sns.violinplot(data=data,x="MACD Bull",y="Close",hue="Price Binary Target",split=True,fill=False)

「建模前的数据切分与缩放」

要把行情序列喂给分类模型,第一步不是直接训练,而是先按时间顺序拆集。用 train_test_split 设 test_size=0.5 且 shuffle=False,保证训练在前、测试在后,避免未来信息泄漏——外汇与贵金属的高杠杆属性下,这种泄漏会让你在回测里看到虚假胜率。 特征分三组:OHLC 组含 Open/High/Low/Close/Bull Bear,MACD 组含 MACD Main/Signal/Bull,All 组是两者合并。目标变量有三个二分类标签:MACD Binary Target、Price Binary Target、All。 缩放用 MinMaxScaler 只在训练集 fit,再 transform 训练与测试,杜绝测试集统计特性混入训练。交叉验证用 TimeSeriesSplit(n_splits=5, gap=look_ahead),gap 把预测前瞻窗口空出来,更贴近实盘滚动推演。 跑完 MLPClassifier(hidden_layer_sizes=(10,6)) 的 cv 后,记录到的准确率很说明问题:OHLC 训练 0.690267、测试 0.684069,远好于 MACD 组(训练 0.507129、测试 0.48669)和 All 组(训练 0.504577、测试 0.487442)。纯价量结构对方向二分类的贡献,倾向高于单一 MACD 派生特征。 开 MT5 导出自家品种的历史 tick 或分钟 OHLC,按上面三组特征复刻切分,能直接验证你手上品种的 OHLC 组是否也守住 0.68 上下的区分度。

MQL5 / C++
class="macro">#Perform train test splits
from sklearn.model_selection class="kw">import train_test_split,TimeSeriesSplit
from sklearn.metrics class="kw">import accuracy_score
train,test = train_test_split(data,test_size=class="num">0.5,shuffle=False)
class="macro">#Let&class="macro">#x27;s scale the data
ohlc_predictors = ["Open","High","Low","Close","Bull Bear"]
macd_predictors = ["MACD Main","MACD Signal","MACD Bull"]
all_predictors  = ohlc_predictors + macd_predictors
cv_predictors   = [ohlc_predictors,macd_predictors,all_predictors]
class="macro">#Define the targets
cv_targets = ["MACD Binary Target","Price Binary Target","All"]
class="macro">#Scaling the data
from sklearn.preprocessing class="kw">import MinMaxScaler
scaler = MinMaxScaler()
scaler.fit(train[all_predictors])
train_scaled = pd.DataFrame(scaler.transform(train[all_predictors]),columns=all_predictors)
test_scaled = pd.DataFrame(scaler.transform(test[all_predictors]),columns=all_predictors)
class="macro">#Import the models we will evaluate
from sklearn.neural_network class="kw">import MLPClassifier,MLPRegressor
from sklearn.linear_model class="kw">import LinearRegression
tscv = TimeSeriesSplit(n_splits=class="num">5,gap=look_ahead)
err_indexes = ["MACD Train","Price Train","All Train","MACD Test","Price Test","All Test"]
class="macro">#Now let us define a table to store our error levels
columns = ["Model Accuracy"]
cv_err = pd.DataFrame(columns=columns,index=err_indexes)
class="macro">#Reset index
train = train.reset_index(drop=True)
test = test.reset_index(drop=True)
class="macro">#Initailize the model
price_model = MLPClassifier(hidden_layer_sizes=(class="num">10,class="num">6))
macd_model  = MLPClassifier(hidden_layer_sizes=(class="num">10,class="num">6))
all_model   = MLPClassifier(hidden_layer_sizes=(class="num">10,class="num">6))
price_acc = []
macd_acc = []
all_acc = []
class="macro">#Cross validate each model twice
for j,(train_index,test_index) in enumerate(tscv.split(train_scaled)):
  class="macro">#Fit the models
  price_model.fit(train_scaled.loc[train_index,ohlc_predictors],train.loc[train_index,"Price Binary Target"])
  macd_model.fit(train_scaled.loc[train_index,all_predictors],train.loc[train_index,"MACD Binary Target"])
  all_model.fit(train_scaled.loc[train_index,all_predictors],train.loc[train_index,"Price Binary Target"])
  class="macro">#Store the accuracy
  price_acc.append(accuracy_score(train.loc[test_index,"Price Binary Target"],price_model.predict(train_scaled.loc[test_index,ohlc_predictors])))
  macd_acc.append(accuracy_score(train.loc[test_index,cv_targets[class="num">0]],macd_model.predict(train_scaled.loc[test_index,all_predictors])))
  all_acc.append(accuracy_score(train.loc[test_index,cv_targets[class="num">1]],all_model.predict(train_scaled.loc[test_index,all_predictors])))
class="macro">#Now we can store our estimates of the model&class="macro">#x27;s error
cv_err.iloc[class="num">0,class="num">0] = np.mean(price_acc)
cv_err.iloc[class="num">1,class="num">0] = np.mean(macd_acc)
cv_err.iloc[class="num">2,class="num">0] = np.mean(all_acc)
class="macro">#Estimating test error
cv_err.iloc[class="num">3,class="num">0] = accuracy_score(test[cv_targets[class="num">1]],price_model.predict(test_scaled[ohlc_predictors]))
cv_err.iloc[class="num">4,class="num">0] = accuracy_score(test[cv_targets[class="num">0]],macd_model.predict(test_scaled[all_predictors]))
cv_err.iloc[class="num">5,class="num">0] = accuracy_score(test[cv_targets[class="num">1]],all_model.predict(test_scaled[all_predictors]))

◍ 用排列重要性和岭系数看特征权重

评估深度神经网络里哪个输入真正管用,可以用排列重要性(permutation importance)。思路很直接:把某一列输入随机打乱,看模型准确率掉多少;掉得多说明该特征重要,掉得接近 0 说明模型根本不靠它。我们在测试集上跑了这个流程,n_repeats=10、random_state=42,结果 MACD 相关列被打乱后准确率变化几乎为 0,意味着在这套模型里 MACD 基本是噪声。 但这个方法有坑。打乱是逐列独立随机的,可能把开盘价洗到比最高价还高,而真实行情里这种样本永远不可能出现。于是模型被放在了 unrealistic 条件下受罚,解释结果要打折扣。再加上神经网络训练本身有随机性,同一数据训两次可能给出差别很大的重要性排序,别把一次分数当铁律。 换个线性视角交叉验证:用 RidgeClassifier(L2 正则把系数往 0 推),数据标准化后系数小的特征就是信息量低的。我们拟合后画系数,最高价和最低价的系数最靠前,和排列重要性里收盘价最显眼形成互补视角——非线性模型最认收盘,线性模型最认高低点。 外汇和贵金属波动受多因子纠缠,这类特征重要性只反映历史样本内的统计倾向,实盘切换周期或品种后可能反转,属高风险验证项,建议自己在 MT5 导出的数据上重跑确认。

MQL5 / C++
class="macro">#Let us <span class="keyword">try</span> assess feature importance
<span class="keyword">from</span> sklearn.inspection class="kw">import permutation_importance
<span class="keyword">from</span> sklearn.linear_model class="kw">import RidgeClassifier
<span class="preprocessor">class="macro">#Let </span>us fit the model
model&nbsp;&nbsp; = MLPClassifier(hidden_layer_sizes=(<span class="number">class="num">10</span>,<span class="number">class="num">6</span>))
model.fit(train_scaled.loc[:,all_predictors],train.loc[:,<span class="class="type">class="kw">string">"Price Binary Target"</span>])
<span class="preprocessor">class="macro">#Calculate </span>permutation importance scores
pi = permutation_importance(
&nbsp;&nbsp;&nbsp;&nbsp;model, test_scaled.loc[:,all_predictors], test.loc[:,<span class="class="type">class="kw">string">"Price Binary Target"</span>], n_repeats=<span class="number">class="num">10</span>, random_state=<span class="number">class="num">42</span>, n_jobs=-<span class="number">class="num">1</span>
)
<span class="preprocessor">class="macro">#Sort </span>the importance scores
sorted_importances_idx = pi.importances_mean.argsort()
importances = pd.DataFrame(
&nbsp;&nbsp;&nbsp;&nbsp;pi.importances[sorted_importances_idx].T,
&nbsp;&nbsp;&nbsp;&nbsp;columns=test_scaled.columns[sorted_importances_idx],
)
<span class="preprocessor">class="macro">#Create </span>the plot
ax = importances.plot.box(vert=False, whis=<span class="number">class="num">10</span>)
ax.set_title(<span class="class="type">class="kw">string">"Permutation Importances(test set)"</span>)
ax.axvline(x=<span class="number">class="num">0</span>, <span class="keyword">class="type">color</span>=<span class="class="type">class="kw">string">"k"</span>, linestyle=<span class="class="type">class="kw">string">"--"</span>)
ax.set_xlabel(<span class="class="type">class="kw">string">"Decrease in accuracy score"</span>)
ax.figure.tight_layout()
<span class="preprocessor">class="macro">#Let </span>us fit the model
model&nbsp;&nbsp; = RidgeClassifier()
model.fit(train_scaled.loc[:,all_predictors],train.loc[:,<span class="class="type">class="kw">string">"Price Binary Target"</span>])
ridge_importance = pd.DataFrame(model.coef_.tolist(),columns=all_predictors)
<span class="preprocessor">class="macro">#Prepare </span>the plot
fig,ax = plt.subplots(figsize=(<span class="number">class="num">10</span>,<span class="number">class="num">5</span>))
sns.barplot(ridge_importance,ax=ax)

调参不是万能药

把表现最好的模型再丢进调优流程,这次结果并不理想。优化算法本身就有盲点,跑完参数搜索不代表模型一定更强,只是在有限组合里逼近较优解。 用 RandomizedSearchCV 做随机搜索,n_iter=100、cv=5、n_jobs=-1,评分用 neg_mean_squared_error。搜索空间覆盖了激活函数、求解器、alpha、tol、学习率策略等十来个维度,隐藏层结构给了 10 种候选。 跑完拿到的最优参数里,solver 是 sgd、activation 是 identity、hidden_layer_sizes 为 (20,10)、early_stopping 开、alpha 压到 1e-07。这组值只是本次采样里误差较低的配置,换一批数据可能就失效。外汇与贵金属波动受事件驱动,用这类模型做价格回归须清楚高风险,参数漂亮不等于实盘能跑赢。

MQL5 / C++
class="macro">#Let&class="macro">#x27;s tune our model further
from sklearn.model_selection class="kw">import RandomizedSearchCV
class="macro">#Reinitialize the model
model  = MLPRegressor(max_iter=class="num">200)
class="macro">#Define the tuner
tuner = RandomizedSearchCV(
        model,
        {
        "activation" : ["relu","logistic","tanh","identity"],
        "solver":["adam","sgd","lbfgs"],
        "alpha":[class="num">0.1,class="num">0.01,class="num">0.001,class="num">0.0001,class="num">0.00001,class="num">0.00001,class="num">0.0000001],
        "tol":[class="num">0.1,class="num">0.01,class="num">0.001,class="num">0.0001,class="num">0.00001,class="num">0.000001,class="num">0.0000001],
        "learning_rate":[&class="macro">#x27;constant&class="macro">#x27;,&class="macro">#x27;adaptive&class="macro">#x27;,&class="macro">#x27;invscaling&class="macro">#x27;],
        "learning_rate_init":[class="num">0.1,class="num">0.01,class="num">0.001,class="num">0.0001,class="num">0.00001,class="num">0.000001,class="num">0.0000001],
        "hidden_layer_sizes":[(class="num">2,class="num">4,class="num">8,class="num">2),(class="num">10,class="num">20),(class="num">5,class="num">10),(class="num">2,class="num">20),(class="num">6,class="num">8,class="num">10),(class="num">1,class="num">5),(class="num">20,class="num">10),(class="num">8,class="num">4),(class="num">2,class="num">4,class="num">8),(class="num">10,class="num">5)],
        "early_stopping":[True,False],
        "warm_start":[True,False],
        "shuffle": [True,False]
        },
        n_iter=class="num">100,
        cv=class="num">5,
        n_jobs=-class="num">1,
        scoring="neg_mean_squared_error"
)
tuner.fit(train.loc[:,ohlc_predictors],train.loc[:,"Price Target"])
tuner.best_params_

「用 SciPy 把连续参数逼到收敛」

在随机搜索给出粗略最优组合后,还可以用 SciPy 的 minimize 对神经网络的连续型超参做全局寻优。这里把时间序列拆成 5 折(TimeSeriesSplit,gap=look_ahead 防止未来泄漏),目标函数直接定义为交叉验证里的平均均方误差,让优化器去找使训练误差更小的稳定输入。 初始点取之前 tuner 给出的 alpha、tol、learning_rate_init,边界放开到 10^-100 到 10^100,方法用 L-BFGS-B。跑完返回的状态是 CONVERGENCE: REL_REDUCTION_OF_F_<=_FACTR*EPSMCH,成功状态 True,最终 fun 压到 3.73e-06,x 落在 [9.939e-08, 9.999e-03, 9.999e-03],只用了 3 次迭代、100 次函数评估就收敛,说明这组连续参数附近误差曲面已接近平坦。 别把收敛当终点 收敛只代表当前目标函数附近没有更优解,不保证样本外也稳。外汇与贵金属波动结构会漂移,这套最优权重拿到实盘前,至少要用留出的一段行情重算一次 CV RMSE,概率上才有参考意义。 代码里把每轮平均误差存进 algorithm_progress,最后散点图标出最低点,你能直接复制去 MT5 导出的特征表上重跑,看自己的 optima_x 落在第几折。

MQL5 / C++
class="macro">#Deeper optimization
from scipy.optimize class="kw">import minimize
from sklearn.metrics class="kw">import mean_squared_error
from sklearn.model_selection class="kw">import TimeSeriesSplit
class="macro">#Define the time series split object
tscv = TimeSeriesSplit(n_splits=class="num">5,gap=look_ahead)
class="macro">#Create a dataframe to store our accuracy
current_error_rate = pd.DataFrame(index = np.arange(class="num">0,class="num">5),columns=["Current Error"])
algorithm_progress = []
class="macro">#Define the objective function
def objective(x):
    class="macro">#The parameter x represents a new value for our neural network&class="macro">#x27;s settings
    model = MLPRegressor(hidden_layer_sizes=tuner.best_params_["hidden_layer_sizes"],
                          early_stopping=tuner.best_params_["early_stopping"],
                          warm_start=tuner.best_params_["warm_start"],
                          max_iter=class="num">500,
                          activation=tuner.best_params_["activation"],
                          learning_rate=tuner.best_params_["learning_rate"],
                          solver=tuner.best_params_["solver"],
                          shuffle=tuner.best_params_["shuffle"],
                          alpha=x[class="num">0],
                          tol=x[class="num">1],
                          learning_rate_init=x[class="num">2]
                          )
    class="macro">#Now we will cross validate the model
    for i,(train_index,test_index) in enumerate(tscv.split(train)):
        class="macro">#Train the model
        model.fit(train.loc[train_index,ohlc_predictors],train.loc[train_index,"Price Target"])
        class="macro">#Measure the RMSE
        current_error_rate.iloc[i,class="num">0] = mean_squared_error(train.loc[test_index,"Price Target"],model.predict(train.loc[test_index,ohlc_predictors]))
    class="macro">#Store the algorithm&class="macro">#x27;s progress
    algorithm_progress.append(current_error_rate.iloc[:,class="num">0].mean())
    class="macro">#Return the Mean CV RMSE
    class="kw">return(current_error_rate.iloc[:,class="num">0].mean())
class="macro">#Define the starting point
pt = [tuner.best_params_["alpha"],tuner.best_params_["tol"],tuner.best_params_["learning_rate_init"]]
bnds = ((class="num">10.00 ** -class="num">100,class="num">10.00 ** class="num">100),
        (class="num">10.00 ** -class="num">100,class="num">10.00 ** class="num">100),
        (class="num">10.00 ** -class="num">100,class="num">10.00 ** class="num">100))
class="macro">#Searching deeper for parameters
result = minimize(objective,pt,method="L-BFGS-B",bounds=bnds)
class="macro">#The result of our optimization
result
class="macro">#Store the optimal coefficients
optimal_weights = result.x
optima_y = min(algorithm_progress)
optima_x = algorithm_progress.index(optima_y)
inputs = np.arange(class="num">0,len(algorithm_progress))
class="macro">#Plot the performance of our optimization procedure
plt.scatter(inputs,algorithm_progress)
plt.plot(optima_x,optima_y,&class="macro">#x27;ro&class="macro">#x27;,class="type">color=&class="macro">#x27;r&class="macro">#x27;)
plt.axvline(x=optima_x,ls=&class="macro">#x27;--&class="macro">#x27;,class="type">color=&class="macro">#x27;red&class="macro">#x27;)

◍ 把训练误差收敛画出来

在 Python 里验证梯度下降是否真的走到极小值,最直接的就是把每轮迭代的训练 MSE 描成曲线。下面这段把最优解对应的纵轴画成红色虚线,一眼能看出算法停在哪儿。 plt.axhline(y=optima_y,ls='--',color='red') 在图上拉一条水平红线,标出理论最优的训练误差值;plt.xlabel("Iterations") 横轴写成迭代次数,方便对照收敛速度;plt.ylabel("Training MSE") 纵轴是训练均方误差,数值越低拟合越好;plt.title("Minimizing Training Error") 给图一个名字,省得和别的回测图混在一起。 跑完这段,若曲线在红线附近抖动而不继续下探,说明学习率可能偏小或已到浮点精度极限;外汇与贵金属模型用这类误差面做特征拟合时,过拟合会让训练 MSE 极低但样本外概率显著恶化,属高风险操作。

MQL5 / C++
plt.axhline(y=optima_y,ls=&class="macro">#x27;--&class="macro">#x27;,class="type">color=&class="macro">#x27;red&class="macro">#x27;)
plt.xlabel("Iterations")
plt.ylabel("Training MSE")
plt.title("Minimizing Training Error")
让小布替你跑这套
这些诊断与小布盯盘的 AIGC 已内置,打开对应品种页即可看到 MACD 背离与特征重要度快照,把重复劳动交给小布,你专注决策。

常见问题

主线杯状转折由动能切换造成,结合快慢线差值斜率可区分趋势延续与衰竭,概率上更稳。
可以,小布在品种页内置了类似本篇的建模快照,省去自己准备数据和训练模型的过程。
实证中相当比例反转无背离先兆,因此模型需加入宏观与波动率特征以降低漏报。
建议纳入 M1 与 H1 多周期值,特征重要性分析显示跨周期差异对预测有边际贡献。