2. 数据源与数据采集：催收系统数据源与外部数据源

催收触达特征工程的第一步是明确数据从哪里来、怎么采、采什么。数据源的质量直接决定后续特征筛选的上限。本章将系统梳理催收场景下的两类核心数据源——催收系统内部数据与外部数据源，并给出数据采集的规范与注意事项。

2.1 催收系统数据源

催收系统内部数据是触达特征最直接、最高频的来源，主要包括通话记录、短信记录和 IVR 记录三大类。

2.2 外部数据源

外部数据源用于补充内部数据无法覆盖的风险画像和关联信息，是提升催收模型泛化能力的关键。

• 征信数据：个人征信报告中的逾期历史、负债率、查询次数、信贷产品类型等。注意征信数据有严格的查询授权和时效要求，通常按 T+1 或 T+7 更新。
• 黑名单数据：包括行业共享黑名单、法院失信被执行人名单、反欺诈名单等。黑名单匹配需支持模糊匹配（姓名+身份证号+手机号多维度交叉）。
• 关联人数据：通过外部数据源获取的紧急联系人、通讯录关联人、同住地址关联人等，用于失联修复和触达网络构建。
• 设备与行为数据：设备指纹、IP 地址、GPS 定位、APP 活跃度等，用于识别多头借贷、设备关联风险。

2.3 数据采集规范与注意事项

数据采集不是简单的“把数据拿过来”，而是需要建立一套标准化的采集规范，确保数据的一致性、完整性和可回溯性。

2.3.1 采集规范清单

2.3.2 常见采集陷阱与避坑

• 时间窗口不一致：通话记录按事件时间采集，征信数据按日更新，两者在特征拼接时容易出现时间错位。建议统一以“催收作业日期”为时间锚点。
• 重复采集与数据膨胀：IVR 记录中同一通电话可能产生多条子记录（按键、转接、挂断），需在 ETL 阶段做聚合，避免特征维度爆炸。
• 外部数据延迟：黑名单数据通常有 1~3 天延迟，征信数据延迟更长。在特征工程中必须加入“数据新鲜度”字段，避免使用过时数据做实时决策。
• 采集失败处理：外部接口调用可能超时或返回空值，需设计重试机制和默认值策略，防止下游特征计算中断。

2.4 数据源架构与采集流程

下图展示了催收触达特征的数据源架构与采集流程，帮助理解数据从产生到进入特征工程的全链路。

2.5 数据采集代码示例（伪代码）

以下是一个简化的数据采集调度伪代码，展示如何从催收系统拉取通话记录并写入特征存储。

# 伪代码：催收通话记录采集调度
def collect_call_records(batch_date):
    """
    采集指定日期的通话记录
    :param batch_date: str, 格式 'YYYY-MM-DD'
    :return: DataFrame
    """
    # 1. 从催收系统 API 拉取数据
    raw_data = call_system_api.get_records(
        start_time=f"{batch_date} 00:00:00",
        end_time=f"{batch_date} 23:59:59",
        fields=["case_id", "agent_id", "contact_phone", 
                "call_start", "call_end", "call_result", "hangup_side"]
    )
    
    # 2. 字段重命名与格式统一
    df = raw_data.rename(columns={
        "call_start": "call_start_time",
        "call_end": "call_end_time"
    })
    df["call_start_time"] = pd.to_datetime(df["call_start_time"])
    df["call_end_time"] = pd.to_datetime(df["call_end_time"])
    
    # 3. 缺失值处理
    df["hangup_side"] = df["hangup_side"].fillna("UNKNOWN")
    df["call_result"] = df["call_result"].fillna("UNKNOWN")
    
    # 4. 敏感信息脱敏
    df["contact_phone"] = df["contact_phone"].apply(
        lambda x: x[:3] + "****" + x[7:] if len(x) == 11 else x
    )
    
    # 5. 添加批次标识
    df["batch_id"] = f"BATCH_{batch_date}"
    df["collect_timestamp"] = datetime.now()
    
    # 6. 写入特征存储
    feature_store.write(df, table="call_records", partition="batch_date")
    
    return df.shape[0]  # 返回采集记录数