| 多重时序高阶算法 | 您所在的位置：网站首页 › 微信公众号二维码链接在哪里生成 › 多重时序高阶算法 |
多重时序高阶算法
|
前言 时间序列几乎无处不在，针对时序的预测也成为一个经典问题。根据时间序列数据的输入和输出格式，时序预测问题可以被更详细的划分。 根据单个时间序列输入变量个数一元时间序列（univariate timeseries），该变量也是需要预测的对象（Target）多元时间序列 （multivariate timeseries），除了需要预测的对象之外，其他变量称为协变量（covariate）根据时间序列需要预测的步长预测单个时间步长，称为单步预测 （single step)预测多个连续时间步长，称为多步预测（multi step）一般的教程就分为这两个维度，如果考虑实际应用，还可以从另一个维度进行划分。 时间序列的个数如果是单个时间序列，一般约定俗成为时间序列 （time series）如果是多个时间序列，则称为多重时间序列 （multi-timeseries）我们在解决时序预测问题时，需要根据问题的描述精细定位。对于一元时间序列，可能经典的ARIMA就能解决问题。当然了，这个划分也不是绝对的，因为一元时间序列通过提取时间相关的特征，转换为多元时间序列。这时候传统的回归算法都可适用，比如Lightgbm。 既然有了Lightgbm，为什么还要DeepAR呢？ 如果说Lightgbm是中国厨师手里的菜刀，大厨用的好的话（预处理和特征工程），雕花也没问题，但前提是大厨。DeepAR更像是德国厨师的刀具组，各司其职，术业专攻，普通人也能通过选择合适的刀具也来快速展示厨艺。 DeepAR 介绍DeepAR 就是专门针对多重时间序列（单）多元（单）多步预测问题的算法。 其中DeepAR要解决问题的痛点是：多重时间序列。比如商品销售预测问题，试想一下电商十多个大类，上千个小类，上百万个商品，难道我们需要建立上百万个模型吗？如果是新品，没有训练数据怎么办（冷启动问题）？ DeepAR的核心是用循环神经网络（RNN）预测多重时间序列。相比于一般的RNN算法，个人感觉它至少有三点优势： 不是直接简单地输出一个预测值，而是输出预测值的一个概率分布。在计算损失函数时，采用的是分布损失(DistributionLoss)。这样做的意义是，当你明确数据分布不是正态，而是长尾分布或者其他分布时，你可以选用更适合该分布的损失函数。另外在计算损失函数时，可以让算法只考虑特定分位的损失，比如0.25分位，0.5分位等。支持直接输入多重时间序列，并且可以通过Embedding 层来学习同类时间序列的共性，方便处理海量时间序列。比如多个类别下上万个产品。自带特征工程，比如根据时间步长选定时间特征，以及Lag特征等DeepAR 轮子哪家强DeepAR 已经算一个经典的算法了，因为几乎每个新的多重时间序列预测算法都会以DeepAR作为baseline。这个经典的算法自然已经在主流的深度学习框架中实现。这里列出比较流行的轮子： 亲儿子 Amazon SageMaker。DeepAR是Amazon 的算法，自然要包含在自家的商业平台SageMaker中。个人觉得它用途过于商业，不适合学习和比赛。aws/sagemaker-python-sdk干儿子 GluonTS 。为了让自家的深度框架MXnet 生态更加完善，Amazon 引入了一系列Gluon 包，针对时序的GluonTS 就是其中一个。不过现在MXnet 已经转到Apache 孵化了。awslabs/gluon-ts非亲非故 PyTorch Forecasting，这是博士大牛开发的轮子，推荐的原因有两个，一是它基于Pytorch-lightning框架，代码很是工整。二是它的文档齐全，案例经典，作者曾在FB实习。jdb78/pytorch-forecasting非亲非故 deepar arrigonialberto86/deepar, 基于tensorflow的版本。我自己用的比较多的是GluonTS 和PyTorch Forecasting，这里对两个稍作比较： GluonTS 算法比较齐全，除了DeepAR，还有其他基于概率预测的时间预测算法，比如TFT。PyTorch Forecasting 目前仅仅支持三个算法。就输入数据格式而言，GluonTS 采用的单个时间序列的配置，也就是输入的时候能清晰的理解和配置每个时间序列的相关信息，比如Target，时不变类别变量，时不变连续变量等。PyTorch Forecasting 采用的数据类型是类似表格型的，所有的时序是按行堆叠的。就成熟度而言，PyTorch Forecasting 和原版DeepAR略有差异，比如没有内置时间序列的特征提取。而GluonTS 算是SageMaker的高仿版，虽然它的文档里面标榜自己和SageMaker 实现思路没有任何关系，反正我不信。因为SageMaker中DeepAR的参数含义和用法完全可以套用在GluonTS的DeepAR中。就深度学习框架而言，GluonTS 基于MXnet，相当于手机中锤子的位置吧，曲高和寡。PyTorch Forecasting 基于高阶Pytorch （Pytorch Lightning）, 相当于小米生态。从使用方面来看，个人觉得PyTorch Forecasting 的使用习惯更好一些。GluonTS 虽然也是Amazon旗下的，但是干儿子的待遇似乎不好，连基本的validation 处理思路都有bug。本文是以GluonTS 为例进行实战，主要是体验原汁原味的DeepAR。 实战数据这里采用山东省第二届数据应用创新创业大赛-临沂分赛场-供水管网压力预测的数据，该比赛目前（2021-02-02）还进展的如火如荼。 比赛链接：山东省数据应用创新创业大赛 比赛任务：通过某新区供水管网的历史压力数据、天气数据和供水管网互通图，预测未来某时间点的压力数据。 比赛的数据：训练集：2018至2019年的30个压力监测点近两年的压力数据、2018年至2019年的天气数据，以及标明了30个压力监测点位置的供水管网互通图。 测试集：以下4段时间的每小时的压力数据、每天的天气数据，需要分别去预测对应日期每小时的压力数据。  评价指标：采用均方误差MSE进行评价 概括一下比赛：主要是任务是预测未来14-21天内，30个站点每个小时的压力值。有用的数据是各个站点过去两年的压力值，天气信息以及管网分布。注意，未来的天气在比赛中是不得使用的。 代码会分享在文末的链接，这里主要针对关键思路进行详解。 数据与环境准备gluonts 和mxnet 都可以通过pip安装，如果需要支持GPU，可以安装mxnet-cu101。 !pip install gluonts !pip install mxnet #!pip install -U mxnet-cu101==1.7.0 #### install GPU version if you have神经网络算法几乎都会指定一下seed，便于结果复现。 r_seed = 21 random.seed(r_seed) mx.random.seed(r_seed) np.random.seed(r_seed)从官网可以下载数据，读取数据是基本步骤。然后把2018年和2019年的数据进行合并处理。需要注意的是，Baseline我没有使用天气信息进行处理，文末我会提到一些天气处理方式。 folder = '/content/gdrive/MyDrive/water_pred/' pressure_2018_df = pd.read_csv(folder+'train_水压数据_2018.csv',encoding='utf8') pressure_2019_df = pd.read_csv(folder+'train_水压数据_2019.csv',encoding='utf8') pressure_2020_df = pd.read_csv(folder+'test_水压数据_2020.csv',encoding='utf8') weather_2018_2019_df = pd.read_csv(folder+'2018-2019气象数据.csv',encoding='utf8') # not used in this baseline weather_2020_df = pd.read_csv(folder+'2020气象数据.csv',encoding='utf8') # not used in this baseline pressure_2018_2019_df = pd.concat([pressure_2018_df,pressure_2019_df],ignore_index=True) to_predict_df_1st = pd.read_csv(folder+'to_predict.csv',encoding='utf8')预览一下训练数据集pressure_2018_2019_df 数据，可以看到数据包含Time 列，站点MeasName列，剩下的H0-H23 列为当天的24小时压力数据。  预览一下提交的数据集格式，未来数据中我们仅仅知道站点和时间（日期+小时）。  另外，管网和测点的分布也是有用信息，比如同一条直线上的测点分布会有相似性，主线上的数据会明显区分于支线，另外还可以考虑同一片区的用水的相似性。  数据预处理数据预处理包含以下部分： 基本琐碎项处理，比如去除“站点”和“H”字样，方便索引。def format_df(df): df['MeasName'] = pd.to_numeric(df['MeasName'].str.lstrip('站点')) df['Time'] =pd.to_datetime(df['Time']) return df有必要对训练数据集进行reshape，让数据变得更加适合分析和建模。这里我将每个站点的数据变换为一列这样每列数据可以快速进行EDA分析有必要对每列数据（每个站点）进行单独的预处理，比如查看缺失值，异常值def reshape_pressure_df(df): # reshape to column-wise df_reshape = pd.DataFrame(None,columns=['DateTime','MeasName','Pressure']) for i in [i for i in range(24)]: per_hour_df = df[['Time','MeasName',f'H{i}']] per_hour_df['Time'] = per_hour_df['Time']+ pd.Timedelta(i,unit='h') per_hour_df.columns=['DateTime','MeasName','Pressure'] df_reshape = pd.concat([df_reshape,per_hour_df]) df_reshape = pd.pivot_table(df_reshape, values='Pressure', index=['DateTime'], columns=['MeasName']) return df_reshape提取时间序列特征对于GluonTS来说，这步不是必须的，因为GluonTS DeepAR会自动提取时间特征。但是如果你更相信自己的特征，可以显式的提取特征，加入到DeepAR模型时间序列特征可以便于EDA分析，尤其是查看周期性，用于确定Lag。当然了，GluonTS DeepAR 也会自动提取Lag。def dt_features(df): df['hour'] = df.index.hour df['day'] = df.index.day df['weekday'] = df.index.dayofweek df['weeks'] = df.index.weekofyear df['month'] = df.index.month df['quarter'] = df.index.quarter return df异常处理，异常值有以下几类：明显超出传感器量程或者不合常理的数值，比如压力小于0或者大于0.5瞬时突变类的数值，这里采用z_score进行筛选连续常值，这类数据一般为传感器故障或者是传感器处于标定状态。通过判断连续时间段的方差即可。def remove_abnormal(df): df[df[G_POST_LIST]=0.5] =None return df def remove_abnormal_z_score(df,by_col='hour',th =3 ): for pos in tqdm(G_POST_LIST): df_zscore=df[[pos,by_col]].copy() f_zscores_all = pd.Series() for i in range(24): f_zscores=(df_zscore[df_zscore[by_col]==i][pos]-df_zscore[df_zscore[by_col]==i][pos].mean())/df_zscore[df_zscore[by_col]==i][pos].std() f_zscores_all= pd.concat([f_zscores_all,f_zscores]) f_zscores_all.sort_index(inplace=True) df_zscore.loc[f_zscores_all.abs()> th,pos] = None # must use loc df[pos] = df_zscore[pos] return df def remove_abnormal_zero_var(df,window =3,th=1e-07): for pos in tqdm(G_POST_LIST): df_pos=df[[pos]].copy() df_pos['var']=df_pos[pos].rolling(window=window,center=False).std() df_pos.loc[df_pos['var'] 确定RNN相关的参数cell_type：默认选择LSTM，你可以选择其他，比如GRUnum_layers：RNN隐含层层数，默认为2num_cells：RNN每层节点数，默认为40dropout_rate：默认为0.1确认输入数据格式的参数freq：需要指定，这里我们freq是“1H”prediction_length：需要预测的步长，这里是21天（21*24）context_length：历史数据长度，也就是需要回看多久的数据，这里采用28天确定输入的协变量的参数：use_feat_dynamic_real：是否启用时变连续变量作为协变量，如果我们调用天气信息作为协变量，需要设置为Trueuse_feat_static_cat：是否采用时不变类别变量作为协变量，这里我们设置为True，因为我们需要提供观测点的信息。对于上文提到的管网和测点分布，如果合理分析，可以提取更多有用的时不变类别信息use_feat_static_real：是否启用时不变连续变量作为协变量，这里我们设置为Falselags_seq：显式设置lag参数，否则系统自动计算lagtime_features：显式设置时间特征，否则系统自动计算DeepAR特有的参数：distr_output: 评估误差时采用的分布，默认是StudentT分布cardinality：Embedding的输入维数embedding_dimension：Embedding的输出维数Trainer相关的参数无非是定义一个trainer，这个trainer需要指定batch_siz,epochs，learning_rate 等学习相关的参数理解了上述参数，就比较容易建立模型。以下是参考代码，里面的超参数需要调试。 trainer = Trainer(epochs=3,batch_size=128,num_batches_per_epoch=50,patience=5) estimator = deepar.DeepAREstimator( freq=freq, prediction_length=prediction_length, trainer=trainer,context_length=context_length,use_feat_dynamic_real =False,num_layers=2,num_cells=40,use_feat_static_cat =True,cardinality=[30,30],embedding_dimension =[11,30],lags_seq= [24,12,8])输入数据的格式相对简单，因为GluonTS对输入数据格式ListDataset要求比较宽松。多重时间序列用list表示，每个时间序列需要指定一个字典： Target：预测对象start：预测起始值FieldName.FEAT_STATIC_CAT：时不变类别变量FieldName. FEAT_DYNAMIC_REAL：时变连续变量等等其他FieldNamepos_group_dict = {1:6,2:0,3:6,4:0,5:6,6:4,7:6,8:0,9:4,10:4,11:5,12:6,13:6,14:2,15:1,16:3,17:7,18:6,19:6,20:6,21:6,22:5,23:4,24:4,25:6,26:5,27:5,28:6,29:6,30:5} meas_dict = {pos:pos for pos in G_POST_LIST} train_ds = ListDataset([{'target': use_diff_2018_2019_df.loc['2018-06':'2019-10',pos], 'start': pd.Timestamp("2018-06-01", freq=freq),FieldName.FEAT_STATIC_CAT:[pos_group_dict[pos],meas_dict[pos]]} for pos in G_POST_LIST] , freq=freq) validate_ds = ListDataset([{'target': use_diff_2018_2019_df.loc['2019-11':'2019-12',pos], 'start': pd.Timestamp("2019-11-01", freq=freq),FieldName.FEAT_STATIC_CAT:[pos_group_dict[pos],meas_dict[pos]]} for pos in G_POST_LIST] , freq=freq)有了模型和数据，剩下的无非就是train了。 predictor = estimator.train(training_data=train_ds)#,validation_data=validate_ds)预测数据与提交结果预测数据时，需要对单个时间序列进行单独预测。我尝试过直接同时预测多个时间序列，但是无法合理解析预测结果。我们需要对不同的站点进行分别预测，另外由于需要预测的时间段包含多个区间，我们需要建立循环进行预测。注意：预测过程也涉及到sampling，所以需要就近设置seed。 supplement_period = [['2020-01-02','2020-01-31'],['2020-03-02','2020-03-31'],['2020-05-02','2020-05-31'],['2020-08-01','2020-08-31']] predict_period = [['2020-02-01','2020-02-21'],['2020-04-01','2020-04-21'],['2020-06-01','2020-06-21'],['2020-09-01','2020-09-21']] pred_2020_all = pd.DataFrame() use_diff_2020_df = pressure_2020_df_diff all_df = pd.DataFrame() print('predicting...') for pos in G_POST_LIST: print(f'predict {pos} pos') for s,p in zip(supplement_period,predict_period): out = pd.DataFrame() mx.random.seed(r_seed) np.random.seed(r_seed) test_data = ListDataset([{'target': use_diff_2020_df.loc[s[0]:s[1],pos], 'start': pd.Timestamp(p[0],freq=freq),FieldName.FEAT_STATIC_CAT:[pos_group_dict[pos],meas_dict[pos]]}],freq=freq) prediction = next(predictor.predict(test_data)) out['DateTime'] = [pd.Timestamp(p[0])+pd.Timedelta(i,unit='H') for i in range(21*24)] yhat_mean = prediction.mean out['pressure_dff'] = yhat_mean.tolist() out['pressure'] = out['pressure_dff'].values+pressure_2020_df.loc[s[0]:s[1],pos].iloc[-21*24:].interpolate().values out['MeasName']=pos all_df = pd.concat([all_df,out],axis=0)关于DeepAR的建模和预测已经结束了，剩下就是针对比赛提交文档进行格式整理。 final_result_for_1st_round = pd.merge(left=to_predict_df_1st[['id','MeasName','DateTime']],right =all_df[['pressure','MeasName','DateTime']],on = ['MeasName','DateTime'],how='left') final_result_for_1st_round[['id','pressure']].to_csv('submit_2021_02_XX.csv',index=False)关于比赛结果本文对于DeepAR的参数进行了阉割，线上分数在6-8之间浮动。如果需要在此基础上进行改善，可以尝试以下方面： 学习超参数的调整，比如batch， learning_rate, epoch等Embedding 超参数调整，Embedding 对于DeepAR是很重要的参数，因为它决定了一个全局模型如何适配多重时序。显式的时间特征和lag特征设置。DeepAR自动计算了lag和时间特征，如果有必要可以自己设置。更多协变量的添加，比如天气信息。本文中不包含天气信息的使用，如果你有好的天气信息使用方法，可以添加。比如温度的分布，以及预测降雨量。我试过直接采用“泄露的”天气数据，会有变化。更多的管网分布信息，以及相关的Embedding模型融合为终极之道。记得改种子！改种子！！改种子！！！代码链接： https://github.com/bingblackbean/water_supply_network_pressure_pred_deepar |
【本文地址】
公司简介
联系我们