TimesFM：谷歌开源时序预测模型解析

TimesFM简介

长期以来时间序列预测还停留在”一个任务一套模型”的阶段，每次换数据集都要重新来过。2024年，Google Research推出了TimesFM（Time Series Foundation Model），将大语言模型”预训练+通用泛化”的范式迁移至时间序列预测，标志着该领域终于迎来了属于自己的”GPT时刻”。

传统时序预测的痛点

传统时间序列预测技术分为统计学模型与深度学习模型两大路线：

• 统计学模型（ARIMA、Prophet、ETS）：在复杂非线性时序、多变量关联场景下预测精度不足，且强依赖人工特征工程，跨场景需重复适配。
• 深度学习模型（LSTM、Temporal Fusion Transformer、PatchTST）：需要海量标注样本训练、依赖高端GPU算力，小业务场景下企业无法承担算力与数据标注成本，同时模型泛化能力较差，更换业务数据集需要重新训练拟合。

TimesFM的诞生

TimesFM（Time Series Foundation Model）是 Google Research 推出的开源时间序列基础模型（Foundation Model）。它借鉴了大语言模型（LLM）的训练思想，在海量时间序列数据上进行预训练，使模型能够学习各种时间序列中的共性规律，例如：

• 趋势（Trend）
• 周期性（Seasonality）
• 波动性（Volatility）
• 突变（Change Point）
• 长期依赖关系

与传统模型不同，TimesFM 不需要针对每个业务重新训练，就能够直接进行预测（Zero-shot Forecasting），并在许多公开数据集上取得接近专门训练模型的效果。

TimesFM 到底是什么？

TimesFM 是谷歌研究院（Google Research）于 2024 年提出、并持续迭代的开源时间序列基础模型，论文《A decoder-only foundation model for time-series forecasting》发表于机器学习顶会 ICML 2024。它的核心定位一句话就能说清：一个 Decoder-Only（仅解码器）Transformer，把时间序列切成连续值的 patch 当作 token，在 1000 亿+ 真实世界时间点上自回归「续写」预训练，从而对任意领域、任意频率的序列实现零样本预测。

它借鉴了大型语言模型（LLM）的成功经验，但关键差异在于：自然语言有离散的「词表」，而时间序列每个时刻是连续浮点数——所以 TimesFM 把连续的时间点分组为 patch，每个 patch 就是一个「词元」。这让模型既能复用 Transformer 的成熟工程能力，又贴合时序数据本身的结构。

为什么这件事重要

时间序列预测是金融、能源、零售、供应链的底层基础设施，但传统做法有个长期痛点：每个数据集都要单独调参、单独训练。ARIMA 的 p/d/q、Transformer 的层数/头数、PatchTST 的 patch size……一个场景一套参数，迁移成本极高。

在 Monash 等权威基准上，TimesFM 的零样本表现已经超越 ARIMA、ETS 等传统统计方法，并与 DeepAR、PatchTST 等专为数据训练的监督深度学习模型相当。它把时序预测的门槛，从「机器学习工程」降到了「调一个 API」。

TimesFM的架构简介

Patch 化的 Decoder-Only Transformer

TimesFM 不是简单套用 LLM，而是针对时序做了精准改造。它的推理流程可以拆成四步：把历史序列切成 patch → 用带残差的 MLP 编码成向量 → 因果自注意力的 Transformer 解码 → 自回归生成未来的 patch。

几个值得记住的设计取舍

• 输入/输出 patch 长度：公开 checkpoint 默认 input=32、output=128。input 太小会”看不见”局部形态，output 太小则长预测累计误差变大。
• PerDimScale：用可学习的逐维注意力缩放，替代固定 1/√d，自动聚焦关键时序特征。
• 翻转不变性：保证 TimesFM(aX+b) = a·TimesFM(X)+b，对正负缩放都成立，相当于无成本的数据增强。
• 归一化：默认对输入做归一化，预测值再乘以对应 patch 的归一化因子回到原始量纲。

版本演进：从 1.0 到 2.5

TimesFM 经历了快速迭代，v2.5（当前稳定版）是一次有意思的”反向瘦身”——参数从 500M 砍到 200M，但上下文从 2048 拉长到 16K，还新增了连续分位数头。

概率化预测：不只给一个点，还给区间

v2.5 通过可选的 30M 连续分位数头（quantile head），输出 10%–90% 共 9 个分位数 + 均值，给出完整的概率分布而非单点估计。这意味着你能直接量化预测的不确定性，支撑库存备货、金融风控等风险决策。

优势与局限：它不是银弹

适用 / 优势

• 快速原型验证的数据科学家
• 缺乏大量标注数据的中小业务
• 百万级 SKU / 传感器批量预测
• 需要置信区间的不确定性量化
• SQL 团队无 ML 设施（BigQuery）
• 200M 轻量，消费级 GPU 可跑

局限 / 不适用

• 需要强解释性的合规报告场景
• 极短序列（< 10 个时间点）
• 多变量因果联合建模（可选 MOIRAI-2）
• 极低延迟实时流式推理（需额外框架）
• 高度专业化领域（罕见病、量子物理）
• 黑盒模型，训练数据不透明

TimesFM的使用

本教程基于官方仓库（google-research/timesfm，最新 timesfm==2.0.2）与 timesfm-forecasting/ 示例整理，所有代码均对应 TimesFM 2.5 的 PyTorch 接口。先装好环境，再逐节照做即可。

安装与环境

TimesFM 同时提供 PyTorch（GPU/CPU/Apple Silicon） 与 Flax/JAX（TPU/GPU 高速推理） 两种后端，协变量功能需要额外依赖。

# 基础预测
pip install timesfm[torch]

# 需要 XReg 协变量时
pip install timesfm[xreg]

# TPU / GPU 上的高速推理
pip install timesfm[flax]

本地开发建议

用 uv 管理虚拟环境：uv venv && source .venv/bin/activate && uv pip install -e .[torch]。若你机器上没有 GPU，PyTorch CPU 版也能跑通示例，只是速度较慢。

微调（LoRA）还需额外装：pip install transformers accelerate peft pandas pyarrow scikit-learn。

加载模型与编译

v2.5 的模型类以显式命名暴露，常用的是 TimesFM_2p5_200M_torch（对应 HuggingFace 权重 google/timesfm-2.5-200m-pytorch）。先用 from_pretrained 加载，再用 compile(ForecastConfig(…)) 固化推理配置。

import torch, numpy as np, timesfm

# Ampere 及以上 GPU 建议开启，提升矩阵乘精度/速度
torch.set_float32_matmul_precision("high")

# 1) 加载预训练权重
model = timesfm.TimesFM_2p5_200M_torch.from_pretrained(
    "google/timesfm-2.5-200m-pytorch"
)

# 2) 编译：固化推理配置（只需一次）
model.compile(timesfm.ForecastConfig(
    max_context=1024,
    max_horizon=256,
    normalize_inputs=True,
    use_continuous_quantile_head=True,
    force_flip_invariance=True,
    infer_is_positive=True,
    fix_quantile_crossing=True,
))

# 3) 预测：返回 (点预测, 分位数)
point, quantiles = model.forecast(horizon=12, inputs=[
    np.linspace(0, 1, 100),
    np.sin(np.linspace(0, 20, 67)),
])
point.shape        # (2, 12)       两条序列各 12 步
quantiles.shape    # (2, 12, 10)   均值 + 10th~90th 共 10 个分位

先跑系统检查（可选但推荐）

官方 Agent 技能要求在首次加载前运行 python scripts/check_system.py，检查内存/显存/磁盘/Python≥3.10，避免大模型把机器拖垮。

ForecastConfig 参数速查

compile() 接受的 ForecastConfig 决定了模型行为。下表列出常用字段：

功能①：基础点预测

最简单的用法：把一条一维序列丢进 inputs，指定 horizon，拿到 point 点预测。即使不开分位数头，也会返回点预测（取均值）。

import numpy as np, timesfm
model = timesfm.TimesFM_2p5_200M_torch.from_pretrained("google/timesfm-2.5-200m-pytorch")
model.compile(timesfm.ForecastConfig(max_context=512, max_horizon=52,
                            normalize_inputs=True))

series = np.sin(np.linspace(0, 20, 200))  # 任意 1-D 数组
point, _ = model.forecast(horizon=24, inputs=[series])
point.shape   # (1, 24)

输入要求

inputs 是一个「序列列表」，每条序列是 1-D 数组（list 或 numpy）。单条序列也要包成 [series]。历史长度建议 ≤ max_context；超过会自动截断到上下文窗口。

功能②：分位数概率预测

把 use_continuous_quantile_head=True 编进配置，forecast 会返回第二个值 quantiles，形状为 (N, H, 10)：第 0 维均值，第 1~9 维依次是 10th、20th、…、90th 分位数。

model.compile(timesfm.ForecastConfig(
    max_context=1024, max_horizon=256,
    normalize_inputs=True,
    use_continuous_quantile_head=True,   # ← 关键开关
    fix_quantile_crossing=True,
))

point, q = model.forecast(horizon=24, inputs=[series])
mean    = q[0, :, 0]   # 均值
lower90 = q[0, :, 1]   # 10th 分位（下界）
upper90 = q[0, :, 9]   # 90th 分位（上界）
# 上界 - 下界 = 80% 预测区间宽度，量化不确定性

概率预测的价值

单点预测无法告诉你「有多大把握」。分位数区间可用于库存安全库存设定、金融风控的 VaR、异常告警阈值等需要量化不确定性的决策。

功能③：外生协变量 XReg

概率预测的价值

单点预测无法告诉你「有多大把握」。分位数区间可用于库存安全库存设定、金融风控的 VaR、异常告警阈值等需要量化不确定性的决策。

功能③：外生协变量 XReg

v2.5 重新引入了协变量支持（pip install timesfm[xreg]）。用 forecast_with_covariates() 接入已知的未来信息——比如促销、价格、节假日，显著提升复杂场景精度。

point, quantiles = model.forecast_with_covariates(
    inputs=inputs,                                  # 历史序列列表
    dynamic_numerical_covariates={"price": price_arrays},     # 动态数值协变量（随步变化）
    dynamic_categorical_covariates={"holiday": holiday_arrays},# 动态类别协变量
    static_categorical_covariates={"region": region_labels},  # 静态类别协变量（每条序列一个标签）
    xreg_mode="xreg + timesfm",            # 或 "timesfm + xreg"
)

• 动态协变量：必须同时覆盖历史上下文窗口和预测窗口两段，不能有缺口。
• 静态协变量：每条序列一个标签（如 region/品类），不随时间变化。
• xreg_mode：控制协变量与时序主干的融合顺序；数据质量好时 “xreg + timesfm” 常用，可都试一遍对比。

功能④：批量多序列预测

把多条序列放进同一个 inputs 列表即可一次批量预测——非常适合「百万级 SKU / 传感器」场景。返回结果的第 0 维就是序列索引。

import pandas as pd, json
df = pd.read_csv("all_stores.csv", parse_dates=["date"], index_col="date")

# 每列是一条时间序列
inputs = [df[col].dropna().values.astype("float32") for col in df.columns]

point, quantiles = model.forecast(horizon=30, inputs=inputs)

results = {col: {"forecast": point[i].tolist(),
                 "lower_80": quantiles[i, :, 1].tolist(),
                 "upper_80": quantiles[i, :, 9].tolist()}
           for i, col in enumerate(df.columns)}
with open("batch_forecasts.json", "w") as f:
    json.dump(results, f, indent=2)

功能⑤：回测 backcast

把 return_backcast=True 编入配置，forecast 会额外返回对历史区间的重建（backcast）。它常用于：① 评估模型对历史拟合得好不好；② 协变量工作流的前处理；③ 与真实值对比做「拟合优度」检查。

model.compile(timesfm.ForecastConfig(
    max_context=1024, max_horizon=256,
    normalize_inputs=True,
    return_backcast=True,     # ← 开启回测
))

point, quantiles, backcast = model.forecast(horizon=24, inputs=[series])
# backcast: 模型对输入历史的重建，形状与输入序列对应
# 可对比 backcast 与真实历史，检查异常/漂移

和「效果评估」区分

backcast 是对历史的重建；真正的预测效果评估是把序列末尾切出来当 ground truth，看未来预测准不准。

功能⑥：异常检测

基于分位数区间即可做轻量异常检测：真实值掉出 80% 预测区间（10th~90th）即判为异常。无需训练，开箱即用。

point, q = model.forecast(horizon=H, inputs=[values])

lower90 = q[0, :, 1]   # 10th 分位
upper90 = q[0, :, 9]   # 90th 分位

actual   = test_values
anomalies = (actual < lower90) | (actual > upper90)   # 布尔数组

功能⑦：预测效果评估

把序列末尾 H 步切出当真值，用常见指标量化准确率；分位数头还能算区间覆盖率（CRPS 的近似代理）。

H = 24
train, actual = values[:-H], values[-H:]
point, quantiles = model.forecast(horizon=H, inputs=[train])
pred = point[0]

mae  = np.mean(np.abs(actual - pred))
rmse = np.sqrt(np.mean((actual - pred) ** 2))
mape = np.mean(np.abs((actual - pred) / actual)) * 100
coverage = np.mean((actual >= quantiles[0, :, 1]) &
                   (actual <= quantiles[0, :, 9])) * 100

print(f"MAE: {mae:.2f} | RMSE: {rmse:.2f} | "
      f"MAPE: {mape:.1f}% | 80% PI Coverage: {coverage:.1f}%")

怎么算「准」

点预测看 MAE/RMSE/MAPE；概率预测看区间覆盖率（理想约 80% 对应 80% PI）。覆盖率明显偏离说明不确定性被低估/高估。

功能⑧：性能调优

200M 模型很轻，但批量大时仍需调优。核心是 per_core_batch_size 与「分块预测」。

批大小经验值

# GPU 8GB  : 64
# GPU 16GB : 128
# CPU 8GB  : 8
# CPU 16GB : 32
model.compile(timesfm.ForecastConfig(
    per_core_batch_size=128, ...))

显存不够就分块

CHUNK = 50
results = []
for i in range(0, len(inputs), CHUNK):
    p, q = model.forecast(
        horizon=H, inputs=inputs[i:i+CHUNK])
    results.append((p, q))

• 精度开关：Ampere+ GPU（A100/RTX3090+）务必set_float32_matmul_precision(“high”)。
• 后端选择：GPU/CPU 用 PyTorch；TPU 或追求极致推理速度用 Flax/JAX。
• 量化/混合精度：部署时可结合 BF16、Flash Attention 进一步降显存、提速度（依后端支持）。

功能⑨：LoRA 微调

零样本够用，但遇到强领域特性（特定零售节律、专业传感器）时，可用 HuggingFace Transformers + PEFT（LoRA） 做参数高效微调。2.5 已作为标准 Transformers 模型发布（google/timesfm-2.5-200m-transformers），原生支持 PEFT 适配器与 Flash Attention。

安装

pip install transformers accelerate peft pandas pyarrow scikit-learn

一键训练（官方脚本）

python finetune_lora.py \
  --epochs 20 --batch_size 64 \
  --lr 5e-5 --lora_r 8 \
  --lora_alpha 16 \
  --context_len 64 \
  --horizon_len 13 \
  --output_dir my-adapter

关键要点

• 不要外部归一化：TimesFM 2.5 自带实例归一化（RevIN），直接喂原始值，别自己先标准化。
• 随机窗口采样：每个训练样本是从序列里随机切的 (context, horizon) 窗口，比固定窗口更高效。
• LoRA 目标：target_modules=”all-linear”把 LoRA 挂到所有线性层；r=4 仅增加约6% 可训练参数（~1.4M / ~232M）就能有效适配新领域。
• 保存与复用：走标准save_pretrained / from_pretrained 工作流；评估用 python finetune_lora.py –eval_only –output_dir <adapter>。

功能⑩：命令行与脚本

仓库自带 CLI 脚本，无需写代码即可对 CSV 做预测或做资源预估。

# 对 CSV 直接预测（自动按列批量）
python scripts/forecast_csv.py input.csv \
    --horizon 24 \
    --date-col date \
    --value-cols sales,revenue \
    --output forecasts.csv

# 系统/资源检查
python scripts/check_system.py \
  --num-series 1000 --context-length 1024 \
  --horizon 24 --batch-size 32
# 仅估算显存（不查机器）
python scripts/check_system.py \
  --num-series 5000 --context-length 2048 \
  --estimate-only

功能⑪：Agent 技能集成

TimesFM 自带符合 Agent Skills 规范的 SKILL.md，任何支持该开放标准的 AI Agent（Claude Code、Cursor、Codex、OpenCode 等）都能直接调用它做预测。

# 安装技能到 Agent 的 skills 目录
cp -r timesfm-forecasting/ ~/.claude/skills/
# 或
cp -r timesfm-forecasting/ ~/.cursor/skills/

装好后，Agent 就能用自然语言触发「读取 CSV → 零样本预测 → 画图/告警」，把 TimesFM 嵌进自动化数据工作流。

功能⑫：Google 产品集成

不想写代码？TimesFM 已落地到谷歌多款产品中，SQL/表格团队零门槛可用：

• BigQuery ML 企业级 SQL 调用，规模化与可靠性兼备，适合百万序列批量预测。
• Google Sheets 在日常表格里直接「预测数据」，运营/分析师免开发。
• Vertex AI Model Garden Docker 化端点，方便被 Agent 编排调用。
• 开源库（本教程）PyTorch/Flax 双后端，可嵌入自有工程栈。

常见问题

• Q：预测长度超过 max_horizon 怎么办？把长 horizon 拆成多段自回归，或调大 max_horizon（分位数模式最高 1000）。
• Q：序列太短（<10 点）能用吗？零样本对极短序列效果有限，建议凑够至少数十个点，或做微调。
• Q：需要 GPU 吗？不需要，CPU 可跑；但批量/长上下文建议 GPU。消费级显卡（如 RTX 4090）流畅运行。
• Q：多变量联合建模？当前为单变量基础模型，多变量因果联合建模可关注 MOIRAI-2 等方案；TimesFM 用 XReg 接入外生变量作为近似。

TimesFM实战代码

from datetime import timedelta

import numpy as np
import pandas as pd

# ---------------------- 1. 配置 ----------------------
CSV_PATH     = "data.csv"
HORIZON      = 90
MAX_CONTEXT  = 2048
EXCLUDE_TODAY = True

OUT_CSV   = "forecast.csv"
OUT_HTML  = "forecast.html"
OUT_PNG   = "forecast.png"
OUT_ANOM  = "forecast_anomalies.csv"

Z_THRESHOLD = 3.5
ROLL_WINDOW = 30
HIST_DAYS   = None

# ！！按实际业务修改，或设为 None 表示不过滤
PLAT    = "your_platform"      # 例如 "天猫"
PRODUCT = "your_product"       # 例如 "手机"


def _read_csv_auto(path):
    """自动探测文件编码"""
    for enc in ("utf-8-sig", "utf-8", "gbk"):
        try:
            return pd.read_csv(path, encoding=enc)
        except (UnicodeDecodeError, UnicodeError):
            continue
    return pd.read_csv(path)


def detect_anomalies(s, window=ROLL_WINDOW, threshold=Z_THRESHOLD):
    """
    基于滚动中位数 + MAD 的稳健 z-score 异常检测。
    返回：异常布尔序列、z 值、滚动期望值
    """
    s = s.astype(float)
    min_periods = max(3, window // 3)

    roll_med = s.rolling(window, center=True, min_periods=min_periods).median()
    roll_mad = (s - roll_med).abs().rolling(window, center=True, min_periods=min_periods).median()
    scaled = roll_mad * 1.4826  # MAD -> 近似标准差

    with np.errstate(divide="ignore", invalid="ignore"):
        z = (s - roll_med) / scaled.replace(0, np.nan)

    z = z.replace([np.inf, -np.inf], np.nan)
    flags = z.abs() > threshold

    return flags.fillna(False).astype(bool), z, roll_med


def main():
    # ---------------------- 2. 读取并过滤目标序列 ----------------------
    df = _read_csv_auto(CSV_PATH)
    df["ds"] = pd.to_datetime(df["ds"]).dt.normalize()

    # 剔除“当天”不完整数据
    if EXCLUDE_TODAY:
        today = pd.Timestamp.today().normalize()
        n_before = len(df)
        df = df[df["ds"] < today]
        print(f"[信息] 已剔除当天({today.date()})不完整数据，移除 {n_before - len(df)} 行")

    # 过滤平台/产品
    sub = df.copy()
    if PLAT and "plat_type" in sub.columns:
        sub = sub[sub["plat_type"] == PLAT]
    if PRODUCT and "product_type" in sub.columns:
        sub = sub[sub["product_type"] == PRODUCT]

    if len(sub) == 0:
        raise ValueError("过滤后数据为空，请检查 PLAT / PRODUCT 或 CSV 列名")

    # 按天聚合：min_count=1 让缺失日期保持 NaN，而不是变成 0
    series = sub.set_index("ds")["y"].astype(float).sort_index()
    daily = series.resample("D").sum(min_count=1)

    # 补成连续日序列并插值
    daily = daily.interpolate(method="linear").ffill().bfill()

    # 截断到最近 MAX_CONTEXT 天
    values = daily.values.astype(np.float32)[-MAX_CONTEXT:]
    last_date = daily.index[-1]
    print(f"[信息] {PLAT}/{PRODUCT} 历史点数={len(daily)}，送入模型点数={len(values)}，"
          f"日期范围 {daily.index[0].date()} ~ {last_date.date()}，均值≈{values.mean():.0f}")

    # ---------------------- 3. 加载 TimesFM 2.5 并预测 ----------------------
    import torch
    import timesfm

    torch.set_float32_matmul_precision("high")

    model = timesfm.TimesFM_2p5_200M_torch.from_pretrained("google/timesfm-2.5-200m-pytorch")
    model.compile(timesfm.ForecastConfig(
        max_context=MAX_CONTEXT,
        max_horizon=HORIZON,
        normalize_inputs=True,
        use_continuous_quantile_head=True,
        force_flip_invariance=True,
        infer_is_positive=True,
        fix_quantile_crossing=True,
    ))

    # forecast 返回 (point, quantiles)
    # quantiles 形状 (1, H, 10)：[0]=均值，[1]~[9]=10th~90th 分位
    point, quantiles = model.forecast(horizon=HORIZON, inputs=[values])
    point = point[0]
    quantiles = quantiles[0]

    # 80% 预测区间
    lower_80 = quantiles[:, 1]   # 10th
    upper_80 = quantiles[:, 9]   # 90th

    point_int = np.clip(np.round(point), 0, None).astype(int)
    lower_int = np.clip(np.round(lower_80), 0, None).astype(int)
    upper_int = np.clip(np.round(upper_80), 0, None).astype(int)

    # ---------------------- 4. 组装未来日期与结果 ----------------------
    future_dates = pd.date_range(
        start=last_date + pd.Timedelta(days=1),
        periods=HORIZON,
        freq="D"
    )
    print(f"[信息] 预测起始日={future_dates[0].date()}，结束日={future_dates[-1].date()}（共 {HORIZON} 天）")

    result = pd.DataFrame({
        "plat_type": PLAT,
        "product_type": PRODUCT,
        "ds": future_dates.strftime("%Y-%m-%d"),
        "y_pred": point_int,
        "lower_80": lower_int,
        "upper_80": upper_int,
    })
    result.to_csv(OUT_CSV, index=False, encoding="utf-8-sig")
    print(f"[完成] 已写出预测结果 -> {OUT_CSV}")
    print(result.head(5).to_string(index=False))

    # ---------------------- 5. 历史数据异常检测 ----------------------
    anom_flags, anom_z, anom_exp = detect_anomalies(daily)

    # 0 / 负值强制判为异常
    zero_flags = daily.astype(float) <= 0
    anom_flags = anom_flags | zero_flags.fillna(False)

    anom_dates = daily.index[anom_flags]
    if len(anom_dates):
        print(f"\n[异常检测] 发现 {len(anom_dates)} 个异常日（|稳健z|>{Z_THRESHOLD} 或 值<=0）：")

        # 期望值缺失时用整体中位数兜底，而不是填 0
        median_fallback = daily.median()
        anom_df = pd.DataFrame({
            "ds": anom_dates.strftime("%Y-%m-%d"),
            "y_actual": daily[anom_flags].astype(int).values,
            "y_expected": anom_exp[anom_flags].fillna(median_fallback).round(0).astype(int).values,
            "robust_z": anom_z[anom_flags].round(2).fillna(0).values,
        })
        anom_df["direction"] = np.where(
            anom_df["robust_z"] > 0,
            "偏高(突刺)",
            "偏低(缺口/异常)"
        )
        print(anom_df.to_string(index=False))
        anom_df.to_csv(OUT_ANOM, index=False, encoding="utf-8-sig")
        print(f"[完成] 异常日明细 -> {OUT_ANOM}")
    else:
        print("\n[异常检测] 未发现明显异常日（|稳健z|均在阈值内，且无非正值）")

    # ---------------------- 6. 可视化 ----------------------
    try:
        import plotly.graph_objects as go

        hist = daily if HIST_DAYS is None else daily.iloc[-HIST_DAYS:]

        fig = go.Figure()
        fig.add_trace(go.Scatter(
            x=hist.index, y=hist.values, mode="lines",
            name="历史单量", line=dict(color="#2563eb", width=1.2)
        ))

        # 预测区间
        fig.add_trace(go.Scatter(
            x=future_dates, y=upper_int, mode="lines",
            line=dict(width=0), showlegend=False, hoverinfo="skip"
        ))
        fig.add_trace(go.Scatter(
            x=future_dates, y=lower_int, mode="lines",
            line=dict(width=0), fill="tonexty",
            fillcolor="rgba(245,158,11,0.25)",
            name="80% 预测区间", hoverinfo="skip"
        ))

        # 点预测
        fig.add_trace(go.Scatter(
            x=future_dates, y=point_int, mode="lines",
            name=f"预测({HORIZON}天)", line=dict(color="#f59e0b", width=1.6)
        ))

        # 异常日
        if len(anom_dates):
            fig.add_trace(go.Scatter(
                x=anom_dates, y=daily[anom_flags].values, mode="markers",
                name="异常日",
                marker=dict(color="#dc2626", size=7, symbol="x"),
                hovertemplate="异常日 %{x}<br>实际=%{y}<extra></extra>"
            ))

        title_suffix = f"{PLAT}·{PRODUCT}" if PLAT and PRODUCT else "业务"
        fig.update_layout(
            title=f"{title_suffix} 未来 {HORIZON} 天单量预测（TimesFM 2.5 零样本）",
            xaxis_title="日期",
            yaxis_title="单量",
            font=dict(family="Microsoft YaHei, SimHei, PingFang SC, sans-serif"),
            hovermode="x unified",
            legend=dict(orientation="h", y=1.02),
            template="plotly_white",
            height=520,
            margin=dict(t=60),
        )
        fig.update_xaxes(
            hoverformat="%Y-%m-%d",
            tickformat="%Y-%m",
            tickangle=-45,
        )

        fig.show()
        fig.write_html(OUT_HTML, include_plotlyjs="inline")
        print(f"[完成] 已保存交互式预测图 -> {OUT_HTML}")

        try:
            fig.write_image(OUT_PNG, width=1400, height=520, scale=1.5)
            print(f"[完成] 已保存静态预测图 -> {OUT_PNG}")
        except Exception as pe:
            print(f"[提示] 未生成 PNG（如需请 pip install kaleido）：{pe}")
    except Exception as e:
        print(f"[提示] 绘图跳过（不影响预测结果）：{e}")


if __name__ == "__main__":
    main()

参考资料

• 论文（ICML 2024）：arXiv:2310.10688
• GitHub 仓库：google-research/timesfm
• HuggingFace 模型集：google/timesfm-release
• 官方博客：Google Research Blog
• 官方文档：google-research.github.io/timesfm