3. 数据采集与预处理：交易数据源（Kafka/API）、特征工程流水线、实时数据清洗与标准化

实时反洗钱（AML）监测模型的准确性高度依赖数据管道的质量与时效性。本章聚焦于从异构交易数据源到标准化特征向量的完整链路，涵盖数据采集、清洗、标准化及特征工程流水线的设计与实现。

3.1 交易数据源接入：Kafka 与 API 网关

实时交易数据通常以高吞吐、低延迟的流式方式产生。主流接入方案包括消息队列（Kafka）和 RESTful API 网关，两者在架构中承担不同角色。

3.2 实时数据清洗与标准化

原始交易数据常包含缺失字段、格式异常、重复记录或非法值。在进入特征工程前，必须经过严格的清洗与标准化处理。

3.2.1 清洗规则清单

• 字段完整性校验： 交易 ID、账户 ID、交易金额、交易时间、交易对手等核心字段不可为空。
• 格式标准化： 时间戳统一为 UTC 毫秒级（Unix epoch ms）；金额统一为最小货币单位（分/厘）；货币代码统一为 ISO 4217 三位字母码。
• 去重处理： 基于交易 ID + 账户 ID + 交易时间窗口（如 5 秒内）进行幂等去重。
• 异常值过滤： 金额为负、交易时间在未来、账户状态异常等数据直接丢弃或进入死信队列（DLQ）。

3.2.2 标准化处理流程

1. 解析与反序列化： 从 Kafka 消费 Avro/JSON 消息，转换为内部统一数据模型（POJO/Dict）。
2. 字段映射： 将不同数据源的字段名映射为标准字段（如 txn_amt → transaction_amount）。
3. 类型转换： 字符串数字转浮点型；日期字符串转时间戳；枚举值转内部编码。
4. 衍生字段计算： 如交易时间所属时段（凌晨/白天/夜间）、交易对手类型（个人/商户/对公）。

DataStream<Transaction> cleanedStream = rawStream
    .filter(txn -> txn.getAmount() != null && txn.getAmount() > 0)
    .filter(txn -> txn.getTimestamp() <= System.currentTimeMillis())
    .map(new MapFunction<RawTransaction, Transaction>() {
        @Override
        public Transaction map(RawTransaction raw) {
            Transaction std = new Transaction();
            std.setTransactionId(raw.getTxnId());
            std.setAccountId(raw.getAcctId());
            std.setAmount(raw.getAmount().doubleValue() / 100.0); // 分转元
            std.setTimestamp(raw.getTimestamp());
            std.setCurrencyCode(raw.getCcy().toUpperCase());
            return std;
        }
    })
    .keyBy(txn -> txn.getTransactionId())
    .process(new DeduplicateFunction(Time.seconds(5)));

3.3 特征工程流水线

特征工程是 AML 模型性能的关键。实时流水线需在低延迟约束下完成特征提取、聚合与编码。

3.3.1 特征分类与计算方式

3.3.2 流水线架构

特征工程流水线通常采用有向无环图（DAG）方式组织，每个节点负责一类特征计算，节点间通过内存通道或状态后端共享中间结果。

3.4 数据质量监控与异常处理

实时流水线必须内置数据质量监控机制，确保特征输入的稳定性。

• 延迟监控： 每条数据从进入 Kafka 到完成特征计算的时间差（端到端延迟），告警阈值通常设为 5 秒。
• 数据倾斜检测： 监控各分区/分片的数据量方差，当某个分区数据量超过均值 3 倍时触发重分区。
• 特征分布漂移： 使用滑动窗口统计特征均值、方差、分位数，与基线对比，超过 3σ 时触发模型重训练或特征回滚。
• 死信队列（DLQ）： 清洗失败、格式异常、无法映射的数据统一写入 Kafka DLQ Topic，供离线分析。

3.5 流水线部署与版本管理

特征工程流水线应作为独立微服务部署，与模型推理服务解耦。推荐使用容器化（Docker + Kubernetes）方式管理。

通过上述数据采集、清洗、标准化与特征工程流水线的完整设计，可以为 AML 实时监测模型提供稳定、高质量的特征输入，从而保障模型推理的准确性与时效性。