多站点多元时间序列预测基线方法开发与实践

1. 多站点多元空气污染时间序列预测的基线方法开发在真实世界的时间序列预测任务中我们常常面临多重挑战多输入变量、多步预测需求以及跨多个物理站点的同步预测要求。EMC数据科学全球黑客马拉松提供的空气质量预测数据集正是这样一个典型案例它记录了多个站点的气象观测数据要求预测未来三天的空气质量指标。作为时间序列预测的第一步建立性能基线至关重要。基线预测策略简单快速被称为朴素方法因为它们几乎不对具体预测问题做任何假设。本文将深入探讨如何为这类多步多元时间序列预测问题开发有效的基线方法。专业提示在实际项目中基线模型的价值常被低估。一个合理的基线不仅能评估后续复杂模型的有效性还能帮助我们发现数据中的潜在问题和模式特征。1.1 问题特性与挑战解析空气质量预测数据集具有几个关键特征多站点结构数据来自多个监测站点每个站点有独立但相关的时序模式多元输入包含温度、气压、风速、风向等多个气象变量非连续预测点需要预测特定时间点的空气质量1, 2, 3, 4, 5, 10, 17, 24, 48, 72小时数据缺失严重大量观测值缺失需要特殊处理数据集按数据块组织每个块包含8天的连续观测其中前5天作为训练数据后3天用于测试。这种结构要求我们的预测方法能够处理不完整的数据块和跨块的信息利用。1.2 朴素预测方法设计原则有效的基线方法应遵循以下设计原则最小假设原则尽可能少地利用问题特定信息可解释性方法简单明了便于理解计算高效相比复杂模型应有显著的速度优势可复现性结果稳定随机性低我们将重点考察两类朴素方法全局方法利用整个训练集的信息局部方法仅使用当前数据块的信息2. 数据准备与测试框架构建2.1 数据集加载与预处理首先我们需要建立完整的数据处理流程。以下是使用Python的关键步骤from numpy import unique, nan, array from pandas import read_csv def load_and_preprocess(file_path): # 加载原始数据 dataset read_csv(file_path, header0) values dataset.values # 按chunkID分组 chunks {} chunk_ids unique(values[:, 1]) # 假设chunkID在第二列 for chunk_id in chunk_ids: chunks[chunk_id] values[values[:, 1] chunk_id] return chunks2.2 训练集/测试集分割策略我们采用以下分割方法每个数据块的前5天120小时作为训练数据后3天72小时作为测试数据剔除没有足够训练或测试数据的数据块实现代码def split_train_test(chunks, cut_point5*24): train, test [], [] for chunk_id, rows in chunks.items(): train_rows rows[rows[:, 2] cut_point] # position_within_chunk在第三列 test_rows rows[rows[:, 2] cut_point] if len(train_rows) 0 and len(test_rows) 0: # 保留chunkID、位置、小时和目标变量 indices [1, 2, 5] list(range(56, rows.shape[1])) train.append(train_rows[:, indices]) test.append(test_rows[:, indices]) return train, test2.3 预测评估框架设计我们使用MAE平均绝对误差作为评估指标与原始比赛保持一致。评估时需注意忽略测试集中缺失值NaN如果预测为NaN但实际值存在将实际值全额计入误差同时计算整体MAE和各预测时间点的MAE评估函数实现def evaluate_forecasts(predictions, testset): total_mae, times_mae 0.0, [0.0]*10 total_c, times_c 0, [0]*10 lead_times [1, 2, 3, 4, 5, 10, 17, 24, 48, 72] for i in range(len(testset)): # 遍历各chunk actual testset[i] predicted predictions[i] for j in range(39): # 39个目标变量 for k in range(10): # 10个预测时间点 if isnan(actual[j, k]): continue error abs(actual[j, k] - predicted[j, k]) if not isnan(predicted[j, k]) else abs(actual[j, k]) total_mae error times_mae[k] error total_c 1 times_c[k] 1 total_mae / total_c times_mae [times_mae[i]/times_c[i] for i in range(10)] return total_mae, times_mae3. 全局朴素预测方法实现3.1 全局平均值预测最简单的基线方法是预测每个目标变量在整个训练集中的平均值def global_mean_forecast(train_chunks, test_input): # 收集所有训练数据 all_train array([row for chunk in train_chunks for row in chunk]) # 计算各目标变量的均值跳过前3列元数据 means [nanmean(all_train[:, 3i]) for i in range(39)] # 生成预测 predictions [] for chunk in test_input: chunk_pred [] for tau in range(10): # 10个预测时间点 chunk_pred.append(means) predictions.append(array(chunk_pred).T) # 转置为[变量][时间]格式 return array(predictions)3.2 按小时分段的全局平均值预测更精细化的方法是考虑一天中不同小时的影响预测每个目标变量在特定小时的全局平均值def global_mean_by_hour_forecast(train_chunks, test_input): # 组织训练数据小时 - 变量 - 值列表 hourly_data {h: [[] for _ in range(39)] for h in range(24)} for chunk in train_chunks: for row in chunk: hour int(row[2]) % 24 # 小时在第3列 for var in range(39): if not isnan(row[3var]): hourly_data[hour][var].append(row[3var]) # 计算各小时各变量的均值 hourly_means {h: [nanmean(vars_data) if vars_data else nan for vars_data in var_lists] for h, var_lists in hourly_data.items()} # 生成预测 predictions [] for chunk in test_input: chunk_pred [] for row in chunk: hour int(row[2]) % 24 chunk_pred.append(hourly_means[hour]) predictions.append(array(chunk_pred).T) return array(predictions)4. 局部朴素预测方法实现4.1 最后观测值持久化预测对于每个数据块使用训练期最后一个有效观测值作为未来所有时间点的预测def last_observation_forecast(train_chunks, test_input): predictions [] for i, chunk in enumerate(train_chunks): # 找出各变量最后一个有效观测值 last_obs [] for var in range(39): # 逆序查找第一个非NaN值 for row in reversed(chunk): if not isnan(row[3var]): last_obs.append(row[3var]) break else: last_obs.append(nan) # 为所有预测时间点重复该值 chunk_pred [last_obs for _ in range(10)] predictions.append(array(chunk_pred)) return array(predictions)4.2 局部平均值预测使用当前数据块训练部分的平均值作为预测def local_mean_forecast(train_chunks, test_input): predictions [] for chunk in train_chunks: # 计算当前chunk各变量的均值 means [nanmean(chunk[:, 3i]) for i in range(39)] # 生成预测 chunk_pred [means for _ in range(10)] predictions.append(array(chunk_pred)) return array(predictions)4.3 按小时分段的局部平均值预测结合数据块内的小时信息预测各小时段的平均值def local_mean_by_hour_forecast(train_chunks, test_input): predictions [] for i, chunk in enumerate(train_chunks): # 组织当前chunk的小时数据 hourly_data {h: [[] for _ in range(39)] for h in range(24)} for row in chunk: hour int(row[2]) % 24 for var in range(39): if not isnan(row[3var]): hourly_data[hour][var].append(row[3var]) # 计算各小时各变量的均值 hourly_means {h: [nanmean(vars_data) if vars_data else nan for vars_data in var_lists] for h, var_lists in hourly_data.items()} # 为测试集的每个小时生成预测 test_hours [int(row[2]) % 24 for row in test_input[i]] chunk_pred [hourly_means[h] for h in test_hours] predictions.append(array(chunk_pred).T) return array(predictions)5. 方法比较与结果分析我们使用上述五种方法在空气质量数据集上进行测试得到如下典型结果具体数值会因数据预处理方式略有差异方法整体MAE最佳预测点最差预测点全局平均值0.31248h (0.289)1h (0.341)全局小时平均值0.28772h (0.263)1h (0.315)最后观测值0.2981h (0.275)72h (0.327)局部平均值0.27624h (0.251)1h (0.302)局部小时平均值0.25424h (0.230)1h (0.283)从结果可以看出考虑小时周期的局部方法表现最佳说明空气质量数据具有明显的日周期特征所有方法在1h的预测误差最大表明短期变化最难捕捉局部方法普遍优于全局方法说明站点特异性特征很重要比赛最佳成绩MAE为0.21058我们的最佳基线与之差距约20%显示了更复杂模型的潜在价值实战经验在真实项目中我们通常会先运行所有基线方法选择表现最好的作为基准。当开发复杂模型时如果不能在验证集上稳定超越基线10-15%则说明模型可能存在问题或特征工程不足。6. 扩展与优化方向基于基线方法的结果我们可以考虑以下优化方向跨站点信息利用同一地区的不同站点数据可能存在相关性气象变量整合将温度、风速等作为辅助变量纳入预测时间特征工程添加星期、节假日等时序特征缺失值插补采用更智能的缺失值处理方式而非简单忽略误差分析针对MAE最大的预测点进行专项优化一个简单的跨站点改进示例def enhanced_local_forecast(train_chunks, test_input, site_vars3): 考虑邻近站点信息的增强版局部预测 predictions [] site_info {} # 存储各站点的位置等信息 # 实现会考虑站点空间关系的预测逻辑 # ... return array(predictions)在实际业务场景中这些基线方法的价值不仅在于提供比较基准还能帮助我们快速验证数据管道、评估问题难度并为后续的特征工程提供方向指引。