时间序列分类问题中的因果推理·综合运用
📘

时间序列分类问题中的因果推理·综合运用

第 3/3 篇
本章目录
  1. X-Learner 怎么把丢弃的信息捡回来
  2. 把元学习器塞进EA的边界
  3. 用多分类器剔除坏样本来降偏
  4. 用 25 次重训练验证因果稳定性
  5. 把工具请下神坛

「X-Learner 怎么把丢弃的信息捡回来」

X-Learner 属于元学习器的一类，核心目标是直接估算 CATE（条件平均处理效应），而不是像 S-Learner、T-Learner 那样只建模反应函数。反应函数描述特质 X 与处理 T 到结果 y 的映射，但这两类模型都没拿真实结果去直接拟合 CATE，X-Learner 把这部分被放弃的信息重新用上了。 具体流程第一步和 T-Learner 一致：按处理变量把数据切成两个独立子集，一组仅含接受处理的单位，另一组仅含未接受处理的单位，各自训练一个模型。随后引入一个倾向得分模型（最朴素可用逻辑回归），用来预测在给定 X 特质下接受处理的概率。 接着基于两个子集分别算出处理效应，再用特征和 CATE 值训练两个模型，最终把这两个模型输出按倾向得分给出的权重做加权融合。在数据集高度不平衡时，这种对子模型加权的能力让 X-Learner 明显比前两者更稳；但若样本量很小，每多训一个模型就多一份拟合噪声，数据不够喂的话反而 S-Learner 更合适。外汇与贵金属行情建模用这类方法需注意：小样本过拟合会带来极高实盘风险。 更上游的 Debiased/orthogonal ML 与 R-learner 不在本篇范围内，有兴趣可自行查阅，实际盯盘策略里很少直接落地。

◍ 把元学习器塞进EA的边界

前面几套元学习器（如 S-Learner、T-Learner）在纸面上能估出 CATE 条件平均处理效应，但文献几乎只停在「效应估出来就好」，没管你拿它怎么过完一整个交易周期。真要做实验，论文默认是研究员自己定方案、再喂给估算器，工具不替你背决策锅。 我把它再推一步：直接把估算器的内部元素拆出来，嵌进一个自动交易系统。输入输出的符号和标签沿用之前那套约定，算法在每次 tick 或 bar 闭合时扫一遍数据，只把识别出的潜在因果关系留下，其余噪声从交易逻辑里剔除。 这里有个实打实的限制——原方法估的是「效应存在与否」，不是「下一根 K 线往哪走」。你在 MT5 里接这套，得自己定处理变量（比如某宏观发布是否发生），否则 EA 连因和果都分不清。外汇与贵金属杠杆高，因果误判会放大亏损，先用历史数据跑通再上实盘。

用多分类器剔除坏样本来降偏

单一分类器评估潜在结果有偏，实际跑算法时不可靠。这套元学习器思路是一次性喂入指定数量的基模型（实验里用 CatBoost），靠多个学习器的交叉验证把噪声样本逼出来。 函数首参是分类器个数，后面跟迭代次数、树深度和 bad_samples_fraction——即最终训练集里要剔除的分类差样本占比。外汇与贵金属波动无序，这类坏样本对应的时刻最好避开交易。 每次循环按 50/50 随机切训练集和验证集，防止单个模型过拟合；整份数据都参与产出估计，但每个分类器只在独有子样本上训练，偏差因此被摊平。预测标签和真实标签对不上的，其索引塞进 BAD_BUY / BAD_SELL，随迭代累积。 所有循环结束后，统计每个索引落入坏样本的次数，只保留超过平均次数乘以 bad_samples_fraction 的那些索引，把它们标成 0.0 踢出最终训练。若阈值设太狠，每个指标都至少误判一次，全量被删，算法就废了。 下面这段是实现核心，逐行拆一下逻辑： def meta_learners(models_number: int, iterations: int, depth: int, bad_samples_fraction: float): # 定义元学习器函数，接收模型数、迭代、深度、坏样本比例 &nbsp;&nbsp;&nbsp;&nbsp;dataset = get_labels(get_prices()) # 取价格并打标签得到原始数据集 &nbsp;&nbsp;&nbsp;&nbsp;data = dataset[(dataset.index < FORWARD) & (dataset.index > BACKWARD)].copy() # 截取前后向窗口内的数据副本 &nbsp;&nbsp;&nbsp;&nbsp;X = data[data.columns[1:-2]] # 取中间列作特征 &nbsp;&nbsp;&nbsp;&nbsp;y = data['labels'] # 取标签列作目标 &nbsp;&nbsp;&nbsp;&nbsp;BAD_BUY = pd.DatetimeIndex([]) # 初始化买入坏样本索引集 &nbsp;&nbsp;&nbsp;&nbsp;BAD_SELL = pd.DatetimeIndex([]) # 初始化卖出坏样本索引集 &nbsp;&nbsp;&nbsp;&nbsp;for i in range(models_number): # 按模型个数循环 &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;X_train, X_val, y_train, y_val = train_test_split(X, y, train_size = 0.5, test_size = 0.5, shuffle = True) # 50/50随机切分 &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;# learn debias model with train and validation subsets &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;meta_m = CatBoostClassifier(iterations = iterations, depth = depth, custom_loss = ['Accuracy'], eval_metric = 'Accuracy', verbose = False, use_best_model = True) # 建CatBoost分类器 &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;meta_m.fit(X_train, y_train, eval_set = (X_val, y_val), plot = False) # 训练并验证 &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;coreset = X.copy() # 复制特征集 &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;coreset['labels'] = y # 附真实标签 &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;coreset['labels_pred'] = meta_m.predict_proba(X)[:, 1] # 取正类概率 &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;coreset['labels_pred'] = coreset['labels_pred'].apply(lambda x: 0 if x < 0.5 else 1) # 概率转0/1预测 &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;# add bad samples of this iteration (bad labels indices) &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;coreset_b = coreset[coreset['labels']==0] # 真实负类子集 &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;coreset_s = coreset[coreset['labels']==1] # 真实正类子集 &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;diff_negatives_b = coreset_b['labels'] != coreset_b['labels_pred'] # 负类误判掩码 &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;diff_negatives_s = coreset_s['labels'] != coreset_s['labels_pred'] # 正类误判掩码 &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;BAD_BUY = BAD_BUY.append(diff_negatives_b[diff_negatives_b == True].index) # 累加买入误判索引 &nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;BAD_SELL = BAD_SELL.append(diff_negatives_s[diff_negatives_s == True].index) # 累加卖出误判索引 &nbsp;&nbsp;&nbsp;&nbsp;to_mark_b = BAD_BUY.value_counts() # 统计各买入索引误判次数 &nbsp;&nbsp;&nbsp;&nbsp;to_mark_s = BAD_SELL.value_counts() # 统计各卖出索引误判次数 &nbsp;&nbsp;&nbsp;&nbsp;marked_idx_b = to_mark_b[to_mark_b > to_mark_b.mean() * bad_samples_fraction].index # 超阈买入索引 &nbsp;&nbsp;&nbsp;&nbsp;marked_idx_s = to_mark_s[to_mark_s > to_mark_s.mean() * bad_samples_fraction].index # 超阈卖出索引 &nbsp;&nbsp;&nbsp;&nbsp;data.loc[data.index.isin(marked_idx_b), 'meta_labels'] = 0.0 # 买入坏样本标0 &nbsp;&nbsp;&nbsp;&nbsp;data.loc[data.index.isin(marked_idx_s), 'meta_labels'] = 0.0 # 卖出坏样本标0 &nbsp;&nbsp;&nbsp;&nbsp;return data[data.columns[1:]] # 返回剔除首列后的数据 在 MT5 里接自己的行情管道跑一遍，调大 models_number 看 BAD 集合收敛速度，可能比单模型稳。

MQL5 / C++
def meta_learners(models_number: <span class="keyword">class="type">int</span>, iterations: <span class="keyword">class="type">int</span>, depth: <span class="keyword">class="type">int</span>, bad_samples_fraction: <span class="keyword">class="type">float</span>):
&nbsp;&nbsp;&nbsp;&nbsp;dataset = get_labels(get_prices())
&nbsp;&nbsp;&nbsp;&nbsp;data = dataset[(dataset.index &lt; FORWARD) &amp; (dataset.index &gt; BACKWARD)].copy()
&nbsp;&nbsp;&nbsp;&nbsp;X = data[data.columns[<span class="number">class="num">1</span>:-<span class="number">class="num">2</span>]]
&nbsp;&nbsp;&nbsp;&nbsp;y = data[<span class="class="type">class="kw">string">&class="macro">#x27;labels&class="macro">#x27;</span>]
&nbsp;&nbsp;&nbsp;&nbsp;BAD_BUY = pd.DatetimeIndex([])
&nbsp;&nbsp;&nbsp;&nbsp;BAD_SELL = pd.DatetimeIndex([])
&nbsp;&nbsp;&nbsp;&nbsp;<span class="keyword">for</span> i <span class="keyword">in</span> range(models_number):
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;X_train, X_val, y_train, y_val = train_test_split(
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;X, y, train_size = <span class="number">class="num">0.5</span>, test_size = <span class="number">class="num">0.5</span>, shuffle = True)
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;# learn debias model with train and validation subsets
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;meta_m = CatBoostClassifier(iterations = iterations,
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;depth = depth,
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;custom_loss = [<span class="class="type">class="kw">string">&class="macro">#x27;Accuracy&class="macro">#x27;</span>],
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;eval_metric = <span class="class="type">class="kw">string">&class="macro">#x27;Accuracy&class="macro">#x27;</span>,
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;verbose = False,
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;use_best_model = True)
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;meta_m.fit(X_train, y_train, eval_set = (X_val, y_val), plot = False)
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;coreset = X.copy()
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;coreset[<span class="class="type">class="kw">string">&class="macro">#x27;labels&class="macro">#x27;</span>] = y
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;coreset[<span class="class="type">class="kw">string">&class="macro">#x27;labels_pred&class="macro">#x27;</span>] = meta_m.predict_proba(X)[:, <span class="number">class="num">1</span>]
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;coreset[<span class="class="type">class="kw">string">&class="macro">#x27;labels_pred&class="macro">#x27;</span>] = coreset[<span class="class="type">class="kw">string">&class="macro">#x27;labels_pred&class="macro">#x27;</span>].apply(lambda x: <span class="number">class="num">0</span> <span class="keyword">if</span> x &lt; <span class="number">class="num">0.5</span> <span class="keyword">else</span> <span class="number">class="num">1</span>)
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;# add bad samples of <span class="keyword">this</span> iteration(bad labels indices)
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;coreset_b = coreset[coreset[<span class="class="type">class="kw">string">&class="macro">#x27;labels&class="macro">#x27;</span>]==<span class="number">class="num">0</span>]
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;coreset_s = coreset[coreset[<span class="class="type">class="kw">string">&class="macro">#x27;labels&class="macro">#x27;</span>]==<span class="number">class="num">1</span>]
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;diff_negatives_b = coreset_b[<span class="class="type">class="kw">string">&class="macro">#x27;labels&class="macro">#x27;</span>] != coreset_b[<span class="class="type">class="kw">string">&class="macro">#x27;labels_pred&class="macro">#x27;</span>]
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;diff_negatives_s = coreset_s[<span class="class="type">class="kw">string">&class="macro">#x27;labels&class="macro">#x27;</span>] != coreset_s[<span class="class="type">class="kw">string">&class="macro">#x27;labels_pred&class="macro">#x27;</span>]
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;BAD_BUY = BAD_BUY.append(diff_negatives_b[diff_negatives_b == True].index)
&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;&nbsp;BAD_SELL = BAD_SELL.append(diff_negatives_s[diff_negatives_s == True].index)
&nbsp;&nbsp;&nbsp;&nbsp;to_mark_b = BAD_BUY.value_counts()
&nbsp;&nbsp;&nbsp;&nbsp;to_mark_s = BAD_SELL.value_counts()
&nbsp;&nbsp;&nbsp;&nbsp;marked_idx_b = to_mark_b[to_mark_b &gt; to_mark_b.mean() * bad_samples_fraction].index
&nbsp;&nbsp;&nbsp;&nbsp;marked_idx_s = to_mark_s[to_mark_s &gt; to_mark_s.mean() * bad_samples_fraction].index
&nbsp;&nbsp;&nbsp;&nbsp;data.loc[data.index.isin(marked_idx_b), <span class="class="type">class="kw">string">&class="macro">#x27;meta_labels&class="macro">#x27;</span>] = <span class="number">class="num">0.0</span>
&nbsp;&nbsp;&nbsp;&nbsp;data.loc[data.index.isin(marked_idx_s), <span class="class="type">class="kw">string">&class="macro">#x27;meta_labels&class="macro">#x27;</span>] = <span class="number">class="num">0.0</span>
&nbsp;&nbsp;&nbsp;&nbsp;<span class="keyword">class="kw">return</span> data[data.columns[<span class="number">class="num">1</span>:]]

「用 25 次重训练验证因果稳定性」

做遗传优化时，光看单轮最佳拟合容易踩坑：若策略本身弱或参数步长过大，大量跑出来的结果会拖垮平均评估，掩盖真实表现。更稳妥的做法是把算法反复训多遍，取平均再和最优比，看偏差是否可控。 这里改了自定义测试器，一次性把列表里所有训好的模型都测一遍。具体跑了 25 次因果推理，每次都是独立模型、随机划分子样本，随机性拉满。 先按 R^2 挑出最佳单模型测试，再把 25 个模型集体丢进 test_all_models。实测平均结果和最佳结果相差无几——说明在受控随机实验下，更可能逼近真实因果关系而非过拟合噪声。 换一组元学习器参数（如树深 5/10、学习率 0.4）再训再测，结论一致：训练历史深度、特征数和其他超参数会影响单轮质量，但模型间差异仍很小。这种跨随机种子的稳定，是该类算法值得信任的点，对外汇/贵金属这类高波动、高风险的品种尤其重要，实盘前务必自己跑一遍验证。

MQL5 / C++
def test_all_models(result: list):
    pr_tst = get_prices()
    X = pr_tst[pr_tst.columns[class="num">1:]]
    pr_tst[&class="macro">#x27;labels&class="macro">#x27;] = class="num">0.5
    pr_tst[&class="macro">#x27;meta_labels&class="macro">#x27;] = class="num">0.5
    
    for i in range(len(result)):
        pr_tst[&class="macro">#x27;labels&class="macro">#x27;] += result[i][class="num">1].predict_proba(X)[:,class="num">1]
        pr_tst[&class="macro">#x27;meta_labels&class="macro">#x27;] += result[i][class="num">2].predict_proba(X)[:,class="num">1]
    pr_tst[&class="macro">#x27;labels&class="macro">#x27;] = pr_tst[&class="macro">#x27;labels&class="macro">#x27;] / (len(result)+class="num">1)
    pr_tst[&class="macro">#x27;meta_labels&class="macro">#x27;] = pr_tst[&class="macro">#x27;meta_labels&class="macro">#x27;] / (len(result)+class="num">1)
    pr_tst[&class="macro">#x27;labels&class="macro">#x27;] = pr_tst[&class="macro">#x27;labels&class="macro">#x27;].apply(lambda x: class="num">0.0 if x < class="num">0.5 else class="num">1.0)
    pr_tst[&class="macro">#x27;meta_labels&class="macro">#x27;] = pr_tst[&class="macro">#x27;meta_labels&class="macro">#x27;].apply(lambda x: class="num">0.0 if x < class="num">0.5 else class="num">1.0)
    class="kw">return tester(pr_tst, plot=plt)
options = []
for i in range(class="num">25):
    print(&class="macro">#x27;Learn &class="macro">#x27; + str(i) + &class="macro">#x27; model&class="macro">#x27;)
    options.append(learn_final_models(meta_learners(class="num">15, class="num">25, class="num">2, class="num">0.3)))
options.sort(key=lambda x: x[class="num">0])
test_model(options[-class="num">1][class="num">1:], plt=True)
test_all_models(options)
options = []
for i in range(class="num">25):
    print(&class="macro">#x27;Learn &class="macro">#x27; + str(i) + &class="macro">#x27; model&class="macro">#x27;)
    options.append(learn_final_models(meta_learners(class="num">5, class="num">10, class="num">1, class="num">0.4)))
options.sort(key=lambda x: x[class="num">0])
test_model(options[-class="num">1][class="num">1:], plt=True)
test_all_models(options)

◍ 把工具请下神坛

因果推理这套东西根子在哲学和心理学，读起来大多能在直觉层面共鸣，但落到量化交易只是时间序列分类里的一枚螺丝。前面给出的 Python 样例（causal_inference.py，11.05 KB）已经证明，替换上篇文章几个函数就能跑通分类实验，并没有神话般的边际收益。 真正值得做的，是打开 MT5 旁边的 Python 环境，把那份 ZIP 里的脚本接进你自己的行情片段，看混淆变量怎么改写样本权重。外汇与贵金属杠杆高、滑点诡变，任何因果结论都只是概率倾向，别把回测里的因果实线当成盘面上的必现路径。 作者后续若挖出新现象自然会放出来，眼下你能验证的，也就是自己跑一遍、换一组特征、比对准确率掉了还是涨了。

常见问题

它先分别给处理组和控制组各训一个基模型，再用它们生成伪结果来训练因果效应模型，从而利用原本被忽略的反事实信息，降低小样本偏差。
元学习器计算开销大，实时跑易卡顿；建议只在盘后做样本重估，或限制重训练频率，避免阻塞下单逻辑。
小布可自动跑多轮重训练并标记效应漂移大的品种，把重复劳动交给它，你只需看哪些信号靠谱。
实战中 3 个以上独立分类器投票剔除离群样本即可，太多会误杀有效样本，反而抬高偏差。
若 25 次中效应符号翻转超 5 次或方差超基线 2 倍，该因果结论大概率不稳定，不建议用于实盘。

相关阅读

小布盯盘 · 交易者自己的 价格行为学 + AIGC 分析工具。小布知识库 持续更新硬核交易干货。

本文由人工智能（AIGC）基于价格行为学进行深度分析，仅为技术方法论的探讨与分享，不构成任何形式的投资建议或交易指令。外汇保证金及贵金属交易具有高风险，并不适合所有投资者；过往表现不代表未来结果，请结合自身风险承受能力独立决策。

© 2026 小布盘 unyet.cn