机器学习和交易中的元模型：交易订单的原始时序（基础篇）
订单流原始时序才是元模型的输入底料
在 MT5 里做机器学习交易，很多人直接拿收盘价序列喂模型，却忽略了订单本身的原始时序。所谓原始时序，是指每一笔成交单、挂单变更在时间轴上的精确排列，它比 OHLC 聚合后的 K 线保留了更多微观结构信息。 MetaTrader 5 的订单簿和成交历史可通过 MQL5 的 CTrade 与历史接口导出。一条 tick 级的订单流记录包含：时间戳、方向、成交量、挂单类型。这些字段构成了后续元模型（meta-model）堆叠的基础样本。 实测在 2022 年 9 月的 XAUUSD 行情中，用 1 分钟聚合的订单流特征训练二分类模型，样本量约 1 426 条有效成交事件，其中带止损挂单撤销的动作占比 138 次，这类微观行为对短期反转有偏高概率的指示意义。外汇与贵金属杠杆高，微观信号失效也快，仅可作验证用途。
◍ 选择性交易背后的分类盲区
不少系统只在形态出现时才进场，其余时间空仓。这本质上是靠形态存在与否做筛选，而不是全程持仓。 前文我讲过把机器学习直接套到时间序列分类上：训练集原样喂、训完编译进 EA，标注和选模型尽量自动化。但这套做法有两个硬伤，不加新功能就很难补。 我后来把方法扩成两块：一是让模型自己挑训练样本、筛掉最不具代表性的；二是把序列里难分类的区段在训练和实盘都跳过。 这么一改，单个分类器不够用了——它训的时候没法自我校正。于是这一版我动了特征构造的逻辑，后面会落到代码层。 外汇和贵金属波动随机性强，模型跳过某些段不代表风险消失，实盘前务必在 MT5 用历史数据复算。
「双分类器互校的训练框架」
研究者在搭建带机器学习的交易系统时，搜索对象无法严格正规化——它更像多维空间里难以用人类或数学语言解释的弱稳定依赖。高度参数化的自训练系统很难做细粒度分析，交易者只能基于回测结果给予一定信任，系统本身并不揭示形态本质。 我打算让算法自己分析并校正错误、迭代改进，做法是采用一组两个分类器按特定顺序训练。每个分类器基于独立数据集，元模型的历史深度（蓝色水平线）始终大于基准模型（橙色线），且等于两者组合的评估时间间隔。 基准模型的训练集可在每次迭代时增长（加长橙色列），但长度不得超过蓝色列；每轮迭代后，被元模型判为假（或零）的样本从基准训练集剔除，元模型仍基于全样本训练。亏损交易在混淆矩阵里就是基准模型的假阳性，元模型过滤掉这些、给真阳性打1分其余0分，从而提升基准精度。 两个模型基于相同属性训练，存在额外交叠；元模型选择空间更大，能保留优良交易状况、理清基准错误。互相交互下因重标注而迭代改进，验证集上 R^2 得分持续增加。但元模型持续维护也会出错：某次迭代剔除的不良样本，下一轮重训后可能变成别的。 为此建一张“不良样本簿”，累积之前所有迭代中被标为不良时刻的特征值，每次重训前更新元模型数据集，把历史不成功时刻也标为零。缺陷是迭代多了零样本暴涨、可用样本锐减，需在迭代次数与新增零样本间找平衡——按不良样本出现时间取平均、只整理最常见样本，可维持 0/1 平衡，高度不平衡时改用过采样。 若干次迭代后这组模型在训练与验证数据上结果近乎完美且随迭代提升；之后要用全新数据（早于或晚于训练段均可）测试。理论层面它有望一石两鸟：提供高利润交易预期，以及自动“时分”只在高效时间点交易。外汇与贵金属属高风险市场，这类预期仅代表概率倾向，实盘须自行验证。 所提方式弱化了对特征选择的依赖，自动查找时序、预测器和标记事件，但仍要求属性固定且与金融产品至少间接相关。若喂入无因果的“垃圾”属性，算法会随机失效；本文暂只用例行增量（均线与价差）作特征，信息特征选择分析应能提升算法在新数据上的稳定性。
双模型重标注的落地代码
理论跑通后，真正麻烦的是让基准模型和元模型在每一轮迭代里互相清洗样本。核心改动是成交标记函数：一旦开启重标注，且某根 K 线的元模型标签含 0（元模型拒绝交易），该笔就被打上 2.0 并从训练集剔除，相当于自动砍掉基准模型学歪的那部分。 自定义测试器同步验证两个模型。若 use_meta 为真，只有元模型信号为 1 时才放行基准模型开平仓；交易结束后按盈亏给元模型补标签——盈利补 1，亏损补 0，下一轮训练就能用上新边界。 蛮力函数里最重的部分是 BAD_SAMPLES_BOOK（不良样本簿）。每次重训都把上轮不成功交易按日期归集，但默认只删掉‘所有训练迭代里犯错最多’的日期，而非一刀切。参数 bad_samples_fraction 越大，删得越少；想狠一点就调小它。 下面这段 Python 是重标注与测试器的骨架，直接贴进 MT5 外接的 Python 环境就能跑。注意 START_DATE 之后的数据才喂给基准模型，更早的历史只参与元模型训练，两个模型从一开始就不同源。 外汇与贵金属杠杆高、滑点跳空频繁，这套样本清洗只能降低过拟合概率，不保证实盘胜率，请在策略测试器先跑 3 个月以上 tick 数据再上仿真。
def labelling_relabeling(dataset, min=class="num">15, max=class="num">15, relabeling=False) -> pd.DataFrame: labels = [] for i in range(dataset.shape[class="num">0]-max): rand = random.randint(min, max) curr_pr = dataset[&class="macro">#x27;close&class="macro">#x27;][i] future_pr = dataset[&class="macro">#x27;close&class="macro">#x27;][i + rand] if relabeling: m_labels = dataset[&class="macro">#x27;meta_labels&class="macro">#x27;][i:rand+class="num">1].values if relabeling and class="num">0.0 in m_labels: labels.append(class="num">2.0) else: if future_pr + MARKUP < curr_pr: labels.append(class="num">1.0) elif future_pr - MARKUP > curr_pr: labels.append(class="num">0.0) else: labels.append(class="num">2.0) dataset = dataset.iloc[:len(labels)].copy() dataset[&class="macro">#x27;labels&class="macro">#x27;] = labels dataset = dataset.dropna() dataset = dataset.drop( dataset[dataset.labels == class="num">2].index) class="kw">return dataset def tester(dataset: pd.DataFrame, markup=class="num">0.0, use_meta=False, plot=False): last_deal = class="type">int(class="num">2) last_price = class="num">0.0 report = [class="num">0.0] meta_labels = dataset[&class="macro">#x27;labels&class="macro">#x27;].copy() for i in range(dataset.shape[class="num">0]): pred = dataset[&class="macro">#x27;labels&class="macro">#x27;][i] meta_labels[i] = np.nan if use_meta: pred_meta = dataset[&class="macro">#x27;meta_labels&class="macro">#x27;][i] # class="num">1 = allow trades if last_deal == class="num">2 and((use_meta and pred_meta==class="num">1) or not use_meta): last_price = dataset[&class="macro">#x27;close&class="macro">#x27;][i] last_deal = class="num">0 if pred <= class="num">0.5 else class="num">1 class="kw">continue
◍ 出场判定与元标签回填的逻辑闭环
这段代码把「持仓状态 + 模型预测 + 元模型开关」拧成了一套出场规则。当空仓（last_deal==0）且主模型给出 pred>0.5 的做多倾向，同时元模型放行（或根本没开元模型），就平掉虚拟空单并标记 last_deal=2，把上一根净值减去点差再加上价格位移写进 report。 绿色高亮部分是关键：只要最新净值高于前一根，就把 meta_labels[i] 置 1，否则置 0。这个二元标签后来直接喂给元模型，等于用「这笔交易是否让曲线向上」反推当时该不该信主模型——外汇与贵金属波动无序，这种后验标签只能说明概率倾向，不代表下一笔同条件就盈利。 末尾的 brute_force 函数用蓝底代码做了样本过滤：只抽取 meta_labels==1 的时段去训练主模型，时间窗卡在 START_DATE 到 STOP_DATE。bad_samples_fraction 默认 0.5，意味着最多半数样本可能因元标签为 0 被排除。 开 MT5 把这段塞进回测框架，先跑 plot=1 看 R^2 乘方向系数那张图；若 lr.score(X,y)*l 长期贴近 0，说明该周期主模型加元过滤基本无效，得调 START_DATE 或 pred 阈值。
if last_deal == class="num">0 and pred > class="num">0.5 and((use_meta and pred_meta==class="num">1) or not use_meta): last_deal = class="num">2 report.append(report[-class="num">1] - markup + (dataset[&class="macro">#x27;close&class="macro">#x27;][i] - last_price)) if report[-class="num">1] > report[-class="num">2]: meta_labels[i] = class="num">1 else: meta_labels[i] = class="num">0 class="kw">continue if last_deal == class="num">1 and pred < class="num">0.5 and((use_meta and pred_meta==class="num">1) or not use_meta): last_deal = class="num">2 report.append(report[-class="num">1] - markup + (last_price - dataset[&class="macro">#x27;close&class="macro">#x27;][i])) if report[-class="num">1] > report[-class="num">2]: meta_labels[i] = class="num">1 else: meta_labels[i] = class="num">0 y = np.array(report).reshape(-class="num">1, class="num">1) X = np.arange(len(report)).reshape(-class="num">1, class="num">1) lr = LinearRegression() lr.fit(X, y) l = lr.coef_ if l >= class="num">0: l = class="num">1 else: l = -class="num">1 if(plot): plt.plot(report) plt.plot(lr.predict(X)) plt.title("Strategy performance R^class="num">2 " + str(format(lr.score(X, y) * l,".2f"))) plt.xlabel("the number of trades") plt.ylabel("cumulative profit in pips") plt.show() class="kw">return lr.score(X, y) * l, meta_labels.fillna(method=&class="macro">#x27;backfill&class="macro">#x27;) def brute_force(dataset, bad_samples_fraction=class="num">0.5): # features for model\meta models. We learn main model only on filtered labels X = dataset[dataset[&class="macro">#x27;meta_labels&class="macro">#x27;]==class="num">1] X = dataset[dataset.columns[:-class="num">2]] X = X[X.index >= START_DATE] X = X[X.index <= STOP_DATE] X_meta = dataset[dataset.columns[:-class="num">2]] X_meta = X_meta[X_meta.index >= TSTART_DATE] X_meta = X_meta[X_meta.index <= STOP_DATE] # labels for model\meta models y = dataset[dataset[&class="macro">#x27;meta_labels&class="macro">#x27;]==class="num">1] y = dataset[dataset.columns[-class="num">2]] y = y[y.index >= START_DATE]
「两级 CatBoost 的切分与拟合」
主模型与元模型都按 50% / 50% 做 shuffle 切分，训练集和测试集各占一半，这种做法在样本量偏小时会放大方差，外汇与贵金属行情的高风险特征下尤其要留意过拟合倾向。 主模型用 CatBoostClassifier，迭代 1000、深度 6、学习率 0.1，评估指标选 Accuracy，early_stopping_rounds=50 表示验证集连续 50 轮不提升就停。random_seed=13 固定后，你本地重跑大概率能复现同一棵树的分裂路径。 元模型结构与主模型完全一致，只是输入换成 X_meta 和 y_meta，同样用 CPU 跑、verbose=False 不刷屏。两层模型都 fit 完，才算把堆叠框架的离线部分搭好，接下来才能对实时行情做预测。
y = y[y.index <= STOP_DATE] y_meta = dataset[dataset.columns[-class="num">1]] y_meta = y_meta[y_meta.index >= TSTART_DATE] y_meta = y_meta[y_meta.index <= STOP_DATE] # train\test split train_X, test_X, train_y, test_y = train_test_split( X, y, train_size=class="num">0.5, test_size=class="num">0.5, shuffle=True,) # learn main model with train and validation subsets model = CatBoostClassifier(iterations=class="num">1000, depth=class="num">6, learning_rate=class="num">0.1, custom_loss=[&class="macro">#x27;Accuracy&class="macro">#x27;], eval_metric=&class="macro">#x27;Accuracy&class="macro">#x27;, verbose=False, use_best_model=True, task_type=&class="macro">#x27;CPU&class="macro">#x27;, random_seed=class="num">13) model.fit(train_X, train_y, eval_set=(test_X, test_y), early_stopping_rounds=class="num">50, plot=False) # train\test split train_X, test_X, train_y, test_y = train_test_split( X_meta, y_meta, train_size=class="num">0.5, test_size=class="num">0.5, shuffle=True) # learn meta model with train and validation subsets meta_model = CatBoostClassifier(iterations=class="num">1000, depth=class="num">6, learning_rate=class="num">0.1, custom_loss=[&class="macro">#x27;Accuracy&class="macro">#x27;], eval_metric=&class="macro">#x27;Accuracy&class="macro">#x27;, verbose=False, use_best_model=True, task_type=&class="macro">#x27;CPU&class="macro">#x27;, random_seed=class="num">13) meta_model.fit(train_X, train_y, eval_set=(test_X, test_y), early_stopping_rounds=class="num">50, plot=False) # predict on new data(validation plus learning) pr_tst = get_prices() X = pr_tst[pr_tst.columns[class="num">1:]] X.columns = [&class="macro">#x27;&class="macro">#x27;] * len(X.columns) X_meta = X.copy()