赞
踩
在这个笔记本中,我们展示了如何使用darts来使用TCNs的示例。
# 导入fix_pythonpath_if_working_locally函数
from utils import fix_pythonpath_if_working_locally
# 调用fix_pythonpath_if_working_locally函数,用于修复本地工作时的Python路径问题
fix_pythonpath_if_working_locally()
# 导入所需的库 %matplotlib inline import numpy as np import pandas as pd import matplotlib.pyplot as plt from pytorch_lightning.callbacks import TQDMProgressBar from darts import TimeSeries, concatenate from darts.utils.callbacks import TFMProgressBar from darts.models import TCNModel, RNNModel from darts.dataprocessing.transformers import Scaler from darts.utils.timeseries_generation import datetime_attribute_timeseries from darts.metrics import mape, r2_score from darts.utils.missing_values import fill_missing_values from darts.datasets import AirPassengersDataset, SunspotsDataset, EnergyDataset # 忽略警告信息 import warnings warnings.filterwarnings("ignore") # 禁用日志记录 import logging logging.disable(logging.CRITICAL) # 定义函数generate_torch_kwargs,返回一个字典,用于设置torch模型的参数 def generate_torch_kwargs(): # 在CPU上运行torch模型,并且除了训练阶段以外,禁用所有模型阶段的进度条 return { "pl_trainer_kwargs": { "accelerator": "cpu", "callbacks": [TFMProgressBar(enable_train_bar_only=True)], } }
# 导入AirPassengersDataset数据集 ts = AirPassengersDataset().load() # 将数据集分为训练集和验证集 train, val = ts.split_after(pd.Timestamp("19580801")) # 创建一个Scaler对象 scaler = Scaler() # 对训练集进行标准化处理 train_scaled = scaler.fit_transform(train) # 对验证集进行标准化处理 val_scaled = scaler.transform(val) # 对整个数据集进行标准化处理 ts_scaled = scaler.transform(ts) # 使用月份作为协变量,进行one-hot编码,生成时间序列 month_series = datetime_attribute_timeseries(ts, attribute="month", one_hot=True)
在下面的单元格中,我们将构建一个TCNModel
。稍后,当我们训练它时,模型将将其内部输入宽度初始化为2,因为我们有一个维度用于时间序列,另一个维度用于协变量月份时间序列(尽管对我们来说这是透明的)。
此外,由于我们稍后想要模拟一个历史预测,预测未来6个月,我们必须确保我们的TCNModel
实例的output_chunk_length
参数至少为6。
如果我们不指定num_layers
参数,它将自动选择为确保每个输出条目的完整历史覆盖的最小数量,即每个输出将依赖于整个输入。
# 定义模型名称 model_name = "TCN_air" # 创建TCN模型 model_air = TCNModel( input_chunk_length=13, # 输入数据的时间步长 output_chunk_length=12, # 输出数据的时间步长 n_epochs=500, # 训练的轮数 dropout=0.1, # dropout的比例 dilation_base=2, # 扩张卷积的基数 weight_norm=True, # 是否使用权重归一化 kernel_size=5, # 卷积核的大小 num_filters=3, # 卷积核的数量 random_state=0, # 随机数种子 save_checkpoints=True, # 是否保存检查点 model_name=model_name, # 模型名称 force_reset=True, # 是否强制重置模型 **generate_torch_kwargs() # 传递给PyTorch的参数 )
# 使用fit()方法训练模型
# 参数:
# series: 训练数据的时间序列,已经进行了缩放处理
# past_covariates: 训练数据的历史协变量,即与时间序列相关的其他特征
# val_series: 验证数据的时间序列,已经进行了缩放处理
# val_past_covariates: 验证数据的历史协变量,即与时间序列相关的其他特征
model_air.fit(
series=train_scaled,
past_covariates=month_series,
val_series=val_scaled,
val_past_covariates=month_series,
)
Training: 0it [00:00, ?it/s]
TCNModel(kernel_size=5, num_filters=3, num_layers=None, dilation_base=2, weight_norm=True, dropout=0.1, input_chunk_length=13, output_chunk_length=12, n_epochs=500, random_state=0, save_checkpoints=True, model_name=TCN_air, force_reset=True, pl_trainer_kwargs={'accelerator': 'cpu', 'callbacks': [<darts.utils.callbacks.TFMProgressBar object at 0x2b00e2a40>]})
我们将从在验证集上表现最佳的检查点中加载模型。
# 加载TCNModel模型,并从检查点中加载最佳模型
model_air = TCNModel.load_from_checkpoint(model_name=model_name, best=True)
让我们看看我们得到了什么样的6个月的预测:
# 使用model_air模型进行历史预测 # 参数: # series:时间序列数据,即待预测的数据 # past_covariates:过去的协变量数据,即过去的相关数据 # start:预测的起始时间 # forecast_horizon:预测的时间范围 # retrain:是否重新训练模型 # verbose:是否显示详细信息 backtest = model_air.historical_forecasts( series=ts_scaled, past_covariates=month_series, start=val_scaled.start_time(), forecast_horizon=6, retrain=False, verbose=True, )
0%| | 0/23 [00:00<?, ?it/s]
# 绘制时间序列数据的缩放版本并标记为“actual”
ts_scaled.plot(label="actual")
# 绘制回测数据并标记为“backtest (H=6)”
backtest.plot(label="backtest (H=6)")
<Axes: xlabel='time'>
让我们尝试在一个更复杂且更大的数据集上使用TCNModel
。
# 加载太阳黑子数据集 series_sunspot = SunspotsDataset().load() # 将数据集分割为训练集和验证集 # 分割点为1940年10月1日 train, val = series_sunspot.split_after(pd.Timestamp("19401001")) # 创建一个缩放器对象 scaler = Scaler() # 对训练集进行缩放处理 train_sp_transformed = scaler.fit_transform(train) # 对验证集进行缩放处理 val_sp_transformed = scaler.transform(val) # 对整个数据集进行缩放处理 series_sp_transformed = scaler.transform(series_sunspot)
# 定义模型名称为"TCN_sun" model_name = "TCN_sun" # 创建TCNModel对象,传入以下参数: # input_chunk_length:输入序列的长度为250 # output_chunk_length:输出序列的长度为36 # n_epochs:训练的轮数为100 # dropout:dropout的比例为0 # dilation_base:扩张系数的基数为2 # weight_norm:是否使用权重归一化为True # kernel_size:卷积核的大小为3 # num_filters:卷积核的数量为6 # nr_epochs_val_period:每隔1轮进行一次验证 # random_state:随机种子为0 # save_checkpoints:是否保存检查点为True # model_name:模型名称为"TCN_sun" # force_reset:是否强制重置为True # **generate_torch_kwargs():使用generate_torch_kwargs()函数生成的其他参数 model_sun = TCNModel( input_chunk_length=250, output_chunk_length=36, n_epochs=100, dropout=0, dilation_base=2, weight_norm=True, kernel_size=3, num_filters=6, nr_epochs_val_period=1, random_state=0, save_checkpoints=True, model_name=model_name, force_reset=True, **generate_torch_kwargs() )
# 使用model_sun模型对train_sp_transformed进行拟合,并使用val_sp_transformed作为验证集进行验证。
Training: 0it [00:00, ?it/s]
TCNModel(kernel_size=3, num_filters=6, num_layers=None, dilation_base=2, weight_norm=True, dropout=0, input_chunk_length=250, output_chunk_length=36, n_epochs=100, nr_epochs_val_period=1, random_state=0, save_checkpoints=True, model_name=TCN_sun, force_reset=True, pl_trainer_kwargs={'accelerator': 'cpu', 'callbacks': [<darts.utils.callbacks.TFMProgressBar object at 0x176bcc790>]})
# 加载TCNModel模型,从指定的checkpoint文件中读取模型参数
model_sun = TCNModel.load_from_checkpoint(model_name=model_name, best=True)
# 导入模型model_sun的historical_forecasts函数 # 该函数用于生成历史预测结果 # 参数series_sp_transformed为时间序列数据 # 参数start为开始时间,使用val_sp_transformed的起始时间 # 参数forecast_horizon为预测时间范围,为12个时间步长 # 参数stride为步长,为12,表示每隔12个时间步长进行一次预测 # 参数last_points_only为False,表示返回所有预测结果 # 参数retrain为False,表示不重新训练模型 # 参数verbose为True,表示输出详细信息 backtest_sp = model_sun.historical_forecasts( series_sp_transformed, start=val_sp_transformed.start_time(), forecast_horizon=12, stride=12, last_points_only=False, retrain=False, verbose=True, ) # 将backtest_sp中的预测结果连接起来 backtest_sp = concatenate(backtest_sp)
0%| | 0/43 [00:00<?, ?it/s]
# 绘制val_sp_transformed的图像,标签为"actual"
val_sp_transformed.plot(label="actual")
# 绘制backtest_sp的图像,标签为"backtest (H=12)"
backtest_sp.plot(label="backtest (H=12)")
# 添加图例
plt.legend()
<matplotlib.legend.Legend at 0x288b5d720>
# 导入EnergyDataset类,通过load()方法加载数据,再通过pd_dataframe()方法将数据转换为pandas的DataFrame格式 df3 = EnergyDataset().load().pd_dataframe() # 将日期和时间拆分,只保留日期,然后按日期分组,计算每一天的平均值,并重置索引 df3_day_avg = ( df3.groupby(df3.index.astype(str).str.split(" ").str[0]).mean().reset_index() ) # 将DataFrame转换为TimeSeries格式,并填充缺失值 series_en = fill_missing_values( TimeSeries.from_dataframe( df3_day_avg, "time", ["generation hydro run-of-river and poundage"] ), "auto", ) # 创建训练集和验证集 train_en, val_en = series_en.split_after(pd.Timestamp("20170901")) # 对数据进行缩放 scaler_en = Scaler() train_en_transformed = scaler_en.fit_transform(train_en) val_en_transformed = scaler_en.transform(val_en) series_en_transformed = scaler_en.transform(series_en) # 将日期作为协变量添加到时间序列中,并进行one-hot编码 day_series = datetime_attribute_timeseries( series_en_transformed, attribute="day", one_hot=True ) # 绘制训练集和验证集的折线图 plt.figure(figsize=(10, 3)) train_en_transformed.plot(label="train") val_en_transformed.plot(label="validation")
<Axes: xlabel='time'>
# 定义模型名称 model_name = "TCN_energy" # 定义TCN模型 model_en = TCNModel( # 输入数据的时间步长 input_chunk_length=365, # 输出数据的时间步长 output_chunk_length=7, # 训练轮数 n_epochs=50, # dropout概率 dropout=0.2, # 空洞卷积的基数 dilation_base=2, # 是否使用权重归一化 weight_norm=True, # 卷积核大小 kernel_size=5, # 卷积核数量 num_filters=8, # 验证集评估周期 nr_epochs_val_period=1, # 随机种子 random_state=0, # 是否保存检查点 save_checkpoints=True, # 模型名称 model_name=model_name, # 是否强制重置模型 force_reset=True, # 生成torch参数 **generate_torch_kwargs() )
# 使用fit()方法训练模型
# 参数:
# series: 训练数据的时间序列
# past_covariates: 过去的协变量数据
# val_series: 验证数据的时间序列
# val_past_covariates: 验证数据的过去协变量数据
model_en.fit(
series=train_en_transformed, # 训练数据的时间序列
past_covariates=day_series, # 过去的协变量数据
val_series=val_en_transformed, # 验证数据的时间序列
val_past_covariates=day_series, # 验证数据的过去协变量数据
)
Training: 0it [00:00, ?it/s]
TCNModel(kernel_size=5, num_filters=8, num_layers=None, dilation_base=2, weight_norm=True, dropout=0.2, input_chunk_length=365, output_chunk_length=7, n_epochs=50, nr_epochs_val_period=1, random_state=0, save_checkpoints=True, model_name=TCN_energy, force_reset=True, pl_trainer_kwargs={'accelerator': 'cpu', 'callbacks': [<darts.utils.callbacks.TFMProgressBar object at 0x2bc722440>]})
# 加载 TCNModel 模型,从指定的模型名称中加载最佳模型
model_en = TCNModel.load_from_checkpoint(model_name=model_name, best=True)
# 对英文模型进行历史预测
backtest_en = model_en.historical_forecasts(
series=series_en_transformed, # 输入的时间序列数据
past_covariates=day_series, # 过去的协变量数据
start=val_en_transformed.start_time(), # 预测开始时间
forecast_horizon=7, # 预测的时间跨度
stride=7, # 步长
last_points_only=False, # 是否只返回最后一个时间点的预测结果
retrain=False, # 是否重新训练模型
verbose=True, # 是否打印详细信息
)
# 将预测结果拼接成一个大的时间序列
backtest_en = concatenate(backtest_en)
0%| | 0/69 [00:00<?, ?it/s]
# 设置图形的大小为10x6
plt.figure(figsize=(10, 6))
# 绘制val_en_transformed的折线图,并设置标签为"actual"
val_en_transformed.plot(label="actual")
# 绘制backtest_en的折线图,并设置标签为"backtest (H=7)"
backtest_en.plot(label="backtest (H=7)")
# 添加图例
plt.legend()
<matplotlib.legend.Legend at 0x2bc3c3970>
Copyright © 2003-2013 www.wpsshop.cn 版权所有,并保留所有权利。