时间序列预测实战：ARIMA、Prophet 与 LSTM

y(t) = 趋势 T(t) + 季节 S(t) + 循环 + 残差 ε(t)
      （长期方向）（固定周期）      （随机噪声）

import pandas as pd
from statsmodels.tsa.seasonal import seasonal_decompose

decomp = seasonal_decompose(series, model='additive', period=12)
decomp.trend.plot(); decomp.seasonal.plot(); decomp.resid.plot()

from statsmodels.tsa.stattools import adfuller

adf, p, *_ = adfuller(series)
print(f'ADF p-value: {p}')
# p < 0.05 → 平稳（拒绝非平稳假设）
# p >= 0.05 → 非平稳，需差分

series_diff = series.diff().dropna()   # 一阶差分

from statsmodels.graphics.tsaplots import plot_acf, plot_pacf

plot_acf(series_diff)    # ACF：拖尾或截尾？
plot_pacf(series_diff)   # PACF：定 AR 阶数 p

from statsmodels.tsa.holtwinters import ExponentialSmoothing

# Holt-Winters：趋势 + 季节
model = ExponentialSmoothing(
    series, trend='add', seasonal='add', seasonal_periods=12
).fit()
pred = model.forecast(12)

p = AR 阶（自回归，用过去值预测）
d = 差分阶（去趋势，见第 2 节）
q = MA 阶（误差移动平均）

from statsmodels.tsa.arima.model import ARIMA
import itertools

# 网格搜索最佳 (p, d, q)，用 AIC 最小
best_aic, best_cfg = float('inf'), None
for p, d, q in itertools.product(range(3), range(2), range(3)):
    try:
        model = ARIMA(series, order=(p, d, q)).fit()
        if model.aic < best_aic:
            best_aic, best_cfg = model.aic, (p, d, q)
    except Exception:
        pass

model = ARIMA(series, order=best_cfg).fit()
pred = model.forecast(12)
print('最佳阶数:', best_cfg, 'AIC:', best_aic)

from prophet import Prophet
import pandas as pd

# 数据格式：ds（日期）+ y（值）
df = pd.DataFrame({'ds': dates, 'y': values})

model = Prophet(
    yearly_seasonality=True,
    weekly_seasonality=True,
    daily_seasonality=False,
    changepoint_prior_scale=0.05,   # 趋势突变灵敏度
)
# 加节假日（电商大促等）
model.add_country_holidays(country_name='CN')

model.fit(df)
future = model.make_future_dataframe(periods=30)   # 未来 30 天
forecast = model.predict(future)

import torch, torch.nn as nn
import numpy as np

# 构造滑窗样本：用前 W 步预测下一步
def make_windows(series, window=12):
    X, y = [], []
    for i in range(len(series) - window):
        X.append(series[i:i+window])
        y.append(series[i+window])
    return np.array(X, dtype=np.float32), np.array(y, dtype=np.float32)

X, y = make_windows(values, window=12)

class TimeLSTM(nn.Module):
    def __init__(self, window=12, hidden=32):
        super().__init__()
        self.lstm = nn.LSTM(1, hidden, batch_first=True)
        self.fc = nn.Linear(hidden, 1)
    def forward(self, x):            # x: (B, W, 1)
        out, _ = self.lstm(x)
        return self.fc(out[:, -1])   # 取最后一步 → 预测

model = TimeLSTM()
# 训练循环同标准流程（归一化、损失 MSE、AdamW）

df['year'] = df['ds'].dt.year
df['month'] = df['ds'].dt.month
df['dayofweek'] = df['ds'].dt.dayofweek
df['is_holiday'] = df['ds'].isin(holidays)
df['lag_1'] = df['y'].shift(1)          # 过去值
df['rolling_7'] = df['y'].rolling(7).mean()   # 滚动均值

# 随机切分会数据泄露（未来信息进训练）→ 必须按时间切
train = df[df['ds'] < '2025-01-01']
test  = df[df['ds'] >= '2025-01-01']

# 滚动验证（walk-forward）：更真实的时序评估
# 不断前移训练窗口预测下一段，累计误差

from sklearn.metrics import mean_absolute_error, mean_squared_error
import numpy as np

mae = mean_absolute_error(y_test, y_pred)
rmse = np.sqrt(mean_squared_error(y_test, y_pred))
# MAPE（百分比误差，业务可读）
mape = np.mean(np.abs((y_test - y_pred) / y_test)) * 100