时间序列分析方法
时间序列分析是处理按时间顺序排列的数据点的统计方法，广泛应用于金融预测、需求规划、异常检测等场景。本文介绍从经典统计方法到深度学习方法的时间序列分析技术体系。
时间序列基本概念
时间序列数据的核心特征是数据点之间存在时间依赖关系。理解这些依赖模式是预测和异常检测的基础。
import pandas as pd
import numpy as np
import matplotlib.pyplot as plt
# 生成示例时间序列
np.random.seed(42)
dates = pd.date_range('2023-01-01', periods=365, freq='D')
trend = np.linspace(100, 150, 365)
seasonality = 20 * np.sin(2 * np.pi * np.arange(365) / 365)
noise = np.random.normal(0, 5, 365)
ts = pd.Series(trend + seasonality + noise, index=dates, name='sales')平稳性（Stationarity）
平稳时间序列的统计特性（均值、方差、自相关）不随时间变化。大多数统计模型要求数据平稳或可以通过差分转化为平稳。
ADF 检验
from statsmodels.tsa.stattools import adfuller
def test_stationarity(timeseries):
"""ADF 平稳性检验"""
result = adfuller(timeseries, autolag='AIC')
print(f'ADF Statistic: {result[0]:.4f}')
print(f'P-value: {result[1]:.4f}')
print(f'Lags Used: {result[2]}')
for key, value in result[4].items():
print(f' Critical Value ({key}): {value:.4f}')
if result[1] < 0.05:
print("结论: 序列平稳 (拒绝单位根假设)")
else:
print("结论: 序列非平稳 (无法拒绝单位根假设)")
test_stationarity(ts)
# 差分使序列平稳
ts_diff = ts.diff().dropna()
test_stationarity(ts_diff)时间序列分解
将时间序列分解为三个基本成分：
- 趋势（Trend）：长期变化方向
- 季节性（Seasonality）：周期性重复的模式
- 残差（Residual）：去除趋势和季节性后的随机波动
from statsmodels.tsa.seasonal import seasonal_decompose
# 加法分解: Y(t) = T(t) + S(t) + R(t)
decomposition = seasonal_decompose(ts, model='additive', period=30)
fig, axes = plt.subplots(4, 1, figsize=(12, 10))
decomposition.observed.plot(ax=axes[0], title='Observed')
decomposition.trend.plot(ax=axes[1], title='Trend')
decomposition.seasonal.plot(ax=axes[2], title='Seasonal')
decomposition.resid.plot(ax=axes[3], title='Residual')
plt.tight_layout()ARIMA 模型
ARIMA(p, d, q) 是经典的时间序列预测模型：
- AR(p)：自回归，使用过去 p 个时间步的值
- I(d)：差分阶数，使序列平稳
- MA(q)：移动平均，使用过去 q 个预测误差
参数选择
from statsmodels.graphics.tsaplots import plot_acf, plot_pacf
from statsmodels.tsa.arima.model import ARIMA
import warnings
# ACF/PACF 图辅助选择 p, q
fig, axes = plt.subplots(1, 2, figsize=(14, 5))
plot_acf(ts_diff, lags=40, ax=axes[0])
plot_pacf(ts_diff, lags=40, ax=axes[1])
plt.tight_layout()
# 自动选择参数（网格搜索）
def auto_arima_search(ts, max_p=5, max_d=2, max_q=5):
best_aic = float('inf')
best_params = None
for p in range(max_p + 1):
for d in range(max_d + 1):
for q in range(max_q + 1):
try:
with warnings.catch_warnings():
warnings.simplefilter("ignore")
model = ARIMA(ts, order=(p, d, q))
results = model.fit()
if results.aic < best_aic:
best_aic = results.aic
best_params = (p, d, q)
except Exception:
continue
return best_params, best_aic
best_params, best_aic = auto_arima_search(ts)
print(f"最优参数: ARIMA{best_params}, AIC: {best_aic:.2f}")模型拟合与预测
# 使用 pmdarima 自动选择参数
import pmdarima as pm
auto_model = pm.auto_arima(
ts,
seasonal=True,
m=30, # 季节周期
d=None, # 自动确定差分阶数
start_p=0, max_p=5,
start_q=0, max_q=5,
trace=True,
error_action='ignore',
suppress_warnings=True,
stepwise=True,
)
print(auto_model.summary())
# 预测
forecast = auto_model.predict(n_periods=30)
conf_int = auto_model.predict(n_periods=30, return_conf_int=True)[1]Prophet
Meta 开发的 Prophet 模型专为业务时间序列设计，自动处理趋势变化点、季节性和节假日效应。
from prophet import Prophet
# 准备数据（Prophet 要求 ds 和 y 列名）
df_prophet = pd.DataFrame({
'ds': ts.index,
'y': ts.values,
})
# 建模
model = Prophet(
changepoint_prior_scale=0.05, # 趋势变化点灵活度
seasonality_prior_scale=10, # 季节性灵活度
holidays_prior_scale=10, # 节假日效应灵活度
yearly_seasonality=True,
weekly_seasonality=True,
daily_seasonality=False,
)
# 添加中国节假日
model.add_country_holidays(country_name='CN')
# 添加自定义季节性
model.add_seasonality(name='monthly', period=30.5, fourier_order=5)
model.fit(df_prophet)
# 预测
future = model.make_future_dataframe(periods=60)
forecast = model.predict(future)
# 可视化
model.plot(forecast)
model.plot_components(forecast)指数平滑（Exponential Smoothing）
指数平滑方法对近期观测值给予更高的权重，权重随时间呈指数递减。
| 方法 | 特点 | 适用场景 |
|---|---|---|
| 简单指数平滑 (SES) | 无趋势、无季节性 | 平稳序列 |
| Holt 线性 | 有趋势、无季节性 | 增长/衰减趋势 |
| Holt-Winters | 有趋势、有季节性 | 大多数业务序列 |
from statsmodels.tsa.holtwinters import ExponentialSmoothing
# Holt-Winters（三指数平滑）
hw_model = ExponentialSmoothing(
ts,
trend='add', # 加法趋势
seasonal='add', # 加法季节性
seasonal_periods=30, # 季节周期
damped_trend=True, # 衰减趋势（避免过度外推）
).fit(optimized=True)
# 预测
hw_forecast = hw_model.forecast(30)
print(f"平滑参数: alpha={hw_model.params['smoothing_level']:.3f}, "
f"beta={hw_model.params['smoothing_trend']:.3f}, "
f"gamma={hw_model.params['smoothing_seasonal']:.3f}")异常检测
from sklearn.ensemble import IsolationForest
# 方法 1：基于统计的异常检测（Z-Score）
def detect_anomalies_zscore(ts, window=30, threshold=3):
rolling_mean = ts.rolling(window=window).mean()
rolling_std = ts.rolling(window=window).std()
z_scores = (ts - rolling_mean) / rolling_std
anomalies = ts[z_scores.abs() > threshold]
return anomalies
# 方法 2：基于 STL 分解的残差
from statsmodels.tsa.seasonal import STL
stl = STL(ts, period=30, robust=True)
result = stl.fit()
residuals = result.resid
threshold = 3 * residuals.std()
anomalies_stl = ts[residuals.abs() > threshold]
# 方法 3：Isolation Forest
def detect_anomalies_iforest(ts, contamination=0.05):
# 构造特征：值本身、差分、滚动统计
features = pd.DataFrame({
'value': ts.values,
'diff': ts.diff().fillna(0).values,
'rolling_mean': ts.rolling(7).mean().fillna(method='bfill').values,
'rolling_std': ts.rolling(7).std().fillna(method='bfill').values,
})
clf = IsolationForest(contamination=contamination, random_state=42)
predictions = clf.fit_predict(features)
anomalies = ts[predictions == -1]
return anomalies预测评估指标
def forecast_metrics(actual, predicted):
"""计算预测评估指标"""
actual = np.array(actual)
predicted = np.array(predicted)
# MAE: 平均绝对误差
mae = np.mean(np.abs(actual - predicted))
# RMSE: 均方根误差
rmse = np.sqrt(np.mean((actual - predicted) ** 2))
# MAPE: 平均绝对百分比误差
mape = np.mean(np.abs((actual - predicted) / actual)) * 100
# sMAPE: 对称 MAPE（解决 MAPE 对零值敏感的问题）
smape = np.mean(2 * np.abs(actual - predicted) / (np.abs(actual) + np.abs(predicted))) * 100
return {
'MAE': mae,
'RMSE': rmse,
'MAPE': f"{mape:.2f}%",
'sMAPE': f"{smape:.2f}%",
}
# 时间序列交叉验证
from sklearn.model_selection import TimeSeriesSplit
tscv = TimeSeriesSplit(n_splits=5)
scores = []
for train_idx, test_idx in tscv.split(ts):
train, test = ts.iloc[train_idx], ts.iloc[test_idx]
model = ARIMA(train, order=(1, 1, 1)).fit()
predictions = model.forecast(len(test))
metrics = forecast_metrics(test.values, predictions.values)
scores.append(metrics)深度学习方法
LSTM
import torch
import torch.nn as nn
class LSTMForecaster(nn.Module):
def __init__(self, input_dim=1, hidden_dim=64, num_layers=2, output_dim=1):
super().__init__()
self.lstm = nn.LSTM(input_dim, hidden_dim, num_layers,
batch_first=True, dropout=0.2)
self.fc = nn.Linear(hidden_dim, output_dim)
def forward(self, x):
# x: (batch, seq_len, features)
lstm_out, (h_n, c_n) = 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 np.array(xs), np.array(ys)
seq_length = 30
X, y = create_sequences(ts.values, seq_length)
X = torch.FloatTensor(X).unsqueeze(-1) # (N, seq_len, 1)
y = torch.FloatTensor(y).unsqueeze(-1) # (N, 1)Transformer 用于时间序列
Transformer 通过 self-attention 机制捕获长距离依赖，在时间序列预测中表现出色。
class TimeSeriesTransformer(nn.Module):
def __init__(self, input_dim=1, d_model=64, nhead=4,
num_layers=3, dim_feedforward=256, pred_len=7):
super().__init__()
self.input_projection = nn.Linear(input_dim, d_model)
self.pos_encoding = nn.Parameter(torch.randn(1, 500, d_model))
encoder_layer = nn.TransformerEncoderLayer(
d_model=d_model,
nhead=nhead,
dim_feedforward=dim_feedforward,
dropout=0.1,
batch_first=True,
)
self.encoder = nn.TransformerEncoder(encoder_layer, num_layers=num_layers)
self.output_projection = nn.Linear(d_model, pred_len)
def forward(self, x):
# x: (batch, seq_len, input_dim)
x = self.input_projection(x)
x = x + self.pos_encoding[:, :x.size(1), :]
encoded = self.encoder(x)
# 使用最后一个时间步的表示预测未来多步
output = self.output_projection(encoded[:, -1, :])
return output方法选择指南
| 场景 | 推荐方法 | 原因 |
|---|---|---|
| 快速基线 | Prophet / ETS | 简单有效，自动处理季节性 |
| 单变量预测 | ARIMA / Prophet | 统计理论扎实 |
| 多变量预测 | VAR / Transformer | 捕获变量间关系 |
| 超长序列 | Transformer | 长距离依赖建模 |
| 异常检测 | STL + Z-Score | 简单可解释 |
| 多序列批量预测 | DeepAR / N-BEATS | 共享模式学习 |
总结
时间序列分析的核心流程是：平稳性检验 -> 分解（趋势、季节性、残差） -> 模型选择 -> 评估。ARIMA 和 Prophet 是经典工具，适合大多数业务场景；LSTM 和 Transformer 在复杂模式和多变量场景中表现更优。评估时应使用时间序列交叉验证，选择与业务目标匹配的指标（MAE 对异常值鲁棒，RMSE 惩罚大误差，MAPE 提供百分比视角）。
贡献者
更新日志
2026/3/14 13:09
查看所有更新日志
9f6c2-于