本次研究是使用GRU模型和GRU-Attention模型对长时间序列温度数据进行预测拟合，对于这两个模型有兴趣的可以去网上了解一下，
首先是日数据预测，由于日数据存在缺失值需要对缺失值进行填补，
在对存在缺失值的数据中我使用三次样方插值对数据进行处理，其代码如下：
import pandas as pd
import numpy as np
from scipy.interpolate import interp1d
# 假设我们有一组带有缺失值的数据
df = pd.read_csv('your_data.csv')
field = df['wd']
field = np.where(field == -999, np.nan, field)
# 创建一个插值函数，使用三次样条插值方法
interpolator = interp1d(np.arange(len(field))[~np.isnan(field)], field[~np.isnan(field)], kind='cubic')
print(type(np.arange(len(field))[~np.isnan(field)]))
# 遍历数据，对缺失值进行插值
filled_data = np.where(np.isnan(field), interpolator(np.arange(len(field))), field)
# 创建一个 DataFrame 对象
df['wd_1'] = filled_data
# 将结果输出到 Excel 文件
df.to_csv('new_zp.csv', index=False)
对于处理好的数据需要对其构建数据集，本次由于数据量不大，未使用DataLoader等模块，并且归一化：
import numpy as np
import pandas as pd
from torch.utils import data
import torch
from matplotlib import pyplot as plt
import torch.nn as nn
import torchvision
import torchvision.transforms as transforms
from sklearn import preprocessing
from sklearn.metrics import mean_squared_error
from sklearn.metrics import mean_absolute_error
from sklearn.metrics import r2_score
from torchsummary import summary
import torch.cuda
loc = r'your_data_after_prefect.csv'
data_csv = pd.read_csv(loc, header=None)
yt = data_csv.iloc[2:-1, 5]
yt_1 = yt.shift(1)
yt_2 = yt.shift(2)
yt_3 = yt.shift(3)
yt_4 = yt.shift(4)
yt_5 = yt.shift(5)
yt_6 = yt.shift(6)
yt_7 = yt.shift(7)
data = pd.concat([yt, yt_1, yt_2, yt_3, yt_4, yt_5,yt_6,yt_7], axis=1)
data.columns = ['yt', 'yt_1', 'yt_2', 'yt_3', 'yt_4', 'yt_5','yt_6','yt_7']
data.head(10)
data = data.dropna()
x1 = np.array(data['yt_1'], dtype=np.float32)
x1 = torch.tensor(x1)
x2 = torch.tensor(np.array(data['yt_2'], dtype=np.float32))
x3 = torch.tensor(np.array(data['yt_3'], dtype=np.float32))
x4 = torch.tensor(np.array(data['yt_4'], dtype=np.float32))
x5 = torch.tensor(np.array(data['yt_5'], dtype=np.float32))
x6 = torch.tensor(np.array(data['yt_6'], dtype=np.float32))
x7 = torch.tensor(np.array(data['yt_7'], dtype=np.float32))
x = torch.cat((x7,x6,x5,x4,x3,x1,x2), dim=0)
x = x.reshape(7, -1).T
y = np.array(data['yt'], dtype=np.float32)
y = y.reshape(len(y), 1)
y = torch.tensor(y)
scaler_x = preprocessing.MinMaxScaler(feature_range=(-1, 1))
scaler_y = preprocessing.MinMaxScaler(feature_range=(-1, 1))
x = scaler_x.fit_transform(x)
y = scaler_y.fit_transform(y)
train_end = 5479
x_train = torch.tensor(x[0:train_end, ], dtype=torch.float32)
y_train = torch.tensor(y[0:train_end, ], dtype=torch.float32)
x_test = torch.tensor(x[train_end + 1:-1], dtype=torch.float32)
y_test = torch.tensor(y[train_end + 1:-1], dtype=torch.float32)
x_train = x_train.reshape(x_train.shape + (1,))
x_test = x_test.reshape(x_test.shape + (1,))
device = torch.device("cuda:0" if torch.cuda.is_available() else "cpu")
我在这里调用了GPU，实际上我的数据量不大，可以不调用GPU，之后就是构建模型和训练，并且对于前面归一化的数据需要进行反归一化，才能进行指标的计算，代码如下：
seed = 2019
np.random.seed(seed)
class GRUModel(nn.Module):
def __init__(self):
super(GRUModel, self).__init__()
self.gru = nn.GRU(input_size=1, hidden_size=32, num_layers=1)
self.fc1 = nn.Linear(32, 16)
self.act1 = nn.Tanh()
self.fc = nn.Linear(16, 4)
self.act2 = nn.Tanh()
self.dense = nn.Linear(4, 1)
def forward(self, x):
out, _ = self.gru(x)
out = self.fc1(out[:,-1,:])
out = self.fc(self.act1(out))
out = self.dense(self.act2(out))
return out
model = GRUModel()
# 计算参数数量
params_count = sum(p.numel() for p in model.parameters() if p.requires_grad)
# 打印模型和参数数量
print(model)
print("Total params: ", params_count)
torch.save(model, 'qh_gru.pt')
#model = torch.load('gru.pt')
criterion = nn.MSELoss()
optimizer = torch.optim.Adam(model.parameters())
batch_size = 64
num_epochs = 50
for epoch in range(num_epochs):
for i in range(0, len(x_train), batch_size):
batch_x = x_train[i:i + batch_size]
batch_y = y_train[i:i + batch_size]
optimizer.zero_grad()
outputs = model(batch_x)
loss = criterion(outputs, batch_y)
loss.backward()
optimizer.step()
print('Epoch: %d, Loss: %f' % (epoch, float(loss)))
model.eval()
with torch.no_grad():
outputs_train = model(x_train)
score_train = criterion(outputs_train, y_train).item()
with torch.no_grad():
outputs_test = model(x_test)
score_test = criterion(outputs_test, y_test).item()
print('In Train MSE=', round(score_train, 5))
print('In Test MSE=', round(score_test, 5))
y_test = scaler_y.inverse_transform(np.array(y_test).reshape((len(y_test), 1)))
predictions = model(x_test).detach().numpy()
predictions = scaler_y.inverse_transform(np.array(predictions).reshape((len(predictions), 1)))
这里计算一些指标来解释拟合效果，并且将结果制图：
rmse = np.sqrt(mean_squared_error(y_test, predictions))
print("RMSE:", rmse)
mae = mean_absolute_error(y_test, predictions)
print("MAE:", mae)
r2 = r2_score(y_test, predictions)
print("R2:", r2)
def calculate_mape(actual, predicted):
if len(actual) != len(predicted):
raise ValueError("actual and predicted lists must have the same length")
if 0 in actual:
raise ValueError("actual list must not contain zero values")
percentage_errors = [abs((actual[i] - predicted[i]) / actual[i]) for i in range(len(actual))]
mape = sum(percentage_errors) * 100 / len(actual)
return mape
mape = calculate_mape(y_test, predictions)
print(f"MAPE: {mape}")
def calculate_IA(observed, predicted):
numerator = np.sum((observed - predicted) ** 2)
denominator = np.sum((np.abs(predicted - np.mean(observed)) + np.abs(observed - np.mean(observed))) ** 2)
ia = 1 - (numerator / denominator)
return ia
ia_value = calculate_IA(y_test, predictions)
print("IA值:", ia_value)
plt.plot(y_test)
plt.plot(predictions)
plt.legend('target', 'prediction')
plt.show()
以上是使用GRU来处理，下面是用GRU-Attention处理的相同数据：
import numpy as np
import pandas as pd
from torch.utils import data
import torch
from matplotlib import pyplot as plt
import torch.nn as nn
import torchvision
import torchvision.transforms as transforms
from sklearn import preprocessing
from sklearn.metrics import mean_squared_error
from sklearn.metrics import mean_absolute_error
from sklearn.metrics import r2_score
loc = r'your_data.csv'
data_csv = pd.read_csv(loc, header=None)
yt = data_csv.iloc[2:-1, 5]
yt_1 = yt.shift(1)
yt_2 = yt.shift(2)
yt_3 = yt.shift(3)
yt_4 = yt.shift(4)
yt_5 = yt.shift(5)
yt_6 = yt.shift(6)
yt_7 = yt.shift(7)
data = pd.concat([yt, yt_1, yt_2, yt_3, yt_4, yt_5,yt_6,yt_7], axis=1)
data.columns = ['yt', 'yt_1', 'yt_2', 'yt_3', 'yt_4', 'yt_5','yt_6','yt_7']
data.head(10)
data = data.dropna()
x1 = np.array(data['yt_1'], dtype=np.float32)
x1 = torch.tensor(x1)
x2 = torch.tensor(np.array(data['yt_2'], dtype=np.float32))
x3 = torch.tensor(np.array(data['yt_3'], dtype=np.float32))
x4 = torch.tensor(np.array(data['yt_4'], dtype=np.float32))
x5 = torch.tensor(np.array(data['yt_5'], dtype=np.float32))
x6 = torch.tensor(np.array(data['yt_6'], dtype=np.float32))
x7 = torch.tensor(np.array(data['yt_7'], dtype=np.float32))
x = torch.cat((x7,x6,x5,x4,x3,x2,x1), dim=0)
x = x.reshape(7, -1).T
y = np.array(data['yt'], dtype=np.float32)
y = y.reshape(len(y), 1)
y = torch.tensor(y)
scaler_x = preprocessing.MinMaxScaler(feature_range=(-1, 1))
scaler_y = preprocessing.MinMaxScaler(feature_range=(-1, 1))
x = scaler_x.fit_transform(x)
y = scaler_y.fit_transform(y)
train_end = 5479
x_train = torch.tensor(x[0:train_end, ], dtype=torch.float32)
y_train = torch.tensor(y[0:train_end, ], dtype=torch.float32)
x_test = torch.tensor(x[train_end + 1:-1], dtype=torch.float32)
y_test = torch.tensor(y[train_end + 1:-1], dtype=torch.float32)
x_train = x_train.reshape(x_train.shape + (1,))
x_test = x_test.reshape(x_test.shape + (1,))
print(x_train.shape) # x_train.shape=torch.Size([5479, 7, 1])
print(y_train.shape) # y_train.shape=torch.Size([5479, 1])
print(x_test.shape) # x_test.shape=torch.Size([5479, 7, 1])
print(y_test.shape) # y_test.shape=torch.Size([5479, 1])
seed = 2019
np.random.seed(seed)
class GRUAttention(nn.Module):
def __init__(self, input_size, hidden_size, attention_size, output_size):
super(GRUAttention, self).__init__()
self.hidden_size = hidden_size
# 定义 GRU 层
self.gru = nn.GRU(input_size, hidden_size, batch_first=True)
# 定义自注意力层
self.query = nn.Linear(hidden_size, attention_size)
self.key = nn.Linear(hidden_size, attention_size)
self.energy = nn.Linear(attention_size,1)
self.tran = nn.Linear(32, 32)
self.fc1 = nn.Linear(32, 16)
self.act1 = nn.Tanh()
self.fc3 = nn.Linear(16, 4)
self.act2 = nn.Tanh()
self.dense = nn.Linear(4, output_size)
# # 定义全连接层
# self.fc = nn.Linear(hidden_size, output_size)
def forward(self, x):
# GRU 步骤
hidden, _ = self.gru(x)
# 自注意力步骤
query = self.query(hidden)
key = self.key(hidden)
energy = self.energy(torch.tanh(query + key))
attention_weights = torch.softmax(energy, dim=1)
attended_hidden = torch.sum(hidden * attention_weights, dim=1)
# out = self.tran(attended_hidden)
out = self.fc1(attended_hidden)
out = self.act1(out)
out = self.fc3(out)
# 全连接层
out = self.dense(self.act2(out))
return out
input_size = 1
hidden_size = 32
attention_size = 32
output_size = 1
model = GRUAttention(input_size, hidden_size, attention_size, output_size)
criterion = nn.MSELoss()
optimizer = torch.optim.Adam(model.parameters())
batch_size =128
num_epochs = 50
for epoch in range(num_epochs):
for i in range(0, len(x_train), batch_size):
batch_x = x_train[i:i + batch_size]
batch_y = y_train[i:i + batch_size]
optimizer.zero_grad()
outputs = model(batch_x)
loss = criterion(outputs, batch_y)
loss.backward()
optimizer.step()
print('Epoch: %d, Loss: %f' % (epoch, float(loss)))
model.eval()
with torch.no_grad():
outputs_train = model(x_train)
score_train = criterion(outputs_train, y_train).item()
with torch.no_grad():
outputs_test = model(x_test)
score_test = criterion(outputs_test, y_test).item()
print('In Train MSE=', round(score_train, 5))
print('In Test MSE=', round(score_test, 5))
y_test = scaler_y.inverse_transform(np.array(y_test).reshape((len(y_test), 1)))
predictions = model(x_test).detach().numpy()
predictions = scaler_y.inverse_transform(np.array(predictions).reshape((len(predictions), 1)))
rmse = np.sqrt(mean_squared_error(y_test, predictions))
print("RMSE:", rmse)
mae = mean_absolute_error(y_test, predictions)
print("MAE:", mae)
r2 = r2_score(y_test, predictions)
print("R2:", r2)
def calculate_mape(actual, predicted):
if len(actual) != len(predicted):
raise ValueError("actual and predicted lists must have the same length")
if 0 in actual:
raise ValueError("actual list must not contain zero values")
percentage_errors = [abs((actual[i] - predicted[i]) / actual[i]) for i in range(len(actual))]
mape = sum(percentage_errors) *100 / len(actual)
return mape
mape = calculate_mape(y_test, predictions)
print("MAPE:", mape)
def calculate_IA(observed, predicted):
numerator = np.sum((observed - predicted) ** 2)
denominator = np.sum((np.abs(predicted - np.mean(observed)) + np.abs(observed - np.mean(observed))) ** 2)
ia = 1 - (numerator / denominator)
return ia
ia_value = calculate_IA(y_test, predictions)
print("IA值:", ia_value)
plt.plot(y_test)
plt.plot(predictions)
plt.legend('target', 'prediction')
plt.show()
当然在对GRU-Attention构建时也可以使用另外一种方法，自己定义Attention类：
class Attention(nn.Module):
def __init__(self,embed_dim):
super(Attention,self).__init__()
self.query = nn.Linear(embed_dim,embed_dim)
self.key = nn.Linear(embed_dim,embed_dim)
self.value = nn.Linear(embed_dim,embed_dim)
self.act = nn.Tanh()
def forward(self,x):
q = self.act(self.query(x))
k = self.act(self.key(x))
v = self.act(self.value(x))
attn_weights = torch.matmul(q,k.transpose(1,2))
attn_weights = nn.functional.softmax(attn_weights,dim=-1)
attended_values = torch.matmul(attn_weights,v)
return attended_values
class GRUModel(nn.Module):
def __init__(self):
super(GRUModel, self).__init__()
self.gru = nn.GRU(input_size=1, hidden_size=32, num_layers=1)
self.attention = Attention(32)
self.fc1 = nn.Linear(32, 16)
self.act1 = nn.Tanh()
self.fc = nn.Linear(16, 4)
self.act2 = nn.Tanh()
self.dense = nn.Linear(4, 1)
def forward(self, x):
print(x.shape)
out, _ = self.gru(x)
out = self.attention(out)
out = self.fc1(out[:,-1,:])
out = self.fc(self.act1(out))
out = self.dense(self.act2(out))
return out
model = GRUModel()
--------------------------------------------------------------------------------------------------------------------------------
以上是针对时间分辨率为天的日数据，下面是对时间周期为4小时的数据的GRU模型：
import numpy as np
import pandas as pd
from torch.utils import data
import torch
from matplotlib import pyplot as plt
import torch.nn as nn
import torchvision
import torchvision.transforms as transforms
from sklearn import preprocessing
from sklearn.metrics import mean_squared_error
from sklearn.metrics import mean_absolute_error
from sklearn.metrics import r2_score
loc = '周期为4小时数据.csv'
data_csv = pd.read_csv(loc, header=None)
yt = data_csv.iloc[1:-1, 1]
yt_1 = yt.shift(1)
yt_2 = yt.shift(2)
yt_3 = yt.shift(3)
yt_4 = yt.shift(4)
yt_5 = yt.shift(5)
yt_6 = yt.shift(6)
yt_7 = yt.shift(7)
data = pd.concat([yt, yt_1, yt_2, yt_3, yt_4, yt_5,yt_6,yt_7], axis=1)
data.columns = ['yt', 'yt_1', 'yt_2', 'yt_3', 'yt_4', 'yt_5','yt_6','yt_7']
data.head(10)
data = data.dropna()
x1 = np.array(data['yt_1'], dtype=np.float32)
x1 = torch.tensor(x1)
x2 = torch.tensor(np.array(data['yt_2'], dtype=np.float32))
x3 = torch.tensor(np.array(data['yt_3'], dtype=np.float32))
x4 = torch.tensor(np.array(data['yt_4'], dtype=np.float32))
x5 = torch.tensor(np.array(data['yt_5'], dtype=np.float32))
x6 = torch.tensor(np.array(data['yt_6'], dtype=np.float32))
x7 = torch.tensor(np.array(data['yt_7'], dtype=np.float32))
x = torch.cat((x7,x6,x5,x4,x3,x2,x1), dim=0)
x = x.reshape(7, -1).T
y = np.array(data['yt'], dtype=np.float32)
y = y.reshape(len(y), 1)
y = torch.tensor(y)
scaler_x = preprocessing.MinMaxScaler(feature_range=(-1, 1))
scaler_y = preprocessing.MinMaxScaler(feature_range=(-1, 1))
x = scaler_x.fit_transform(x)
y = scaler_y.fit_transform(y)
train_end = 4466
x_train = torch.tensor(x[0:train_end, ], dtype=torch.float32)
y_train = torch.tensor(y[0:train_end, ], dtype=torch.float32)
x_test = torch.tensor(x[train_end + 1:-1], dtype=torch.float32)
y_test = torch.tensor(y[train_end + 1:-1], dtype=torch.float32)
x_train = x_train.reshape(x_train.shape + (1,))
x_test = x_test.reshape(x_test.shape + (1,))
seed = 2019
np.random.seed(seed)
class GRUModel(nn.Module):
def __init__(self):
super(GRUModel, self).__init__()
self.gru = nn.GRU(input_size=1, hidden_size=32, num_layers=1)
self.fc1 = nn.Linear(32, 16)
self.act1 = nn.Tanh()
self.fc = nn.Linear(16, 4)
self.act2 = nn.Tanh()
self.dense = nn.Linear(4, 1)
def forward(self, x):
out, _ = self.gru(x)
out = self.fc1(out[:,-1,:])
out = self.fc(self.act1(out))
out = self.dense(self.act2(out))
return out
# model = GRUModel()
# torch.save(model, 'gru2.pt')
model = torch.load('gru2.pt')
criterion = nn.MSELoss()
optimizer = torch.optim.Adam(model.parameters())
batch_size = 48
num_epochs = 50
for epoch in range(num_epochs):
for i in range(0, len(x_train), batch_size):
batch_x = x_train[i:i + batch_size]
batch_y = y_train[i:i + batch_size]
optimizer.zero_grad()
outputs = model(batch_x)
loss = criterion(outputs, batch_y)
loss.backward()
optimizer.step()
print('Epoch: %d, Loss: %f' % (epoch, float(loss)))
model.eval()
with torch.no_grad():
outputs_train = model(x_train)
score_train = criterion(outputs_train, y_train).item()
with torch.no_grad():
outputs_test = model(x_test)
score_test = criterion(outputs_test, y_test).item()
print('In Train MSE=', round(score_train, 5))
print('In Test MSE=', round(score_test, 5))
y_test = scaler_y.inverse_transform(np.array(y_test).reshape((len(y_test), 1)))
predictions = model(x_test).detach().numpy()
predictions = scaler_y.inverse_transform(np.array(predictions).reshape((len(predictions), 1)))
rmse = np.sqrt(mean_squared_error(y_test, predictions))
print("RMSE:", rmse)
mae = mean_absolute_error(y_test, predictions)
print("MAE:", mae)
r2 = r2_score(y_test, predictions)
print("R2:", r2)
def calculate_mape(actual, predicted):
if len(actual) != len(predicted):
raise ValueError("actual and predicted lists must have the same length")
if 0 in actual:
raise ValueError("actual list must not contain zero values")
percentage_errors = [abs((actual[i] - predicted[i]) / actual[i]) for i in range(len(actual))]
mape = sum(percentage_errors) *100 / len(actual)
return mape
mape = calculate_mape(y_test, predictions)
print("MAPE:", mape)
def calculate_IA(observed, predicted):
numerator = np.sum((observed - predicted) ** 2)
denominator = np.sum((np.abs(predicted - np.mean(observed)) + np.abs(observed - np.mean(observed))) ** 2)
ia = 1 - (numerator / denominator)
return ia
ia_value = calculate_IA(y_test, predictions)
print("IA值:", ia_value)
plt.plot(y_test)
plt.plot(predictions)
plt.legend('target', 'prediction')
plt.show()
GRU-Attention模型：
import numpy as np
import pandas as pd
from torch.utils import data
import torch
from matplotlib import pyplot as plt
import torch.nn as nn
import torchvision
import torchvision.transforms as transforms
from sklearn import preprocessing
from sklearn.metrics import mean_squared_error
from sklearn.metrics import mean_absolute_error
from sklearn.metrics import r2_score
loc = 'D:/研一作业_全/dataset/zhangpin (2).csv'
data_csv = pd.read_csv(loc, header=None)
yt = data_csv.iloc[1:-1, 1]
yt_1 = yt.shift(1)
yt_2 = yt.shift(2)
yt_3 = yt.shift(3)
yt_4 = yt.shift(4)
yt_5 = yt.shift(5)
yt_6 = yt.shift(6)
yt_7 = yt.shift(7)
data = pd.concat([yt, yt_1, yt_2, yt_3, yt_4, yt_5,yt_6,yt_7], axis=1)
data.columns = ['yt', 'yt_1', 'yt_2', 'yt_3', 'yt_4', 'yt_5','yt_6','yt_7']
data.head(10)
data = data.dropna()
x1 = np.array(data['yt_1'], dtype=np.float32)
x1 = torch.tensor(x1)
x2 = torch.tensor(np.array(data['yt_2'], dtype=np.float32))
x3 = torch.tensor(np.array(data['yt_3'], dtype=np.float32))
x4 = torch.tensor(np.array(data['yt_4'], dtype=np.float32))
x5 = torch.tensor(np.array(data['yt_5'], dtype=np.float32))
x6 = torch.tensor(np.array(data['yt_6'], dtype=np.float32))
x7 = torch.tensor(np.array(data['yt_7'], dtype=np.float32))
x = torch.cat((x7,x6,x5,x4,x3,x2,x1), dim=0)
x = x.reshape(7, -1).T
y = np.array(data['yt'], dtype=np.float32)
y = y.reshape(len(y), 1)
y = torch.tensor(y)
scaler_x = preprocessing.MinMaxScaler(feature_range=(-1, 1))
scaler_y = preprocessing.MinMaxScaler(feature_range=(-1, 1))
x = scaler_x.fit_transform(x)
y = scaler_y.fit_transform(y)
train_end = 5241
x_train = torch.tensor(x[0:train_end, ], dtype=torch.float32)
y_train = torch.tensor(y[0:train_end, ], dtype=torch.float32)
x_test = torch.tensor(x[train_end + 1:-1], dtype=torch.float32)
y_test = torch.tensor(y[train_end + 1:-1], dtype=torch.float32)
x_train = x_train.reshape(x_train.shape + (1,))
x_test = x_test.reshape(x_test.shape + (1,))
seed = 2019
np.random.seed(seed)
class GRUAttention(nn.Module):
def __init__(self, input_size, hidden_size, attention_size, output_size):
super(GRUAttention, self).__init__()
self.hidden_size = hidden_size
# 定义 GRU 层
self.gru = nn.GRU(input_size, hidden_size, batch_first=True)
# 定义自注意力层
self.query = nn.Linear(hidden_size, attention_size)
self.key = nn.Linear(hidden_size, attention_size)
self.energy = nn.Linear(attention_size,1)
self.tran = nn.Linear(32, 32)
self.fc1 = nn.Linear(32, 16)
self.act1 = nn.Tanh()
self.fc3 = nn.Linear(16, 4)
self.act2 = nn.Tanh()
self.dense = nn.Linear(4, output_size)
# # 定义全连接层
# self.fc = nn.Linear(hidden_size, output_size)
def forward(self, x):
# GRU 步骤
hidden, _ = self.gru(x)
# 自注意力步骤
query = self.query(hidden)
key = self.key(hidden)
energy = self.energy(torch.tanh(query + key))
attention_weights = torch.softmax(energy, dim=1)
attended_hidden = torch.sum(hidden * attention_weights, dim=1)
# out = self.tran(attended_hidden)
out = self.fc1(attended_hidden)
out = self.act1(out)
out = self.fc3(out)
# 全连接层
out = self.dense(self.act2(out))
return out
input_size = 1
hidden_size = 32
attention_size = 32
output_size = 1
# model = GRUAttention(input_size, hidden_size, attention_size, output_size)
# class GRUAttentionModel(nn.Module):
# def __init__(self):
# super(GRUAttentionModel, self).__init__()
# self.gru = nn.GRU(input_size=1, hidden_size=32, batch_first = True)
# self.attention =nn.Linear(32,32)
# self.fc1 = nn.Linear(32, 16)
# self.act1 = nn.Tanh()
# self.fc = nn.Linear(16, 4)
# self.act2 = nn.Tanh()
# self.dense = nn.Linear(4, 1)
# def forward(self, x):
# out, hidden = self.gru(x)
# print(out.shape)
# print(hidden.shape)
# attention_weights = torch.softmax(self.attention(out), dim=1)
# out = torch.sum(attention_weights * out, dim=1)
# out = self.fc1(self.act1(out))
# out = self.fc(self.act2(out))
# out = self.dense(out)
# return out
# model = GRUAttentionModel()
# class GRUWithAttention(nn.Module):
# def __init__(self, input_dim, hidden_dim, output_dim):
# super(GRUWithAttention, self).__init__()
# self.gru = nn.GRU(input_dim, hidden_dim, bidirectional=True)
# self.attention = nn.Linear(hidden_dim * 2, 1)
# self.fc = nn.Linear(hidden_dim * 2, output_dim)
# def forward(self, x):
# output, _ = self.gru(x)
# attention_weights = torch.softmax(self.attention(output), dim=0)
# context = torch.sum(output * attention_weights, dim=0)
# prediction = self.fc(context)
# return prediction
# model = GRUAttentionModel()
# 数据准备
# input_dim = 1 # 输入特征维度
# hidden_dim = 32 # GRU的隐藏层维度
# output_dim = 1 # 输出特征维度
# torch.save(model, 'gru-attention.pt')
model = torch.load('gru-attention1.pt')
criterion = nn.MSELoss()
optimizer = torch.optim.Adam(model.parameters())
batch_size = 48
num_epochs = 100
for epoch in range(num_epochs):
for i in range(0, len(x_train), batch_size):
batch_x = x_train[i:i + batch_size]
batch_y = y_train[i:i + batch_size]
optimizer.zero_grad()
outputs = model(batch_x)
loss = criterion(outputs, batch_y)
loss.backward()
optimizer.step()
print('Epoch: %d, Loss: %f' % (epoch, float(loss)))
model.eval()
with torch.no_grad():
outputs_train = model(x_train)
score_train = criterion(outputs_train, y_train).item()
with torch.no_grad():
outputs_test = model(x_test)
score_test = criterion(outputs_test, y_test).item()
print('In Train MSE=', round(score_train, 5))
print('In Test MSE=', round(score_test, 5))
y_test = scaler_y.inverse_transform(np.array(y_test).reshape((len(y_test), 1)))
predictions = model(x_test).detach().numpy()
predictions = scaler_y.inverse_transform(np.array(predictions).reshape((len(predictions), 1)))
rmse = np.sqrt(mean_squared_error(y_test, predictions))
print("RMSE:", rmse)
mae = mean_absolute_error(y_test, predictions)
print("MAE:", mae)
r2 = r2_score(y_test, predictions)
print("R2:", r2)
def calculate_mape(actual, predicted):
if len(actual) != len(predicted):
raise ValueError("actual and predicted lists must have the same length")
if 0 in actual:
raise ValueError("actual list must not contain zero values")
percentage_errors = [abs((actual[i] - predicted[i]) / actual[i]) for i in range(len(actual))]
mape = sum(percentage_errors) * 100 / len(actual)
return mape
mape = calculate_mape(y_test, predictions)
print("MAPE:", mape)
def calculate_IA(observed, predicted):
numerator = np.sum((observed - predicted) ** 2)
denominator = np.sum((np.abs(predicted - np.mean(observed)) + np.abs(observed - np.mean(observed))) ** 2)
ia = 1 - (numerator / denominator)
return ia
ia_value = calculate_IA(y_test, predictions)
print("IA值:", ia_value)
plt.plot(y_test)
plt.plot(predictions)
plt.legend('target', 'prediction')
plt.show()
y_test = np.ravel(y_test)
predictions = np.ravel(predictions)
fit = np.polyfit(y_test, predictions, 1)
fit_line = np.polyval(fit, y_test)
plt.scatter(y_test, predictions, label='Data')
plt.plot(y_test, fit_line, color='red', label='Fit Line')
plt.xlabel("real_value")
plt.ylabel("prediction_value")
# 设置图标题和显示R方
plt.title(f"Scatter Map (R-squared = {r2:.4f})")
# 显示图例
plt.legend()
# 显示图形
plt.show()
--------------------------------------------------------------------------------------------------------------------------------
以下是对单变量时间分辨率为1小时的GRU模型预测:
import numpy as np
import pandas as pd
from torch.utils import data
import torch
from matplotlib import pyplot as plt
import torch.nn as nn
import torchvision
import torchvision.transforms as transforms
from sklearn import preprocessing
from sklearn.metrics import mean_squared_error
from sklearn.metrics import mean_absolute_error
from sklearn.metrics import r2_score
loc = 'D:/研一作业_全/dataset/zhangpin (2).csv'
data_csv = pd.read_csv(loc, header=None)
yt = data_csv.iloc[1:-1, 1]
yt_1 = yt.shift(1)
yt_2 = yt.shift(2)
yt_3 = yt.shift(3)
yt_4 = yt.shift(4)
yt_5 = yt.shift(5)
yt_6 = yt.shift(6)
yt_7 = yt.shift(7)
data = pd.concat([yt, yt_1, yt_2, yt_3, yt_4, yt_5,yt_6,yt_7], axis=1)
data.columns = ['yt', 'yt_1', 'yt_2', 'yt_3', 'yt_4', 'yt_5','yt_6','yt_7']
data.head(10)
data = data.dropna()
x1 = np.array(data['yt_1'], dtype=np.float32)
x1 = torch.tensor(x1)
x2 = torch.tensor(np.array(data['yt_2'], dtype=np.float32))
x3 = torch.tensor(np.array(data['yt_3'], dtype=np.float32))
x4 = torch.tensor(np.array(data['yt_4'], dtype=np.float32))
x5 = torch.tensor(np.array(data['yt_5'], dtype=np.float32))
x6 = torch.tensor(np.array(data['yt_6'], dtype=np.float32))
x7 = torch.tensor(np.array(data['yt_7'], dtype=np.float32))
x = torch.cat((x7,x6,x5,x4,x3,x2,x1), dim=0)
x = x.reshape(7, -1).T
y = np.array(data['yt'], dtype=np.float32)
y = y.reshape(len(y), 1)
y = torch.tensor(y)
scaler_x = preprocessing.MinMaxScaler(feature_range=(-1, 1))
scaler_y = preprocessing.MinMaxScaler(feature_range=(-1, 1))
x = scaler_x.fit_transform(x)
y = scaler_y.fit_transform(y)
train_end = 5241
x_train = torch.tensor(x[0:train_end, ], dtype=torch.float32)
y_train = torch.tensor(y[0:train_end, ], dtype=torch.float32)
x_test = torch.tensor(x[train_end + 1:-1], dtype=torch.float32)
y_test = torch.tensor(y[train_end + 1:-1], dtype=torch.float32)
x_train = x_train.reshape(x_train.shape + (1,))
x_test = x_test.reshape(x_test.shape + (1,))
seed = 2019
np.random.seed(seed)
class GRUModel(nn.Module):
def __init__(self):
super(GRUModel, self).__init__()
self.gru = nn.GRU(input_size=1, hidden_size=32, num_layers=1)
self.fc1 = nn.Linear(32, 16)
self.act1 = nn.Tanh()
self.fc = nn.Linear(16, 4)
self.act2 = nn.Tanh()
self.dense = nn.Linear(4, 1)
def forward(self, x):
out, _ = self.gru(x)
out = self.fc1(out[:,-1,:])
out = self.fc(self.act1(out))
out = self.dense(self.act2(out))
return out
model = GRUModel()
torch.save(model, 'gru_tem.pt')
# model = torch.load('gru2.pt')
criterion = nn.MSELoss()
optimizer = torch.optim.Adam(model.parameters())
batch_size = 48
num_epochs = 50
for epoch in range(num_epochs):
for i in range(0, len(x_train), batch_size):
batch_x = x_train[i:i + batch_size]
batch_y = y_train[i:i + batch_size]
optimizer.zero_grad()
outputs = model(batch_x)
loss = criterion(outputs, batch_y)
loss.backward()
optimizer.step()
print('Epoch: %d, Loss: %f' % (epoch, float(loss)))
model.eval()
with torch.no_grad():
outputs_train = model(x_train)
score_train = criterion(outputs_train, y_train).item()
with torch.no_grad():
outputs_test = model(x_test)
score_test = criterion(outputs_test, y_test).item()
print('In Train MSE=', round(score_train, 5))
print('In Test MSE=', round(score_test, 5))
y_test = scaler_y.inverse_transform(np.array(y_test).reshape((len(y_test), 1)))
predictions = model(x_test).detach().numpy()
predictions = scaler_y.inverse_transform(np.array(predictions).reshape((len(predictions), 1)))
y_test = np.ravel(y_test)
predictions = np.ravel(predictions)
fit = np.polyfit(y_test, predictions, 1)
fit_line = np.polyval(fit, y_test)
plt.scatter(y_test, predictions, label='Data')
plt.plot(y_test, fit_line, color='red', label='Fit Line')
plt.xlabel("real_value")
plt.ylabel("prediction_value")
# 设置图标题和显示R方
plt.title(f"Scatter Map (R-squared = {r2:.4f})")
# 显示图例
plt.legend()
# 显示图形
plt.show()
---------------------------------------------------------------------------------------------------------------------------------
多变量GRU模型预测：
import numpy as np
import pandas as pd
from torch.utils import data
import torch
from matplotlib import pyplot as plt
import torch.nn as nn
import torchvision
import torchvision.transforms as transforms
from sklearn import preprocessing
from sklearn.metrics import mean_squared_error
from sklearn.metrics import mean_absolute_error
from sklearn.metrics import r2_score
loc = 'data.csv'
data_csv = pd.read_csv(loc).values
data_csv = data_csv[0:,1:]
print(data_csv.shape)
data_csv
x_t = data_csv[0:,1:]
ss_X = preprocessing.MinMaxScaler().fit(x_t)
x_ss_t = ss_X.transform(x_t)
print(x_ss_t)
y_t = data_csv[0:,0]
y_t = y_t.reshape(len(y_t),1)
ss_Y = preprocessing.MinMaxScaler().fit(y_t)
y_ss_t = ss_Y.transform(y_t)
print(y_ss_t)
data_csv_ss = np.concatenate((y_ss_t,x_ss_t),axis=1)
print(data_csv_ss.shape)
print(data_csv_ss)
def split_data(data,timestep,input_size):
dataX = []
dataY = []
for index in range(len(data) - timestep):
dataX.append(data[index+1:index+timestep+1][:,1:])
dataY.append(data[index][0])
dataX = np.array(dataX)
dataY = np.array(dataY)
print(dataX.shape)
print(dataX)
print(dataY.shape)
train_size = 5232
x_train = dataX[:train_size,:,:].reshape(-1,timestep,input_size)
y_train = dataY[:train_size].reshape(-1,1)
x_test = dataX[train_size:,:,:].reshape(-1,timestep,input_size)
y_test = dataY[train_size:].reshape(-1,1)
return [x_train,y_train,x_test,y_test]
timestep = 24
input_size = 6
x_train,y_train,x_test,y_test = split_data(data_csv_ss,timestep,input_size)
x_train = torch.Tensor(x_train)
y_train = torch.Tensor(y_train)
x_test = torch.Tensor(x_test)
y_test = torch.Tensor(y_test)
class GRUModel(nn.Module):
def __init__(self):
super(GRUModel, self).__init__()
self.gru = nn.GRU(input_size=6, hidden_size=32, num_layers=1)
self.fc1 = nn.Linear(32, 16)
self.act1 = nn.Tanh()
self.fc = nn.Linear(16, 4)
self.act2 = nn.Tanh()
self.dense = nn.Linear(4, 1)
def forward(self, x):
out, _ = self.gru(x)
out = self.fc1(out[:,-1,:])
out = self.fc(self.act1(out))
out = self.dense(self.act2(out))
return out
model = GRUModel()
# torch.save(model, 'gru2.pt')
# model = torch.load('gru2.pt')
criterion = nn.MSELoss()
optimizer = torch.optim.Adam(model.parameters())
batch_size = 48
num_epochs = 50
for epoch in range(num_epochs):
for i in range(0, len(x_train), batch_size):
batch_x = x_train[i:i + batch_size]
batch_y = y_train[i:i + batch_size]
optimizer.zero_grad()
outputs = model(batch_x)
loss = criterion(outputs, batch_y)
loss.backward()
optimizer.step()
print('Epoch: %d, Loss: %f' % (epoch, float(loss)))
model.eval()
with torch.no_grad():
outputs_train = model(x_train)
score_train = criterion(outputs_train, y_train).item()
with torch.no_grad():
outputs_test = model(x_test)
score_test = criterion(outputs_test, y_test).item()
print('In Train MSE=', round(score_train, 5))
print('In Test MSE=', round(score_test, 5))
y_test = ss_Y.inverse_transform(np.array(y_test).reshape((len(y_test), 1)))
predictions = model(x_test).detach().numpy()
predictions = ss_Y.inverse_transform(np.array(predictions).reshape((len(predictions), 1)))
rmse = np.sqrt(mean_squared_error(y_test, predictions))
print("RMSE:", rmse)
mae = mean_absolute_error(y_test, predictions)
print("MAE:", mae)
r2 = r2_score(y_test, predictions)
print("R2:", r2)
def calculate_mape(actual, predicted):
if len(actual) != len(predicted):
raise ValueError("actual and predicted lists must have the same length")
if 0 in actual:
raise ValueError("actual list must not contain zero values")
percentage_errors = [abs((actual[i] - predicted[i]) / actual[i]) for i in range(len(actual))]
mape = sum(percentage_errors) *100 / len(actual)
return mape
mape = calculate_mape(y_test, predictions)
print("MAPE:", mape)
def calculate_IA(observed, predicted):
numerator = np.sum((observed - predicted) ** 2)
denominator = np.sum((np.abs(predicted - np.mean(observed)) + np.abs(observed - np.mean(observed))) ** 2)
ia = 1 - (numerator / denominator)
return ia
ia_value = calculate_IA(y_test, predictions)
print("IA值:", ia_value)
plt.plot(y_test)
plt.plot(predictions)
plt.legend('target', 'prediction')
plt.show()
多变量GRU-Attention：
import numpy as np
import pandas as pd
from torch.utils import data
import torch
from matplotlib import pyplot as plt
import torch.nn as nn
import torchvision
import torchvision.transforms as transforms
from sklearn import preprocessing
from sklearn.metrics import mean_squared_error
from sklearn.metrics import mean_absolute_error
from sklearn.metrics import r2_score
loc = 'data.csv'
data_csv = pd.read_csv(loc).values
data_csv = data_csv[0:,1:]
data_csv
x_t = data_csv[0:,1:]
ss_X = preprocessing.MinMaxScaler().fit(x_t)
x_ss_t = ss_X.transform(x_t)
y_t = data_csv[0:,0]
y_t = y_t.reshape(len(y_t),1)
ss_Y = preprocessing.MinMaxScaler().fit(y_t)
y_ss_t = ss_Y.transform(y_t)
data_csv_ss = np.concatenate((y_ss_t,x_ss_t),axis=1)
def split_data(data,timestep,input_size):
dataX = []
dataY = []
for index in range(len(data) - timestep):
dataX.append(data[index+1:index+timestep+1][:,1:])
dataY.append(data[index][0])
dataX = np.array(dataX)
dataY = np.array(dataY)
print(dataX.shape)
print(dataX)
print(dataY.shape)
train_size = 5232
x_train = dataX[:train_size,:,:].reshape(-1,timestep,input_size)
y_train = dataY[:train_size].reshape(-1,1)
x_test = dataX[train_size:,:,:].reshape(-1,timestep,input_size)
y_test = dataY[train_size:].reshape(-1,1)
return [x_train,y_train,x_test,y_test]
timestep = 24
input_size = 6
x_train,y_train,x_test,y_test = split_data(data_csv_ss,timestep,input_size)
x_train = torch.Tensor(x_train)
y_train = torch.Tensor(y_train)
x_test = torch.Tensor(x_test)
y_test = torch.Tensor(y_test)
class GRUAttention(nn.Module):
def __init__(self, input_size, hidden_size, attention_size, output_size):
super(GRUAttention, self).__init__()
self.hidden_size = hidden_size
# 定义 GRU 层
self.gru = nn.GRU(input_size, hidden_size, batch_first=True)
# 定义自注意力层
self.query = nn.Linear(hidden_size, attention_size)
self.key = nn.Linear(hidden_size, attention_size)
self.energy = nn.Linear(attention_size,1)
self.fc1 = nn.Linear(32, 16)
self.act1 = nn.Tanh()
self.fc3 = nn.Linear(16, 4)
self.act2 = nn.Tanh()
self.dense = nn.Linear(4, output_size)
def forward(self, x):
# GRU 步骤
hidden, _ = self.gru(x)
# 自注意力步骤
query = self.query(hidden)
key = self.key(hidden)
energy = self.energy(torch.tanh(query + key))
attention_weights = torch.softmax(energy, dim=1)
attended_hidden = torch.sum(hidden * attention_weights, dim=1)
out = self.fc1(attended_hidden)
out = self.act1(out)
out = self.fc3(out)
# 全连接层
out = self.dense(self.act2(out))
return out
input_size = 6
hidden_size = 32
attention_size = 32
output_size = 1
model = GRUAttention(input_size, hidden_size, attention_size, output_size)
criterion = nn.MSELoss()
optimizer = torch.optim.Adam(model.parameters())
batch_size = 48
num_epochs = 50
for epoch in range(num_epochs):
for i in range(0, len(x_train), batch_size):
batch_x = x_train[i:i + batch_size]
batch_y = y_train[i:i + batch_size]
optimizer.zero_grad()
outputs = model(batch_x)
loss = criterion(outputs, batch_y)
loss.backward()
optimizer.step()
print('Epoch: %d, Loss: %f' % (epoch, float(loss)))
model.eval()
with torch.no_grad():
outputs_train = model(x_train)
score_train = criterion(outputs_train, y_train).item()
with torch.no_grad():
outputs_test = model(x_test)
score_test = criterion(outputs_test, y_test).item()
print('In Train MSE=', round(score_train, 5))
print('In Test MSE=', round(score_test, 5))
y_test = ss_Y.inverse_transform(np.array(y_test).reshape((len(y_test), 1)))
predictions = model(x_test).detach().numpy()
predictions = ss_Y.inverse_transform(np.array(predictions).reshape((len(predictions), 1)))
rmse = np.sqrt(mean_squared_error(y_test, predictions))
print("RMSE:", rmse)
mae = mean_absolute_error(y_test, predictions)
print("MAE:", mae)
r2 = r2_score(y_test, predictions)
print("R2:", r2)
def calculate_mape(actual, predicted):
if len(actual) != len(predicted):
raise ValueError("actual and predicted lists must have the same length")
if 0 in actual:
raise ValueError("actual list must not contain zero values")
percentage_errors = [abs((actual[i] - predicted[i]) / actual[i]) for i in range(len(actual))]
mape = sum(percentage_errors) * 100 / len(actual)
return mape
mape = calculate_mape(y_test, predictions)
print("MAPE:", mape)
def calculate_IA(observed, predicted):
numerator = np.sum((observed - predicted) ** 2)
denominator = np.sum((np.abs(predicted - np.mean(observed)) + np.abs(observed - np.mean(observed))) ** 2)
ia = 1 - (numerator / denominator)
return ia
ia_value = calculate_IA(y_test, predictions)
print("IA值:", ia_value)
plt.plot(y_test)
plt.plot(predictions)
plt.legend('target', 'prediction')
plt.show()
y_test = np.ravel(y_test)
predictions = np.ravel(predictions)
fit = np.polyfit(y_test, predictions, 1)
fit_line = np.polyval(fit, y_test)
plt.scatter(y_test, predictions, label='Data')
plt.plot(y_test, fit_line, color='red', label='Fit Line')
plt.xlabel("real_value")
plt.ylabel("prediction_value")
# 设置图标题和显示R方
plt.title(f"Scatter Map (R-squared = {r2:.4f})")
# 显示图例
plt.legend()
# 显示图形
plt.show()