使用TensorFlow预测标普500价格结果不佳，求排查问题

问题描述 #

我正在用TensorFlow深入学习机器学习，当前项目为预测标普500（S&P500）指数未来10天的价格。编写的代码运行后预测结果表现极差，不仅波动异常剧烈，甚至出现负值。尝试增加epochs的数量，但结果仍不理想，求排查代码中的遗漏错误。

原始代码：

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

plt.style.use("fivethirtyeight")

file_path = "gspc.csv"
df = pd.read_csv(file_path)

df["Date"] = pd.to_datetime(df["Date"])
df.set_index("Date", inplace=True)

data = df.filter(["Close"])
dataset = data.values
training_data_len = math.ceil(len(dataset) * 0.8)

scaler = MinMaxScaler(feature_range=(0, 1))
scaled_data = scaler.fit_transform(dataset)

train_data = scaled_data[0:training_data_len, :]
x_train = []
y_train = []

for i in range(60, len(train_data)):
    x_train.append(train_data[i - 60:i, 0])
    y_train.append(train_data[i, 0])

x_train, y_train = np.array(x_train), np.array(y_train)
x_train = np.reshape(x_train, (x_train.shape[0], x_train.shape[1], 1))

model = Sequential()
model.add(LSTM(50, return_sequences=True, input_shape=(x_train.shape[1], 1)))
model.add(LSTM(50, return_sequences=False))
model.add(Dense(25))
model.add(Dense(1))

model.compile(optimizer="adam", loss="mean_squared_error")

model.fit(x_train, y_train, batch_size=1, epochs=50)

test_data = scaled_data[training_data_len - 60:, :]
x_test = []
y_test = dataset[training_data_len:, :]
for i in range(60, len(test_data)):
    x_test.append(test_data[i-60:i, 0])

x_test = np.array(x_test)
x_test = np.reshape(x_test, (x_test.shape[0], x_test.shape[1], 1))

predictions = model.predict(x_test)
predictions = scaler.inverse_transform(predictions)

rmse = np.sqrt(np.mean(predictions - y_test)**2)

train = data[:training_data_len]
valid = data[training_data_len:]
valid["Predictions"] = predictions

last_60_days = data[-60:].values
last_60_days_scaled = scaler.transform(last_60_days)

prediction_list = []

for i in range(10):
    x_input = last_60_days_scaled.reshape((1, 60, 1))

    next_price = model.predict(x_input)[0, 0]

    prediction_list.append(next_price)
    last_60_days = np.append(last_60_days[1:], [[next_price]], axis=0)
    last_60_days_scaled = scaler.transform(last_60_days)

prediction_list = scaler.inverse_transform(np.array(prediction_list).reshape(-1, 1))

print(prediction_list)

核心错误分析 #

1. 滚动预测的缩放逻辑错误
   循环中你将模型输出的缩放后价格next_price直接拼接到原始价格数组last_60_days，再重新调用scaler.transform处理混合数据。原始价格和缩放后数据分布完全不同，这会导致后续输入模型的数据严重偏离训练时的分布，最终引发预测值异常波动甚至负值。

2. RMSE计算逻辑错误
   当前代码中rmse = np.sqrt(np.mean(predictions - y_test)**2)计算的是平均误差的平方根，并非正确的均方根误差（RMSE）。正确公式应为先计算每个误差的平方，取均值后再开平方：np.sqrt(np.mean((predictions - y_test)**2))。错误的评估指标会让你无法准确判断模型真实性能。

3. 训练参数易导致过拟合
   使用batch_size=1训练LSTM极易引发过拟合，且训练效率极低；同时未设置验证集监控训练过程，无法及时发现过拟合或欠拟合问题。

修正后的代码 #

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

plt.style.use("fivethirtyeight")

file_path = "gspc.csv"
df = pd.read_csv(file_path)

df["Date"] = pd.to_datetime(df["Date"])
df.set_index("Date", inplace=True)

data = df.filter(["Close"])
dataset = data.values
training_data_len = math.ceil(len(dataset) * 0.8)

scaler = MinMaxScaler(feature_range=(0, 1))
scaled_data = scaler.fit_transform(dataset)

train_data = scaled_data[0:training_data_len, :]
x_train = []
y_train = []

for i in range(60, len(train_data)):
    x_train.append(train_data[i - 60:i, 0])
    y_train.append(train_data[i, 0])

x_train, y_train = np.array(x_train), np.array(y_train)
x_train = np.reshape(x_train, (x_train.shape[0], x_train.shape[1], 1))

model = Sequential()
model.add(LSTM(50, return_sequences=True, input_shape=(x_train.shape[1], 1)))
model.add(Dropout(0.2))  # 添加Dropout降低过拟合
model.add(LSTM(50, return_sequences=False))
model.add(Dropout(0.2))
model.add(Dense(25))
model.add(Dense(1))

model.compile(optimizer="adam", loss="mean_squared_error")

# 调整batch_size并添加验证集，降低过拟合风险
model.fit(x_train, y_train, batch_size=32, epochs=50, validation_split=0.1)

test_data = scaled_data[training_data_len - 60:, :]
x_test = []
y_test = dataset[training_data_len:, :]
for i in range(60, len(test_data)):
    x_test.append(test_data[i-60:i, 0])

x_test = np.array(x_test)
x_test = np.reshape(x_test, (x_test.shape[0], x_test.shape[1], 1))

predictions = model.predict(x_test)
predictions = scaler.inverse_transform(predictions)

# 修正RMSE计算逻辑
rmse = np.sqrt(np.mean((predictions - y_test)**2))
print(f"RMSE: {rmse}")

train = data[:training_data_len]
valid = data[training_data_len:].copy()  # 避免SettingWithCopyWarning
valid["Predictions"] = predictions

# 修正滚动预测的缩放逻辑
last_60_days_scaled = scaler.transform(data[-60:].values)
prediction_list = []

for i in range(10):
    x_input = last_60_days_scaled.reshape((1, 60, 1))
    next_price_scaled = model.predict(x_input)[0, 0]
    prediction_list.append(next_price_scaled)
    # 直接拼接缩放后的值，保持数据分布一致
    last_60_days_scaled = np.append(last_60_days_scaled[1:], [[next_price_scaled]], axis=0)

# 最后统一逆缩放得到原始价格
prediction_list = scaler.inverse_transform(np.array(prediction_list).reshape(-1, 1))

print(prediction_list)

# 可选：绘制验证集和预测结果
plt.figure(figsize=(16,8))
plt.title('Model')
plt.xlabel('Date', fontsize=18)
plt.ylabel('Close Price USD ($)', fontsize=18)
plt.plot(train['Close'])
plt.plot(valid[['Close', 'Predictions']])
plt.legend(['Train', 'Val', 'Predictions'], loc='lower right')
plt.show()

额外优化建议 #

• 尝试不同时间步长：60天是常用值，可尝试30天或90天，观察模型性能变化。
• 使用双向LSTM：替换普通LSTM为Bidirectional(LSTM(...))，捕捉双向时间序列特征。
• 添加早停机制：使用keras.callbacks.EarlyStopping(monitor='val_loss', patience=5)，当验证集loss不再下降时自动停止训练，避免过拟合。

内容的提问来源于stack exchange，提问作者FaBrown