上手 TimesFM-3：长表数据、协变量与滚动回测

这篇能做出什么

读完后你手上会有一条能跑通的流水线：把一份宽表销售/流量数据整理成规范长表，做零样本多序列批量预测，把静态协变量和未来已知协变量注入预测结果，最后用滚动回测给出每个序列的 MAE、MASE、分位数损失和区间覆盖率。

具体产出四个东西：

• 一份通过校验的长表（unique_id, ds, y 加协变量列），缺口补齐、频率统一；
• 一个批量预测函数，输入长表和预测步长，输出点预测 + 分位数预测的宽表；
• 一套协变量注入逻辑，未来才知道的变量（促销、节假日）不会泄漏到训练窗；
• 一份滚动回测报告，能回答"这个零样本模型在我的数据上到底行不行"。

先澄清一个容易混淆的点：TimesFM 在架构上是单序列基础模型，它吃一段一维历史，吐这一段未来的分布。"多变量"在这套流程里靠两条路实现——一是多序列批量预测（几百上千条序列共享同一个模型，无需训练），二是通过协变量把序列之间的外部信息接进来。理解这一点，后面的数据组织方式才说得通。

前置条件清单

环境

• Python 环境（版本以官方文档当前版本为准），建议用虚拟环境隔离；
• 依赖：pandas、numpy、torch、scikit-learn，以及 TimesFM 官方库。安装包名和模型权重获取方式以官方仓库 README 为准，本文不写死。

硬件

• CPU 能跑通小批量（几十条序列），速度尚可接受；
• 序列规模上到几百条以上、或者要反复回测，建议用 GPU。显存占用与批大小、上下文长度正相关。

数据

• 每条序列的历史长度不少于模型要求的上下文长度（TimesFM 系列支持的最大上下文随版本提升，具体上限以官方文档为准）；
• 时间戳等间隔、无重复、时区统一；
• 缺失值策略先定好：是补还是截断，别到预测阶段再临时处理。

心理准备

• 零样本不等于不用评估。基础模型的强项是"开箱可用"，不是"开箱即准"，回测环节不能省。

第一步：把模型加载包进适配层

各版本 TimesFM 的类名、compile 参数、分位点列表都可能变。与其在教程里写死，不如把加载逻辑收进一个文件，其余代码只依赖这个文件里的两个函数。

```python

model_adapter.py

"""把 TimesFM 的版本差异挡在这一层。

官方 quickstart 通常是两步：加载 checkpoint，然后用 compile 设定

max_context / max_horizon；预测时调用 forecast(horizon, inputs)。

若你的版本类名或参数名不同，只改这个文件即可。

"""

import numpy as np

import timesfm

def load_model(max_context: int = 512, max_horizon: int = 128):

model = timesfm.TimesFm(...) # 参数按官方 README 填写

model.compile(...) # 参数按官方 README 填写

return model

def forecast_batch(model, inputs, horizon: int):

"""inputs: List[np.ndarray]，每个元素是一条序列的一维历史。

返回 (point, quantiles)

point: (n_series, horizon)

quantiles: (n_series, horizon, n_quantiles)

"""

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

return np.asarray(point, dtype=float), np.asarray(quantiles, dtype=float)

```

compile 里通常需要设置最大上下文和最大预测步长。预测步长不要超过 max_horizon，超了要么报错，要么只能靠滚动外推（把上一段的预测当输入再预测一次），误差会累积，回测时尤其要避免。

第二步：宽表转长表并做 schema 校验

拿到手的原始数据多半是宽表：一行一个日期，一列一条序列。

```python

import pandas as pd

示例宽表：index 是日期，列是各门店

ds store_a store_b store_c

2024-01-01 120 80 45

2024-01-02 132 91 52

def wide_to_long(df_wide: pd.DataFrame, id_name: str = "unique_id",

time_name: str = "ds", value_name: str = "y") -> pd.DataFrame:

df = df_wide.copy()

df.index.name = time_name

long = (

df.reset_index()

.melt(id_vars=time_name, var_name=id_name, value_name=value_name)

.sort_values([id_name, time_name])

.reset_index(drop=True)

)

return long

```

转完立刻校验，别把脏数据带到后面：

```python

def validate_long(df: pd.DataFrame, id_col="unique_id", time_col="ds",

value_col="y", freq="D") -> pd.DataFrame:

report = {}

1) 主键唯一

dup = df.duplicated([id_col, time_col]).sum()

report["duplicated_keys"] = int(dup)

2) 时间戳可解析

df = df.copy()

df[time_col] = pd.to_datetime(df[time_col])

3) 频率一致性：用每个序列的时间差众数判断

diffs = df.groupby(id_col)[time_col].diff().dropna()

report["mode_gap"] = str(diffs.mode().iloc[0]) if len(diffs) else None

4) 每序列长度分布

lens = df.groupby(id_col)[time_col].size()

report["n_series"] = int(len(lens))

report["len_min"] = int(lens.min())

report["len_median"] = float(lens.median())

5) 缺失率

report["missing_rate"] = float(df[value_col].isna().mean())

print(report)

assert dup == 0, "存在重复的 (unique_id, ds)，先聚合再去重"

return df

```

补齐时间轴（有些序列中间缺了几天，直接喂给模型会让"第 7 个点"的含义错位）：

```python

def fill_gaps(df, id_col="unique_id", time_col="ds",

value_col="y", freq="D") -> pd.DataFrame:

out = []

for sid, g in df.groupby(id_col, sort=False):

g = g.set_index(time_col).sort_index()

full_idx = pd.date_range(g.index.min(), g.index.max(), freq=freq)

g = g.reindex(full_idx)

g[id_col] = sid

out.append(g.rename_axis(time_col).reset_index())

return pd.concat(out, ignore_index=True)

```

补齐后 y 会出现新的 NaN。日频零售数据用线性插值加前后填充通常够用；如果是强季节性数据，插值窗口别太长，否则等于凭空造数据。

第三步：把协变量分成三类

这一步做错，后面全盘皆输。按"预测时刻能不能拿到"来划分：

第三类绝对不能进入预测期的特征矩阵。这是时间序列建模最常见的泄漏点：离线回测时整张表都在手上，一不小心就把未来值当成了已知量，线上跑起来指标直接崩。

```python

STATIC_COLS = ["city_tier", "store_area"]

FUTURE_KNOWN_COLS = ["is_promo", "is_holiday", "planned_price"]

PAST_ONLY_COLS = ["footfall_actual", "temp_actual"] # 只允许用在历史窗

```

第四步：组装批量推理的输入

对每条序列取最后 context_len 个点作为输入。要点有四个：

1. 长度不够的序列直接跳过，别补零也别补均值——补出来的假历史会污染预测；

2. 尾部缺失单独处理，末尾如果是 NaN，要么截到最后一个有效点，要么显式填充并记录；

3. 按批切分，批大小从 32 开始试，显存不够就减半；

4. 记录顺序，批处理和打乱是天然一对，务必保证输出顺序能对回 unique_id。

```python

import numpy as np

def build_inputs(long_df, context_len, id_col="unique_id",

time_col="ds", value_col="y"):

ids, inputs, stats = [], [], {"skipped_short": 0, "skipped_all_nan": 0}

for sid, g in long_df.groupby(id_col, sort=False):

y = g.sort_values(time_col)[value_col].to_numpy(dtype=float)

y = pd.Series(y).ffill().bfill().to_numpy() # 中间缺口填充

if len(y) < context_len:

stats["skipped_short"] += 1

continue

window = y[-context_len:]

if np.isnan(window).all():

stats["skipped_all_nan"] += 1

continue

ids.append(sid)

inputs.append(window)

print(stats)

return ids, inputs

```

第五步：跑零样本预测并拼回长表

```python

import numpy as np

import pandas as pd

from model_adapter import load_model, forecast_batch

QUANTILES = [0.1, 0.2, 0.3, 0.4, 0.5, 0.6, 0.7, 0.8, 0.9] # 以官方默认为准

def batch_forecast(long_df, horizon, context_len=512,

batch_size=64, freq="D", **model_kwargs):

model = load_model(max_context=context_len, max_horizon=horizon)

ids, inputs = build_inputs(long_df, context_len)

all_point, all_q = [], []

for i in range(0, len(inputs), batch_size):

chunk = inputs[i:i + batch_size]

point, q = forecast_batch(model, chunk, horizon)

all_point.append(point)

all_q.append(q)

print(f"done {min(i + batch_size, len(inputs))}/{len(inputs)}")

point = np.vstack(all_point)

q = np.vstack(all_q)

生成未来时间戳：以每条序列最后一个观测日为锚点

last_ds = (long_df.sort_values("ds")

.groupby("unique_id")["ds"].max()

.reindex(ids))

frames = []

for k, sid in enumerate(ids):

future_ds = pd.date_range(last_ds.iloc[k], periods=horizon + 1, freq=freq)[1:]

df = pd.DataFrame({

"unique_id": sid,

"ds": future_ds,

"yhat": point[k],

})

for j, qq in enumerate(QUANTILES):

df[f"q{int(qq * 100):02d}"] = q[k, :, j]

frames.append(df)

return pd.concat(frames, ignore_index=True)

```

分位点顺序务必和模型输出对齐。不同版本默认返回的分位点集合可能不同（有的是 10 个、有的带中位数），用 len(QUANTILES) 和输出张量最后一维做一次断言，对不上就照官方文档改。

区间预测常见的毛病是分位数交叉（q10 > q50），可视化前做一次单调化修正：

```python

def fix_crossing(df, qcols):

df = df.copy()

arr = df[qcols].to_numpy()

df[qcols] = np.sort(arr, axis=1) # 简单可靠的单调化

return df

```

第六步：把协变量接进预测

有两种接法，按你所用版本的能力选。

接法 A：模型原生支持协变量

部分版本支持在预测时直接传入协变量矩阵。具体参数名、矩阵形状、静态变量怎么编码（one-hot 还是 embedding）都得以官方文档为准。这条路省事，缺点是版本间差异大，升级时容易踩坑。

接法 B：模型无关的外挂回归

不依赖模型内部实现，做法是：先跑基础预测，再用历史窗口拟合一个线性模型，把协变量对未来走势的修正量叠加上去。核心思路是拟合协变量的相对贡献，而不是让回归去替代基础预测。

```python

from sklearn.linear_model import Ridge

import numpy as np

def xreg_adjust(y_hist, X_hist, X_future, alpha=1.0):

"""返回 (horizon,) 的调整量。

X_hist: (len(y_hist), n_features) 历史期协变量

X_future: (horizon, n_features) 预测期可得协变量

"""

X_hist = np.nan_to_num(X_hist, nan=0.0)

X_future = np.nan_to_num(X_future, nan=0.0)

reg = Ridge(alpha=alpha, fit_intercept=True)

reg.fit(X_hist, y_hist)

contrib_hist = reg.predict(X_hist)

contrib_future = reg.predict(X_future)

只取"相对历史平均水平的增量"，避免把整体水平重复叠加

return contrib_future - contrib_hist.mean()

```

用法：

```python

yhat_adj = yhat_base + xreg_adjust(y_hist, X_hist, X_future)

```

三个注意点：

• 只用第三类之外的变量（静态 + 未来已知），历史窗里也不能混入未来信息；
• alpha 要用回测选，不要在测试集上调；
• 协变量效果不稳定时，宁可不加。基础模型的零样本结果本身就常能打平甚至超过一个调得不好的回归。

第七步：滚动回测

这是判断"能不能用"的唯一手段。做法是多个起点、固定步长往前推：每个 fold 用 [0, origin) 做输入，预测 [origin, origin+horizon)，和真实值比。

```python

def make_folds(series_len, context_len, horizon, n_folds=4, step=None):

step = step or horizon

last_origin = series_len - horizon

folds = []

for k in range(n_folds):

origin = last_origin - k * step

if origin - context_len < 0:

break

folds.append({"ctx_start": origin - context_len,

"origin": origin,

"end": origin + horizon})

return folds[::-1] # 从早到晚

```

指标三件套：

```python

def mae(y_true, y_pred):

return float(np.mean(np.abs(y_true - y_pred)))

def mase(y_true, y_pred, y_insample, seasonality=7):

naive = np.mean(np.abs(y_insample[seasonality:] - y_insample[:-seasonality]))

return float("nan") if naive == 0 else mae(y_true, y_pred) / naive

def wql(y_true, q_pred, quantiles):

"""加权分位数损失，q_pred: (horizon, n_quantiles)"""

total = []

for j, q in enumerate(quantiles):

e = y_true - q_pred[:, j]

total.append(np.maximum(q * e, (q - 1) * e))

return float(2 * np.mean(total))

def coverage(y_true, lo, hi):

return float(np.mean((y_true >= lo) & (y_true <= hi)))

```

把三者串起来跑完整回测：

```python

def rolling_backtest(long_df, horizon, context_len=512,

n_folds=4, seasonality=7, quantile_idx=(0, 8)):

model = load_model(max_context=context_len, max_horizon=horizon)

rows = []

for sid, g in long_df.groupby("unique_id", sort=False):

g = g.sort_values("ds")

y_all = g["y"].ffill().bfill().to_numpy(dtype=float)

ds_all = g["ds"].to_numpy()

for f in make_folds(len(y_all), context_len, horizon, n_folds):

ctx = y_all[f["ctx_start"]:f["origin"]]

truth = y_all[f["origin"]:f["end"]]

if len(truth) < horizon or np.isnan(truth).any():

continue

point, q = forecast_batch(model, [ctx], horizon)

point, q = point[0], q[0]

rows.append({

"unique_id": sid,

"origin": ds_all[f["origin"]],

"mae": mae(truth, point),

"mase": mase(truth, point, ctx, seasonality),

"wql": wql(truth, q, QUANTILES),

"cov80": coverage(truth, q[:, quantile_idx[0]], q[:, quantile_idx[1]]),

})

df = pd.DataFrame(rows)

summary = df.groupby("unique_id")[["mae", "mase", "wql", "cov80"]].mean()

return df, summary

```

看结果时有几个判断标准：

• MASE < 1：比季节朴素法强，这是基础门槛；
• cov80 在 0.7–0.9 之间：区间大致靠谱。远低于 0.8 说明区间太窄，远高于 0.8 说明过于保守；
• 分序列看，不要只看全局均值。少数长尾序列崩盘会被平均值掩盖，按 MASE 排序看最后 10 条。

常见坑与排错

预测步长超过 max_horizon

有些实现会静默截断，有些会报错。先在 compile 阶段把 max_horizon 设到你要的最大步长，别指望自动扩展。

时间戳不规整

跨时区、夏令时、月末日期，都可能导致 pd.date_range 生成的未来时间戳和真实业务日期对不上。统一转成固定时区、固定频率再进模型。

尾部缺失没处理，预测整体偏移

最后几个点是 NaN 时，模型可能把 NaN 当 0 或直接算进归一化统计量，预测水平会莫名偏移一个台阶。在 build_inputs 里显式截断到最后一个有效点。

分位数交叉

本文的排序修正能解决展示问题，但它改变的是事后结果。如果大面积交叉，说明该序列的分布预测本身不稳定，优先看数据质量。

协变量泄漏

最典型的现象：回测指标好得不真实（MASE 小于 0.5），上线后一落千丈。检查方法是把所有"仅历史可知"的变量置零再跑一遍，指标如果暴涨，说明之前吃到了不该吃的信息。

批大小导致 OOM

显存和 batch_size × context_len 大致成正比。回测时 fold 数量多，累计占用容易翻倍，建议回测阶段把 batch_size 调小。

尺度差异巨大的序列混在一批

模型内部一般会做归一化，但如果自己先做了 log 变换或除以量纲，记得在拼回结果时逐序列还原，别全局统一还原。

季节性对齐

horizon 和业务周期（7 天 / 30 天）不匹配时，回测的 fold 起点会系统性地落在周内同一天，评估结果有偏。让 step 等于周期，或随机化起点后多跑几轮。

下一步建议

先建基线再谈优化。季节朴素法、按序列均值、以及一个梯度提升树模型，三者作为对照。TimesFM 零样本只有在明显打赢这三者时才值得继续投入。

做概率校准。回测里如果把 cov80 长期偏离 0.8，可以按序列做分位数缩放（用历史覆盖率反推缩放系数），这是投入产出比很高的一步。

考虑微调。零样本在大促、断货、新品这类分布外场景容易失手。有足够历史时，用少量样本做微调通常比堆协变量有效，但需要独立验证集，且要按官方文档的微调流程走。

分层一致性。门店预测加总不等于区域预测时，用自顶向下或自底向上的调和方式统一，这在报表场景几乎必做。

上线的三件事：定时任务里固定随机种子和模型版本；每天记录预测值和分位区间，用于事后对账；监控 MASE 和覆盖率的漂移，超过阈值触发重训或回退。零样本模型的优势是免训练，但它同样会随数据分布漂移而失效，别把它当成一次性配置。