4. 数据加载：使用 pandas 读取 CSV / Excel 格式的保单与理赔数据

数据加载是赔付率监控与尖峰识别的基础环节。在实际生产环境中，保单数据与理赔数据通常以 CSV 或 Excel 格式存储在数据仓库或本地文件系统中。本节将演示如何使用 pandas 库高效、稳定地加载这两种格式的数据，并处理常见的加载异常。

4.1 数据源结构概览

在开始编码之前，需要明确两个核心数据集的字段结构。下表列出了典型的保单主表与理赔明细表的关键字段：

4.2 使用 pandas 读取 CSV 文件

CSV 是精算分析中最常见的文本格式。使用 pd.read_csv() 时，需要关注编码、分隔符、日期解析和数据类型优化。以下代码演示了标准加载流程：

import pandas as pd

# 读取保单数据
df_policy = pd.read_csv(
    'policy_data.csv',
    encoding='utf-8',
    sep=',',
    parse_dates=['underwrite_date'],
    dtype={
        'policy_id': 'string',
        'premium': 'float64',
        'product_line': 'category'
    }
)

# 读取理赔数据
df_claim = pd.read_csv(
    'claim_data.csv',
    encoding='utf-8',
    sep=',',
    parse_dates=['claim_date'],
    dtype={
        'claim_id': 'string',
        'policy_id': 'string',
        'claim_amount': 'float64'
    }
)

print(f"保单数据加载完成：{df_policy.shape[0]} 条记录")
print(f"理赔数据加载完成：{df_claim.shape[0]} 条记录")

4.3 使用 pandas 读取 Excel 文件

Excel 文件在业务部门中广泛使用，但加载时需注意多工作表、合并单元格和空行问题。推荐使用 pd.read_excel() 配合 sheet_name 参数：

# 读取 Excel 中的指定工作表
df_policy_excel = pd.read_excel(
    'policy_data.xlsx',
    sheet_name='保单信息',
    header=0,
    usecols='A:F',
    dtype={
        'policy_id': 'string',
        'premium': 'float64'
    }
)

# 读取理赔工作表，跳过前 2 行说明行
df_claim_excel = pd.read_excel(
    'claim_data.xlsx',
    sheet_name='理赔明细',
    skiprows=2,
    parse_dates=['claim_date']
)

print(f"Excel 保单数据加载完成：{df_policy_excel.shape[0]} 条记录")
print(f"Excel 理赔数据加载完成：{df_claim_excel.shape[0]} 条记录")

4.4 数据加载后的快速验证

加载完成后，必须执行一组快速验证，确保数据完整性。以下为推荐的验证清单：

• 记录数核对： 使用 df.shape 确认行数与源文件一致。
• 缺失值检查： 使用 df.isnull().sum() 定位关键字段的缺失情况。
• 日期范围确认： 使用 df['claim_date'].min() 和 .max() 检查时间跨度是否合理。
• 唯一性校验： 使用 df['policy_id'].nunique() 检查保单主键是否重复。

# 快速验证示例
print("缺失值统计：")
print(df_policy.isnull().sum())

print("\n理赔日期范围：")
print(f"最早：{df_claim['claim_date'].min()}")
print(f"最晚：{df_claim['claim_date'].max()}")

print("\n保单唯一性：")
print(f"唯一保单数：{df_policy['policy_id'].nunique()}")
print(f"总记录数：{df_policy.shape[0]}")

4.5 数据加载流程架构

下图展示了从原始文件到内存 DataFrame 的完整加载流程，包含异常处理与验证环节：

4.6 常见加载异常与处理策略

在实际项目中，数据加载阶段常遇到以下三类异常。建议在代码中统一捕获并记录日志：

import logging

logging.basicConfig(level=logging.INFO)

def safe_read_csv(filepath, **kwargs):
    """带异常处理的 CSV 加载函数"""
    try:
        df = pd.read_csv(filepath, **kwargs)
        logging.info(f"成功加载 {filepath}，共 {df.shape[0]} 行")
        return df
    except UnicodeDecodeError:
        logging.warning(f"UTF-8 解码失败，尝试 GBK 编码: {filepath}")
        kwargs['encoding'] = 'gbk'
        return pd.read_csv(filepath, **kwargs)
    except Exception as e:
        logging.error(f"加载失败: {filepath}，错误: {str(e)}")
        raise