时序预测实战：从 ARIMA 到时序基础模型

import numpy as np
import pandas as pd

def make_series(n=365*2, period=7, trend=0.02, noise=1.0):
    """构造含趋势与周期的演示序列。"""
    t = np.arange(n)
    y = trend * t + 10 * np.sin(2 * np.pi * t / period)
    y += noise * np.random.randn(n)
    return pd.Series(y, index=pd.date_range("2024-01-01", periods=n, freq="D"))

series = make_series()
print(series.head())

import pandas as pd
from statsmodels.tsa.arima.model import ARIMA
from statsmodels.tsa.seasonal import seasonal_decompose

# 季节分解：把序列拆成 trend / seasonal / resid
series = make_series()
result = seasonal_decompose(series, model="additive", period=7)
trend, seasonal, resid = result.trend, result.seasonal, result.resid

# ARIMA 拟合与预测
model = ARIMA(series, order=(1, 1, 1), seasonal_order=(1, 1, 1, 7))
fit = model.fit()
forecast = fit.forecast(steps=14)   # 预测未来 14 天
print(forecast)

from statsmodels.tsa.stattools import adfuller

def check_stationarity(series: pd.Series) -> bool:
    result = adfuller(series.dropna())
    print(f"ADF p-value = {result[1]:.4f}")
    return result[1] < 0.05   # True 表示平稳

print("原始序列平稳?", check_stationarity(series))
diff = series.diff().dropna()
print("一阶差分后平稳?", check_stationarity(diff))

import lightgbm as lgb
import numpy as np
import pandas as pd

def build_features(df: pd.DataFrame, lags=[1, 7, 14], windows=[7, 30]):
    df = df.copy()
    df["dayofweek"] = df.index.dayofweek
    df["month"] = df.index.month
    df["is_holiday"] = df.index.isin(holidays)
    for lag in lags:
        df[f"lag_{lag}"] = df["y"].shift(lag)
    for w in windows:
        df[f"rolling_mean_{w}"] = df["y"].rolling(w).mean()
        df[f"rolling_std_{w}"] = df["y"].rolling(w).std()
    return df.dropna()

train = build_features(df_train)
X, y = train.drop("y", axis=1), train["y"]

model = lgb.LGBMRegressor(
    n_estimators=800, learning_rate=0.05, num_leaves=63,
    subsample=0.9, colsample_bytree=0.9, reg_alpha=0.1,
    random_state=42,
)
model.fit(X, y, eval_set=[(X_val, y_val)],
          callbacks=[lgb.early_stopping(50)])
print("特征重要度 TOP5:", model.feature_importances_)

from sklearn.model_selection import TimeSeriesSplit

# 时序交叉验证：保证只学过去、不偷看未来
tscv = TimeSeriesSplit(n_splits=5)
for train_idx, val_idx in tscv.split(X):
    X_train, X_val = X.iloc[train_idx], X.iloc[val_idx]
    y_train, y_val = y.iloc[train_idx], y.iloc[val_idx]
    # 训练 + 评估（绝不 shuffle）

import torch
import torch.nn as nn

class TCNBlock(nn.Module):
    """膨胀因果卷积块：dilation 指数扩大感受野。"""
    def __init__(self, in_ch, out_ch, kernel_size=3, dilation=2):
        super().__init__()
        self.conv = nn.Conv1d(
            in_ch, out_ch, kernel_size, dilation=dilation,
            padding=(kernel_size - 1) * dilation,
        )
        self.chomp = nn.ConstantPad1d((0, -(kernel_size - 1) * dilation), 0)
        self.relu = nn.ReLU()
        self.dropout = nn.Dropout(0.2)

    def forward(self, x):
        return self.dropout(self.relu(self.chomp(self.conv(x))))

# 3 层 TCN，dilation = 1,2,4 → 感受野 = 1 + 2*(2^0+2^1+2^2)*kernel

from sklearn.preprocessing import StandardScaler

scaler = StandardScaler()
X_train_norm = scaler.fit_transform(X_train)   # 只用训练集 fit
X_val_norm = scaler.transform(X_val)           # 验证集复用同一缩放器
X_test_norm = scaler.transform(X_test)

# TimesFM（Google）时序基础模型：Zero-shot 预测
import torch
from transformers import TimesFMForPrediction

model = TimesFMForPrediction.from_pretrained("google/timesfm-2.0-500m")
context = torch.tensor([series_values], dtype=torch.float32)
output = model.generate(
    input_ids=context,          # 历史序列（context）
    num_return_sequences=1,
    max_new_tokens=14,          # 预测未来 14 步
    temperature=0.0,            # 确定性预测
)
print(output[0].tolist())

def recursive_forecast(model, history, steps, exog_future=None):
    """递归预测：上一步预测值作为下一步的滞后特征。"""
    preds = []
    for s in range(steps):
        features = build_last_window(history, s, exog_future)
        yhat = model.predict(features)[0]
        preds.append(yhat)
        history = np.append(history, yhat)   # 用预测值续历史
    return preds

import numpy as np

def pinball_loss(y_true, y_pred, q=0.8):
    diff = y_true - y_pred
    return np.mean(np.where(diff >= 0, q * diff, (q - 1) * diff))

# 输出 80% 分位预测：只关心「以 80% 概率不被超卖的库存量」
inventory_q80 = model.predict(x)   # 模型输出 q=0.8 分位
loss = pinball_loss(y_true, inventory_q80, q=0.8)

# 生产滚动预测服务的简化实现（FastAPI）
from fastapi import FastAPI
import joblib

app = FastAPI()
model = joblib.load("forecast_model.pkl")

@app.post("/forecast")
def forecast(body: dict):
    """输入：最新历史序列 + 外生变量；输出：未来 14 步预测 + 分位数。"""
    history = body["history"]            # 最近窗口的观测值
    exog = body.get("exog", {})          # 促销、节假日等
    point = model.predict(history, exog)          # 点预测
    q_low = model.predict_quantile(history, exog, 0.1)   # 10% 分位
    q_high = model.predict_quantile(history, exog, 0.9)  # 90% 分位
    return {"forecast": point, "interval": [q_low, q_high]}

# 生产监控告警配置示例
monitoring:
  metrics:
    - mape_7d: { threshold: 0.25, action: "warn" }
    - mape_30d: { threshold: 0.35, action: "retrain" }
  drift:
    feature_psi: { threshold: 0.25, action: "investigate" }
  retraining:
    schedule: "daily"          # 每日滚动重训
    data_window: "90d"         # 用最近 90 天数据