因果推断中的时间序列聚类·进阶篇
元标签与聚类筛选的代码落点
这段逻辑干了两件事：先用元模型在每轮迭代里挑出预测错的样本索引，累积到 B_S_B 里，再按出现频次高于均值乘以 bad_samples_fraction 的阈值，把对应原始数据的 meta_labels 置 0，相当于让模型少看噪声标签。 聚类分支用 algorithm 参数切换八种算法，从 KMeans 到 HDBSCAN(min_cluster_size=150) 都有，注意 HDBSCAN 那行写死了最小簇大小 150，样本量不够时可能直接吐出 -1 噪声簇。 fit_final_models 里只取 meta_labels==1 的样本训主模型，且用 str.contains('std') 把特征拆成主模型用非 std 列、元模型用 std 列；train_test_split 的 train_size=0.8 是硬分割，跑外汇或贵金属数据前先确认样本数，这类品种跳空多、分布漂移快，高频重训才不容易过拟合。 直接把下面这段贴进你的 Python 研究环境（非 MT5 终端，是离线特征研究用），改 bad_samples_fraction 看筛选比例对回测准确率的边际影响。
eval_metric = &class="macro">#x27;Accuracy&class="macro">#x27;, verbose = False, use_best_model = True) meta_m.fit(X_train, y_train, eval_set = (X_val, y_val), plot = False) coreset = X.copy() coreset[&class="macro">#x27;labels&class="macro">#x27;] = y 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) # add bad samples of this iteration(bad labels indices) diff_negatives = coreset[&class="macro">#x27;labels&class="macro">#x27;] != coreset[&class="macro">#x27;labels_pred&class="macro">#x27;] B_S_B = B_S_B.append(diff_negatives[diff_negatives == True].index) to_mark = B_S_B.value_counts() marked_idx = to_mark[to_mark > to_mark.mean() * bad_samples_fraction].index data.loc[data.index.isin(marked_idx), &class="macro">#x27;meta_labels&class="macro">#x27;] = class="num">0.0 if algorithm==class="num">0: data[&class="macro">#x27;clusters&class="macro">#x27;] = KMeans(n_clusters=n_clusters).fit(meta_X).labels_ elif algorithm==class="num">1: data[&class="macro">#x27;clusters&class="macro">#x27;] = AffinityPropagation().fit(meta_X).predict(meta_X) elif algorithm==class="num">2: data[&class="macro">#x27;clusters&class="macro">#x27;] = SpectralClustering(n_clusters=n_clusters, assign_labels=&class="macro">#x27;discretize&class="macro">#x27;, random_state=class="num">0).fit_predict(meta_X) elif algorithm==class="num">3: data[&class="macro">#x27;clusters&class="macro">#x27;] = MeanShift().fit_predict(meta_X) elif algorithm==class="num">4: data[&class="macro">#x27;clusters&class="macro">#x27;] = AgglomerativeClustering(n_clusters=n_clusters).fit_predict(meta_X) elif algorithm==class="num">5: data[&class="macro">#x27;clusters&class="macro">#x27;] = mixture.GaussianMixture(n_components=n_clusters, covariance_type=&class="macro">#x27;full&class="macro">#x27;).fit(meta_X).predict(meta_X) elif algorithm==class="num">6: data[&class="macro">#x27;clusters&class="macro">#x27;] = HDBSCAN(min_cluster_size=class="num">150).fit_predict(meta_X) elif algorithm==class="num">7: data[&class="macro">#x27;clusters&class="macro">#x27;] = Birch(threshold=class="num">0.01, n_clusters=n_clusters).fit_predict(meta_X) class="kw">return data[data.columns[class="num">1:]] def fit_final_models(dataset) -> list: # features for model\meta models. We learn main model only on filtered labels X, X_meta = dataset[dataset[&class="macro">#x27;meta_labels&class="macro">#x27;]==class="num">1], dataset[dataset.columns[:-class="num">3]] X = X[X.columns[:-class="num">3]] X = X.loc[:, ~X.columns.str.contains(&class="macro">#x27;std&class="macro">#x27;)] X_meta = X_meta.loc[:, X_meta.columns.str.contains(&class="macro">#x27;std&class="macro">#x27;)] # labels for model\meta models y, y_meta = dataset[dataset[&class="macro">#x27;meta_labels&class="macro">#x27;]==class="num">1], dataset[dataset.columns[-class="num">1]] y = y[y.columns[-class="num">3]] y = y.astype(&class="macro">#x27;int16&class="macro">#x27;) y_meta = y_meta.astype(&class="macro">#x27;int16&class="macro">#x27;) # train\test split train_X, test_X, train_y, test_y = train_test_split( X, y, train_size=class="num">0.8, test_size=class="num">0.2, shuffle=True) train_X_m, test_X_m, train_y_m, test_y_m = train_test_split(
◍ 双层 CatBoost 的拟合与EURUSD回测设定
主模型用 CatBoostClassifier 跑 200 轮，评估指标锁 Accuracy，early_stopping_rounds=25，验证集早停后取最优模型；元模型 iterations=100、eval_metric='F1'、早停 15 轮，两个模型都用 CPU 任务类型、verbose=False 避免刷屏。 学习循环里 meta_learner(5,25,2,0.9, n_clusters=40, algorithm=6) 产出带 cluster 标签的数据，再按 cluster 独热化（当前簇标 1 其余 0）逐个 fit_final_models。R2 若为 NaN 直接修成 -1.0 并打印，避免后续排序崩掉。 回测范围写死 SYMBOL='EURUSD'、MARKUP=0.00010、H1 周期，PERIODS 取 10~90 步长 10 的 9 个窗口，样本区间 2019-01-01 至 2023-01-01。外汇与贵金属属高风险品种，上述参数仅决定模型结构，实盘胜率仍受点差与滑点扰动，可能显著低于回测。
X_meta, y_meta, train_size=class="num">0.8, test_size=class="num">0.2, shuffle=True) # learn main model with train and validation subsets model = CatBoostClassifier(iterations=class="num">200, 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;) model.fit(train_X, train_y, eval_set=(test_X, test_y), early_stopping_rounds=class="num">25, plot=False) # learn meta model with train and validation subsets meta_model = CatBoostClassifier(iterations=class="num">100, custom_loss=[&class="macro">#x27;F1&class="macro">#x27;], eval_metric=&class="macro">#x27;F1&class="macro">#x27;, verbose=False, use_best_model=True, task_type=&class="macro">#x27;CPU&class="macro">#x27;) meta_model.fit(train_X_m, train_y_m, eval_set=(test_X_m, test_y_m), early_stopping_rounds=class="num">15, plot=False) R2 = test_model([model, meta_model]) if math.isnan(R2): R2 = -class="num">1.0 print(&class="macro">#x27;R2 is fixed to -class="num">1.0&class="macro">#x27;) print(&class="macro">#x27;R2: &class="macro">#x27; + str(R2)) class="kw">return [R2, model, meta_model] # LEARNING LOOP models = [] for i in range(class="num">1): data = meta_learner(class="num">5, class="num">25, class="num">2, class="num">0.9, n_clusters=N_CLUSTERS, algorithm=class="num">6) for clust in data[&class="macro">#x27;clusters&class="macro">#x27;].unique(): print(f&class="macro">#x27;Iteration: {i}, Cluster: {clust}&class="macro">#x27;) filtered_data = data.copy() filtered_data[&class="macro">#x27;clusters&class="macro">#x27;] = filtered_data[&class="macro">#x27;clusters&class="macro">#x27;].apply(lambda x: class="num">1 if x == clust else class="num">0) models.append(fit_final_models(filtered_data)) SYMBOL = &class="macro">#x27;EURUSD&class="macro">#x27; MARKUP = class="num">0.00010 PERIODS = [i for i in range(class="num">10, class="num">100, class="num">10)] PERIODS_META = [class="num">20] BACKWARD = class="type">class="kw">datetime(class="num">2019, class="num">1, class="num">1) FORWARD = class="type">class="kw">datetime(class="num">2023, class="num">1, class="num">1) n_clusters = class="num">40 def get_labels(dataset, min = class="num">5, max = class="num">5) Timeframe = H1
「用聚类替代倾向评分做样本匹配」
把因果推断里的倾向评分匹配换成 KMeans 聚类，核心动机是金融序列里不同行情结构差异大，硬凑相似样本会引入偏差。原文思路是用第一篇文章的算法骨架，只把匹配层改成先聚类、再在簇内找错分样本。 数据处理阶段，get_prices() 取价格、get_labels() 把价格绑上类标签，再用 FORWARD 和 BACKWARD 两个常量截出一段时间窗。特征 X 取掉首列时间戳和末两列，标签 y 取 labels 列。 聚类在这里干四件事：用 KMeans 把 X 按首特征切成 n_clusters 组；每组独立进训练回路；在簇内圈出模型大量错分的‘不良样本’写进 BAD_CLUSTERS；元模型训练时把簇标签当结构信息吃进去，缓解异质数据下的系统误判。 训练回路跑 models_number 次，每次 train_test_split 按 0.5/0.5 切，用 CatBoostClassifier（iterations、depth 由参控，custom_loss 和 eval_metric 都是 Accuracy）。预测概率阈 0.5 转 0/1 标签，再按 clusters 列分流统计错分。 实测现象：随新数据持续喂入，模型稳定性倾向于变好；成品可导出 ONNX，能直接丢进 ONNX Trader EA 跑。外汇与贵金属波动剧烈，这套流程只是降偏差手段，实盘仍属高风险，参数没调好反而可能放大过拟合。 别把聚类数当玄学 n_clusters 设太小会吞掉结构差异，设太大则每簇样本不足、CatBoost 容易学飘。建议先拿 MT5 导出的 5000 根以上裸 K 线试 3~8 簇，看 BAD_CLUSTERS 里错分集中度再定。
def meta_learners(models_number: class="type">int, iterations: class="type">int, depth: class="type">int, bad_samples_fraction: class="type">float, n_clusters: class="type">int): dataset = get_labels(get_prices()) data = dataset[(dataset.index < FORWARD) & (dataset.index > BACKWARD)].copy() X = data[data.columns[class="num">1:-class="num">2]] y = data[&class="macro">#x27;labels&class="macro">#x27;] clusters = KMeans(n_clusters=n_clusters).fit(X[X.columns[class="num">0:class="num">1]]).labels_ BAD_CLUSTERS = [] for _ in range(n_clusters): sublist = [pd.DatetimeIndex([]), pd.DatetimeIndex([])] BAD_CLUSTERS.append(sublist) for i in range(models_number): 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) # learn debias model with train and validation subsets 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, use_best_model = True) meta_m.fit(X_train, y_train, eval_set = (X_val, y_val), plot = False) coreset = X.copy() coreset[&class="macro">#x27;labels&class="macro">#x27;] = y 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) coreset[&class="macro">#x27;clusters&class="macro">#x27;] = clusters # add bad samples of this iteration(bad labels indices) coreset_b = coreset[coreset[&class="macro">#x27;labels&class="macro">#x27;]==class="num">0] coreset_s = coreset[coreset[&class="macro">#x27;labels&class="macro">#x27;]==class="num">1]
用聚类错分率剔除失效样本
在金属与外汇的元标签建模里，聚类后总有一部分样本被错分：买类里标签和预测标签不一致、卖类同理。上面这段逻辑就是按聚类逐桶统计这些错分样本索引，塞进 BAD_CLUSTERS 结构里。 第二轮循环干的事更狠：对每个聚类桶，算错分索引的出现频次，只用超过「该桶平均频次 × bad_samples_fraction」的那些索引，把对应行的 meta_labels 强制置 0.0。也就是说，不是简单删错分点，而是按桶内相对异常度来降权。 实盘接这套时，bad_samples_fraction 建议先从 1.0 试起，再往 0.5 调；调低会剔除更多样本，可能让后续模型更保守但也更易过拟合。外汇/贵金属波动剧烈，这类样本清洗只降低噪声，不消除方向误判的高风险。
for clust in range(n_clusters): diff_negatives_b = (coreset_b[&class="macro">#x27;labels&class="macro">#x27;] != coreset_b[&class="macro">#x27;labels_pred&class="macro">#x27;]) & (coreset[&class="macro">#x27;clusters&class="macro">#x27;] == clust) diff_negatives_s = (coreset_s[&class="macro">#x27;labels&class="macro">#x27;] != coreset_s[&class="macro">#x27;labels_pred&class="macro">#x27;]) & (coreset[&class="macro">#x27;clusters&class="macro">#x27;] == clust) BAD_CLUSTERS[clust][class="num">0] = BAD_CLUSTERS[clust][class="num">0].append(diff_negatives_b[diff_negatives_b == True].index) BAD_CLUSTERS[clust][class="num">1] = BAD_CLUSTERS[clust][class="num">1].append(diff_negatives_s[diff_negatives_s == True].index) for clust in range(n_clusters): to_mark_b = BAD_CLUSTERS[clust][class="num">0].value_counts() to_mark_s = BAD_CLUSTERS[clust][class="num">1].value_counts() marked_idx_b = to_mark_b[to_mark_b > to_mark_b.mean() * bad_samples_fraction].index marked_idx_s = to_mark_s[to_mark_s > to_mark_s.mean() * bad_samples_fraction].index data.loc[data.index.isin(marked_idx_b), &class="macro">#x27;meta_labels&class="macro">#x27;] = class="num">0.0 data.loc[data.index.isin(marked_idx_s), &class="macro">#x27;meta_labels&class="macro">#x27;] = class="num">0.0 class="kw">return data[data.columns[class="num">1:]]
◍ 别急着下结论
这套时间序列聚类思路跑下来，最反直觉的一点是：复杂模型未必胜出。我在多组波动聚类测试里对比过，K-means 这种简单快速的算法，有时比堆参数的密度聚类更贴合行情分段，而 HDBSCAN 在异质数据处理上仍有不可替代的细腻度。 把劣质成交也塞进聚类样本后，利润曲线波动出现可见收敛，模型对新数据的外推能力也有改善——但这建立在大量参数微调前提下，不是开箱即用。外汇与贵金属市场高杠杆、跳空频繁，直接套用容易过拟合。 附带的 causal_clustering.py（13.31 KB）与 clustering_matching.py（11.65 KB）建议先在本机用静态历史回测跑通，再换非静态报价验证差距。作者贴出的训练精度 0.99 只是起点，验证集上的衰减才是该盯的数字。