3. 数据采集与治理：商户基础信息、交易数据、外部征信数据、数据清洗与标准化流程
商户风险分群评分模型的精度，80% 取决于数据质量，20% 取决于算法。本章聚焦于风险评分所需的数据源体系、采集策略以及治理流程，确保后续沙盘推演所使用的数据具备完整性、一致性与时效性。
3.1 数据源全景：三大核心维度
商户风险画像需要从 基础信息、交易行为、外部征信 三个维度进行立体采集。单一维度的数据极易产生偏差，只有多源交叉验证才能有效识别套现、洗钱、虚假交易等风险。
| 数据维度 | 典型字段 | 采集频率 | 风险关联度 |
|---|---|---|---|
| 基础信息 | 营业执照、法人身份、经营地址、行业类目、注册时长 | 首次入驻 + 年度更新 | ★★★★★（身份真实性） |
| 交易数据 | 交易金额、笔数、客单价、时段分布、退款率、失败率 | 每日 / 实时 | ★★★★★（行为异常） |
| 外部征信 | 司法涉诉、行政处罚、多头借贷、工商变更、黑名单 | 每周 / 事件驱动 | ★★★★☆（信用风险） |
实战提示： 基础信息中的“行业类目”是分群的第一层过滤器。例如，虚拟商品、珠宝、游戏点卡等行业的平均风险权重是餐饮行业的 3-5 倍，在评分卡初始阶段应赋予更高权重。
3.2 数据清洗与标准化流程
原始数据通常包含缺失值、异常值、重复记录和格式不一致问题。以下流程是沙盘推演前必须执行的标准化治理步骤：
- 字段解析与映射：将不同来源的字段名统一映射到标准字典（如
merchant_id→mch_id）。 - 缺失值处理：连续型变量用中位数或行业均值填充；分类型变量用“未知”或众数填充；缺失率超过 60% 的字段直接剔除。
- 异常值截断：对交易金额、笔数等连续变量，采用 3σ 或 IQR 方法进行截断，防止极端值扭曲评分分布。
- 重复记录去重：以商户唯一标识（统一社会信用代码或商户号）为主键，保留最新一条记录。
- 格式标准化：日期统一为
YYYY-MM-DD，金额统一为分（整数），手机号、身份证号做脱敏处理。 - 逻辑校验：例如注册日期不能晚于当前日期，交易金额不能为负数，退款率不能大于 100%。
避坑指南： 在沙盘推演中，很多团队直接使用原始交易数据建模，忽略了“退款率”字段中的分母为零的情况。如果某商户当月交易笔数为 0，退款率应标记为 NULL 而非 0%，否则会错误地降低该商户的风险评分。
3.3 数据治理架构：从采集到特征存储
下图展示了商户风险评分系统中数据采集与治理的完整链路，涵盖离线批处理与实时流处理两条路径。
3.4 数据质量评估指标
在数据进入特征工程之前，必须通过以下四个维度的质量评估。沙盘推演中，建议将质量评分作为数据治理阶段的交付物之一。
| 评估维度 | 指标定义 | 合格阈值 | 治理动作 |
|---|---|---|---|
| 完整性 | 非空字段占比 | ≥ 95% | 缺失率高的字段剔除或填充 |
| 准确性 | 逻辑校验通过率 | ≥ 98% | 异常记录标记或修正 |
| 一致性 | 跨源字段匹配率 | ≥ 90% | 统一字典映射 |
| 时效性 | 数据延迟时间 | ≤ 24h（离线）/ ≤ 5s（实时） | 优化采集管道 |
沙盘推演建议： 在实战演练中，可以故意引入 5%-10% 的脏数据（如重复商户、异常金额、缺失法人信息），让学员体验数据治理对评分模型稳定性的影响。通常，未经治理的数据集会导致模型 KS 值下降 0.15-0.25。
3.5 数据标准化代码示例
以下 Python 代码片段展示了商户交易数据清洗与标准化的核心逻辑，可直接用于沙盘推演的数据预处理环节。
import pandas as pd
import numpy as np
def clean_merchant_data(df):
"""
商户交易数据清洗与标准化
"""
# 1. 去重：保留最新记录
df = df.sort_values('update_time').drop_duplicates(subset='merchant_id', keep='last')
# 2. 缺失值处理
df['refund_rate'].fillna(df['refund_rate'].median(), inplace=True)
df['industry_code'].fillna('UNKNOWN', inplace=True)
# 3. 异常值截断（交易金额：3σ）
mean_amt = df['trans_amount'].mean()
std_amt = df['trans_amount'].std()
df['trans_amount'] = df['trans_amount'].clip(0, mean_amt + 3*std_amt)
# 4. 格式标准化
df['trans_date'] = pd.to_datetime(df['trans_date']).dt.strftime('%Y-%m-%d')
df['trans_amount'] = (df['trans_amount'] * 100).astype(int) # 元转分
# 5. 逻辑校验
df = df[(df['trans_amount'] >= 0) & (df['refund_rate'] <= 1.0)]
return df