4. 原始数据清洗：缺失值处理、异常值检测与修正、重复数据去重、数据格式统一、时间戳标准化

原始反洗钱数据（交易流水、客户信息、账户行为日志）进入特征工程流水线之前，必须经过系统化的清洗。清洗质量直接决定后续特征的可信度与模型稳定性。本章聚焦五个核心清洗环节：缺失值处理、异常值检测与修正、重复数据去重、数据格式统一、时间戳标准化。

4.1 缺失值处理

缺失值在反洗钱数据中普遍存在，原因包括：系统采集故障、字段非必填、数据源异构、历史数据迁移遗漏。处理策略分为三类：

# 缺失值处理示例（Python 伪代码）
import pandas as pd
import numpy as np

df = pd.read_csv('aml_transactions.csv')

# 1. 删除缺失率超过 80% 的列
threshold = 0.8
df.dropna(thresh=len(df) * (1 - threshold), axis=1, inplace=True)

# 2. 对数值型字段用中位数填充
numeric_cols = ['amount', 'account_balance']
for col in numeric_cols:
    df[col].fillna(df[col].median(), inplace=True)

# 3. 对类别型字段用 "UNKNOWN" 填充，并生成缺失标记
categorical_cols = ['counterparty_name', 'occupation']
for col in categorical_cols:
    df[col + '_missing'] = df[col].isna().astype(int)
    df[col].fillna('UNKNOWN', inplace=True)

4.2 异常值检测与修正

反洗钱数据中的异常值可能是真正的洗钱行为（如大额拆分交易），也可能是数据录入错误（如金额多一个零）。检测方法需区分全局异常与局部异常。

# 异常值检测与修正示例
from scipy import stats
import numpy as np

# IQR 法检测
Q1 = df['amount'].quantile(0.25)
Q3 = df['amount'].quantile(0.75)
IQR = Q3 - Q1
lower_bound = Q1 - 1.5 * IQR
upper_bound = Q3 + 1.5 * IQR

# 标记异常
df['amount_anomaly'] = ((df['amount'] < lower_bound) | (df['amount'] > upper_bound)).astype(int)

# 修正：对异常值进行截断（Winsorize）
df['amount_corrected'] = df['amount'].clip(lower=lower_bound, upper=upper_bound)

4.3 重复数据去重

重复数据会导致特征分布偏移、模型过拟合、统计指标失真。反洗钱场景中的重复分为三类：

• 完全重复：所有字段完全相同。直接删除，保留一条。
• 部分重复：关键字段相同（如交易流水号），非关键字段不同。需合并或保留最新记录。
• 业务重复：同一笔交易被不同系统重复上报。需根据交易哈希或唯一标识去重。

# 重复数据去重示例
# 1. 完全重复去重
df.drop_duplicates(inplace=True)

# 2. 按交易流水号去重，保留最新记录
df.sort_values('timestamp', ascending=False, inplace=True)
df.drop_duplicates(subset=['transaction_id'], keep='first', inplace=True)

# 3. 业务重复：按客户+金额+时间窗口（1小时）去重
df['time_window'] = df['timestamp'].dt.floor('H')
df.drop_duplicates(subset=['customer_id', 'amount', 'time_window'], keep='first', inplace=True)

4.4 数据格式统一

异构数据源（核心银行系统、支付网关、风控平台）的字段格式差异巨大。格式统一是特征工程的前置条件。

• 数值格式：统一为浮点数或整数，去除千分位分隔符、货币符号。
• 字符串格式：统一大小写、去除前后空格、标准化缩写（如“USD”统一为“美元”）。
• 类别编码：将“男/女”、“M/F”、“1/0”统一为同一编码体系。
• 地址格式：将“北京市朝阳区”与“北京朝阳”统一为省-市-区三级结构。

# 数据格式统一示例
# 1. 金额字段去除货币符号和千分位
df['amount'] = df['amount'].str.replace('[\$,¥,€]', '', regex=True)
df['amount'] = df['amount'].str.replace(',', '').astype(float)

# 2. 字符串统一小写并去空格
df['customer_name'] = df['customer_name'].str.strip().str.lower()

# 3. 性别字段标准化
gender_map = {'M': '男', 'F': '女', '1': '男', '0': '女', 'male': '男', 'female': '女'}
df['gender'] = df['gender'].map(gender_map).fillna('未知')

4.5 时间戳标准化

时间戳是反洗钱特征工程中最关键的维度之一。不同系统可能使用不同格式：2024-01-15 14:30:00、01/15/2024 2:30 PM、1705312200（Unix 时间戳）。标准化步骤：

1. 统一解析：将所有时间戳转为 datetime 对象。
2. 统一时区：全部转为 UTC 或业务所在时区（如 Asia/Shanghai）。
3. 提取时间特征：从时间戳中提取小时、星期、月份、是否节假日等特征。
4. 计算时间差：计算交易时间与上次交易、开户时间、预警时间的时间间隔。

# 时间戳标准化示例
import pandas as pd

# 1. 统一解析（自动推断格式）
df['timestamp'] = pd.to_datetime(df['timestamp'], errors='coerce')

# 2. 时区转换（假设原始为 Asia/Shanghai，转为 UTC）
df['timestamp_utc'] = df['timestamp'].dt.tz_localize('Asia/Shanghai', ambiguous='NaT').dt.tz_convert('UTC')

# 3. 提取时间特征
df['hour'] = df['timestamp_utc'].dt.hour
df['day_of_week'] = df['timestamp_utc'].dt.dayofweek
df['is_weekend'] = df['day_of_week'].isin([5, 6]).astype(int)
df['month'] = df['timestamp_utc'].dt.month

# 4. 计算时间差（与上次交易的时间间隔）
df.sort_values(['customer_id', 'timestamp_utc'], inplace=True)
df['time_since_last_tx'] = df.groupby('customer_id')['timestamp_utc'].diff().dt.total_seconds()

4.6 清洗质量验证

清洗完成后，必须进行质量验证，确保清洗过程没有引入新的问题：

• 缺失率检查：清洗后各字段缺失率应为 0（填充策略）或可控（标记策略）。
• 异常值分布：清洗后异常值比例应显著下降，且修正后的值在业务合理范围内。
• 重复率检查：去重后重复率应接近 0，且样本总量在预期范围内。
• 格式一致性：随机抽样 100 条，人工检查格式是否统一。
• 时间戳连续性：检查时间戳是否在合理时间范围内，无未来时间或过时数据。

# 清洗质量验证示例
print(f"清洗后缺失率: \n{df.isnull().sum() / len(df)}")
print(f"异常值比例: {df['amount_anomaly'].mean():.2%}")
print(f"重复率: {df.duplicated(subset=['transaction_id']).mean():.2%}")
print(f"时间戳范围: {df['timestamp_utc'].min()} ~ {df['timestamp_utc'].max()}")