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], keepFalse)] 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], markero, linestyle-, colorb) plt.title(油价时间序列与缺失值) plt.xlabel(日期) plt.ylabel(油价) plt.grid(True) plt.show()常见的缺失模式包括连续多日缺失可能由于节假日零星单日缺失数据采集问题特定周期缺失如周末2. 重复记录的系统化处理方案处理重复记录并非简单的去重操作而需要综合考虑业务逻辑和数据完整性。2.1 基于业务规则的解决方案在零售销售场景中同一产品在同一商店的同一天理论上只应有一条记录。因此我们可以采用以下处理流程确认重复性质验证是否为真正的数据重复而非业务上的合理情况选择保留策略保留第一条记录假设为原始记录保留最后一条记录假设为最新修正合并重复记录对数值字段取平均或求和# 实施去重方案 def handle_duplicates(df, strategyfirst): 处理重复记录的三种策略 :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], keepstrategy) return df2.2 处理后的验证与影响评估去重操作后必须验证数据完整性和一致性# 验证去重结果 df_cleaned handle_duplicates(df_train, strategyfirst) 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], axis1, keys[原始数据, 清洗后数据])3. 油价缺失值的智能填充策略油价缺失值的处理需要更精细的方法因为简单的填充可能导致时间序列模式失真。3.1 时间序列特定的填充方法对比我们评估了四种常见的填充方法填充方法性能对比表方法原理优点缺点适用场景前向填充使用前一天的值简单快速忽略趋势变化短期连续缺失线性插值两点间线性估计捕捉局部趋势对波动敏感中等长度缺失季节性填充使用同期历史数据保持周期模式需要足够历史数据周期性明显的数据预测模型建立预测模型估计最准确计算成本高关键指标的大段缺失3.2 基于实际数据的填充实现针对本案例我们实现两种主要方法def fill_oil_prices(df, methodffill): 油价缺失值填充实现 :param df: 原始油价数据框 :param method: 填充方法 (ffill, interpolate, seasonal) :return: 处理后的数据框 df df.sort_values(date) if method ffill: df[dcoilwtico] df[dcoilwtico].fillna(methodffill) elif method interpolate: df[dcoilwtico] df[dcoilwtico].interpolate(methodlinear) 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(methodffill) return df.drop([year, day_of_year], axis1, errorsignore)3.3 填充效果的量化评估为了客观评估不同填充方法的效果我们可以利用完整数据段进行模拟测试# 模拟缺失测试函数 def evaluate_fill_method(df, method, missing_rate0.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, methodmethod) # 计算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 特征工程与数据质量的相互作用高质量的基础数据能够提升特征工程的效果时间序列特征如滑动窗口统计量对数据连续性敏感交叉特征店铺属性与销售数据的关联需要一致的数据目标变量定义销售额的统计依赖于准确的记录# 清洗后的特征工程示例 def create_features(df): 基于清洗后的数据创建时间序列特征 df df.sort_values([store_nbr, family, date]) # 滞后特征 for lag in [7, 14, 28]: df[fsales_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_periods1).mean()) # 季节性特征 df[day_of_week] pd.to_datetime(df[date]).dt.dayofweek df[month] pd.to_datetime(df[date]).dt.month return df4.2 模型性能对比实验为了量化数据清洗的价值我们设计了一个对比实验实验设计对照组使用原始数据直接建模实验组1仅处理重复记录实验组2仅处理油价缺失值实验组3完整清洗流程# 模型评估函数 def evaluate_model(X_train, y_train, X_test, y_test): 评估随机森林模型在不同数据质量下的表现 model RandomForestRegressor(n_estimators100, random_state42) 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, methodinterpolate) df pd.merge(df, df_oil_clean, ondate, howleft) else: df handle_duplicates(df_train) df_oil_clean fill_oil_prices(df_oil, methodinterpolate) df pd.merge(df, df_oil_clean, ondate, howleft) # 特征工程与训练测试分割 df_featured create_features(df) X df_featured.drop([sales, date], axis1) y df_featured[sales] X_train, X_test, y_train, y_test train_test_split(X, y, test_size0.2, random_state42) # 评估模型 experiment_results[condition] evaluate_model(X_train, y_train, X_test, y_test)5. 数据清洗检查清单与最佳实践基于本案例的经验我们总结出一套系统化的数据清洗检查清单5.1 通用数据质量检查项完整性检查缺失值比例统计时间序列连续性验证关键字段空值检测一致性检查重复记录识别业务规则验证如销售不应为负跨表关联完整性准确性检查异常值检测Z-score、IQR方法统计分布合理性与外部数据源比对5.2 时间序列数据特殊考量针对时间序列数据还需额外关注时间维度完整性是否存在间隔或跳跃季节性模式保持填充方法是否破坏周期特征趋势一致性局部处理是否导致全局失真# 时间序列完整性检查函数 def check_time_continuity(df, date_coldate, freqD): 检查时间序列的连续性 :param df: 数据框 :param date_col: 日期列名 :param freq: 预期频率 (D每日,M每月等) :return: 缺失的时间段 df df.sort_values(date_col) full_range pd.date_range(startdf[date_col].min(), enddf[date_col].max(), freqfreq) missing_dates full_range.difference(df[date_col]) return missing_dates5.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(subsetkey_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