import pandas as pd import numpy as np from sklearn.preprocessing import MinMaxScaler from keras.models import Sequential from keras.layers import LSTM, Dense import matplotlib.pyplot as plt

读取数据

df = pd.read_csv('data.csv')

提取需要预测的时间序列数据

data = df['value'].values data = data.reshape(-1, 1)

数据归一化

scaler = MinMaxScaler(feature_range=(0, 1)) data = scaler.fit_transform(data)

划分训练集和测试集

train_size = int(len(data) * 0.8) test_size = len(data) - train_size train, test = data[0:train_size, :], data[train_size:len(data), :]

创建时间步长

time_steps = 10

将时间序列转化为监督学习问题

def create_dataset(dataset, time_steps=1): data_x, data_y = [], [] for i in range(len(dataset)-time_steps-1): a = dataset[i:(i+time_steps), 0] data_x.append(a) data_y.append(dataset[i + time_steps, 0]) return np.array(data_x), np.array(data_y)

train_x, train_y = create_dataset(train, time_steps) test_x, test_y = create_dataset(test, time_steps)

将输入数据转化为LSTM所需的格式 [样本数, 时间步长, 特征数]

train_x = np.reshape(train_x, (train_x.shape[0], train_x.shape[1], 1)) test_x = np.reshape(test_x, (test_x.shape[0], test_x.shape[1], 1))

创建LSTM模型

model = Sequential() model.add(LSTM(units=50, return_sequences=True, input_shape=(time_steps, 1))) model.add(LSTM(units=50)) model.add(Dense(units=1)) model.compile(optimizer='adam', loss='mean_squared_error') model.fit(train_x, train_y, epochs=50, batch_size=32)

预测测试集

test_predict = model.predict(test_x) test_predict = scaler.inverse_transform(test_predict)

可视化预测结果

plt.plot(test_y, label='Actual') plt.plot(test_predict, label='Predicted') plt.legend() plt.show(