第1章：数据源与采集——合规数据的“源头活水”

大家好，我是老张。干合规数据分析这些年，我最大的感触就是：数据质量决定了分析的天花板。你算法再牛，模型再炫，源头数据是脏的、乱的、不全的，那结果就是“垃圾进，垃圾出”。今天咱们就来聊聊，合规数据到底从哪来，怎么采，以及如何做第一道“体检”。

1.1 合规数据的三大来源

我个人习惯把合规数据来源分成三类：交易数据、客户数据、第三方数据。这三类数据各有各的脾气，咱们一个一个说。

1. 交易数据——合规分析的“主食”

交易数据是合规分析最核心的数据源。说白了，就是每一笔业务发生的记录。比如银行转账、证券买卖、保险理赔、电商订单等等。这类数据的特点是：量大、频率高、结构化程度高。

我在项目中遇到过一家支付公司，每天处理上千万笔交易。他们的交易数据表里，字段多达200多个。但真正用于合规分析的，其实就那么十几个核心字段：交易ID、交易时间、交易金额、交易对手、交易渠道、IP地址、设备指纹等。

2. 客户数据——合规分析的“灵魂”

客户数据，也叫KYC（Know Your Customer）数据。包括客户的身份信息、联系方式、职业、收入、风险等级等。这类数据的特点是：相对稳定、更新频率低、但敏感度高。

嗯，这里要注意：客户数据是反洗钱、反欺诈的基石。你想想看，如果连客户是谁都搞不清楚，怎么判断交易是否异常？

我曾经遇到过一个案例：某客户在系统里注册了三个账户，用的都是同一个身份证号，但名字却不一样——一个写“张三”，一个写“张叁”，还有一个写“Zhang San”。这就是典型的客户数据质量问题，根源在于录入环节没有做校验。

3. 第三方数据——合规分析的“外援”

第三方数据，说白了就是你自己没有，需要从外部买或换的数据。比如：

• 黑名单数据：制裁名单、政治人物名单、通缉犯名单等
• 工商数据：企业注册信息、股权结构、法人代表等
• 舆情数据：新闻、社交媒体、论坛等公开信息
• 征信数据：央行征信、百行征信等

第三方数据最大的问题是时效性和准确性。我见过不少公司买了黑名单数据，结果半年没更新，漏掉了最新的制裁对象。所以，用第三方数据一定要关注数据源的更新频率和覆盖范围。

1.2 数据采集方法与工具

数据采集，说白了就是把数据从各个源头“搬”到你的分析环境里。方法有很多，但核心就三个字：快、准、稳。

1. 批量采集——最传统也最常用

批量采集，就是定时（比如每天凌晨）把数据从源系统导出，然后加载到数据仓库或分析平台。常用的工具有：

• Sqoop：适合关系型数据库（MySQL、Oracle）到Hadoop的批量传输
• DataX：阿里巴巴开源的异构数据源同步工具，支持多种数据源
• Kettle：一款开源的ETL工具，图形化界面，适合非技术人员

# 一个简单的Sqoop导入示例
sqoop import \
  --connect jdbc:mysql://localhost:3306/compliance \
  --username root \
  --password secret \
  --table transaction_log \
  --target-dir /data/compliance/transaction \
  --fields-terminated-by ',' \
  --num-mappers 4

2. 实时采集——应对“快”场景

有些合规场景需要实时分析，比如在线支付的反欺诈、实时交易监控。这时候就需要实时采集技术。常用的工具有：

• Kafka：分布式消息队列，高吞吐、低延迟，是实时数据采集的“标配”
• Flume：专门用于日志数据的采集和聚合
• Canal：阿里巴巴开源的MySQL binlog解析工具，可以实时捕获数据库变更

我个人的经验是：能用Kafka就别用其他。Kafka的生态太成熟了，从采集到处理到存储，一条龙服务。而且它的“重放”机制特别有用——万一数据消费出错了，可以从某个时间点重新消费，不用重新采集。

3. API采集——对接第三方数据

第三方数据通常通过API接口提供。比如调用工商数据的API查询企业信息，或者调用黑名单API校验客户身份。API采集的关键是：限流、重试、缓存。

# Python调用第三方API示例
import requests
import time

def fetch_blacklist(name, id_number):
    url = "https://api.blacklist.com/v1/check"
    payload = {
        "name": name,
        "id_number": id_number
    }
    headers = {
        "Authorization": "Bearer YOUR_API_KEY"
    }
    
    # 重试机制：最多重试3次
    for i in range(3):
        try:
            response = requests.post(url, json=payload, headers=headers, timeout=5)
            if response.status_code == 200:
                return response.json()
            else:
                time.sleep(2 ** i)  # 指数退避
        except Exception as e:
            print(f"请求失败，第{i+1}次重试: {e}")
            time.sleep(2 ** i)
    
    return None

1.3 数据质量初筛——别让脏数据毁了你的分析

数据采集进来之后，别急着分析。先做一道“体检”，把明显有问题的数据筛出来。我总结了一个“四步筛查法”：

第一步：完整性检查

检查必填字段是否有空值。比如交易数据里的交易金额、交易时间、交易对手，这些字段如果为空，那这条数据基本就是废的。

-- SQL示例：检查交易数据的完整性
SELECT 
  COUNT(*) AS total_records,
  SUM(CASE WHEN amount IS NULL THEN 1 ELSE 0 END) AS missing_amount,
  SUM(CASE WHEN trans_time IS NULL THEN 1 ELSE 0 END) AS missing_time,
  SUM(CASE WHEN counterparty IS NULL THEN 1 ELSE 0 END) AS missing_counterparty
FROM transaction_log
WHERE dt = '2024-01-01';

第二步：准确性检查

检查数据是否符合业务规则。比如：

• 交易金额不能为负数（除非是退款）
• 客户年龄不能超过150岁
• 身份证号必须符合校验规则

我曾经遇到过一个奇葩数据：某客户的出生日期是“1900-01-01”，年龄124岁。这明显是系统默认值没改过来。这种数据如果不筛掉，后面分析出来的“老年客户交易特征”全是错的。

第三步：一致性检查

检查不同数据源之间的数据是否一致。比如：

• 交易数据里的客户ID，在客户数据表里是否存在
• 同一笔交易，在核心系统和风控系统里的金额是否一致

我建议用交叉验证的方式来做一致性检查。比如，把交易数据按客户ID分组，然后和客户数据表做关联，找出那些“有交易无客户”的孤儿数据。

第四步：时效性检查

检查数据是否在合理的时间范围内。比如：

• 交易时间不能是未来时间
• 数据采集时间不能比交易时间还早（这明显是系统时钟出了问题）

知识体系总览

下面这张图，是我梳理的本章知识体系。你可以把它当作一个“地图”，随时回来对照。

本章小结

好了，咱们来捋一捋今天的内容：

• 数据来源：交易数据是主食，客户数据是灵魂，第三方数据是外援。三者缺一不可。
• 采集方法：批量采集用Sqoop/DataX，实时采集用Kafka，API采集注意限流和缓存。
• 质量初筛：完整性、准确性、一致性、时效性，四步筛查法帮你过滤掉80%的脏数据。

记住一句话：数据采集阶段多花一小时，分析阶段就能省下一天。别嫌麻烦，这是合规数据分析的“基本功”。

---