3. 数据清洗与预处理：缺失值处理、异常值检测、重复数据去重、数据类型统一

在交易风控场景中，原始数据通常来自多个异构源（交易流水、用户画像、设备指纹、外部黑名单等），存在大量脏数据。数据清洗与预处理是特征工程的基石，直接影响后续模型训练的收敛速度与泛化能力。本节将系统讲解四大核心操作：缺失值处理、异常值检测、重复数据去重、数据类型统一。

3.1 缺失值处理

缺失值在风控数据中极为常见，原因包括：用户未填写、系统采集失败、数据同步延迟、字段不适用等。处理策略需结合缺失率、字段含义与业务逻辑。

import pandas as pd
import numpy as np

# 示例：交易流水数据缺失值处理
df = pd.DataFrame({
    'amount': [1000, 2000, np.nan, 1500, 3000],
    'age': [25, np.nan, 30, 28, np.nan],
    'province': ['北京', '上海', np.nan, '广州', '深圳']
})

# 1. 连续变量：中位数填充
df['amount'].fillna(df['amount'].median(), inplace=True)

# 2. 离散变量：众数填充
df['province'].fillna(df['province'].mode()[0], inplace=True)

# 3. 构造缺失指示特征
df['age_missing'] = df['age'].isna().astype(int)
df['age'].fillna(df['age'].median(), inplace=True)

print(df)

3.2 异常值检测

交易风控中的异常值通常对应欺诈交易、数据录入错误或极端用户行为。常用检测方法包括统计法、距离法与模型法。

• Z-Score 法： 适用于近似正态分布，阈值通常取 3 或 3.5。
• IQR 四分位距法： 适用于偏态分布，定义下界 Q1 - 1.5×IQR，上界 Q3 + 1.5×IQR。
• 孤立森林： 适用于高维数据，无分布假设，风控场景常用。
• 业务规则法： 如单笔交易金额 > 日均交易额 10 倍，直接标记为异常。

# IQR 法检测异常值
def detect_outliers_iqr(series):
    Q1 = series.quantile(0.25)
    Q3 = series.quantile(0.75)
    IQR = Q3 - Q1
    lower_bound = Q1 - 1.5 * IQR
    upper_bound = Q3 + 1.5 * IQR
    return (series < lower_bound) | (series > upper_bound)

df['amount_outlier'] = detect_outliers_iqr(df['amount'])
print(df[df['amount_outlier']])

3.3 重复数据去重

重复数据可能来自数据源重复采集、系统重试机制或恶意刷单。去重策略需区分完全重复与部分重复（如同一用户短时间内多次提交）。

• 完全重复去重： 所有字段完全一致，直接保留一条。
• 基于主键去重： 按交易ID、用户ID+时间戳等唯一标识去重。
• 模糊去重： 使用编辑距离或相似度阈值，适用于文本型字段（如地址、设备ID）。

# 完全重复去重
df.drop_duplicates(inplace=True)

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

# 模糊去重示例（使用 Levenshtein 距离）
from rapidfuzz import fuzz

def fuzzy_dedup(series, threshold=90):
    unique = []
    for item in series:
        if not any(fuzz.ratio(item, u) >= threshold for u in unique):
            unique.append(item)
    return unique

3.4 数据类型统一

数据类型不一致会导致计算错误、内存浪费或模型报错。常见问题包括：数值型存为字符串、时间戳格式不统一、分类变量编码混乱。

# 数据类型统一示例
df['amount'] = pd.to_numeric(df['amount'].str.replace(',', ''), errors='coerce')
df['transaction_time'] = pd.to_datetime(df['transaction_time'], format='mixed')
df['province'] = df['province'].str.strip().str.lower().astype('category')
df['is_fraud'] = df['is_fraud'].map({'Y': 1, 'N': 0}).astype(int)

3.5 实战案例：交易数据清洗完整流程

以下代码整合了本节所有知识点，展示一个完整的交易风控数据清洗函数。

def clean_transaction_data(df):
    """
    交易风控数据清洗完整流水线
    """
    # 1. 数据类型统一
    df['amount'] = pd.to_numeric(df['amount'], errors='coerce')
    df['timestamp'] = pd.to_datetime(df['timestamp'], format='mixed')
    df['user_id'] = df['user_id'].astype(str).str.strip()

    # 2. 缺失值处理
    df['amount'].fillna(df['amount'].median(), inplace=True)
    df['amount_missing'] = df['amount'].isna().astype(int)

    # 3. 异常值检测（IQR）
    Q1 = df['amount'].quantile(0.25)
    Q3 = df['amount'].quantile(0.75)
    IQR = Q3 - Q1
    lower = Q1 - 1.5 * IQR
    upper = Q3 + 1.5 * IQR
    df['amount_outlier'] = ((df['amount'] < lower) | (df['amount'] > upper)).astype(int)

    # 4. 重复数据去重
    df.drop_duplicates(subset=['user_id', 'timestamp'], keep='first', inplace=True)

    # 5. 重置索引
    df.reset_index(drop=True, inplace=True)

    return df

# 使用示例
df_clean = clean_transaction_data(df)
print(f"清洗后数据量: {len(df_clean)} 条")
print(f"异常交易占比: {df_clean['amount_outlier'].mean():.2%}")