Python数据分析实战：Pandas数据清洗与特征工程全流程

数据清洗和特征工程通常占据数据分析师60%-80%的工作时间。这篇文章用一个真实场景（电商用户行为数据），带你走通从脏数据到建模就绪数据的完整流程。

一、数据背景

假设我们有一份电商平台的用户行为日志，包含以下字段：

• user_id：用户ID
• event_type：行为类型（view/add_cart/purchase）
• product_id：商品ID
• price：商品价格
• timestamp：行为发生时间
• user_age：用户年龄
• city：用户城市
• device：设备类型

这份数据存在缺失值、异常值、重复记录等问题。我们的目标是：清理数据并构建可用于用户转化预测的特征矩阵。

二、数据加载与初步探索

import pandas as pd
import numpy as np
from datetime import datetime
import warnings
warnings.filterwarnings('ignore')

# 加载数据
df = pd.read_csv('user_behavior.csv', parse_dates=['timestamp'])

print(f"数据集形状: {df.shape}")
print(f"\n数据类型:\n{df.dtypes}")
print(f"\n缺失值统计:\n{df.isnull().sum()}")
print(f"\n基本统计:\n{df.describe()}")

三、缺失值处理

3.1 缺失值诊断

在动手填充之前，先判断缺失模式。缺失值分三种：

• 完全随机缺失（MCAR）：缺失与任何变量无关
• 随机缺失（MAR）：缺失与其他变量有关
• 非随机缺失（MNAR）：缺失与自身值有关

# 创建缺失值指示矩阵
missing_matrix = df.isnull().astype(int)

# 检查缺失值之间的相关性
missing_corr = missing_matrix.corr()
print("缺失值相关性矩阵:")
print(missing_corr[missing_corr > 0.3])

# 按城市分组查看 price 缺失率
price_missing_by_city = df.groupby('city')['price'].apply(
    lambda x: x.isnull().mean()
).sort_values(ascending=False)
print(f"\n各城市price缺失率:\n{price_missing_by_city.head(10)}")

3.2 分策略填充

# 策略1：价格用中位数填充（受异常值影响小）
df['price'] = df['price'].fillna(df['price'].median())

# 策略2：年龄用分组中位数填充（按城市分组）
df['user_age'] = df.groupby('city')['user_age'].transform(
    lambda x: x.fillna(x.median())
)

# 策略3：城市为空直接删除（关键特征不能缺失）
df = df.dropna(subset=['city'])

# 策略4：设备类型用"未知"填充
df['device'] = df['device'].fillna('unknown')

print(f"缺失值处理后的数据形状: {df.shape}")
print(f"剩余缺失值:\n{df.isnull().sum()}")

四、异常值检测与处理

# 使用IQR方法检测异常值
def detect_outliers_iqr(data, column):
    """使用1.5*IQR规则检测异常值"""
    Q1 = data[column].quantile(0.25)
    Q3 = data[column].quantile(0.75)
    IQR = Q3 - Q1
    lower_bound = Q1 - 1.5 * IQR
    upper_bound = Q3 + 1.5 * IQR
    outliers = data[(data[column] < lower_bound) | (data[column] > upper_bound)]
    return outliers, lower_bound, upper_bound

# 检测价格异常值
price_outliers, price_lower, price_upper = detect_outliers_iqr(df, 'price')
print(f"价格异常值数量: {len(price_outliers)}")
print(f"正常价格范围: [{price_lower:.2f}, {price_upper:.2f}]")

# 检测年龄异常值
age_outliers, age_lower, age_upper = detect_outliers_iqr(df, 'user_age')
print(f"年龄异常值数量: {len(age_outliers)}")
print(f"正常年龄范围: [{age_lower:.0f}, {age_upper:.0f}]")

# 截断法处理（保留数据，限制范围）
df['price'] = df['price'].clip(lower=price_lower, upper=price_upper)
df['user_age'] = df['user_age'].clip(lower=0, upper=100)

print(f"异常值处理后的数据形状: {df.shape}")

五、重复值处理

# 检查完全重复行
dup_count = df.duplicated().sum()
print(f"完全重复行数: {dup_count}")

# 检查按user_id + timestamp的近似重复
near_dup = df.groupby(['user_id', 'timestamp']).size()
near_dup_count = (near_dup > 1).sum()
print(f"同一用户同一时间戳的重复记录: {near_dup_count} 组")

# 保留每组的第一条
df = df.drop_duplicates(subset=['user_id', 'timestamp'], keep='first')
print(f"去重后数据形状: {df.shape}")

六、特征工程

6.1 时间特征构造

# 提取时间维度特征
df['hour'] = df['timestamp'].dt.hour
df['day_of_week'] = df['timestamp'].dt.dayofweek
df['is_weekend'] = df['day_of_week'].isin([5, 6]).astype(int)
df['day_period'] = pd.cut(
    df['hour'],
    bins=[0, 6, 12, 18, 24],
    labels=['凌晨', '上午', '下午', '晚上']
)
print(df[['hour', 'day_of_week', 'is_weekend', 'day_period']].head())

6.2 用户行为聚合特征

# 按用户聚合行为特征
user_features = df.groupby('user_id').agg(
    total_actions=('event_type', 'count'),
    view_count=('event_type', lambda x: (x == 'view').sum()),
    cart_count=('event_type', lambda x: (x == 'add_cart').sum()),
    purchase_count=('event_type', lambda x: (x == 'purchase').sum()),
    unique_products=('product_id', 'nunique'),
    avg_price=('price', 'mean'),
    max_price=('price', 'max'),
    active_days=('timestamp', lambda x: x.dt.date.nunique())
).reset_index()

# 构造衍生特征
user_features['cart_rate'] = user_features['cart_count'] / user_features['view_count'].replace(0, 1)
user_features['purchase_rate'] = user_features['purchase_count'] / user_features['view_count'].replace(0, 1)
user_features['avg_daily_actions'] = user_features['total_actions'] / user_features['active_days'].replace(0, 1)

# 目标变量：是否产生购买
user_features['has_purchase'] = (user_features['purchase_count'] > 0).astype(int)

print(f"用户特征矩阵形状: {user_features.shape}")
print(f"购买转化率: {user_features['has_purchase'].mean():.2%}")
print(f"\n特征概览:\n{user_features.describe()}")

6.3 类别特征编码

# City编码：使用频次编码处理高基数特征
city_freq = df['city'].value_counts(normalize=True)
df['city_freq_enc'] = df['city'].map(city_freq)

# Device编码：One-Hot编码（类别少）
device_dummies = pd.get_dummies(df['device'], prefix='device', dtype=int)
df = pd.concat([df, device_dummies], axis=1)

# 时段编码：Label Encoding
period_map = {'凌晨': 0, '上午': 1, '下午': 2, '晚上': 3}
df['day_period_enc'] = df['day_period'].map(period_map)

print("编码后的特征列:")
print([col for col in df.columns if col.startswith('device_') or 'enc' in col])

七、特征质量验证

# 检查特征方差（去除常数值特征）
low_variance_cols = []
for col in user_features.select_dtypes(include=[np.number]).columns:
    if user_features[col].std() < 1e-6:
        low_variance_cols.append(col)
print(f"需移除的低方差特征: {low_variance_cols}")

# 检查特征间相关性
feature_cols = ['total_actions', 'view_count', 'cart_count', 'purchase_count',
                'unique_products', 'avg_price', 'max_price', 'active_days',
                'cart_rate', 'avg_daily_actions']

correlation_matrix = user_features[feature_cols].corr()
high_corr_pairs = []
for i in range(len(feature_cols)):
    for j in range(i+1, len(feature_cols)):
        if abs(correlation_matrix.iloc[i, j]) > 0.9:
            high_corr_pairs.append((feature_cols[i], feature_cols[j],
                                    correlation_matrix.iloc[i, j]))

print(f"高相关特征对(>0.9): {high_corr_pairs}")

# 检查目标变量分布
import matplotlib.pyplot as plt

fig, ax = plt.subplots(figsize=(8, 4))
user_features['has_purchase'].value_counts().plot(kind='bar', ax=ax)
ax.set_title('Purchase Distribution')
ax.set_xlabel('Has Purchase')
ax.set_ylabel('Count')
plt.tight_layout()
plt.savefig('purchase_distribution.png', dpi=100)
plt.show()
print("✅ 特征质量验证完成")

八、完整Pipeline封装

class ECommerceDataPipeline:
    """电商用户行为数据清洗与特征工程Pipeline"""
    
    def __init__(self):
        self.price_median = None
        self.age_medians = {}  # 按城市分组的年龄中位数
        self.price_bounds = None  # (lower, upper)
        self.city_freq = None
        self.city_enc_mapping = {}
        
    def fit(self, df):
        """从训练数据中学习转换参数"""
        # 学习价格统计量
        self.price_median = df['price'].median()
        Q1, Q3 = df['price'].quantile([0.25, 0.75])
        IQR = Q3 - Q1
        self.price_bounds = (Q1 - 1.5 * IQR, Q3 + 1.5 * IQR)
        
        # 学习各城市年龄中位数
        self.age_medians = df.groupby('city')['user_age'].median().to_dict()
        
        # 学习城市频次编码
        self.city_freq = df['city'].value_counts(normalize=True).to_dict()
        
        return self
    
    def transform(self, df):
        """对数据应用清洗和特征工程转换"""
        df = df.copy()
        
        # 缺失值处理
        df['price'] = df['price'].fillna(self.price_median)
        df['user_age'] = df.apply(
            lambda row: self.age_medians.get(row['city'], 30)
            if pd.isna(row['user_age']) else row['user_age'],
            axis=1
        )
        df = df.dropna(subset=['city'])
        df['device'] = df['device'].fillna('unknown')
        
        # 异常值处理
        df['price'] = df['price'].clip(*self.price_bounds)
        df['user_age'] = df['user_age'].clip(0, 100)
        
        # 重复值
        df = df.drop_duplicates(subset=['user_id', 'timestamp'], keep='first')
        
        return df
    
    def fit_transform(self, df):
        return self.fit(df).transform(df)


# 使用示例
pipeline = ECommerceDataPipeline()
df_clean = pipeline.fit_transform(df)

# 验证清洗结果
print("=== 数据清洗验证 ===")
print(f"原始数据行数: {len(df)}")
print(f"清洗后行数: {len(df_clean)}")
print(f"数据损失率: {(1 - len(df_clean)/len(df)):.2%}")
print(f"缺失值数量: {df_clean.isnull().sum().sum()}")
print("✅ Pipeline执行成功")

九、总结与最佳实践

这套数据清洗与特征工程流程的核心要点：

• 缺失值处理：先诊断再填充，不同特征用不同策略（中位数/分组中位数/删除）
• 异常值处理：IQR方法检测 + 截断法处理，保留数据而非简单删除
• 特征工程：时间特征（小时/星期/时段）、行为聚合特征（频次/率值）、类别编码（频次/One-Hot）
• 特征验证：检查低方差特征和共线性，保证模型输入质量
• Pipeline封装：用fit/transform模式避免数据泄露，确保生产环境可复现

实际工作中，数据清洗不是一次性操作，而是迭代过程。建议用Jupyter Notebook做探索，确定清洗策略后再封装为Pipeline。这样既保留了探索的灵活性，又确保了上线的可靠性。

完整的代码和示例数据集已推送到GitHub，需要的话评论区留言获取。