4. 数据准备与清洗：保单数据、理赔数据、核保数据、外部数据的整合与清洗规则

赔付率坏样本归因的起点是数据。原始数据通常分散在多个业务系统，且存在缺失、异常、口径不一致等问题。本章将系统讲解如何将保单、理赔、核保及外部数据进行整合，并建立标准化的清洗规则，为后续的赔付率计算与坏样本识别奠定可靠基础。

4.1 数据源全景与核心字段

赔付率分析所需的数据主要来自四个内部系统及一个外部数据源。下表列出了各数据源的核心字段及其在归因分析中的用途：

4.2 数据整合流程与架构

数据整合不是简单的拼接，而是需要经过抽取、清洗、关联、聚合四个阶段。下图展示了从原始数据到分析宽表的完整架构：

4.3 清洗规则详解

清洗规则是数据质量的保障。根据赔付率分析的特殊要求，我们将清洗规则分为以下五类：

4.3.1 缺失值处理规则

• 关键字段不可缺失：保单号、起保日期、险种代码、赔款金额。若缺失，直接剔除该记录并记录日志。
• 可插补字段：NCD系数（按行业均值插补）、车龄（按车型平均车龄插补）、地区代码（按邮编映射）。
• 标记字段：核保人ID、特约条款等，缺失时标记为“未知”，不参与归因分析。

4.3.2 异常值截断规则

• 赔款金额：单次赔款超过保额120%的，视为异常，需人工复核；超过保额200%的，直接剔除。
• 保费：零保费或负保费记录，直接剔除（批单除外）。
• 日期逻辑：出险日期早于起保日期或晚于终保日期+90天的，标记为异常，需人工确认。

4.3.3 格式统一规则

• 日期格式：统一为YYYY-MM-DD。
• 金额单位：统一为元，保留两位小数。
• 险种代码：统一为6位数字编码，不足前补零。
• 地区代码：统一为6位行政区划代码。

4.3.4 重复记录去重规则

• 保单数据：按保单号+险种代码去重，保留最新批改版本。
• 理赔数据：按报案号去重，同一报案号多条记录时，取结案日期最新的记录。
• 核保数据：按保单号去重，保留最后一次核保决策。

4.3.5 日期对齐与暴露期计算规则

• 已赚保费计算：按日计算，起保日到评估日或终保日的实际天数。
• 理赔归属：按出险日期归属到对应的保单年度，而非结案日期。
• 延迟理赔处理：对于未决赔款，使用链梯法或Bornhuetter-Ferguson法进行IBNR预估。

4.4 数据整合后的质量校验

整合完成后，必须进行以下质量校验，确保数据可用于赔付率分析：

1. 记录数校验：保单数据与核保数据关联后，记录数偏差不应超过0.5%。
2. 金额汇总校验：理赔数据中的赔款总额，应与财务系统中的已决赔款总额一致，偏差不超过1%。
3. 赔付率合理性校验：计算出的整体赔付率应与业务感知或历史趋势一致，偏差超过5个百分点需排查原因。
4. 缺失率报告：生成各字段缺失率报告，缺失率超过10%的字段需评估是否纳入分析。

下图展示了数据清洗前后赔付率分布的变化，直观体现清洗对数据质量的提升效果：

4.5 清洗规则代码示例

以下Python代码展示了清洗规则的核心实现逻辑，使用Pandas库进行数据清洗：

import pandas as pd
import numpy as np

def clean_claim_data(df_claim):
    """
    理赔数据清洗函数
    """
    # 1. 缺失值处理：关键字段不可缺失
    df_claim = df_claim.dropna(subset=['claim_id', 'policy_id', 'accident_date', 'claim_amount'])
    
    # 2. 异常值截断：赔款金额超过保额120%需标记
    df_claim['claim_amount_ratio'] = df_claim['claim_amount'] / df_claim['sum_insured']
    df_claim['claim_flag'] = np.where(df_claim['claim_amount_ratio'] > 1.2, '异常', '正常')
    
    # 3. 日期逻辑校验：出险日期必须在保单有效期内
    df_claim = df_claim.merge(df_policy[['policy_id', 'start_date', 'end_date']], on='policy_id', how='left')
    df_claim = df_claim[(df_claim['accident_date'] >= df_claim['start_date']) & 
                        (df_claim['accident_date'] <= df_claim['end_date'] + pd.Timedelta(days=90))]
    
    # 4. 重复记录去重：按报案号去重，保留结案日期最新的
    df_claim = df_claim.sort_values('close_date', ascending=False).drop_duplicates(subset='claim_id')
    
    # 5. 格式统一：日期格式统一
    df_claim['accident_date'] = pd.to_datetime(df_claim['accident_date']).dt.strftime('%Y-%m-%d')
    
    return df_claim

def build_analysis_wide_table(df_policy, df_claim, df_underwrite, df_external):
    """
    构建分析宽表
    """
    # 保单维度聚合
    df_policy_agg = df_policy.groupby('policy_id').agg({
        'premium': 'sum',
        'start_date': 'min',
        'end_date': 'max'
    }).reset_index()
    
    # 理赔维度聚合
    df_claim_agg = df_claim.groupby('policy_id').agg({
        'claim_amount': 'sum',
        'claim_id': 'count'
    }).rename(columns={'claim_id': 'claim_count'}).reset_index()
    
    # 多表关联
    df_wide = df_policy_agg.merge(df_claim_agg, on='policy_id', how='left')
    df_wide = df_wide.merge(df_underwrite, on='policy_id', how='left')
    df_wide = df_wide.merge(df_external, on='policy_id', how='left')
    
    # 填充缺失值
    df_wide['claim_amount'] = df_wide['claim_amount'].fillna(0)
    df_wide['claim_count'] = df_wide['claim_count'].fillna(0)
    
    # 计算已赚保费和赔付率
    df_wide['earned_premium'] = df_wide['premium'] * (评估天数 / 保险期限)
    df_wide['loss_ratio'] = df_wide['claim_amount'] / df_wide['earned_premium']
    
    return df_wide

4.6 数据清洗质量报告模板

每次数据清洗后，应生成标准化的质量报告，记录清洗前后的数据变化。以下为报告的核心指标：

通过系统化的数据准备与清洗，我们能够获得高质量的分析数据，为后续的赔付率驱动坏样本归因提供坚实的数据基础。下一阶段将基于清洗后的数据，进行赔付率的计算与分解。