重构经典策略(第十部分):人工智能(AI)能否为MACD提供动力?·进阶篇
(2/3)· 当滞后指标遇上机器学习,欧元兑美元上的实证拆解揭示特征权重与调参路径
MACD 读数里藏着的 nonlinearity 陷阱
把未来价格水平对当前 MACD 读数画成散点图,能看出两者确实有关联,但关系不是一条直线——图里呈现的是非线性且结构杂乱的形态,单看 MACD 变化方向,没法直接判定后续是偏多还是偏空的价格表现。 三维散点把 MACD 主线、信号线同欧元兑美元收盘价叠在一起,边界模糊到没法清晰切分样本,分类预期数据会很吃力。唯一能合理推断的是:当 MACD 冲到极端值后,市场倾向于快速向中心区回归。外汇与贵金属杠杆高,这种回归只是概率倾向,不是确定性规律。 小提琴图对比了 MACD 上行与下行两组后续价格分布,蓝色轮廓几乎重合,内嵌箱线图的均值也基本一样。也就是说,指标处在看涨或看跌状态,对未来价格水平的统计分布影响微弱,别把方向信号当因果。 下面这段 Python(seaborn + matplotlib)就是上面三张图的生成逻辑,直接在 MT5 导出的 CSV 上跑就能复现:
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 上下的区分度。
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 导出的数据上重跑确认。
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 = 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( 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( pi.importances[sorted_importances_idx].T, 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 = 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。这组值只是本次采样里误差较低的配置,换一批数据可能就失效。外汇与贵金属波动受事件驱动,用这类模型做价格回归须清楚高风险,参数漂亮不等于实盘能跑赢。
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 落在第几折。
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 极低但样本外概率显著恶化,属高风险操作。
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")