3. 数据采集与清洗：催收系统数据源与清洗规则

催收触达异常波动的根因定位，高度依赖底层数据的完整性与准确性。本章聚焦催收系统三大核心数据源（CTI、IVR、催收系统日志）的采集要点，以及数据清洗的三大规则（去重、异常值处理、时间窗口对齐），确保看板指标可追溯、可归因。

3.1 催收系统三大数据源

催收触达数据分散在多个子系统，采集时需明确各数据源的定义、字段及采集粒度。

3.2 数据清洗三大规则

原始数据存在重复、异常、时间不同步等问题，必须经过清洗才能进入看板计算。以下为催收场景下的清洗规则。

3.2.1 去重规则

• CTI去重：以“坐席工号 + 外呼号码 + 通话开始时间（精确到秒）”为唯一键，保留最早一条记录。
• IVR去重：以“客户号码 + IVR流程ID + 进入时间（精确到秒）”为唯一键，剔除重复按键轨迹。
• 催收日志去重：以“案件ID + 操作类型 + 操作时间（精确到秒）”为唯一键，合并相同操作。

3.2.2 异常值处理

• 通话时长异常：通话时长 < 0秒 或 > 3600秒（1小时）视为异常，标记为“无效通话”，不参与接通率计算。
• IVR按键异常：按键序列包含非数字字符或长度超过20位，视为“按键异常”，剔除该次交互。
• 时间戳异常：操作时间早于案件创建时间或晚于当前时间+5分钟，视为“时间戳漂移”，需回退至最近一次有效时间戳。

3.2.3 时间窗口对齐

CTI、IVR、催收日志的时间戳来自不同系统，存在毫秒级偏差。对齐规则如下：

1. 统一基准：以催收系统日志的时间戳为基准（通常为数据库写入时间）。
2. CTI对齐：CTI通话开始时间与催收日志中的“外呼操作时间”匹配，允许偏差 ≤ 2秒。
3. IVR对齐：IVR进入时间与催收日志中的“IVR触发时间”匹配，允许偏差 ≤ 1秒。
4. 未匹配处理：超出偏差范围的记录标记为“时间窗口未对齐”，单独存储供人工核查。

3.3 数据采集与清洗流程架构

以下SVG图展示从数据源采集到清洗输出的完整链路，包含三大数据源、清洗规则模块及最终事实表。

3.4 清洗规则代码示例（Python伪代码）

以下为催收数据清洗的核心逻辑，包含去重、异常值过滤及时间窗口对齐。

import pandas as pd
from datetime import timedelta

# 1. 加载原始数据
cti_df = pd.read_csv('cti_records.csv')
ivr_df = pd.read_csv('ivr_records.csv')
log_df = pd.read_csv('collection_log.csv')

# 2. 去重规则
cti_df.drop_duplicates(subset=['agent_id', 'customer_phone', 'call_start_time'], keep='first', inplace=True)
ivr_df.drop_duplicates(subset=['customer_phone', 'ivr_flow_id', 'enter_time'], keep='first', inplace=True)
log_df.drop_duplicates(subset=['case_id', 'operation_type', 'operation_time'], keep='first', inplace=True)

# 3. 异常值处理
cti_df = cti_df[(cti_df['call_duration'] >= 0) & (cti_df['call_duration'] <= 3600)]
ivr_df = ivr_df[ivr_df['key_sequence'].str.match(r'^[0-9]{1,20}$')]
log_df = log_df[(log_df['operation_time'] >= log_df['case_create_time']) &
                (log_df['operation_time'] <= pd.Timestamp.now() + timedelta(minutes=5))]

# 4. 时间窗口对齐（以催收日志为基准）
merged = pd.merge_asof(
    log_df.sort_values('operation_time'),
    cti_df.sort_values('call_start_time'),
    left_on='operation_time',
    right_on='call_start_time',
    direction='nearest',
    tolerance=pd.Timedelta(seconds=2)
)
merged['time_window_aligned'] = merged['call_start_time'].notna()

3.5 数据质量监控指标

清洗完成后，需建立数据质量监控看板，确保后续分析可信。