📚 课程目录第3章坏样本数据采集

3. 坏样本数据采集：数据源分类、采集频率与数据质量初筛

坏样本归因分析的第一步，也是最容易被低估的一步，是数据采集。如果采集阶段就混入了系统性偏差或低质量数据，后续所有的特征工程和归因结论都会建立在沙丘之上。本章聚焦数据源的三大分类、采集频率与时效性的权衡策略，以及数据质量初筛的核心方法。

3.1 数据源分类：三大来源与特征对比

在工业级归因系统中，坏样本的数据来源通常分为三类：内部日志、外部三方数据和埋点数据。三类数据在采集方式、覆盖范围、可信度和延迟上存在显著差异。

数据源类型	典型示例	采集方式	覆盖范围	可信度	采集延迟
内部日志	服务器访问日志、数据库慢查询日志、应用错误日志	服务端自动记录	全量用户行为	高（内部可控）	秒级～分钟级
外部三方数据	设备指纹、IP 信誉库、黑名单、风控评分	API 调用 / 文件同步	仅覆盖调用方	中（依赖供应商）	分钟级～天级
埋点数据	页面点击流、曝光事件、自定义事件	客户端 SDK 上报	前端用户交互	中（易受网络/设备影响）	秒级～小时级

实践建议：坏样本归因中，内部日志通常作为“黄金基线”用于交叉验证；埋点数据对用户行为还原最细，但需注意上报丢失和延迟；外部三方数据适合补充设备级和网络级特征，但需关注数据新鲜度。

3.2 数据源采集架构总览

下图展示了三类数据源在典型归因系统中的采集路径与汇聚关系。内部日志通过 Flume/Filebeat 采集，埋点数据通过 SDK → Kafka 管道，外部三方数据通过定时任务或 API 网关接入，最终统一进入数据湖或实时数仓。

3.3 采集频率与时效性

不同数据源对采集频率和时效性的要求差异巨大。错误的选择会导致特征滞后或系统过载。下表总结了常见场景下的推荐配置：

数据源	推荐采集频率	时效性要求	典型延迟	适用场景
服务器错误日志	实时（流式）	秒级	< 5 秒	实时风控、故障检测
用户点击埋点	实时（流式）	秒级～分钟级	< 30 秒	用户行为归因
设备指纹	按需 + 缓存	分钟级	< 5 分钟	设备维度特征
IP 信誉库	每日全量 + 增量	天级	1 天	网络环境评估
黑名单 / 灰名单	每日全量	天级	1 天	离线批量归因

常见陷阱：不要对所有数据源统一使用“实时采集”。外部三方数据（如 IP 库）的更新频率通常为天级，实时调用只会增加延迟和成本，不会提升特征质量。建议按数据源的自然更新周期设置采集频率。

3.4 数据质量初筛：四维检查清单

在数据进入特征工程管线之前，必须执行一次快速的质量初筛。初筛不是深度清洗，而是快速过滤明显不可用的数据。推荐从以下四个维度展开：

完整性检查：关键字段（如用户 ID、时间戳、事件类型）是否存在大量空值？空值率超过 30% 的字段应标记为“低质量”。
一致性检查：同一实体的字段间是否存在逻辑矛盾？例如“注册时间 > 当前时间”或“订单金额为负数”。
时效性检查：数据时间戳是否在合理范围内？是否存在大量“未来时间戳”或“过期超过 7 天”的数据？
唯一性检查：主键或业务键是否存在重复？重复率超过 5% 的数据源需要排查采集逻辑。

3.5 数据质量初筛流程

下图展示了从原始数据到可入特征工程管线的完整初筛流程。每个检查环节都对应一个明确的“通过/拒绝”决策点。

3.6 初筛阶段的代码示例

以下 Python 代码展示了如何对采集到的日志数据执行四维初筛。实际生产环境中，建议将检查逻辑封装为独立的 Data Quality Checker 模块。

import pandas as pd
import numpy as np
from datetime import datetime, timedelta

def quality_initial_screen(df: pd.DataFrame, 
                           key_col: str = 'user_id',
                           time_col: str = 'event_time',
                           null_threshold: float = 0.3,
                           dup_threshold: float = 0.05,
                           max_time_lag_days: int = 7) -> dict:
    """
    数据质量初筛：返回各维度检查结果与整体判定。
    """
    result = {}
    total = len(df)
    
    # 1. 完整性检查
    null_ratio = df[key_col].isnull().mean()
    result['null_ratio'] = round(null_ratio, 4)
    result['null_pass'] = null_ratio < null_threshold
    
    # 2. 一致性检查：时间戳不应为未来
    now = datetime.now()
    future_count = (df[time_col] > now).sum()
    result['future_time_count'] = future_count
    result['consistency_pass'] = future_count == 0
    
    # 3. 时效性检查：数据不应过于陈旧
    cutoff = now - timedelta(days=max_time_lag_days)
    stale_count = (df[time_col] < cutoff).sum()
    result['stale_count'] = stale_count
    result['freshness_pass'] = stale_count / total < 0.1
    
    # 4. 唯一性检查
    dup_ratio = df[key_col].duplicated().mean()
    result['dup_ratio'] = round(dup_ratio, 4)
    result['dup_pass'] = dup_ratio < dup_threshold
    
    # 综合判定
    result['overall_pass'] = all([
        result['null_pass'],
        result['consistency_pass'],
        result['freshness_pass'],
        result['dup_pass']
    ])
    
    return result

# 使用示例
sample_df = pd.DataFrame({
    'user_id': ['u1', 'u2', None, 'u4', 'u1'],
    'event_time': [
        datetime(2025, 3, 1, 10, 0),
        datetime(2025, 3, 2, 12, 0),
        datetime(2025, 3, 1, 14, 0),
        datetime(2025, 3, 3, 8, 0),
        datetime(2025, 3, 1, 10, 0)
    ]
})
report = quality_initial_screen(sample_df)
print(report)

  输出解读：上述示例中，由于存在空值（user_id 缺失）和重复键（u1 出现两次），overall_pass 将返回 False。此时该批次数据应被标记为“待人工审核”或“退回采集层修复”，而非直接进入特征工程。

3.7 本章小结

数据源分类：内部日志（高可信）、埋点数据（细粒度）、外部三方（补充维度），三者需协同使用。
采集频率：按数据源的自然更新周期设置，避免“一刀切”实时采集导致资源浪费。
质量初筛：完整性、一致性、时效性、唯一性四维检查，是坏样本归因的第一道防线。
工程落地：初筛逻辑应自动化、可配置，并输出明确的通过/拒绝信号。