序列模型是指一類特別設(shè)計(jì)來處理序列數(shù)據(jù)的神經(jīng)網(wǎng)絡(luò)模型。序列數(shù)據(jù)指的是數(shù)據(jù)中的每個(gè)元素都有先后順序，比如時(shí)間序列數(shù)據(jù)（股票價(jià)格、天氣變化等）、自然語言文本（句子中的單詞順序）、語音信號等。

1 統(tǒng)計(jì)工具

前面介紹了卷積神經(jīng)網(wǎng)絡(luò)架構(gòu)，但是在處理序列數(shù)據(jù)時(shí)，需要新的神經(jīng)網(wǎng)絡(luò)架構(gòu)，下面以股票價(jià)格為例：

我們用$x_t$表示價(jià)格，其中$t$表示時(shí)間步(time step)，也就是在時(shí)間步$t$時(shí)觀察到的價(jià)格$x_t$，我們通過下列公式來表示我們預(yù)測第$t$日的價(jià)格：
$$x_t～P(x_t\mid x_{t?1},\dots ,x_1).$$
即，在已知 $1$ 到 $t?1$ 的價(jià)格，求第 $t$ 天的價(jià)格的概率分布。

1.1 自回歸模型

為了實(shí)現(xiàn)這個(gè)預(yù)測，可以使用自回歸模型：假設(shè)當(dāng)前值 $y_t$ 與過去的值$y_{t?1},y_{t?2},...y_{t?p}$ 之間存在線性關(guān)系，一般形式為 ：

其中：

大致分為兩種策略：
①自回歸模型： 假設(shè)在現(xiàn)實(shí)情況下相當(dāng)長的序列$x_{t?1},\dots ,x_1$可能是沒價(jià)值的，因此我們只需要滿足某個(gè)長度為$\tau$的時(shí)間跨度， 即使用觀測序列$x_{t?1},\dots ,x_{t?\tau }$。也就是說過長的歷史序列可能并不必要，因此只需要關(guān)注較短的一段歷史數(shù)據(jù)即可。因?yàn)橹豢紤]觀測值本身，所以叫自回歸模型

②隱變量自回歸模型： 即保留一些對過去觀測的總結(jié) $h_t$，這個(gè)“總結(jié)”是無法直觀解釋的，它是模型自助捕捉的內(nèi)部關(guān)系依賴，然后同時(shí)更新預(yù)測值${\hat{x}}_t$和$h_t$，即變?yōu)橄铝惺阶?#xff1a;$${\hat{x}}_t=P(x_t\mid h_t)和h_t=g(h_{t?1},x_{t?1})$$由于$h_t$$h_t$從未被觀測到，這類模型也被稱為隱變量自回歸模型，這里做出一個(gè)假設(shè)，即序列本身的動(dòng)力學(xué)（數(shù)據(jù)隨時(shí)間演變的方式）不會(huì)改變，意味著我們可以用過去的數(shù)據(jù)來推斷未來的趨勢，因?yàn)槲覀兗俣ɑ镜膭?dòng)態(tài)規(guī)則是一致的。因此，整個(gè)序列的概率值可以表示為一系列條件概率的乘積：
$$P(x_1,\dots ,x_T)=\prod _{t=1}^{T}P(x_t\mid x_{t?1},\dots ,x_1).$$
注意，如果我們處理的是離散的對象（如單詞）， 而不是連續(xù)的數(shù)字，則上述的考慮仍然有效。我們需要使用分類器而不是回歸模型來估計(jì)

1.2 馬爾可夫模型

馬爾可夫條件： 在自回歸模型中，如果 $t$ 時(shí)刻的數(shù)值，只與 $x_{t?1},\dots ,x_{t?\tau }$ 有關(guān)，而不是整個(gè)過去的序列，則稱其滿足馬爾可夫條件。

如果 $\tau =1$ ，則得到了一個(gè)一階馬爾可夫模型，$P(x)$由如下公式表示：
$$P(x_1,\dots ,x_T)=\prod _{t=1}^{T}P(x_t\mid x_{t?1})\text{?當(dāng)?}P(x_1\mid x_0)=P(x_1).$$
若當(dāng)假設(shè) $x_t$ 僅是離散值時(shí)，可以使用動(dòng)態(tài)規(guī)劃可以沿著馬爾可夫鏈精確地計(jì)算結(jié)果。

2 訓(xùn)練、預(yù)測

下面我們將用一個(gè)正弦函數(shù)和一些噪聲生成1000個(gè)序列數(shù)據(jù)，并使用自回歸模型進(jìn)行訓(xùn)練和預(yù)測

2.1 生成數(shù)據(jù)

import torch
from torch import nn
import matplotlib.pyplot as plt
import numpy as np
from torch.utils.data import TensorDataset, DataLoaderT=1000
time=torch.arange(1,T+1,dtype=torch.float32)
x=torch.sin(0.01*time)+torch.normal(0,0.2,(T,))
# 繪制折線圖
plt.plot(time, x)
plt.xlabel('Time')
plt.ylabel('Value')
plt.title('Time Series Data')
plt.show()

運(yùn)行結(jié)果

2.2 構(gòu)造數(shù)據(jù)集

我們是準(zhǔn)備用$y_t=F(X_t)$，其中$X_t=[x_{t?\tau },\dots ,x_{t?1}]$，我們這里假設(shè)$\tau =4$，即用前四個(gè)數(shù)據(jù)來預(yù)測下一個(gè)數(shù)據(jù)，但是這樣的話，前 $4$ 個(gè)數(shù)據(jù)就沒有歷史樣本去描述了，一般的做法是直接舍棄，或者用零序列去填充。

這里我們用600個(gè)數(shù)據(jù)進(jìn)行訓(xùn)練，剩余的用于預(yù)測。

構(gòu)建數(shù)據(jù)集時(shí)，使用滑動(dòng)窗口去構(gòu)建：

# 構(gòu)造數(shù)據(jù)集
tau=4# 初始化特征矩陣，因?yàn)榍八膫€(gè)值就是當(dāng)前值的特征
features = torch.zeros((T - tau, tau))
for i in range(T - tau): # 用滑動(dòng)窗口進(jìn)行構(gòu)建features[i,:]=x[i:tau+i]
print('features:',features.shape)
print(features[:5])labels = x[tau:].reshape((-1, 1))
print('labels:',labels.shape)
print(labels[:5])batch_size = 16
n = 600  # 只有前600個(gè)樣本用于訓(xùn)練
dataset = TensorDataset(features[:n], labels[:n])
train_iter = DataLoader(dataset, batch_size=batch_size, shuffle=False)

運(yùn)行結(jié)果

2.3 構(gòu)造模型進(jìn)行訓(xùn)練

# 構(gòu)造模型
def init_weights(m):if type(m)==nn.Linear:nn.init.xavier_uniform_(m.weight)def net():net=nn.Sequential(nn.Linear(4,10),nn.ReLU(),nn.Linear(10,1))net.apply(init_weights)return net# 評估模型在給定數(shù)據(jù)集上的損失
def evaluate_loss(net, data_iter, loss):"""評估模型在給定數(shù)據(jù)集上的損失"""net.eval()  # 設(shè)置模型為評估模式total_loss = 0.0with torch.no_grad():  # 不計(jì)算梯度for X, y in data_iter:y_hat = net(X)l = loss(y_hat, y)total_loss += l.sum().item()  # 計(jì)算總損失net.train()  # 恢復(fù)模型為訓(xùn)練模式return total_loss / len(data_iter.dataset)loss=nn.MSELoss(reduction='none')
lr=0.01
net=net()
optimzer=torch.optim.Adam(net.parameters(),lr)
loss_sum=[]
num_epoch=20
def train(net,num_epoch,train_iter,loss,optimzer,loss_sum):for epoch in range(num_epoch):for x,y in train_iter:optimzer.zero_grad()l=loss(net(x),y)l.sum().backward()optimzer.step()temp=evaluate_loss(net,train_iter,loss)loss_sum.append(temp)print("epoch ",epoch+1,": loss:",temp)train(net,num_epoch,train_iter,loss,optimzer,loss_sum)# 繪制折線圖
plt.plot(range(num_epoch), loss_sum)
plt.xlabel('epoch')
plt.ylabel('loss')
plt.show()

運(yùn)行結(jié)果

2.4 預(yù)測

# 使用模型進(jìn)行預(yù)測
def predict(net, data_iter):net.eval()  # 設(shè)置模型為評估模式predictions = []with torch.no_grad():  # 不計(jì)算梯度for X, y in data_iter:y_hat = net(X)predictions.extend(y_hat.numpy())net.train()  # 恢復(fù)模型為訓(xùn)練模式return predictions# 獲取測試集的預(yù)測結(jié)果
predictions = predict(net, test_iter)# 繪制預(yù)測結(jié)果與真實(shí)值的對比圖
true_values = labels[n:].numpy()
plt.plot(true_values, label='True Values')
plt.plot(predictions, label='Predictions')
plt.xlabel('Time')
plt.ylabel('Value')
plt.legend()
plt.show()

運(yùn)行結(jié)果

2.5 多步預(yù)測

# 多步預(yù)測
def multistep_predict(net, data_iter, steps):net.eval()  multistep_predictions = []with torch.no_grad():  for X, y in data_iter:current_features = X.clone()for _ in range(steps):'''在每一步中，模型用 current_features 作為輸入，并預(yù)測出 y_hat。然后將 y_hat 拼接到 current_features 的末尾，同時(shí)移除 current_features 的第一個(gè)時(shí)間步，保持輸入長度不變。這樣，y_hat 成為下一步的輸入'''y_hat = net(current_features)current_features = torch.cat([current_features[:, 1:], y_hat], dim=1)multistep_predictions.extend(y_hat.numpy())net.train() return multistep_predictions# 獲取測試集的不同步數(shù)的多步預(yù)測結(jié)果
steps = [4, 16, 32]
multistep_predictions = {step: multistep_predict(net, test_iter, step) for step in steps}# 繪制結(jié)果
plt.figure(figsize=(12, 6))  # 設(shè)置圖像的寬度為12英寸，高度為6英寸
plt.plot(true_values, label='True Values')
plt.plot(ones_predictions, label='1-step Predictions')
for step, preds in multistep_predictions.items():plt.plot(preds, label=f'{step}-step Predictions')
plt.xlabel('Time')
plt.ylabel('Value')
plt.legend()
plt.show()

上述的多步預(yù)測是迭代預(yù)測法，即用自己預(yù)測數(shù)據(jù)再去預(yù)測下一個(gè)數(shù)據(jù)，另一種方法是seq2seq，后面在介紹，迭代預(yù)測法如下圖所示：