时间序列预测¶
概述¶
时间序列预测是分析和预测按时间顺序排列的数据点的技术，广泛应用于股票预测、天气预报、销量预测等场景。
时间序列基础¶
时间序列组成¶
平稳性检验¶
from statsmodels.tsa.stattools import adfuller
def test_stationarity(series):
result = adfuller(series)
print(f'ADF统计量: {result[0]:.4f}')
print(f'p值: {result[1]:.4f}')
print('平稳' if result[1] < 0.05 else '非平稳')
经典方法¶
移动平均（MA）¶
import pandas as pd
def moving_average(series, window=7):
return series.rolling(window=window).mean()
ARIMA模型¶
from statsmodels.tsa.arima.model import ARIMA
# ARIMA(p, d, q)
# p: 自回归阶数
# d: 差分阶数
# q: 移动平均阶数
model = ARIMA(train_data, order=(1, 1, 1))
fitted_model = model.fit()
predictions = fitted_model.forecast(steps=30)
Prophet（Facebook）¶
from prophet import Prophet
# 数据格式：ds(日期), y(值)
df = pd.DataFrame({'ds': dates, 'y': values})
model = Prophet(yearly_seasonality=True)
model.fit(df)
future = model.make_future_dataframe(periods=30)
forecast = model.predict(future)
深度学习方法¶
LSTM预测¶
import torch
import torch.nn as nn
class LSTMPredictor(nn.Module):
def __init__(self, input_size=1, hidden_size=64, num_layers=2):
super().__init__()
self.lstm = nn.LSTM(input_size, hidden_size, num_layers, batch_first=True)
self.fc = nn.Linear(hidden_size, 1)
def forward(self, x):
lstm_out, _ = self.lstm(x)
out = self.fc(lstm_out[:, -1, :])
return out
# 数据准备
def create_sequences(data, seq_length):
xs, ys = [], []
for i in range(len(data) - seq_length):
xs.append(data[i:i+seq_length])
ys.append(data[i+seq_length])
return torch.FloatTensor(xs), torch.FloatTensor(ys)
Transformer预测¶
class TimeSeriesTransformer(nn.Module):
def __init__(self, input_size=1, d_model=64, nhead=4, num_layers=2):
super().__init__()
self.input_proj = nn.Linear(input_size, d_model)
self.pos_encoding = PositionalEncoding(d_model)
encoder_layer = nn.TransformerEncoderLayer(d_model, nhead)
self.transformer = nn.TransformerEncoder(encoder_layer, num_layers)
self.fc = nn.Linear(d_model, 1)
def forward(self, x):
x = self.input_proj(x)
x = self.pos_encoding(x)
x = self.transformer(x)
return self.fc(x[:, -1, :])
实战：销量预测¶
import numpy as np
import pandas as pd
from sklearn.preprocessing import MinMaxScaler
# 数据加载
df = pd.read_csv('sales.csv')
scaler = MinMaxScaler()
scaled_data = scaler.fit_transform(df[['sales']])
# 创建序列
seq_length = 30
X, y = create_sequences(scaled_data, seq_length)
# 划分训练集
split = int(0.8 * len(X))
X_train, X_test = X[:split], X[split:]
y_train, y_test = y[:split], y[split:]
# 训练模型
model = LSTMPredictor()
criterion = nn.MSELoss()
optimizer = torch.optim.Adam(model.parameters(), lr=0.001)
for epoch in range(50):
model.train()
pred = model(X_train)
loss = criterion(pred, y_train)
optimizer.zero_grad()
loss.backward()
optimizer.step()
# 预测
model.eval()
with torch.no_grad():
predictions = model(X_test)
predictions = scaler.inverse_transform(predictions.numpy())
评估指标¶
from sklearn.metrics import mean_absolute_error, mean_squared_error
def evaluate_forecast(y_true, y_pred):
mae = mean_absolute_error(y_true, y_pred)
rmse = np.sqrt(mean_squared_error(y_true, y_pred))
mape = np.mean(np.abs((y_true - y_pred) / y_true)) * 100
return {'MAE': mae, 'RMSE': rmse, 'MAPE': mape}