4. 数据加载与初步探索：使用Pandas读取CSV/Excel/数据库数据、info()/describe()/head()/shape方法、数据类型检查

在风控数据质量治理中，数据加载是异常值检测的第一步。只有正确、高效地将原始数据读入内存，才能进行后续的探索与清洗。本节将系统讲解如何使用Pandas加载常见数据源，并利用核心方法快速掌握数据全貌。

4.1 数据加载：从文件到DataFrame

Pandas支持多种数据源读取，风控场景中最常用的是CSV、Excel和关系型数据库。下表对比了三种方式的适用场景与核心参数：

import pandas as pd
from sqlalchemy import create_engine

# 1. 读取CSV（指定分隔符、编码、日期列）
df_csv = pd.read_csv('transaction_2024.csv', sep=',', encoding='utf-8',
                     parse_dates=['trans_time'], dtype={'user_id': str})

# 2. 读取Excel（指定sheet和列范围）
df_excel = pd.read_excel('rule_config.xlsx', sheet_name='anti_fraud',
                         usecols='A:F')

# 3. 读取数据库（使用SQLAlchemy连接）
engine = create_engine('mysql+pymysql://user:pass@host:3306/risk_db')
df_sql = pd.read_sql('SELECT * FROM feature_store WHERE dt = "2024-01-01"',
                     con=engine)

4.2 初步探索：四步掌握数据全貌

数据加载完成后，必须通过系统化的探索方法快速建立对数据集的认知。以下四个方法构成了风控数据初步探索的标准流程：

1. df.shape — 获取行数和列数，判断数据规模是否在预期范围内。
2. df.info() — 查看每列的非空计数、数据类型和内存占用，快速定位缺失字段和类型异常。
3. df.head() — 预览前5行数据，直观检查字段内容是否符合业务预期。
4. df.describe() — 生成数值型字段的统计摘要（均值、标准差、分位数），初步识别离群值。

# 加载示例风控数据集
df = pd.read_csv('risk_sample.csv')

# 1. 数据规模
print(f"数据集维度: {df.shape}")  # 输出: (100000, 25)

# 2. 数据类型与缺失概览
df.info()

# 3. 预览前5行
print(df.head())

# 4. 数值字段统计摘要
print(df.describe())

4.3 数据类型检查：风控字段的“身份验证”

数据类型错误是风控数据质量中最隐蔽的问题之一。例如，用户ID被读成浮点数、交易时间被读成字符串、金额字段出现非数字字符等。Pandas提供了两种核心检查方式：

# 数据类型检查实战
print("各字段类型：")
print(df.dtypes)

# 检查是否存在异常类型
numeric_cols = df.select_dtypes(include=['int64', 'float64']).columns.tolist()
object_cols = df.select_dtypes(include=['object']).columns.tolist()

print(f"\n数值列（共{len(numeric_cols)}个）：{numeric_cols}")
print(f"文本列（共{len(object_cols)}个）：{object_cols}")

# 强制类型转换（风控常用）
df['user_id'] = df['user_id'].astype(str)
df['trans_time'] = pd.to_datetime(df['trans_time'], errors='coerce')

4.4 综合实践：风控数据集加载与探索模板

以下代码整合了本节所有核心方法，形成风控数据加载与初步探索的标准模板，可直接用于实际项目：

def load_and_explore_risk_data(file_path, date_col='trans_time'):
    """
    风控数据集加载与初步探索模板
    参数:
        file_path: 数据文件路径（CSV格式）
        date_col: 日期列名，用于自动解析
    返回:
        df: 加载后的DataFrame
        report: 数据质量初步报告（字典）
    """
    # 1. 加载数据（显式指定类型）
    df = pd.read_csv(
        file_path,
        encoding='utf-8',
        parse_dates=[date_col] if date_col else False,
        dtype={'user_id': str, 'order_id': str},
        low_memory=False
    )

    # 2. 初步探索
    report = {
        'shape': df.shape,
        'columns': list(df.columns),
        'dtypes': df.dtypes.to_dict(),
        'missing_count': df.isnull().sum().to_dict(),
        'missing_pct': (df.isnull().sum() / len(df) * 100).round(2).to_dict(),
        'numeric_summary': df.describe().to_dict()
    }

    # 3. 类型检查与修正
    for col in df.select_dtypes(include=['object']).columns:
        if 'time' in col.lower() or 'date' in col.lower():
            df[col] = pd.to_datetime(df[col], errors='coerce')

    # 4. 输出关键信息
    print(f"✅ 数据加载完成：{report['shape'][0]}行 × {report['shape'][1]}列")
    print(f"⚠️  缺失字段数：{sum(1 for v in report['missing_count'].values() if v > 0)}")

    return df, report

# 使用示例
df_risk, report = load_and_explore_risk_data('risk_transactions.csv')
print(report['numeric_summary'])