4. Pandas 数据清洗：缺失值处理、重复值处理、数据类型转换、异常值过滤

数据清洗是数据分析流程中最耗时但最关键的一环。原始数据通常包含缺失值、重复记录、错误的数据类型以及异常数值，这些问题会直接影响 KPI 指标计算的准确性和监控大屏的可靠性。本章将系统讲解 Pandas 中四大核心清洗操作，确保数据进入分析管道前达到“干净、一致、可用”的状态。

4.1 缺失值处理

缺失值在 Pandas 中通常表示为 NaN（Not a Number）或 None。处理策略包括：删除缺失行/列、填充固定值或统计量、以及向前/向后填充。

import pandas as pd
import numpy as np

# 示例数据：包含缺失值的销售记录
df = pd.DataFrame({
    '日期': ['2025-01-01', '2025-01-02', '2025-01-03', '2025-01-04'],
    '销售额': [1200, np.nan, 1500, np.nan],
    '订单量': [30, 25, np.nan, 28],
    '客户评分': [4.5, 4.0, 3.8, np.nan]
})

# 1. 查看缺失概况
print("缺失数量：\n", df.isnull().sum())

# 2. 填充销售额缺失值为该列均值
df['销售额'] = df['销售额'].fillna(df['销售额'].mean())

# 3. 订单量缺失值用前向填充（假设时间序列）
df['订单量'] = df['订单量'].fillna(method='ffill')

# 4. 客户评分缺失值用固定值 4.0 填充
df['客户评分'] = df['客户评分'].fillna(4.0)

print("清洗后数据：\n", df)

4.2 重复值处理

重复值会导致统计指标（如总和、均值）被放大，尤其在订单记录、用户行为日志中常见。Pandas 提供基于全部列或指定子集的去重功能。

# 示例数据：包含重复行
df_dup = pd.DataFrame({
    '用户ID': [101, 102, 101, 103, 102],
    '购买金额': [200, 350, 200, 400, 350],
    '时间戳': ['2025-01-01', '2025-01-01', '2025-01-01', '2025-01-02', '2025-01-01']
})

# 1. 检测重复行
print("完全重复行数：", df_dup.duplicated().sum())

# 2. 查看重复行
print("重复行详情：\n", df_dup[df_dup.duplicated(keep=False)])

# 3. 删除完全重复行（保留第一次出现）
df_clean = df_dup.drop_duplicates()
print("去重后数据：\n", df_clean)

# 4. 基于指定列去重（例如：用户ID + 时间戳）
df_clean_subset = df_dup.drop_duplicates(subset=['用户ID', '时间戳'], keep='last')
print("基于用户ID和时间戳去重（保留最后一条）：\n", df_clean_subset)

4.3 数据类型转换

数据类型不匹配是数据清洗中的“隐形杀手”。常见问题包括：数值列被读为字符串、日期列未解析为 datetime 类型、布尔值存储为 0/1 整数。正确的数据类型能提升计算效率并避免隐式错误。

# 示例：常见数据类型问题与修复
df_types = pd.DataFrame({
    '日期': ['2025/01/01', '2025/01/02', '2025/01/03'],
    '销售额': ['1,200', '1,500', '1,800'],
    '是否达标': ['是', '否', '是'],
    '城市代码': [1, 2, 1]
})

# 查看原始类型
print("原始类型：\n", df_types.dtypes)

# 1. 日期转换
df_types['日期'] = pd.to_datetime(df_types['日期'], format='%Y/%m/%d')

# 2. 销售额：去除逗号并转为浮点数
df_types['销售额'] = pd.to_numeric(df_types['销售额'].str.replace(',', ''))

# 3. 是否达标：映射为布尔值
df_types['是否达标'] = df_types['是否达标'].map({'是': True, '否': False})

# 4. 城市代码：转为类别型
df_types['城市代码'] = df_types['城市代码'].astype('category')

print("转换后类型：\n", df_types.dtypes)
print("数据预览：\n", df_types)

4.4 异常值过滤

异常值是指明显偏离数据分布的值，可能由录入错误、传感器故障或特殊事件引起。常用检测方法包括：Z-Score 法、IQR（四分位距）法、业务规则过滤。

# 示例：使用 IQR 法过滤异常值
df_outlier = pd.DataFrame({
    '店铺': ['A', 'B', 'C', 'D', 'E'],
    '日销售额': [5000, 5200, 4800, 30000, 5100]  # D 店明显异常
})

# 计算四分位数
Q1 = df_outlier['日销售额'].quantile(0.25)
Q3 = df_outlier['日销售额'].quantile(0.75)
IQR = Q3 - Q1

lower_bound = Q1 - 1.5 * IQR
upper_bound = Q3 + 1.5 * IQR

print(f"正常范围：{lower_bound:.0f} ~ {upper_bound:.0f}")

# 过滤异常值
df_normal = df_outlier[
    (df_outlier['日销售额'] >= lower_bound) & 
    (df_outlier['日销售额'] <= upper_bound)
]

print("过滤后数据：\n", df_normal)

# 使用 Z-Score 方法（scipy 可选）
from scipy import stats
z_scores = stats.zscore(df_outlier['日销售额'])
df_z_clean = df_outlier[abs(z_scores) < 3]
print("Z-Score 过滤后：\n", df_z_clean)

4.5 综合清洗流程

在实际项目中，上述四个步骤通常按固定顺序执行。以下是一个完整的清洗管道示例，可直接用于 KPI 监控数据的预处理阶段。

def clean_kpi_data(df):
    """
    完整的 KPI 数据清洗函数
    步骤顺序：类型转换 → 缺失值处理 → 重复值处理 → 异常值过滤
    """
    df_clean = df.copy()
    
    # Step 1: 数据类型转换
    for col in df_clean.select_dtypes(include=['object']).columns:
        # 尝试转为数值
        df_clean[col] = pd.to_numeric(df_clean[col], errors='ignore')
    
    # Step 2: 缺失值处理（数值列用中位数填充，类别列用众数填充）
    for col in df_clean.select_dtypes(include=['float64', 'int64']).columns:
        df_clean[col] = df_clean[col].fillna(df_clean[col].median())
    for col in df_clean.select_dtypes(include=['object', 'category']).columns:
        df_clean[col] = df_clean[col].fillna(df_clean[col].mode()[0] if not df_clean[col].mode().empty else '未知')
    
    # Step 3: 重复值处理（基于全部列去重）
    df_clean = df_clean.drop_duplicates()
    
    # Step 4: 异常值过滤（IQR 法，仅对数值列）
    for col in df_clean.select_dtypes(include=['float64', 'int64']).columns:
        Q1 = df_clean[col].quantile(0.25)
        Q3 = df_clean[col].quantile(0.75)
        IQR = Q3 - Q1
        lower = Q1 - 1.5 * IQR
        upper = Q3 + 1.5 * IQR
        # 将异常值替换为边界值（也可选择标记或删除）
        df_clean[col] = df_clean[col].clip(lower, upper)
    
    return df_clean

# 使用示例
raw_data = pd.read_csv('kpi_daily.csv')
cleaned_data = clean_kpi_data(raw_data)
print(f"清洗前：{raw_data.shape}，清洗后：{cleaned_data.shape}")