因果推理中的倾向性评分·进阶篇
（2/3）·高波动与低波动样本混在一起，ATE 估计为何悄悄偏向了错误方向
「用逆概率加权把样本拉回总体」
逆概率加权（IPW）核心就一步：拿每个样本接受处理的概率倒数当权重，概率越低的观测反而吃得越重。这么做能把处理组和对照组里被协变量带偏的分布重新拉平，让加权后的数据集更接近总体人群结构。 实操上先估倾向性评分——即在已观测协变量下某个样本被处理的概率，再取倒数得到权重，把每个观测值乘上去。处理组和对照组都乘，不偏袒任一边，目的是消掉协变量造成的混杂。 一个容易被忽略的点：IPW 成立的前提是所有相关混杂变量都已测到并进了评分模型。模型质量差，加权后偏差只会换种形式存在。外汇与贵金属市场高阶因子常不可观测，直接套 IPW 有高估平衡度的风险。 从等式看，左边处理组、右边对照组，简单均值比较其实等价于倒数加权均值比较。它构造出一个和原样本等规模的“虚拟群体”，其中左侧每个人都接受处理；同时高度重视那些本没被处理、但特征像已处理样本的观察值。开 MT5 做策略回测时，可用该法对少数类样本提权，验证分组净值曲线是否收敛。
用倾向得分切分样本再训分类器
拿到倾向得分 e(x) 后，别急着直接算 ATE。你可以把它当新特征喂给后续分类器，也可以按 e(x) 高低把样本分层，用来抵消混淆变量带来的偏差。MT5 上做这段实验时，建议先跑一遍 Logistic 回归拿 e(x)，再观察分层后各类别样本量是否塌缩到极少数。 我们真正想找的是对处理反应好或差的亚群，而不是全体平均效果。做法上，先在第一批数据训一个分类器，把预测误差小的样本留下训最终模型；误差大的扔进第二个子组，再训一个分类器专门区分好样本和坏样本。这样整体 ATE 反而没那么重要，重点在子群匹配。 评估阶段必须用没参与训练的新数据跑算法结果。另外，在随机数据上训一组模型，看它们的平均速度：独立模型平均得分越高，对单个模型的信心才越站得住。外汇与贵金属杠杆高、滑点随机，这套流程在实盘前务必用历史 tick 复验。
◍ 把倾向评分塞进元学习器后的三组对照
直接把倾向性评分（PSM 输出概率）当作额外特征喂给元学习器，思路是把样本按「被分到处理组还是对照组」的倾向拆开加权。具体落地要训两个模型：先用 CatBoost 训一个 PSM 分类器，拿到每条样本属于处理组的概率；再把这批概率拼回原始协变量，送进第二个模型去预测买卖标签。这么做省掉了显式匹配，机器学习自己把倾向差异吸收进估计里。 我跑了三套变体各 25 个模型做横向比对。第一套「无匹配」把 PSM 概率直接并入元学习器特征，最佳得分 0.96、平均 0.69；第二套「按预测标签倾向分层」只训一个 PSM，按概率分箱筛掉预测差的层，最佳 0.94、平均 0.85；第三套贴近理论假设随机分配处理，再叠加 IPW（逆概率加权）后最佳 0.97、平均 0.84，是不加 IPW（0.95 / 0.77）里平均最稳的一组。外汇与贵金属属高风险品种，上述 OOS 得分仅反映历史样本内建模质量，实盘迁移可能明显衰减。 别把 0.97 当成天花板。小样本下 IPW 只是把最差层权重压下去，换交易品种或周期很可能均值回落到 0.7 附近，验证时建议先固定随机种子复跑 25 遍看方差。 下面这段是第三套里随机分处理组并训 PSM 的核心切片，注意 y_T1 / y_T0 用 1 和 0 填 T 列后再拼回索引，PSM 和 meta 都用 CatBoost 但前者 predict_proba 取第二列作倾向。
X_train, X_val, y_train, y_val = train_test_split( X, y, train_size = class="num">0.5, test_size = class="num">0.5, shuffle = True) # randomly assign treated and control y_T1 = pd.DataFrame(y_train) y_T1[&class="macro">#x27;T&class="macro">#x27;] = class="num">1 y_T1 = y_T1.drop([&class="macro">#x27;labels&class="macro">#x27;], axis=class="num">1) y_T0 = pd.DataFrame(y_val) y_T0[&class="macro">#x27;T&class="macro">#x27;] = class="num">0 y_T0 = y_T0.drop([&class="macro">#x27;labels&class="macro">#x27;], axis=class="num">1) y_TT = pd.concat([y_T1, y_T0]) y_TT = y_TT.sort_index() X_trainT, X_valT, y_trainT, y_valT = train_test_split( X, y_TT, train_size = class="num">0.5, test_size = class="num">0.5, shuffle = True) # fit propensity model PSM = CatBoostClassifier(iterations = iterations, depth = depth, custom_loss = [&class="macro">#x27;Accuracy&class="macro">#x27;], eval_metric = &class="macro">#x27;Accuracy&class="macro">#x27;, use_best_model=True, early_stopping_rounds=class="num">15, verbose = False).fit(X_trainT, y_trainT, eval_set = (X_valT, y_valT), plot = False) # predict probabilities train_proba = PSM.predict_proba(X_train)[:, class="num">1] val_proba = PSM.predict_proba(X_val)[:, class="num">1] # fit meta-learner meta_m = CatBoostClassifier(iterations = iterations, depth = depth, custom_loss = [&class="macro">#x27;Accuracy&class="macro">#x27;], eval_metric = &class="macro">#x27;Accuracy&class="macro">#x27;, verbose = False,
「倾向得分分层筛掉难样本」
在元学习器产出预测概率后，把样本按倾向得分（propensity）切桶是最直接的清洗手段。下面这段把 coreset 按 bins_number 个等距区间分组，逐桶比较预测一致样本数是否少于错误样本的 coefficient 倍，若条件成立就把该桶内判错索引收进 B_S_B。 coreset['bin'] = pd.cut(coreset['propensity'], bins) for val in range(len(coreset['bin'].unique())): values = coreset.loc[coreset['bin'] == coreset['bin'].unique()[val]] diff_negatives = values['labels'] != values['labels_pred'] if len(diff_negatives[diff_negatives == False]) < (len(diff_negatives[diff_negatives == True]) * coefficient): B_S_B = B_S_B.append(diff_negatives[diff_negatives == True].index) 上述循环里 coefficient 是人工设定的宽松度，调小它会更激进地把整桶判错样本踢出训练集。回测中若把 coefficient 从 1.0 降到 0.5，B_S_B 收录的索引数通常翻倍，后续模型在验证集的 AUC 可能抬升 1~3 个百分点，但也伴随过拟合倾向。 外层还跑了 25 轮 learn_final_models，每轮用 models_number=5、depth=2、bad_samples_fraction=0.5 去迭代，保留最优 best_res 对应的模型组。外汇与贵金属行情噪声大，这种样本清洗在高杠杆环境下属于高风险操作，实盘前务必在 MT5 历史数据上重跑确认。
bins = np.linspace(lower_bound, upper_bound, num=bins_number) coreset[&class="macro">#x27;bin&class="macro">#x27;] = pd.cut(coreset[&class="macro">#x27;propensity&class="macro">#x27;], bins) for val in range(len(coreset[&class="macro">#x27;bin&class="macro">#x27;].unique())): values = coreset.loc[coreset[&class="macro">#x27;bin&class="macro">#x27;] == coreset[&class="macro">#x27;bin&class="macro">#x27;].unique()[val]] diff_negatives = values[&class="macro">#x27;labels&class="macro">#x27;] != values[&class="macro">#x27;labels_pred&class="macro">#x27;] if len(diff_negatives[diff_negatives == False]) < (len(diff_negatives[diff_negatives == True]) * coefficient): B_S_B = B_S_B.append(diff_negatives[diff_negatives == True].index) options = [] best_res = class="num">0.0 for i in range(class="num">25):
把倾向得分切成桶再做分层校验
模型训完别急着信，得把样本按倾向得分分层看看预测和真实标签在每一段里是否对得上。下面这段直接构造带 propensity 的 coreset，并用 np.linspace 在 lower_bound 到 upper_bound 之间均分出 bins_number 个区间，外汇与贵金属行情受流动性突变影响大，分层能暴露出模型在极端倾向区的失真，属于高风险场景下的必要核查。 coreset 里先拷 X，再挂 labels 和 PSM 预测的概率第二列；若开了 Use_IPW，就把 propensity 转成逆概率权重，大于 0.5 取 1/x 否则取 1/(1-x)，最后 round(3) 截断三位。meta_m 的预测概率同样取第二列，按 0.5 阈值转成 0/1 的 labels_pred。 用 pd.cut 把 propensity 丢进 bins 生成 'bin' 列，遍历每个桶：抽出该桶全部行，算 labels 与 labels_pred 不一致的样本 mask diff_negatives。当一致样本数小于不一致样本数乘以 coefficient（原文设 1.0）时，这一层就判定为不可靠，后续可据此丢掉或重训。原代码里 models_number=5、iterations=25、depth=2、bad_samples_fraction=0.5、bins_number=25、lower_bound=0.3、upper_bound=0.7，开 MT5 接 Python 脚本文火验证时可直接照抄这些参数看分层覆盖率。
print(&class="macro">#x27;Learn &class="macro">#x27; + str(i) + &class="macro">#x27; model&class="macro">#x27;) options.append(learn_final_models(meta_learners(models_number=class="num">5, iterations=class="num">25, depth=class="num">2, bad_samples_fraction=class="num">0.5, bins_number=class="num">25, lower_bound=class="num">0.3, upper_bound=class="num">0.7, coefficient=class="num">1.0))) if options[-class="num">1][class="num">0] > best_res: best_res = options[-class="num">1][class="num">0] print("BEST: " + str(best_res)) options.sort(key=lambda x: x[class="num">0]) test_model(options[-class="num">1][class="num">1:], plt=True) # create daatset with predicted values coreset = X.copy() coreset[&class="macro">#x27;labels&class="macro">#x27;] = y coreset[&class="macro">#x27;propensity&class="macro">#x27;] = PSM.predict_proba(X)[:, class="num">1] if Use_IPW: coreset[&class="macro">#x27;propensity&class="macro">#x27;] = coreset[&class="macro">#x27;propensity&class="macro">#x27;].apply(lambda x: class="num">1 / x if x > class="num">0.5 else class="num">1 / (class="num">1 - x)) coreset[&class="macro">#x27;propensity&class="macro">#x27;] = coreset[&class="macro">#x27;propensity&class="macro">#x27;].round(class="num">3) coreset[&class="macro">#x27;labels_pred&class="macro">#x27;] = meta_m.predict_proba(X)[:, class="num">1] coreset[&class="macro">#x27;labels_pred&class="macro">#x27;] = coreset[&class="macro">#x27;labels_pred&class="macro">#x27;].apply(lambda x: class="num">0 if x < class="num">0.5 else class="num">1) bins = np.linspace(lower_bound, upper_bound, num=bins_number) coreset[&class="macro">#x27;bin&class="macro">#x27;] = pd.cut(coreset[&class="macro">#x27;propensity&class="macro">#x27;], bins) for val in range(len(coreset[&class="macro">#x27;bin&class="macro">#x27;].unique())): values = coreset.loc[coreset[&class="macro">#x27;bin&class="macro">#x27;] == coreset[&class="macro">#x27;bin&class="macro">#x27;].unique()[val]] diff_negatives = values[&class="macro">#x27;labels&class="macro">#x27;] != values[&class="macro">#x27;labels_pred&class="macro">#x27;] if len(diff_negatives[diff_negatives == False]) < (len(diff_negatives[diff_negatives == True]) * coefficient):