Store Sales 数据清洗实战：处理 4.2 万条重复记录与 43 天油价缺失值

Store Sales 数据清洗实战：处理 4.2 万条重复记录与 43 天油价缺失值

当面对真实世界的数据集时，数据清洗往往是数据分析过程中最耗时但也最关键的环节。Kaggle 上的 Store Sales 时间序列预测竞赛提供了一个绝佳的机会，让我们能够深入探讨如何处理实际业务场景中常见的数据质量问题。本文将聚焦两个最具挑战性的数据问题：基于(date, store_nbr, family)组合的 4.2 万条重复记录，以及油价时间序列中 43 天的缺失值。

1. 数据质量问题的识别与影响分析

在开始任何数据清洗操作之前，理解数据问题的本质及其对分析的影响至关重要。Store Sales 数据集包含多个关联表格，我们需要特别关注训练数据（train.csv）和油价数据（oil.csv）。

1.1 重复记录的检测与验证

首先，我们需要确认重复记录的存在及其性质。在 Pandas 中，可以使用以下代码检测基于关键字段的重复记录：

# 检查基于日期、商店编号和产品类别的重复记录 duplicates = df_train[df_train.duplicated(subset=['date', 'store_nbr', 'family'], keep=False)] print(f"发现 {len(duplicates)} 条潜在重复记录")

验证这些重复记录是否完全相同：

# 检查重复记录是否在所有字段上都相同 identical_duplicates = duplicates.groupby(['date', 'store_nbr', 'family']).filter(lambda x: x.nunique().sum() == len(x.columns)) print(f"完全相同的重复记录数量: {len(identical_duplicates)}")

重复记录的影响矩阵

影响维度	短期影响	长期影响
数据准确性	销售统计偏高	模型训练偏差
存储效率	资源浪费	处理时间增加
分析结果	统计指标失真	预测准确性下降

1.2 油价缺失值的模式分析

油价数据的缺失并非随机，我们需要分析其缺失模式：

# 可视化油价缺失情况 import matplotlib.pyplot as plt plt.figure(figsize=(12, 6)) plt.plot(df_oil['date'], df_oil['dcoilwtico'], marker='o', linestyle='-', color='b') plt.title('油价时间序列与缺失值') plt.xlabel('日期') plt.ylabel('油价') plt.grid(True) plt.show()

常见的缺失模式包括：

• 连续多日缺失（可能由于节假日）
• 零星单日缺失（数据采集问题）
• 特定周期缺失（如周末）

2. 重复记录的系统化处理方案

处理重复记录并非简单的去重操作，而需要综合考虑业务逻辑和数据完整性。

2.1 基于业务规则的解决方案

在零售销售场景中，同一产品在同一商店的同一天理论上只应有一条记录。因此，我们可以采用以下处理流程：

1. 确认重复性质：验证是否为真正的数据重复而非业务上的合理情况
2. 选择保留策略：
   • 保留第一条记录（假设为原始记录）
   • 保留最后一条记录（假设为最新修正）
   • 合并重复记录（对数值字段取平均或求和）

# 实施去重方案 def handle_duplicates(df, strategy='first'): """ 处理重复记录的三种策略 :param df: 原始数据框 :param strategy: 'first'保留第一条，'last'保留最后一条，'mean'取平均值 :return: 处理后的数据框 """ if strategy == 'mean': # 对数值字段取平均值 numeric_cols = df.select_dtypes(include=['int64', 'float64']).columns df = df.groupby(['date', 'store_nbr', 'family'])[numeric_cols].mean().reset_index() else: df = df.drop_duplicates(subset=['date', 'store_nbr', 'family'], keep=strategy) return df

2.2 处理后的验证与影响评估

去重操作后，必须验证数据完整性和一致性：

# 验证去重结果 df_cleaned = handle_duplicates(df_train, strategy='first') print(f"原始记录数: {len(df_train)}") print(f"去重后记录数: {len(df_cleaned)}") print(f"去除重复比例: {(len(df_train)-len(df_cleaned))/len(df_train):.2%}") # 检查关键统计指标变化 original_stats = df_train['sales'].describe() cleaned_stats = df_cleaned['sales'].describe() pd.concat([original_stats, cleaned_stats], axis=1, keys=['原始数据', '清洗后数据'])

3. 油价缺失值的智能填充策略

油价缺失值的处理需要更精细的方法，因为简单的填充可能导致时间序列模式失真。

3.1 时间序列特定的填充方法对比

我们评估了四种常见的填充方法：

填充方法性能对比表

方法	原理	优点	缺点	适用场景
前向填充	使用前一天的值	简单快速	忽略趋势变化	短期连续缺失
线性插值	两点间线性估计	捕捉局部趋势	对波动敏感	中等长度缺失
季节性填充	使用同期历史数据	保持周期模式	需要足够历史数据	周期性明显的数据
预测模型	建立预测模型估计	最准确	计算成本高	关键指标的大段缺失

3.2 基于实际数据的填充实现

针对本案例，我们实现两种主要方法：

def fill_oil_prices(df, method='ffill'): """ 油价缺失值填充实现 :param df: 原始油价数据框 :param method: 填充方法 ('ffill', 'interpolate', 'seasonal') :return: 处理后的数据框 """ df = df.sort_values('date') if method == 'ffill': df['dcoilwtico'] = df['dcoilwtico'].fillna(method='ffill') elif method == 'interpolate': df['dcoilwtico'] = df['dcoilwtico'].interpolate(method='linear') elif method == 'seasonal': # 使用前一年同期数据填充 df['year'] = pd.to_datetime(df['date']).dt.year df['day_of_year'] = pd.to_datetime(df['date']).dt.dayofyear for idx, row in df[df['dcoilwtico'].isna()].iterrows(): same_day_prev_year = df[(df['day_of_year'] == row['day_of_year']) & (df['year'] == row['year']-1)] if not same_day_prev_year.empty: df.at[idx, 'dcoilwtico'] = same_day_prev_year['dcoilwtico'].values[0] # 剩余缺失值用前向填充 df['dcoilwtico'] = df['dcoilwtico'].fillna(method='ffill') return df.drop(['year', 'day_of_year'], axis=1, errors='ignore')

3.3 填充效果的量化评估

为了客观评估不同填充方法的效果，我们可以利用完整数据段进行模拟测试：

# 模拟缺失测试函数 def evaluate_fill_method(df, method, missing_rate=0.1): """ 通过人工制造缺失评估填充方法效果 :param df: 完整数据框 :param method: 填充方法 :param missing_rate: 模拟缺失比例 :return: RMSE评估结果 """ # 随机选择部分数据点作为"缺失" test_df = df.copy() mask = np.random.rand(len(test_df)) < missing_rate test_df.loc[mask, 'dcoilwtico'] = np.nan # 记录真实值 true_values = df.loc[mask, 'dcoilwtico'] # 应用填充方法 filled_df = fill_oil_prices(test_df, method=method) # 计算RMSE rmse = np.sqrt(mean_squared_error(true_values, filled_df.loc[mask, 'dcoilwtico'])) return rmse # 多次测试取平均值 methods = ['ffill', 'interpolate', 'seasonal'] results = {method: np.mean([evaluate_fill_method(df_oil, method) for _ in range(5)]) for method in methods}

4. 数据清洗决策对后续建模的影响

数据清洗不是孤立的过程，我们需要考虑其对后续分析和建模的连锁影响。

4.1 特征工程与数据质量的相互作用

高质量的基础数据能够提升特征工程的效果：

1. 时间序列特征：如滑动窗口统计量对数据连续性敏感
2. 交叉特征：店铺属性与销售数据的关联需要一致的数据
3. 目标变量定义：销售额的统计依赖于准确的记录

# 清洗后的特征工程示例 def create_features(df): """基于清洗后的数据创建时间序列特征""" df = df.sort_values(['store_nbr', 'family', 'date']) # 滞后特征 for lag in [7, 14, 28]: df[f'sales_lag_{lag}'] = df.groupby(['store_nbr', 'family'])['sales'].shift(lag) # 滚动窗口特征 df['rolling_7d_mean'] = df.groupby(['store_nbr', 'family'])['sales'].transform( lambda x: x.rolling(7, min_periods=1).mean()) # 季节性特征 df['day_of_week'] = pd.to_datetime(df['date']).dt.dayofweek df['month'] = pd.to_datetime(df['date']).dt.month return df

4.2 模型性能对比实验

为了量化数据清洗的价值，我们设计了一个对比实验：

实验设计

1. 对照组：使用原始数据直接建模
2. 实验组1：仅处理重复记录
3. 实验组2：仅处理油价缺失值
4. 实验组3：完整清洗流程

# 模型评估函数 def evaluate_model(X_train, y_train, X_test, y_test): """评估随机森林模型在不同数据质量下的表现""" model = RandomForestRegressor(n_estimators=100, random_state=42) model.fit(X_train, y_train) preds = model.predict(X_test) return { 'MAE': mean_absolute_error(y_test, preds), 'RMSE': np.sqrt(mean_squared_error(y_test, preds)), 'R2': r2_score(y_test, preds) } # 执行对比实验 experiment_results = {} for condition in ['raw', 'dedup_only', 'oil_only', 'full_clean']: if condition == 'raw': df = df_train.copy() elif condition == 'dedup_only': df = handle_duplicates(df_train) elif condition == 'oil_only': df = df_train.copy() df_oil_clean = fill_oil_prices(df_oil, method='interpolate') df = pd.merge(df, df_oil_clean, on='date', how='left') else: df = handle_duplicates(df_train) df_oil_clean = fill_oil_prices(df_oil, method='interpolate') df = pd.merge(df, df_oil_clean, on='date', how='left') # 特征工程与训练测试分割 df_featured = create_features(df) X = df_featured.drop(['sales', 'date'], axis=1) y = df_featured['sales'] X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42) # 评估模型 experiment_results[condition] = evaluate_model(X_train, y_train, X_test, y_test)

5. 数据清洗检查清单与最佳实践

基于本案例的经验，我们总结出一套系统化的数据清洗检查清单：

5.1 通用数据质量检查项

1. 完整性检查

   • 缺失值比例统计
   • 时间序列连续性验证
   • 关键字段空值检测

2. 一致性检查

   • 重复记录识别
   • 业务规则验证（如销售不应为负）
   • 跨表关联完整性

3. 准确性检查

   • 异常值检测（Z-score、IQR方法）
   • 统计分布合理性
   • 与外部数据源比对

5.2 时间序列数据特殊考量

针对时间序列数据，还需额外关注：

• 时间维度完整性：是否存在间隔或跳跃
• 季节性模式保持：填充方法是否破坏周期特征
• 趋势一致性：局部处理是否导致全局失真

# 时间序列完整性检查函数 def check_time_continuity(df, date_col='date', freq='D'): """ 检查时间序列的连续性 :param df: 数据框 :param date_col: 日期列名 :param freq: 预期频率 ('D'每日,'M'每月等) :return: 缺失的时间段 """ df = df.sort_values(date_col) full_range = pd.date_range(start=df[date_col].min(), end=df[date_col].max(), freq=freq) missing_dates = full_range.difference(df[date_col]) return missing_dates

5.3 自动化监控建议

对于生产环境，建议建立自动化数据质量监控：

1. 定期检查脚本：运行核心质量指标计算
2. 异常警报机制：设置关键指标的阈值告警
3. 数据血统追踪：记录所有清洗操作的完整日志

# 自动化监控示例类 class DataQualityMonitor: def __init__(self, df): self.df = df self.results = {} def check_completeness(self): """检查数据完整性""" self.results['missing_rates'] = self.df.isnull().mean() def check_uniqueness(self, key_columns): """检查关键字段唯一性""" dup_rate = self.df.duplicated(subset=key_columns).mean() self.results['duplication_rate'] = dup_rate def generate_report(self): """生成质量报告""" report = { 'overview': f"数据集包含 {len(self.df)} 行, {len(self.df.columns)} 列", 'completeness': self.results.get('missing_rates', {}), 'uniqueness': self.results.get('duplication_rate', 0) } return report