4. 数据采集与ETL流程

在授信评分沙盘中，数据是模型的基础。本章聚焦于如何使用Python的pandas库完成从数据源采集、合并到清洗的全流程ETL（Extract, Transform, Load）。掌握这些技能，能确保后续评分模型建立在干净、一致的数据之上。

4.1 数据源读取

pandas支持从多种数据源直接读取数据，以下是授信评分场景中最常用的三种方式：

数据源类型	pandas函数	关键参数	授信评分典型场景
CSV文件	pd.read_csv()	filepath_or_buffer, encoding, sep	客户基本信息表、历史还款记录
Excel文件	pd.read_excel()	io, sheet_name, header	银行流水、财务报表、人工录入的审批表
关系型数据库	pd.read_sql()	sql, con (数据库连接)	从风控系统、核心业务系统抽取数据

import pandas as pd
from sqlalchemy import create_engine

# 1. 读取CSV
df_customer = pd.read_csv('customer_base.csv', encoding='utf-8')

# 2. 读取Excel（指定sheet）
df_transaction = pd.read_excel('transaction_2024.xlsx', sheet_name='Sheet1')

# 3. 读取数据库
engine = create_engine('postgresql://user:pass@localhost:5432/credit_db')
df_loan = pd.read_sql('SELECT * FROM loan_records WHERE status = \'active\'', con=engine)

4.2 数据合并

授信评分通常需要将多个来源的数据整合成一张宽表。pandas提供两种核心合并方式：

4.2.1 merge（类似SQL的JOIN）

用于按关键字段横向拼接数据，如将客户基本信息与贷款记录按客户ID合并。

# 按客户ID进行左连接
df_merged = pd.merge(df_customer, df_loan, on='customer_id', how='left')

how参数	行为	授信评分使用场景
inner	仅保留两表都匹配的记录	已有完整信息的客户评分
left	保留左表全部记录，右表无匹配则填充NaN	以客户主表为中心，补充外部数据
outer	保留两表所有记录	数据完整性校验、发现缺失关联

4.2.2 concat（纵向堆叠）

用于将结构相同的多张表按行拼接，如合并不同月份的还款记录。

# 合并1月和2月的交易数据
df_jan = pd.read_excel('trans_jan.xlsx')
df_feb = pd.read_excel('trans_feb.xlsx')
df_all = pd.concat([df_jan, df_feb], ignore_index=True)

4.3 数据清洗

原始数据通常存在缺失、异常、重复等问题，必须清洗后才能用于建模。以下是授信评分场景中的三大清洗任务：

4.3.1 缺失值处理

处理方法	pandas实现	适用场景
删除缺失行	df.dropna(subset=['关键字段'])	缺失比例小（<5%），且字段为建模必需
均值/中位数填充	df['col'].fillna(df['col'].median())	连续型变量（如收入、年龄）
众数填充	df['col'].fillna(df['col'].mode()[0])	分类变量（如学历、婚姻状况）
前向/后向填充	df.fillna(method='ffill')	时间序列数据（如月度还款记录）

4.3.2 异常值检测与处理

授信评分中，异常值可能代表欺诈或数据录入错误。常用方法：

• 业务规则法：年龄 < 18 或 > 100、收入为负数等直接判定为异常。
• 统计法（IQR）：超出 Q1-1.5*IQR 或 Q3+1.5*IQR 的值视为异常。
• 处理策略：对明显错误值直接删除或置为NaN后填充；对极端但合理的值（如高收入客户）进行截尾处理（winsorize）。

# IQR法检测异常值
Q1 = df['income'].quantile(0.25)
Q3 = df['income'].quantile(0.75)
IQR = Q3 - Q1
lower_bound = Q1 - 1.5 * IQR
upper_bound = Q3 + 1.5 * IQR

# 标记异常
df['income_outlier'] = (df['income'] < lower_bound) | (df['income'] > upper_bound)

4.3.3 重复值处理

重复数据会导致模型过拟合或统计偏差，必须识别并去重。

# 检查重复行
print(df.duplicated().sum())

# 基于指定列去重（保留第一条）
df_clean = df.drop_duplicates(subset=['customer_id', 'loan_date'], keep='first')

4.4 完整ETL流程示例

以下代码整合了从读取、合并到清洗的完整流程，可直接用于授信评分沙盘的数据准备阶段：

# ========== 完整ETL流程 ==========
import pandas as pd

# 1. 读取数据
df_base = pd.read_csv('customer_base.csv', encoding='utf-8')
df_loan = pd.read_sql('SELECT * FROM loan_records', con=engine)

# 2. 合并数据
df = pd.merge(df_base, df_loan, on='customer_id', how='left')

# 3. 清洗缺失值
df['income'].fillna(df['income'].median(), inplace=True)
df['education'].fillna(df['education'].mode()[0], inplace=True)

# 4. 清洗异常值（收入截尾）
upper_cap = df['income'].quantile(0.99)
df['income'] = df['income'].clip(upper=upper_cap)

# 5. 清洗重复值
df.drop_duplicates(subset='customer_id', keep='last', inplace=True)

# 6. 输出清洗后数据概况
print(f'清洗后数据量：{len(df)} 条')
print(f'缺失值统计：\n{df.isnull().sum()}')