4. 数据准备与标注：正负样本定义、标注规范、数据划分策略与质量校验

数据是反欺诈图谱模型的基石。在构建和评估模型之前，必须对原始数据进行清洗、定义明确的标签、制定统一的标注规范，并采用合理的划分策略。本章将系统讲解数据准备与标注的核心环节，确保后续评估工作的科学性和可复现性。

4.1 正负样本定义

在反欺诈场景中，正样本通常指“欺诈行为”，负样本指“正常行为”。但实际业务中，欺诈的定义往往具有时效性和业务依赖性。以下是常见的正负样本定义原则：

4.2 标注规范

标注规范是保证数据一致性的关键。对于图谱反欺诈任务，标注不仅涉及节点（实体）的标签，还涉及边（关系）的标签。以下是推荐的标注规范框架：

4.2.1 节点标注规范

• 实体类型： 明确标注每个节点的实体类型（如：用户、设备、IP、手机号、银行卡、地址）。
• 欺诈标签： 每个节点可标注“欺诈”、“正常”或“未知”。对于“欺诈”节点，需记录欺诈类型（如：盗用、恶意注册、洗钱）。
• 属性完整性： 标注节点的重要属性（如注册时间、交易金额、地理位置），缺失值需用统一占位符（如“UNKNOWN”）标记。

4.2.2 边（关系）标注规范

• 关系类型： 明确边的语义（如：登录、交易、通话、转账、共享设备）。
• 时间戳： 每条边必须标注发生时间，用于时间序列划分和时序特征构建。
• 关系权重： 可标注边的强度（如交易金额、通话时长、共享次数），用于图谱算法中的边加权。

4.2.3 标注一致性检查

• 交叉验证： 同一批样本至少由两名标注员独立标注，计算标注一致性（Cohen's Kappa 或 Fleiss' Kappa），Kappa 值应 ≥ 0.8。
• 争议仲裁： 建立争议样本仲裁机制，由资深风控专家最终裁定。
• 标注日志： 记录每次标注的修改历史、标注员ID、标注时间，便于追溯和审计。

4.3 数据划分策略

数据划分直接影响评估结果的可靠性。对于反欺诈图谱数据，推荐结合时间序列划分和随机划分两种策略，以适应不同评估目标。

4.3.1 时间序列划分

反欺诈场景具有明显的时效性——欺诈手法随时间演变。时间序列划分能模拟模型在未来的真实表现，是评估泛化能力的黄金标准。

4.3.2 随机划分

随机划分适用于评估模型在静态数据集上的拟合能力和特征有效性。在反欺诈中，随机划分通常用于模型调参和特征选择阶段，但不能替代时间序列划分。

4.3.3 划分策略对比

4.4 数据质量校验

数据质量直接影响模型效果和评估结论的可信度。在数据准备阶段，必须建立系统化的质量校验流程。以下是推荐的校验清单：

4.4.1 完整性校验

• 缺失值检测： 检查节点属性和边属性的缺失比例，对缺失率超过30%的字段需评估是否保留。
• 孤立节点检测： 在图谱中，孤立节点（无边连接的节点）可能表示数据采集不完整，需确认是否应保留。
• 时间戳完整性： 确保所有边都有有效的时间戳，无时间戳的边应标记为“未知时间”并单独处理。

4.4.2 一致性校验

• 实体对齐： 同一实体在不同数据源中的ID是否一致？如不一致，需建立实体映射表。
• 关系对称性： 对于双向关系（如好友关系），检查A→B和B→A是否同时存在。
• 标签一致性： 同一节点在不同时间窗口内的标签是否矛盾？如一个用户既被标为“欺诈”又被标为“正常”，需人工复核。

4.4.3 准确性校验

• 异常值检测： 检查属性值是否在合理范围内（如交易金额为负数、注册时间在未来）。
• 重复数据检测： 检查是否存在完全重复的节点或边，去重后保留唯一记录。
• 标签噪声检测： 随机抽取5%-10%的标注样本进行二次复核，计算标注准确率（应 ≥ 95%）。

4.4.4 时效性校验

• 数据新鲜度： 检查数据的时间分布，确保训练集和测试集的时间窗口覆盖当前业务周期。
• 概念漂移检测： 比较不同时间窗口内正样本的分布变化，若欺诈模式发生显著变化，需重新标注或调整模型。

4.5 实战案例：数据准备流程示例

以下是一个简化的数据准备流程代码示例，展示如何实现时间序列划分和数据质量校验：

import pandas as pd
import numpy as np
from sklearn.model_selection import train_test_split

# 假设已有图谱边数据，包含时间戳和标签
edges = pd.read_csv('graph_edges.csv')
edges['timestamp'] = pd.to_datetime(edges['timestamp'])

# 1. 数据质量校验
print("缺失值统计：")
print(edges.isnull().sum())
print(f"重复边数量：{edges.duplicated().sum()}")
print(f"时间范围：{edges['timestamp'].min()} 至 {edges['timestamp'].max()}")

# 2. 时间序列划分
edges_sorted = edges.sort_values('timestamp')
train_end = int(len(edges_sorted) * 0.6)
val_end = int(len(edges_sorted) * 0.8)

train_data = edges_sorted.iloc[:train_end]
val_data = edges_sorted.iloc[train_end:val_end]
test_data = edges_sorted.iloc[val_end:]

print(f"训练集时间范围：{train_data['timestamp'].min()} - {train_data['timestamp'].max()}")
print(f"验证集时间范围：{val_data['timestamp'].min()} - {val_data['timestamp'].max()}")
print(f"测试集时间范围：{test_data['timestamp'].min()} - {test_data['timestamp'].max()}")

# 3. 随机划分（用于特征验证）
X = edges.drop('label', axis=1)
y = edges['label']
X_train_rand, X_test_rand, y_train_rand, y_test_rand = train_test_split(
    X, y, test_size=0.3, random_state=42, stratify=y
)
print(f"随机划分 - 训练集大小：{len(X_train_rand)}, 测试集大小：{len(X_test_rand)}")