经典时间序列模型 DeepAR 预测股票趋势
经典时间序列模型 DeepAR 预测股票趋势
DeepAR 预测算法
在时间序列预测领域，根据历史数据预测未来值的能力至关重要。因此，先进的机器学习算法已变得不可或缺。DeepAR 是一种功能强大的算法，它在处理复杂的时间模式和生成准确预测方面备受关注。特别适用于需要同时预测多个相关时间序列的场景，使其成为金融、电子商务和供应链管理等各个领域的重要工具。本文将讨论 DeepAR 预测算法，并将其用于时间序列预测。
什么是 DeepAR？
亚马逊公司开发了一种最先进的概率预测算法，称为Deep Autoregressive或DeepAR。它是专门用于捕捉未来预测中固有不确定性的深度学习模型。与传统预测方法不同，DeepAR提供了未来值的概率分布，使决策者能够评估可能的结果范围，并做出更明智的决策。
DeepAR 的工作原则
DeepAR 的一些主要工作原理如下：
- 自回归架构：DeepAR 采用自回归神经网络架构，每个时间步的预测都取决于历史观测数据和模型自身过去预测的组合。这使该算法能够捕捉时间序列数据中更复杂的依赖关系，使其善于处理具有复杂模式和趋势的序列。
- 嵌入分类特征：DeepAR 可以无缝整合与时间序列数据相关的分类特征信息。这是通过使用嵌入来实现的，嵌入可将分类变量转换为连续向量。这些特征的加入增强了模型辨别数据中的模式和关系的能力，尤其是在外部因素影响时间序列的情况下。
- 时间关注机制：为了有效权衡历史数据中不同时间点的重要性，DeepAR 采用了时间关注机制。该机制可使模型关注时间序列的相关部分，并根据数据中存在的模式动态调整其关注度。
- 分位数损失训练：DeepAR 采用概率方法进行训练，以最小化分位数损失。这意味着对模型进行了优化，以生成预测区间，代表未来可能值的范围及相关置信度。这种概率框架在决策过程中尤为重要，它能让决策者对与预测相关的不确定性有细致入微的了解。
DeepAR 基础使用
安装所需模块
首先，我们将为运行时安装所有必要的 Python 模块。
!pip install gluonts
!pip install --upgrade mxnet==1.6.0
!pip install "gluonts[torch]"
导入所需库函数
现在，我们将导入所有需要的 Python 库，如 NumPy、Pandas、Matplotlib 和 OS 等。
import numpy as np
import pandas as pd
import os
import matplotlib as mpl
import matplotlib.pyplot as plt
from gluonts.torch.model.deepar import DeepAREstimator
from gluonts.dataset.common import ListDataset
from gluonts.dataset.field_names import FieldName
from gluonts.evaluation.backtest import make_evaluation_predictions
from tqdm.autonotebook import tqdm
from gluonts.evaluation import Evaluator
from typing import Dict
数据集加载和预处理
现在我们将加载两个简单的数据集，因为 DeepAR 主要用于多时间序列预测。然后，我们将对数据集进行切分，使其平均分布并合并。然后将合并后的数据集分成训练集和测试集。
df_comed = pd.read_csv("/content/COMED_hourly.csv", parse_dates=True) # https://www.kaggle.com/datasets/robikscube/hourly-energy-consumption/data?select=COMED_hourly.csv
df_dom = pd.read_csv("/content/DOM_hourly.csv", parse_dates=True) # https://www.kaggle.com/datasets/robikscube/hourly-energy-consumption/data?select=DOM_hourly.csv
# 数据集获取：关注公众号：数据STUDIO 后台回复 云朵君
# Slicing the datasets to make equal lengh data
df_comed = df_comed.loc[df_comed["Datetime"]
> '2011-12-31'].reset_index(drop=True)
df_dom = df_dom.loc[df_dom["Datetime"] > '2011-12-31'].reset_index(drop=True)
df_comed = df_comed.T
df_comed.columns = df_comed.iloc[0]
df_comed = df_comed.drop(df_comed.index[0])
df_comed['Station_Name'] = "COMED"
df_comed = df_comed.reset_index(drop=True)
df_dom = df_dom.T
df_dom.columns = df_dom.iloc[0]
df_dom = df_dom.drop(df_dom.index[0])
df_dom['Station_Name'] = "DOM"
df_dom = df_dom.reset_index(drop=True)
df_all = pd.concat([df_comed, df_dom], axis=0)
df_all = df_all.set_index("Station_Name")
df_all = df_all.reset_index()
ts_code = df_all['Station_Name'].astype('category').cat.codes.values
freq = "1H" # rate at which dataset is sampled
start_train = pd.Timestamp("2011-12-31 01:00:00", freq=freq)
start_test = pd.Timestamp("2016-06-10 18:00:00", freq=freq)
prediction_lentgh = 24 * 1 # Our prediction Length is 1 Day
# training and testing sets split
df_train = df_all.iloc[:, 1:40000].values
df_test = df_all.iloc[:, 40000:].values
定义模型
现在，我们将通过定义 DeepAR 估计模型的各种超参数来初始化该模型，具体参数如下：
- freq：该参数定义时间序列数据的频率。它表示时间序列在一个周期内的时间步数。我们的数据每天都有观测结果，因此频率由 freq 变量决定。
- context_length：上下文长度，该参数设置了模型用于学习历史数据中的模式和依赖关系的时间步数。这里设置为 (24 * 5)，表示模型回顾的时间段相当于 5 天（假设每个时间步长对应一个小时）。
- prediction_length：预测长度，该参数指定模型应生成多远的预测。它决定了预测范围的长度。
- cardinality：该参数是一个列表，表示数据集中每个分类特征的类别数。
- num_layer：层数，它决定了神经网络架构的层数。在我们的例子中，模型配置为 2 层。
- dropout_rate：这是一种正则化技术，有助于防止过度拟合。它表示在训练过程中丢弃的输入单元的比例。0.25 表示每次更新时，25% 的输入单元将被随机设置为零。
- trainer_kwargs：这是一个字典，包含训练过程中的附加参数。在我们的例子中，它包括 “max_epochs”：16，用于设置最大训练历元数。一个历元是对整个训练数据集的一次完整遍历。
estimator = DeepAREstimator(freq=freq,
# context length is number of time steps will look back(5 days in a week)
context_length=24 * 5,
prediction_length=prediction_lentgh,
cardinality=[1],
num_layers=2,
dropout_rate=0.25,
trainer_kwargs={'max_epochs': 16}
)
数据集准备
对于每个深度学习模型，都需要根据模型的输入类型准备原始数据集。在这里，我们也需要重新定义用于训练和测试的原始数据集。
train_ds = ListDataset([
{
FieldName.TARGET: target,
FieldName.START: start_train,
FieldName.FEAT_STATIC_CAT: fsc
}
for (target, fsc) in zip(df_train,
ts_code.reshape(-1, 1))
], freq=freq)
test_ds = ListDataset([
{
FieldName.TARGET: target,
FieldName.START: start_test,
FieldName.FEAT_STATIC_CAT: fsc
}
for (target, fsc) in zip(df_test,
ts_code.reshape(-1, 1))
], freq=freq)
模型训练
准备通过刚刚准备好的训练数据集来训练 DeepAR 估算器。
predictor = estimator.train(training_data=train_ds, num_workers=4)
预测
训练完成后，我们需要绘制预测值和实际值的对比图。DeepAR 支持置信区间，我们可以使用预测值的中位数作为最接近的预测。在图中，实际预测值和预测值之间的差距代表了置信区间（如 75% 或 90% 的预测置信度等）。
forecast_it, ts_it = make_evaluation_predictions(
dataset=test_ds,
predictor=predictor,
num_samples=100,
)
print("Gathering time series conditioning values ...")
tss = list(tqdm(ts_it, total=len(df_test)))
print("Gathering time series predictions ...")
forecasts = list(tqdm(forecast_it, total=len(df_test)))
def plot_prob_forecasts(ts_entry, forecast_entry):
plot_length = prediction_lentgh
prediction_intervals = (0.5, 0.8)
legend = ["observations", "median prediction"] + \
[f"{k*100}% prediction interval" for k in prediction_intervals][::-1]
fig, ax = plt.subplots(1, 1, figsize=(10, 7))
ts_entry[-plot_length:].plot(ax=ax, color='blue', label='observations')
# Extract the median prediction
median = np.median(forecast_entry, axis=0)
ax.plot(ts_entry.index[-plot_length:], median[-plot_length:],
color='orange', label='median prediction')
# Extract the prediction intervals if available
if len(forecast_entry) > 1:
lower, upper = np.percentile(forecast_entry, q=[(1 - k) * 100 / 2 for k in prediction_intervals], axis=0), np.percentile(
forecast_entry, q=[(1 + k) * 100 / 2 for k in prediction_intervals], axis=0)
# Ensure lower and upper are 1-D arrays
lower, upper = lower[-plot_length:], upper[-plot_length:]
plt.grid(which="both")
plt.legend(legend, loc="upper left")
plt.show()
for i in tqdm(range(2)):
ts_entry = tss[i]
forecast_entry = np.array(forecasts[i].samples)
plot_prob_forecasts(ts_entry, forecast_entry)
图中重复的次数越多，它就越准确。
评估
DeepAR 本身提供多种性能指标，如量化损失、MAPE 等。
evaluator = Evaluator(quantiles=[0.1, 0.5, 0.9])
agg_metrics, item_metrics = evaluator(iter(tss), iter(forecasts), num_series=len(df_test))
item_metrics
item_id forecast_start MSE abs_error abs_target_sum abs_target_mean seasonal_error MASE MAPE sMAPE num_masked_target_values ND MSIS QuantileLoss[0.1] Coverage[0.1] QuantileLoss[0.5] Coverage[0.5] QuantileLoss[0.9] Coverage[0.9]
0 None 2018-06-19 23:00 318517.437500 11284.473633 297067.0 12377.791667 777.236948 0.604946 0.038610 0.037733 0.0 0.037986 3.896573 3385.887109 0.041667 11284.473633 0.583333 6079.214258 1.000000
1 None 2018-06-19 23:00 835246.333333 16977.113281 414494.0 17270.583333 909.647006 0.777642 0.040786 0.042074 0.0 0.040959 4.620998 9115.450977 0.000000 16977.114258 0.166667 6252.767773 0.708333
因此，我们可以看到，每种情况下的误差都非常小。
我们可以得出结论，DeepAR 是一种有效的深度学习模型，可用于预测问题。不过，要想获得更准确的预测，还需要对模型进行更多的历时训练。
DeepAR 预测股票
DeepAR 建立了一个全局模型，适用于多步骤预测、多序列预测，并能提供具有不确定性的预测。我们用沃尔玛商店每周销售额的多个时间序列测试了 DeepAR 的预测能力。
在金融市场预测领域，股票价格时间序列分析一直是一个具有挑战性和价值的研究方向。传统的预测模型如ARIMA等往往难以很好地捕捉股价的非线性动态和复杂波动模式。而深度学习模型由于其强大的非线性拟合能力和对多元时序数据的建模优势，为股价预测提供了新的解决方案。
DeepAR是亚马逊推出的一种基于神经网络的概率性时间序列预测模型，它专门针对具有类似趋势和周期性的多元时间序列数据进行了优化，非常适合对一篮子股票的联合价格序列进行建模和预测。DeepAR的一个关键优势是，它能够同时输出未来多个时间步长的概率预测区间，而不仅仅是点估计值。
这种概率预测能力对于金融风险管理至关重要。股票价格的波动性往往会受到宏观经济指标、企业运营状况、市场情绪等多种因素的影响，存在较大的不确定性。如果我们能够量化这种不确定性，就可以更好地评估潜在的下行风险，制定相应的风险对冲策略。
此外，通过捕捉长期的价格趋势和周期性模式，DeepAR的多期概率预测结果也能为投资者制定复杂的交易策略提供依据，如趋势跟踪、均值回归或动量交易等。相比于简单的点预测，概率预测区间能更全面地反映未来的风险收益情况。
# Installation of gluonTS
!pip uninstall numpy # Downgrade numpy to 1.23
!pip3 install mxnet-mkl==1.6.0 numpy==1.23.1
!pip install gluonts==0.14.2
!pip install yfinance
在本研究中，我们选取了以下大盘股 2020 年至 2024 年的每日价格。
import yfinance as yf
data = yf.download(["SPY","MSFT","AMZN","GOOG","PGR"], start="2020-01-01", end="2024-12-31")
data = data[[('Adj Close', 'AMZN'),('Adj Close', 'GOOG'),('Adj Close', 'MSFT'),('Adj Close', 'PGR'),('Adj Close', 'SPY')]]
data.columns = ['AMZN','GOOG','MSFT','PGR','SPY']
img
绘制它们的曲线，观察它们的动向。
%matplotlib inline
import pandas as pd
import numpy as np
from matplotlib import pyplot as plt
data.plot(figsize=(12, 4))
plt.legend(loc='upper left')
plt.title("Stocks")
然后，我们将给出的时间序列分为训练数据和测试数据。我们将把 85% 的数据作为 “实时”训练数据，其余 15% 作为 “非实时”测试数据。
print("The time series has", data.shape[0], "data points")
len_train = int(data.shape[0] * 0.85)
train_data = data[0:len_train]
test_data = data[len_train:]
[train_data.shape, test_data.shape]
时间序列有 1039 个数据点。(883,5) 和 (156,5)。
时间序列数据应包含开始日期、目标数据和数据频率。GluonTS 数据格式要求包含这三个要素。在下面的代码中，我们将数据集转换为与 GluonTS 兼容的格式。代码会计算最小日期作为起始日期，并将列作为目标。
# Prepare the data for deepAR format
from gluonts.dataset.common import ListDataset
from gluonts.dataset.field_names import FieldName
def to_deepar_format(dataframe, freq):
start_index = dataframe.index.min()
data = [{
FieldName.START: start_index,
FieldName.TARGET: dataframe[c].values,
}
for c in dataframe.columns]
print(data[0])
return ListDataset(data, freq=freq)
train_data_lds = to_deepar_format(train_data, 'D')
test_data_lds = to_deepar_format(test_data, 'D')
建模代码的行数相当简洁。下面的代码使用了 DeepAREstimator() 函数，我将进行相应的解释。
# api: https://ts.gluon.ai/stable/api/gluonts/gluonts.mx.model.deepar.html
# paper: Salinas, David, Valentin Flunkert, and Jan Gasthaus. “DeepAR: Probabilistic forecasting with autoregressive recurrent networks.” arXiv preprint arXiv:1704.04110 (2017).
from gluonts.mx.model.deepar import DeepAREstimator
#from gluonts.torch.model.deepar.estimator import DeepAREstimator
from gluonts.mx.trainer import Trainer
prediction_length = 30
context_length = 30
num_cells = 100
num_layers = 8
epochs= 150
freq="D" # Our data is daily
estimator = DeepAREstimator(freq=freq,
context_length=context_length,
prediction_length=prediction_length,
num_layers=num_layers,
num_cells=num_cells,
cardinality=[1],
trainer=Trainer(epochs=epochs))
predictor = estimator.train(train_data_lds)
完成建模任务后，我们可以进行预测。在验证时间序列模型时，我们会使用测试数据集进行预测，并评估预测性能。这个过程可以封装在“make_evaluation_prediction”函数中。该函数预测测试数据的最后一个窗口，并评估模型的性能。它首先从测试数据中提取长度为“prediction_length”的最后一个窗口，然后使用剩余的测试数据生成预测结果。最后，将预测值与最终窗口内的实际值进行比较。
from gluonts.evaluation.backtest import make_evaluation_predictions
forecast_it, ts_it = make_evaluation_predictions(
dataset=test_data_lds,
predictor=predictor,
)
tss = list(ts_it)
forecasts = list(forecast_it)
我们需要提供概率预测。GluonTS 使用概率分布生成概率预测，捕捉未来预测中的不确定性，让用户能够量化结果范围。GluonTS 默认使用高斯分布进行蒙特卡罗模拟。GluonTS 估算高斯分布的平均值 (μ) 和标准偏差 (σ)，前者表示点预测，后者表示预测的不确定性水平。GluonTS 还可以使用其他概率分布，如学生 t 分布、负二项分布和伽马分布。
将预测结果可视化。
for k in range(len(forecasts)):
fig, ax1 = plt.subplots(1, 1, figsize=(12, 4))
forecasts[k].plot(ax = ax1)
tss[k].plot(ax = ax1)
ax1.get_legend().remove()
plt.grid(which="both")
plt.title("Stock" + data.columns[k])
plt.show()
预测值与实际值基本一致。虽然预测结果很有希望，但我们仍然需要对预测进行仔细审查，因为股市价格总是波动不定。
下面的代码将生成评估指标。为了节省空间，这里我们不输出指标。
from gluonts.evaluation import Evaluator
evaluator = Evaluator()
agg_metrics, item_metrics = evaluator(iter(tss), iter(forecasts), num_series=len(test_data_lds))
import json
print(json.dumps(agg_metrics, indent=4))