训练特征和标签怎么合并？按样本编号连接并检查一对一关系

特征表和标签表分别导出后，应按稳定样本编号连接，不能假定两个文件的第 n 行一定属于同一条样本。先确认两侧编号唯一且不缺失，再用 merge(validate='one_to_one')连接，通过外连接和 indicator=True检查两侧未匹配的编号，最后才生成训练用的 X 和 y。

本文假设每个样本只有一行特征和一个分类标签，标签表列名为 label。示例编号与数值为人为构造，已在 Windows、Python 3.11.15、pandas 3.0.6 环境运行，演示的是连接和拒绝机制，没有真实业务标签或训练效果。

为什么先外连接，再决定能否训练

pandas merge 文档说明，one_to_one 检查两侧连接键的唯一性；indicator 产生 left_only、right_only和 both三种来源标记。外连接保留两侧键的并集，能让漏标或孤立标签显式出现。直接内连接只留下交集，容易把未匹配样本的丢失隐藏掉。

编号要保留约定类型。若编号有前导零，读 CSV 时对两张表都使用 dtype={'sample_id': 'string'}。不要先一张转整数、另一张留字符串，然后因为连接失败而随意删除字符。pandas 对两侧空键可能进行相互匹配，因此本例在连接之前主动拒绝空编号。

先建立独立环境。macOS/Linux 的激活命令是 source .venv/bin/activate。

python -m venv .venv
.venv\Scripts\activate
python -m pip install pandas==3.0.6

完整示例：拒绝未匹配，再接受已核对标签

把代码保存为 join_training_rows.py，运行 python join_training_rows.py。第一次故意让特征编号 103 没有标签、标签编号 104 没有特征；第二次使用已纠正的标签表；最后分别注入左侧和右侧重复编号，验证两种重复都被拒绝。

import pandas as pd

def join_for_training(features, labels, audit_file):
    for name, table in [('features', features), ('labels', labels)]:
        if table['sample_id'].isna().any():
            raise ValueError(f'{name}: missing sample_id')
        duplicated = table.loc[table.duplicated('sample_id', keep=False), 'sample_id'].tolist()
        if duplicated:
            raise ValueError(f'{name}: duplicate sample_id {duplicated}')
    overlap = (set(features.columns) & set(labels.columns)) - {'sample_id'}
    if overlap:
        raise ValueError(f'ambiguous columns: {sorted(overlap)}')
    merged = features.merge(labels, on='sample_id', how='outer',
                            validate='one_to_one', indicator=True, sort=True)
    merged.to_csv(audit_file, index=False, encoding='utf-8-sig')
    unmatched = merged.loc[merged['_merge'] != 'both', ['sample_id', '_merge']]
    if not unmatched.empty:
        raise ValueError('unmatched rows: ' + unmatched.to_json(orient='records'))
    if merged['label'].isna().any():
        raise ValueError('missing label values')
    return merged.drop(columns='_merge')

features = pd.DataFrame({'sample_id': [101, 102, 103],
                         'age': [22, 34, 46], 'visits': [1, 3, 5]})
labels = pd.DataFrame({'sample_id': [102, 101, 104], 'label': [1, 0, 1]})
try:
    join_for_training(features, labels, 'join_audit_unmatched.csv')
except ValueError as error:
    print('unmatched_rejected:', str(error))
else:
    raise AssertionError('Unmatched records were accepted')
audit = pd.read_csv('join_audit_unmatched.csv')
assert audit.loc[audit['_merge'] == 'left_only', 'sample_id'].tolist() == [103]
assert audit.loc[audit['_merge'] == 'right_only', 'sample_id'].tolist() == [104]
corrected_labels = pd.DataFrame({'sample_id': [102, 101, 103], 'label': [1, 0, 1]})
training = join_for_training(features, corrected_labels, 'join_audit_ok.csv')
X = training[['age', 'visits']]
y = training['label']
assert len(training) == 3
assert training['sample_id'].tolist() == [101, 102, 103]
assert y.tolist() == [0, 1, 1]
assert 'sample_id' not in X.columns
training.to_csv('training_rows.csv', index=False, encoding='utf-8-sig')
print('training_rows:', len(training))
print('sample_label_pairs:', list(zip(training['sample_id'], y)))
for name, left, right in [
    ('features', pd.concat([features, features.iloc[[0]]], ignore_index=True), corrected_labels),
    ('labels', features, pd.concat([corrected_labels, corrected_labels.iloc[[0]]], ignore_index=True))
]:
    try:
        join_for_training(left, right, 'must_not_write.csv')
    except ValueError as error:
        print('duplicate_' + name + '_rejected:', str(error))
    else:
        raise AssertionError('Duplicate sample key was accepted')

读审计表，确认每行标签归属

实际输出如下。标签表原本的顺序为 102、101、103，连接后仍按编号对应到正确标签。

unmatched_rejected: unmatched rows: [{"sample_id":103,"_merge":"left_only"},{"sample_id":104,"_merge":"right_only"}]
training_rows: 3
sample_label_pairs: [(101, 0), (102, 1), (103, 1)]
duplicate_features_rejected: features: duplicate sample_id [101, 101]
duplicate_labels_rejected: labels: duplicate sample_id [102, 102]

第一次连接保存 join_audit_unmatched.csv后抛错；审计表中 103 为 left_only，104 为 right_only。第二次保存 join_audit_ok.csv，三行都为 both，训练配对为 (101, 0)、(102, 1)、(103, 1)。最终 training_rows.csv有三行，X 只包含 age 和 visits，编号没有混入模型特征。

左右重复编号均触发拒绝，错误里列出所有重复出现的键。duplicated 官方文档说明，keep=False会标出每组重复的所有行，本例用它保留冲突线索；不能仅靠去重掩盖标签冲突。

查看未匹配审计表应有四行；纠正后审计表应有三行、零条 left_only/right_only。再按样本编号抽查原始特征与标签，核对标签含义和版本。one_to_one 能验证键关系，但不会判断人工标签本身是否正确。

遇到真实数据的三种分歧怎样处理

1. 同一样本有多次标注：如果训练要求一条标签，先由已批准的复核规则解决冲突，再生成最终标签表；保留标注人、时间和原值，不能简单 drop_duplicates。
2. 同一样本有多个时间点：sample_id 可能不足以做唯一键。应使用真实含义明确的组合键，例如 sample_id 与 observation_time，并确保标签的可用时间与特征窗口一致。
3. 只训练有标签样本：可以在审核后有意识地保留交集，但必须记录排除编号、原因和数量，检查排除是否集中在某类样本。本文默认零未匹配才继续，需要修改策略时要同步改验收规则。

实际接入时，先检查两张表都含 sample_id、标签表含 label，再执行函数。对同名但不同含义的非键列，应先显式改名；本例会拒绝这类冲突，避免默认后缀掩盖来源混乱。若前面发生键重复、缺失或未匹配错误，应停止训练数据生成，补齐资料后重新连接并保存新审计证据。

连接通过后可以立刻训练吗？

还要核对标签取值范围、特征类型、缺失值策略，以及训练和测试的划分方式。本文完成的是样本归属与一对一结构验证；它没有验证标签准确率、特征是否泄漏目标或数据是否代表未来使用场景。