3. 数据预处理与清洗故障：字段缺失、数据异常值、重复数据、数据标准化错误

在反洗钱（AML）推理链路中，数据预处理与清洗是决定模型准确性的基石。原始交易数据通常来自多个异构系统，包含大量噪声、缺失值和格式不一致问题。若清洗环节出现故障，后续的规则引擎、异常检测模型和关联分析将产生大量误报或漏报。本章系统梳理四大类清洗故障的根因、排查方法与修复策略。

3.1 字段缺失故障

字段缺失是最常见的数据质量问题，表现为关键字段（如交易对手名称、交易币种、IP 地址）为空或为 NULL。缺失原因包括：源系统未采集、接口传输丢包、ETL 映射遗漏、字段被误删除。

-- 示例：检查交易对手名称缺失率
SELECT
  COUNT(*) AS total_records,
  SUM(CASE WHEN counterparty_name IS NULL OR counterparty_name = '' THEN 1 ELSE 0 END) AS missing_count,
  ROUND(100.0 * SUM(CASE WHEN counterparty_name IS NULL OR counterparty_name = '' THEN 1 ELSE 0 END) / COUNT(*), 2) AS missing_pct
FROM aml_transactions
WHERE transaction_date >= '2025-01-01';

3.2 数据异常值故障

异常值指明显偏离正常分布的数据点，如单笔交易金额为 0 元、负数、或超出正常范围 100 倍以上。异常值会严重扭曲统计指标和模型权重，导致误将正常交易判为可疑，或掩盖真正的洗钱行为。

• 数值异常： 金额为负数、金额为 0、金额超过 99.9% 分位数 5 倍以上。
• 时间异常： 交易时间在未来、交易时间早于客户出生日期、时间戳为 1970-01-01。
• 类别异常： 交易币种代码不在 ISO 4217 列表中、国家代码为“ZZ”。
• 频率异常： 同一账户 1 分钟内交易次数超过 100 笔（疑似机器刷单）。

# Python 示例：使用 IQR 方法检测金额异常值
import pandas as pd

df = pd.read_parquet('transactions.parquet')
Q1 = df['amount'].quantile(0.25)
Q3 = df['amount'].quantile(0.75)
IQR = Q3 - Q1
lower_bound = Q1 - 3 * IQR
upper_bound = Q3 + 3 * IQR

anomalies = df[(df['amount'] < lower_bound) | (df['amount'] > upper_bound)]
print(f"检测到异常值数量: {len(anomalies)}")

3.3 重复数据故障

重复数据导致同一笔交易被多次计分，造成虚假的关联强度和高频交易误判。重复原因包括：源系统重复推送、ETL 未做去重、时间窗口内同一交易被多次采集。

3.4 数据标准化错误

数据标准化错误指不同源系统的数据格式、编码、单位不一致，导致模型无法正确解析。例如：金额字段有的以“元”为单位，有的以“万元”为单位；日期格式有“YYYY-MM-DD”和“MM/DD/YYYY”两种；国家代码有“CN”和“CHN”混用。

-- 示例：日期格式标准化检查
SELECT
  transaction_id,
  raw_date,
  CASE
    WHEN raw_date ~ '^\d{4}-\d{2}-\d{2}$' THEN 'ISO标准'
    WHEN raw_date ~ '^\d{2}/\d{2}/\d{4}$' THEN '美式格式'
    WHEN raw_date ~ '^\d{8}$' THEN '紧凑格式'
    ELSE '未知格式'
  END AS date_format_type
FROM aml_transactions
WHERE raw_date IS NOT NULL
LIMIT 20;

3.5 综合排查流程

当数据预处理环节出现故障时，建议按以下步骤进行系统性排查：

1. 数据概要分析： 统计各字段缺失率、唯一值数量、数据类型、最大值/最小值。
2. 字段级校验： 对每个字段执行预定义的校验规则（非空、范围、格式、枚举值）。
3. 跨字段一致性检查： 例如交易金额与交易币种是否匹配、交易时间与客户注册时间是否合理。
4. 重复检测： 使用业务主键 + 时间窗口进行精确和模糊去重。
5. 标准化验证： 抽样检查标准化后的数据是否符合目标 schema 定义。
6. 修复与重跑： 根据故障类型执行修复脚本，重新执行数据预处理管道。