AI时间序列预测完全教程
1. 时间序列预测概述与应用场景
时间序列是按时间顺序排列的数据点序列，其中每个数据点与特定时间戳关联。时间序列预测的核心目标是基于历史观测值推断未来的数据趋势。这一任务在工业界有着极为广泛的应用：
- 能源领域：电力负荷预测、风电出力预测、光伏功率预测
- 金融领域：股价走势、汇率波动、交易量预估
- 供应链：库存需求预测、物流流量预估
- 物联网：设备健康状态监测、传感器数据预测
- 气象科学：温度、降水、空气质量预测
时间序列数据通常包含四个核心成分：趋势（Trend）、季节性（Seasonality）、周期性（Cyclicity） 和 噪声（Noise）。理解这些成分对于选择合适的建模方法至关重要。
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
# 生成一个包含多种成分的示例时间序列
np.random.seed(42)
dates = pd.date_range('2020-01-01', periods=1000, freq='D')
trend = np.linspace(10, 50, 1000)
seasonal = 10 * np.sin(2 * np.pi * np.arange(1000) / 365.25)
noise = np.random.normal(0, 2, 1000)
values = trend + seasonal + noise
ts = pd.Series(values, index=dates)
ts.plot(title='合成时间序列：趋势 + 季节性 + 噪声', figsize=(12, 4))
plt.ylabel('值')
plt.tight_layout()
plt.show()
2. 传统方法回顾
2.1 ARIMA 模型
ARIMA（自回归积分移动平均模型）是最经典的线性时间序列模型，由三部分组成：AR(p) 自回归项、I(d) 差分阶数、MA(q) 移动平均项。
from statsmodels.tsa.arima.model import ARIMA
from statsmodels.tsa.stattools import adfuller
# ADF平稳性检验
result = adfuller(ts.values)
print(f'ADF统计量: {result[0]:.4f}')
print(f'p值: {result[1]:.4f}')
# 拟合ARIMA模型
model = ARIMA(ts.values, order=(2, 1, 2))
fitted = model.fit()
print(fitted.summary())
# 预测未来30天
forecast = fitted.forecast(steps=30)
print(f'未来30天预测值: {forecast[:5]}...')
2.2 Prophet
由 Meta 开发的 Prophet 擅长处理具有强季节性效应和节假日效应的商业时间序列。其核心思想是将时间序列分解为趋势、季节性和节假日三个可加成分。
from prophet import Prophet
# Prophet要求特定的列名格式
df_prophet = pd.DataFrame({
'ds': ts.index,
'y': ts.values
})
model = Prophet(
yearly_seasonality=True,
weekly_seasonality=False,
daily_seasonality=False,
changepoint_prior_scale=0.05
)
model.fit(df_prophet)
future = model.make_future_dataframe(periods=30)
forecast = model.predict(future)
model.plot(forecast)
plt.title('Prophet 预测结果')
plt.show()
2.3 ETS 模型
ETS（误差-趋势-季节性）模型基于指数平滑原理，通过加法或乘法方式组合各成分。
from statsmodels.tsa.holtwinters import ExponentialSmoothing
ets_model = ExponentialSmoothing(
ts.values,
trend='add',
seasonal='add',
seasonal_periods=365
).fit()
ets_forecast = ets_model.forecast(30)
print(f'ETS预测均值: {np.mean(ets_forecast):.2f}')
3. 深度学习方法
3.1 LSTM（长短期记忆网络）
LSTM 通过门控机制解决了标准 RNN 的梯度消失问题，能够捕捉长期依赖关系。其核心组件包括遗忘门、输入门和输出门。
import torch
import torch.nn as nn
from torch.utils.data import Dataset, DataLoader
class TimeSeriesDataset(Dataset):
"""将时间序列转换为监督学习格式"""
def __init__(self, series, seq_len, pred_len):
self.series = torch.FloatTensor(series)
self.seq_len = seq_len
self.pred_len = pred_len
def __len__(self):
return len(self.series) - self.seq_len - self.pred_len + 1
def __getitem__(self, idx):
x = self.series[idx:idx + self.seq_len]
y = self.series[idx + self.seq_len:idx + self.seq_len + self.pred_len]
return x.unsqueeze(-1), y
class LSTMForecaster(nn.Module):
def __init__(self, input_size=1, hidden_size=64, num_layers=2, pred_len=30):
super().__init__()
self.lstm = nn.LSTM(input_size, hidden_size, num_layers,
batch_first=True, dropout=0.2)
self.fc = nn.Sequential(
nn.Linear(hidden_size, hidden_size // 2),
nn.ReLU(),
nn.Linear(hidden_size // 2, pred_len)
)
def forward(self, x):
lstm_out, _ = self.lstm(x)
last_hidden = lstm_out[:, -1, :]
return self.fc(last_hidden)
# 数据准备
seq_len, pred_len = 60, 30
data_normalized = (ts.values - ts.values.mean()) / ts.values.std()
train_data = data_normalized[:800]
test_data = data_normalized[740:] # 留出overlap
train_dataset = TimeSeriesDataset(train_data, seq_len, pred_len)
train_loader = DataLoader(train_dataset, batch_size=32, shuffle=True)
# 训练
model = LSTMForecaster(pred_len=pred_len)
optimizer = torch.optim.Adam(model.parameters(), lr=1e-3)
criterion = nn.MSELoss()
for epoch in range(50):
total_loss = 0
for x_batch, y_batch in train_loader:
pred = model(x_batch)
loss = criterion(pred, y_batch)
optimizer.zero_grad()
loss.backward()
optimizer.step()
total_loss += loss.item()
if (epoch + 1) % 10 == 0:
print(f'Epoch {epoch+1}, Loss: {total_loss/len(train_loader):.6f}')
3.2 GRU（门控循环单元）
GRU 是 LSTM 的简化变体，将遗忘门和输入门合并为更新门，参数更少、训练更快，在某些场景下性能与 LSTM 相当。
class GRUForecaster(nn.Module):
def __init__(self, input_size=1, hidden_size=64, num_layers=2, pred_len=30):
super().__init__()
self.gru = nn.GRU(input_size, hidden_size, num_layers,
batch_first=True, dropout=0.2)
self.fc = nn.Linear(hidden_size, pred_len)
def forward(self, x):
gru_out, _ = self.gru(x)
return self.fc(gru_out[:, -1, :])
3.3 TCN（时间卷积网络）
TCN 使用因果膨胀卷积确保不泄露未来信息，同时通过膨胀机制实现大范围感受野，适合并行计算。
class CausalConv1d(nn.Module):
"""因果卷积：确保输出只依赖于当前和过去的时间步"""
def __init__(self, in_channels, out_channels, kernel_size, dilation):
super().__init__()
self.padding = (kernel_size - 1) * dilation
self.conv = nn.Conv1d(in_channels, out_channels, kernel_size,
dilation=dilation, padding=self.padding)
def forward(self, x):
out = self.conv(x)
return out[:, :, :-self.padding] if self.padding > 0 else out
class TCNBlock(nn.Module):
def __init__(self, in_ch, out_ch, kernel_size=3, dilation=1):
super().__init__()
self.conv1 = CausalConv1d(in_ch, out_ch, kernel_size, dilation)
self.conv2 = CausalConv1d(out_ch, out_ch, kernel_size, dilation * 2)
self.relu = nn.ReLU()
self.dropout = nn.Dropout(0.2)
self.residual = nn.Conv1d(in_ch, out_ch, 1) if in_ch != out_ch else nn.Identity()
def forward(self, x):
residual = self.residual(x)
out = self.dropout(self.relu(self.conv1(x)))
out = self.dropout(self.relu(self.conv2(out)))
return self.relu(out + residual)
4. Transformer 时间序列预测
4.1 Informer
Informer 针对长序列时间序列预测进行了三项关键优化：ProbSparse 自注意力机制（降低复杂度从 O(L²) 到 O(L·logL)）、自注意力蒸馏（逐层减半序列长度）、生成式解码器（一次性输出多步预测）。
class ProbSparseAttention(nn.Module):
"""ProbSparse自注意力：只计算Top-u个最活跃的Query"""
def __init__(self, d_model, n_heads, factor=5):
super().__init__()
self.n_heads = n_heads
self.d_k = d_model // n_heads
self.factor = factor
self.W_Q = nn.Linear(d_model, d_model)
self.W_K = nn.Linear(d_model, d_model)
self.W_V = nn.Linear(d_model, d_model)
self.out_proj = nn.Linear(d_model, d_model)
def forward(self, x):
B, L, D = x.shape
Q = self.W_Q(x).view(B, L, self.n_heads, self.d_k).transpose(1, 2)
K = self.W_K(x).view(B, L, self.n_heads, self.d_k).transpose(1, 2)
V = self.W_V(x).view(B, L, self.n_heads, self.d_k).transpose(1, 2)
# 计算采样数量
u = max(1, int(self.factor * np.log(L)))
# 计算每个Query的稀疏度度量
K_mean = K.mean(dim=2, keepdim=True)
scores = torch.matmul(Q, K.transpose(-2, -1)) / np.sqrt(self.d_k)
sparsity = scores.max(dim=-1).values - scores.mean(dim=-1)
# 选择Top-u个最活跃的Query
_, top_indices = sparsity.topk(u, dim=-1)
top_indices = top_indices.unsqueeze(-1).expand(-1, -1, -1, self.d_k)
Q_top = Q.gather(2, top_indices)
# 计算注意力
attn = torch.matmul(Q_top, K.transpose(-2, -1)) / np.sqrt(self.d_k)
attn = torch.softmax(attn, dim=-1)
out_top = torch.matmul(attn, V)
# 将结果散回原始位置
out = V.clone()
for b in range(B):
for h in range(self.n_heads):
out[b, h, top_indices[b, h, :, 0]] = out_top[b, h]
out = out.transpose(1, 2).contiguous().view(B, L, D)
return self.out_proj(out)
4.2 Autoformer
Autoformer 引入了自相关机制替代传统注意力，通过频域计算序列间的相似性，同时内嵌了序列分解模块（趋势-季节性分解），使模型能更清晰地捕捉不同频率成分。
4.3 PatchTST
PatchTST 将时间序列分割为固定长度的 patch（类似 ViT 的 token），然后对 patch 序列应用标准 Transformer。这种方法保留了局部语义信息，大幅降低了计算复杂度。
class PatchEmbedding(nn.Module):
"""将时间序列切分为Patch并嵌入"""
def __init__(self, seq_len, patch_len, d_model):
super().__init__()
self.patch_len = patch_len
self.num_patches = seq_len // patch_len
self.proj = nn.Linear(patch_len, d_model)
self.pos_embed = nn.Parameter(torch.randn(1, self.num_patches, d_model))
def forward(self, x):
# x: (B, seq_len, 1)
B = x.shape[0]
# 切分为patch: (B, num_patches, patch_len)
patches = x.unfold(1, self.patch_len, self.patch_len).squeeze(-1)
return self.proj(patches) + self.pos_embed
class PatchTST(nn.Module):
def __init__(self, seq_len=96, patch_len=16, d_model=128, n_heads=8, n_layers=3, pred_len=30):
super().__init__()
self.patch_embed = PatchEmbedding(seq_len, patch_len, d_model)
encoder_layer = nn.TransformerEncoderLayer(d_model, n_heads, dim_feedforward=d_model*4, batch_first=True)
self.encoder = nn.TransformerEncoder(encoder_layer, n_layers)
self.head = nn.Linear(d_model * (seq_len // patch_len), pred_len)
def forward(self, x):
patches = self.patch_embed(x)
encoded = self.encoder(patches)
return self.head(encoded.reshape(encoded.shape[0], -1))
5. 时序大模型
2024年以来，时间序列预测进入了基础模型时代。这些模型在海量时间序列数据上预训练，展现出强大的零样本和少样本预测能力。
5.1 TimesFM（Google）
TimesFM 采用 Decoder-only Transformer 架构，在超过 1000 亿个真实世界时间点上预训练。其核心创新在于将时间序列值离散化为 token，然后以自回归方式预测。
import timesfm
# 加载预训练的TimesFM模型
tfm = timesfm.TimesFm(
hparams=timesfm.TimesFmHparams(
backend='gpu',
per_core_batch_size=32,
horizon_len=128,
num_layers=50,
context_len=2048,
),
checkpoint=timesfm.TimesFmCheckpoint(
huggingface_repo_id='google/timesfm-1.0-200m'
)
)
# 零样本预测
input_ts = [ts.values[:200].tolist()]
forecast = tfm.forecast(
horizon=30,
inputs=input_ts,
)
print(f'TimesFM预测结果 shape: {forecast.shape}')
5.2 Amazon Chronos
Chronos 基于 T5 架构，将时间序列值量化为离散 bin，以语言模型的方式进行概率预测。其独特之处在于能够输出完整的预测分布而非点估计。
from chronos import ChronosPipeline
pipeline = ChronosPipeline.from_pretrained(
'amazon/chronos-t5-large',
device_map='auto',
torch_dtype=torch.float32
)
# 概率预测
context = torch.tensor(ts.values[:200]).unsqueeze(0)
forecast = pipeline.predict(
context=context,
prediction_length=30,
num_samples=100, # 采样100条轨迹
)
# forecast shape: (1, num_samples, prediction_length)
median = np.median(forecast[0].numpy(), axis=0)
lower = np.percentile(forecast[0].numpy(), 10, axis=0)
upper = np.percentile(forecast[0].numpy(), 90, axis=0)
print(f'中位数预测: {median[:5]}')
5.3 Salesforce Moirai
Moirai 采用统一的多变量时间序列 Transformer，支持任意数量的变量和不同频率的数据。它通过混合分布（Student-t、Normal、Negative Binomial）建模不同特征的数据分布。
6. 多变量时间序列预测
多变量时间序列包含多个相互关联的变量，预测时需要同时建模变量内部的时序依赖和变量间的交叉依赖。
class MultiVarTransformer(nn.Module):
"""多变量时间序列Transformer预测器"""
def __init__(self, n_vars, seq_len, pred_len, d_model=128, n_heads=4, n_layers=2):
super().__init__()
self.var_embed = nn.Linear(n_vars, d_model)
self.pos_embed = nn.Parameter(torch.randn(1, seq_len, d_model))
encoder_layer = nn.TransformerEncoderLayer(
d_model, n_heads, dim_feedforward=d_model*4, batch_first=True
)
self.encoder = nn.TransformerEncoder(encoder_layer, n_layers)
self.head = nn.Linear(d_model * seq_len, pred_len * n_vars)
self.n_vars = n_vars
self.pred_len = pred_len
def forward(self, x):
# x: (B, seq_len, n_vars)
x = self.var_embed(x) + self.pos_embed
encoded = self.encoder(x)
out = self.head(encoded.reshape(encoded.shape[0], -1))
return out.reshape(-1, self.pred_len, self.n_vars)
# 示例：多变量预测
n_vars = 5
multi_data = np.random.randn(500, n_vars).astype(np.float32)
model = MultiVarTransformer(n_vars=n_vars, seq_len=60, pred_len=30)
x = torch.tensor(multi_data[:60]).unsqueeze(0)
pred = model(x)
print(f'多变量预测输出 shape: {pred.shape}') # (1, 30, 5)
变量间关系建模可以采用图神经网络，将每个变量视为图节点，通过学习自适应邻接矩阵来捕捉变量间的动态相关性。
7. 异常检测与突变点识别
7.1 基于重建误差的异常检测
使用自编码器学习正常模式，重建误差大的时间点即为潜在异常。
class TSAutoEncoder(nn.Module):
def __init__(self, seq_len, n_features, latent_dim=16):
super().__init__()
self.encoder = nn.Sequential(
nn.Linear(seq_len * n_features, 64),
nn.ReLU(),
nn.Linear(64, latent_dim)
)
self.decoder = nn.Sequential(
nn.Linear(latent_dim, 64),
nn.ReLU(),
nn.Linear(64, seq_len * n_features)
)
self.seq_len = seq_len
self.n_features = n_features
def forward(self, x):
B = x.shape[0]
z = self.encoder(x.reshape(B, -1))
out = self.decoder(z)
return out.reshape(B, self.seq_len, self.n_features)
# 异常检测流程
def detect_anomalies(model, data, seq_len, threshold_percentile=95):
model.eval()
errors = []
with torch.no_grad():
for i in range(len(data) - seq_len):
x = torch.tensor(data[i:i+seq_len]).unsqueeze(0).float()
recon = model(x)
mse = ((x - recon) ** 2).mean().item()
errors.append(mse)
threshold = np.percentile(errors, threshold_percentile)
anomalies = np.where(np.array(errors) > threshold)[0]
return anomalies, errors, threshold
7.2 突变点检测
突变点（Changepepoint）是时间序列统计特性发生显著变化的时间点。常用方法包括 CUSUM、Bayesian Online Changepoint Detection（BOCPD）等。
def cusum_detect(data, threshold=5, drift=0):
"""CUSUM累积和突变点检测"""
mean = np.mean(data)
std = np.std(data)
normalized = (data - mean) / std
pos_cusum = np.zeros(len(data))
neg_cusum = np.zeros(len(data))
changepoints = []
for i in range(1, len(data)):
pos_cusum[i] = max(0, pos_cusum[i-1] + normalized[i] - drift)
neg_cusum[i] = max(0, neg_cusum[i-1] - normalized[i] - drift)
if pos_cusum[i] > threshold or neg_cusum[i] > threshold:
changepoints.append(i)
pos_cusum[i] = 0
neg_cusum[i] = 0
return changepoints
8. 特征工程与数据预处理
优秀的特征工程往往比模型选择更重要。以下是时间序列预测中常用的技术：
def create_time_features(df, freq='D'):
"""从时间索引中提取丰富的时间特征"""
features = pd.DataFrame(index=df.index)
features['hour'] = df.index.hour if hasattr(df.index, 'hour') else 0
features['dayofweek'] = df.index.dayofweek
features['dayofmonth'] = df.index.day
features['month'] = df.index.month
features['quarter'] = df.index.quarter
features['year'] = df.index.year
features['dayofyear'] = df.index.dayofyear
# 周末标记
features['is_weekend'] = (features['dayofweek'] >= 5).astype(int)
# 正弦余弦编码（捕捉周期性）
features['month_sin'] = np.sin(2 * np.pi * features['month'] / 12)
features['month_cos'] = np.cos(2 * np.pi * features['month'] / 12)
features['dow_sin'] = np.sin(2 * np.pi * features['dayofweek'] / 7)
features['dow_cos'] = np.cos(2 * np.pi * features['dayofweek'] / 7)
return features
def add_lag_features(series, lags=[1, 7, 14, 28]):
"""添加滞后特征"""
df = pd.DataFrame({'value': series})
for lag in lags:
df[f'lag_{lag}'] = df['value'].shift(lag)
# 滚动统计特征
for window in [7, 14, 30]:
df[f'rolling_mean_{window}'] = df['value'].rolling(window).mean()
df[f'rolling_std_{window}'] = df['value'].rolling(window).std()
return df.dropna()
关键预处理步骤：
- 缺失值处理：前向填充、线性插值、KNN 插值
- 平稳化：差分、对数变换、Box-Cox 变换
- 归一化：Z-score 标准化、Min-Max 归一化（注意要在训练集上 fit，避免数据泄露）
- 异常值处理：基于 IQR 或 Z-score 的截断或替换
9. 模型评估与交叉验证
时间序列数据具有时间顺序，不能随机划分训练集和测试集。需要使用时间序列交叉验证（滚动窗口或扩展窗口）。
from sklearn.metrics import mean_absolute_error, mean_squared_error
def time_series_cross_validate(model_cls, data, n_splits=5,
train_ratio=0.6, seq_len=60, pred_len=30):
"""时间序列滚动窗口交叉验证"""
n = len(data)
step = (n - int(n * train_ratio)) // n_splits
metrics = []
for i in range(n_splits):
train_end = int(n * train_ratio) + i * step
test_start = train_end - seq_len
test_end = min(train_end + pred_len, n)
train_data = data[:train_end]
test_data = data[test_start:test_end]
# 训练与预测（简化示例）
# model = model_cls(...)
# model.fit(train_data)
# pred = model.predict(test_data[-seq_len:])
# 计算指标
# mae = mean_absolute_error(actual, pred)
# rmse = np.sqrt(mean_squared_error(actual, pred))
# metrics.append({'fold': i, 'mae': mae, 'rmse': rmse})
return metrics
def evaluate_forecast(actual, predicted):
"""全面的预测评估指标"""
mae = mean_absolute_error(actual, predicted)
rmse = np.sqrt(mean_squared_error(actual, predicted))
mape = np.mean(np.abs((actual - predicted) / (actual + 1e-8))) * 100
# 对称MAPE
smape = np.mean(2 * np.abs(actual - predicted) / (np.abs(actual) + np.abs(predicted) + 1e-8)) * 100
return {'MAE': mae, 'RMSE': rmse, 'MAPE': mape, 'SMAPE': smape}
常用评估指标：
- MAE：平均绝对误差，对异常值不敏感
- RMSE：均方根误差，对大误差更敏感
- MAPE：平均绝对百分比误差，直观但接近零值时不稳定
- SMAPE：对称 MAPE，解决 MAPE 的不对称问题
10. 实战案例：电力负荷预测系统
以下构建一个完整的电力负荷预测系统，整合数据处理、特征工程、模型训练和评估。
import torch
import torch.nn as nn
import numpy as np
import pandas as pd
from torch.utils.data import Dataset, DataLoader
class PowerLoadPredictor(nn.Module):
"""结合LSTM和注意力机制的电力负荷预测模型"""
def __init__(self, n_features, hidden_size=128, n_layers=2, pred_len=24):
super().__init__()
self.lstm = nn.LSTM(n_features, hidden_size, n_layers,
batch_first=True, dropout=0.3)
self.attention = nn.MultiheadAttention(hidden_size, num_heads=4, batch_first=True)
self.norm = nn.LayerNorm(hidden_size)
self.fc = nn.Sequential(
nn.Linear(hidden_size, 64),
nn.ReLU(),
nn.Dropout(0.2),
nn.Linear(64, pred_len)
)
def forward(self, x):
lstm_out, _ = self.lstm(x)
attn_out, weights = self.attention(lstm_out, lstm_out, lstm_out)
out = self.norm(lstm_out + attn_out)
return self.fc(out[:, -1, :]), weights
class PowerLoadPipeline:
"""电力负荷预测完整流水线"""
def __init__(self, seq_len=168, pred_len=24): # 7天历史 -> 预测24小时
self.seq_len = seq_len
self.pred_len = pred_len
self.model = None
def preprocess(self, raw_data, weather_data=None):
"""数据预处理与特征构建"""
df = pd.DataFrame({'load': raw_data})
# 时间特征
df['hour'] = df.index.hour
df['dow'] = df.index.dayofweek
df['is_weekend'] = (df['dow'] >= 5).astype(int)
# 滞后特征
for lag in [24, 48, 168]:
df[f'lag_{lag}'] = df['load'].shift(lag)
# 滚动特征
df['rolling_24h_mean'] = df['load'].rolling(24).mean()
df['rolling_24h_std'] = df['load'].rolling(24).std()
if weather_data is not None:
df = df.join(weather_data)
return df.dropna()
def train(self, train_data, epochs=100, lr=1e-3):
"""训练模型"""
n_features = train_data.shape[1]
self.model = PowerLoadPredictor(n_features, pred_len=self.pred_len)
optimizer = torch.optim.AdamW(self.model.parameters(), lr=lr, weight_decay=1e-4)
scheduler = torch.optim.lr_scheduler.CosineAnnealingLR(optimizer, T_max=epochs)
criterion = nn.HuberLoss()
for epoch in range(epochs):
self.model.train()
total_loss = 0
for i in range(len(train_data) - self.seq_len - self.pred_len):
x = torch.tensor(train_data[i:i+self.seq_len]).unsqueeze(0).float()
y = torch.tensor(train_data[i+self.seq_len:i+self.seq_len+self.pred_len, 0]).unsqueeze(0).float()
pred, _ = self.model(x)
loss = criterion(pred, y)
optimizer.zero_grad()
loss.backward()
torch.nn.utils.clip_grad_norm_(self.model.parameters(), 1.0)
optimizer.step()
total_loss += loss.item()
scheduler.step()
if (epoch + 1) % 20 == 0:
print(f'Epoch {epoch+1}, Avg Loss: {total_loss/(len(train_data)-self.seq_len-self.pred_len):.6f}')
def predict(self, recent_data):
"""单次预测"""
self.model.eval()
with torch.no_grad():
x = torch.tensor(recent_data[-self.seq_len:]).unsqueeze(0).float()
pred, attn_weights = self.model(x)
return pred.squeeze().numpy(), attn_weights
# 使用示例
# pipeline = PowerLoadPipeline(seq_len=168, pred_len=24)
# processed = pipeline.preprocess(hourly_load_series)
# pipeline.train(processed.values, epochs=100)
# forecast, weights = pipeline.predict(processed.values)
11. 生产部署与在线学习
11.1 模型服务化
from fastapi import FastAPI
from pydantic import BaseModel
import torch
import numpy as np
app = FastAPI()
model_cache = {}
class ForecastRequest(BaseModel):
series: list[float]
horizon: int = 24
class ForecastResponse(BaseModel):
predictions: list[float]
confidence_lower: list[float]
confidence_upper: list[float]
@app.post('/forecast', response_model=ForecastResponse)
async def forecast(request: ForecastRequest):
model = model_cache.get('power_load')
series = np.array(request.series)
# 预处理
mean, std = series.mean(), series.std()
normalized = (series - mean) / std
# 推理
x = torch.tensor(normalized).unsqueeze(0).unsqueeze(-1).float()
with torch.no_grad():
pred, _ = model(x)
pred_np = pred.squeeze().numpy() * std + mean
# 简单置信区间（基于历史残差）
residual_std = np.std(series[-168:]) * 0.1
return ForecastResponse(
predictions=pred_np.tolist(),
confidence_lower=(pred_np - 1.96 * residual_std).tolist(),
confidence_upper=(pred_np + 1.96 * residual_std).tolist()
)
11.2 在线学习策略
在生产环境中，数据分布会随时间漂移（concept drift），模型需要持续更新。推荐策略：
- 滑动窗口重训练：定期（如每天）用最近 N 天数据全量重训练
- 增量微调：以较小学习率在新数据上继续训练现有模型
- 模型集成：维护多个不同时间窗口训练的模型，动态加权融合
- 漂移检测：监控预测误差分布，当误差显著增大时触发重训练
class OnlineLearner:
"""在线学习管理器"""
def __init__(self, model, lr=1e-4, drift_threshold=2.0):
self.model = model
self.optimizer = torch.optim.Adam(model.parameters(), lr=lr)
self.drift_threshold = drift_threshold
self.recent_errors = []
def update(self, x, y_true):
"""用新数据增量更新模型"""
self.model.train()
pred, _ = self.model(x.unsqueeze(0))
loss = nn.MSELoss()(pred, y_true.unsqueeze(0))
self.optimizer.zero_grad()
loss.backward()
self.optimizer.step()
return loss.item()
def check_drift(self, error):
"""检测概念漂移"""
self.recent_errors.append(error)
if len(self.recent_errors) > 100:
self.recent_errors.pop(0)
if len(self.recent_errors) >= 50:
baseline = np.mean(self.recent_errors[:len(self.recent_errors)//2])
recent = np.mean(self.recent_errors[len(self.recent_errors)//2:])
if recent > baseline * self.drift_threshold:
return True
return False
11.3 监控指标
生产部署需要监控的关键指标：
- 预测精度滑动窗口：MAE/MAPE 的 24h/7d 滚动值
- 预测延迟：P50/P95/P99 推理耗时
- 数据质量：缺失率、异常值比例、数据新鲜度
- 模型漂移：预测残差分布的变化
时间序列预测是一个不断演进的领域。从经典的统计方法到深度学习，再到如今的基础模型，每一代方法都有其适用场景。在实际项目中，建议先用传统方法建立基线，再逐步引入复杂模型，始终以业务指标（而非纯学术指标）衡量模型价值。
