3. 数据采集层架构：实时与离线数据源接入、API网关设计、数据清洗策略

数据采集层是商户风险评分通道的“感官系统”，负责从分散的异构数据源中获取原始信息，并完成格式统一、质量校验与初步加工。本章聚焦三类核心设计：实时与离线数据源的接入方式、API网关的流量治理与协议转换、数据清洗的策略与工程实现。

3.1 实时与离线数据源接入

商户风险评分需要同时依赖实时事件（如交易行为、登录异常）与离线特征（如工商信息、历史黑名单）。两类数据在时效性、吞吐量、接入协议上差异显著，需采用不同的架构组件。

3.2 API网关设计

API网关是数据采集层的统一入口，负责协议转换、流量控制、鉴权与路由。在商户风险评分场景中，网关需要同时支持外部数据源（如第三方征信）与内部微服务的数据请求。

3.2.1 网关核心功能

• 协议适配：将外部HTTP/JSON请求转换为内部gRPC或Thrift调用，或反向转换。
• 限流熔断：按商户等级、API Key、IP维度设置QPS阈值，防止突发流量击穿下游。
• 鉴权与签名校验：验证请求的AppId、Secret、时间戳与签名，防止重放攻击。
• 请求/响应转换：统一字段命名规范（如 snake_case 转 camelCase），过滤敏感字段。
• 链路追踪：注入TraceId，串联网关→采集服务→清洗管道→评分引擎的全链路日志。

3.2.2 网关路由策略

3.3 数据清洗策略

原始数据存在缺失、重复、格式错误、异常值等问题，必须经过清洗才能进入特征工程。清洗策略按处理阶段分为前置过滤、字段级清洗与后置校验三层。

3.3.1 前置过滤（丢弃无效数据）

• 空记录过滤：关键字段（商户ID、交易金额、时间戳）全部为空的记录直接丢弃。
• 格式校验：检查手机号、身份证号、银行卡号的正则格式，不符合则标记为“格式异常”并进入人工审核队列。
• 时间范围过滤：时间戳超出当前时间前后24小时的记录视为异常，拒绝入库。

3.3.2 字段级清洗（标准化与补全）

3.3.3 后置校验（质量监控与告警）

• 完整性校验：每条记录必填字段的填充率是否达到99.9%以上。
• 唯一性校验：按商户ID+交易流水号去重，重复记录只保留第一条。
• 波动性校验：单日数据量环比波动超过±30%时触发告警，排查采集链路是否异常。

3.4 数据清洗代码示例（Python）

以下是一个简化的清洗算子实现，展示字段标准化与异常标记逻辑：

import re
import time
from datetime import datetime

def clean_merchant_record(record: dict) -> dict:
    """清洗单条商户记录，返回清洗后数据与异常标记列表"""
    anomalies = []
    cleaned = record.copy()

    # 1. 商户名称清洗：去空格、全角转半角
    name = cleaned.get('merchant_name', '')
    name = name.strip()
    name = name.replace('　', ' ')  # 全角空格转半角
    name = re.sub(r'[^\w\u4e00-\u9fff]', '', name)  # 只保留汉字、字母、数字
    cleaned['merchant_name'] = name

    # 2. 金额标准化：统一转为分（单位：元转分）
    amount = cleaned.get('amount', 0)
    if isinstance(amount, float) or isinstance(amount, str):
        try:
            amount = int(float(amount) * 100)
        except (ValueError, TypeError):
            anomalies.append('amount_format_error')
            amount = 0
    cleaned['amount_cents'] = amount

    # 3. 时间字段统一为Unix毫秒
    ts = cleaned.get('event_time', '')
    if isinstance(ts, str):
        try:
            dt = datetime.strptime(ts, '%Y-%m-%d %H:%M:%S')
            cleaned['event_time_ms'] = int(dt.timestamp() * 1000)
        except ValueError:
            anomalies.append('time_format_error')
            cleaned['event_time_ms'] = int(time.time() * 1000)
    else:
        cleaned['event_time_ms'] = int(ts) if ts else int(time.time() * 1000)

    # 4. 枚举字段映射
    gender_map = {'男': 'MALE', '女': 'FEMALE', 'M': 'MALE', 'F': 'FEMALE'}
    raw_gender = cleaned.get('gender', 'UNKNOWN')
    cleaned['gender'] = gender_map.get(raw_gender.upper(), 'UNKNOWN')

    cleaned['anomalies'] = anomalies
    return cleaned