从趋势中看见未来,从周期中理解规律
时间序列(Time Series)是按时间顺序排列的数据点序列,广泛存在于金融、气象、销售、IoT等领域。时间序列分析的目标是理解过去、预测未来。
| 方法 | 类型 | 适用场景 | 优点 | 缺点 |
|---|---|---|---|---|
| 移动平均 | 平滑 | 短期预测 | 简单 | 滞后 |
| 指数平滑 | 平滑 | 趋势+季节 | 自适应权重 | 参数敏感 |
| ARIMA | 统计 | 非季节平稳 | 理论基础强 | 需平稳 |
| SARIMA | 统计 | 季节性数据 | 处理季节 | 参数多 |
| Prophet | 混合 | 业务预测 | 自动调参 | 精度一般 |
| LSTM | 深度学习 | 复杂非线性 | 强大 | 需大数据 |
季节分解将时间序列拆分为趋势+季节+残差三个部分,是最基础也最重要的分析步骤。
import numpy as np
import pandas as pd
from statsmodels.tsa.seasonal import seasonal_decompose
np.random.seed(42)
# 生成模拟时间序列: 趋势 + 季节 + 噪声
n = 200
t = np.arange(n)
trend = 0.05 * t + 10 # 线性趋势
season = 5 * np.sin(2 * np.pi * t / 12) # 12期季节性
noise = np.random.randn(n) * 1.5
data = trend + season + noise
dates = pd.date_range('2020-01', periods=n, freq='ME')
ts = pd.Series(data, index=dates)
print(f"时间序列: {len(ts)}期, 均值={ts.mean():.2f}")
# 季节分解(加法模型)
decomposition = seasonal_decompose(ts, model='additive', period=12)
print(f"趋势范围: [{decomposition.trend.dropna().min():.2f}, "
f"{decomposition.trend.dropna().max():.2f}]")
print(f"季节范围: [{decomposition.seasonal.min():.2f}, "
f"{decomposition.seasonal.max():.2f}]")
print(f"残差范围: [{decomposition.resid.dropna().min():.2f}, "
f"{decomposition.resid.dropna().max():.2f}]")
ARIMA要求数据平稳(均值、方差不随时间变化)。ADF检验是判断平稳性的标准方法。
from statsmodels.tsa.stattools import adfuller
# 原始序列ADF检验
result = adfuller(ts.dropna())
print(f"原始序列 ADF统计量={result[0]:.4f}, p={result[1]:.4f}")
print(f"→ {'非平稳 ❌' if result[1] > 0.05 else '平稳 ✅'}")
# 一阶差分
ts_diff = ts.diff().dropna()
result_diff = adfuller(ts_diff)
print(f"\n一阶差分 ADF统计量={result_diff[0]:.4f}, p={result_diff[1]:.6f}")
print(f"→ {'非平稳 ❌' if result_diff[1] > 0.05 else '平稳 ✅'}")
# 差分1次后变平稳 → d=1
ACF(自相关函数)和PACF(偏自相关函数)是确定ARIMA参数p和q的关键工具。
from statsmodels.tsa.stattools import acf, pacf
acf_vals = acf(ts_diff, nlags=15)
pacf_vals = pacf(ts_diff, nlags=15)
print(f"ACF前5阶: {np.round(acf_vals[:5], 3)}")
print(f"PACF前5阶: {np.round(pacf_vals[:5], 3)}")
# 找显著滞后阶(超出95%置信区间)
threshold = 1.96 / np.sqrt(len(ts_diff))
acf_sig = [i for i in range(1, len(acf_vals)) if abs(acf_vals[i]) > threshold]
pacf_sig = [i for i in range(1, len(pacf_vals)) if abs(pacf_vals[i]) > threshold]
print(f"显著ACF滞后阶: {acf_sig}")
print(f"显著PACF滞后阶: {pacf_sig}")
# → 提示AR和MA的阶数范围
from statsmodels.tsa.arima.model import ARIMA
from sklearn.metrics import mean_absolute_error, mean_squared_error
# 划分训练/测试集
train_size = int(len(ts) * 0.8)
train, test = ts[:train_size], ts[train_size:]
print(f"训练集: {len(train)}期, 测试集: {len(test)}期")
# 拟合ARIMA(1,1,1)
model = ARIMA(train, order=(1, 1, 1))
fitted = model.fit()
print(f"ARIMA(1,1,1): AIC={fitted.aic:.2f}, BIC={fitted.bic:.2f}")
# 预测
forecast = fitted.forecast(steps=len(test))
mae = mean_absolute_error(test, forecast)
rmse = np.sqrt(mean_squared_error(test, forecast))
mape = np.mean(np.abs((test - forecast) / test)) * 100
print(f"\n预测指标: MAE={mae:.4f}, RMSE={rmse:.4f}, MAPE={mape:.2f}%")
# 对比不同ARIMA参数
results = []
for p, d, q in [(0,1,1), (1,1,0), (1,1,1), (2,1,1), (1,1,2)]:
try:
m = ARIMA(train, order=(p,d,q)).fit()
fc = m.forecast(steps=len(test))
mae_v = mean_absolute_error(test, fc)
results.append((f"ARIMA({p},{d},{q})", m.aic, mae_v))
print(f"ARIMA({p},{d},{q}): AIC={m.aic:.2f}, MAE={mae_v:.4f}")
except:
pass
best = min(results, key=lambda x: x[1])
print(f"\n最佳(AIC): {best[0]}, AIC={best[1]:.2f}")