先看成果：跑完这套流程你会得到什么
假设你手上有一份“每日订单量”的表格，从今天起要预测未来 14 天。传统做法是训练 ARIMA 之类的模型、调参、评估，一轮下来半天没了；零样本时序模型省掉的正是这一步——把历史序列按约定格式传进去，直接拿回未来若干步的点预测和区间。
跟着这篇走完，你会得到四样东西：
1. 一份可复用的数据接入脚本，把 CSV 或数据库里的原始表变成一条规则时间序列（缺失日期补齐、重复时间戳去重、时区统一）。
2. 一个带超时、重试、缓存的 API 调用封装，模型标识和接口路径全部走环境变量，换模型不用改代码。
3. 一次真正跑通的零样本预测：给定历史 180 天，输出未来 14 天的 P10/P50/P90。
4. 一套滚动回测（rolling-origin）流程，把你的预测和“朴素法”“季节朴素法”摆在同一条时间轴上比 MAE、sMAPE、MASE，并输出区间覆盖率——用来回答“这模型到底比现在的做法好在哪”。
先说清楚一件事：Timestral 的模型标识、接口路径、请求字段名这些细节，以官方文档当前版本为准。下面的代码用环境变量和一层适配器把细节隔离掉，你按文档填进去就行，业务逻辑和回测代码不用动。
前置条件清单
开工前确认这几项，能省掉后面 80% 的返工：
- 一份带时间戳的历史数据。至少覆盖 2 个完整业务周期。做日粒度预测，建议给到 6 个月以上；做周粒度，给到 2 年以上。数据太短，零样本模型也只能瞎猜。
- 明确的时间粒度。日、周、小时，选一个，全流程统一。别在源表里是小时、中途聚合成日、最后又按小时评估。
- 一个 API Key 和一个能访问外网的运行环境。多数时序基础模型走托管 API；如果你所在的环境走不了外网，需要先确认官方是否提供私有化或本地推理方案，同样以官方文档为准。
- Python 3.10+ 环境，装好
pandas、numpy、requests、matplotlib（画图用，可选）。 - 一张评估口径备忘。提前写下：预测步长 H 是多少、多久回测一次、用哪个指标做决策。这三件事在评估过程中很容易改变，所以先固定下来。
依赖安装：
```bash
python -m venv .venv && source .venv/bin/activate
pip install pandas numpy requests matplotlib
```
环境变量（写完放进 .env，别提交到 git）：
```bash
export TS_API_BASE="https://<官方文档给出的接口域名>"
export TS_API_PATH="/<官方文档给出的预测接口路径>"
export TS_MODEL_ID="<官方文档给出的时序模型标识>"
export TS_API_KEY="<你的 key>"
```
第 1 步：把预测任务写成一张“任务卡”
零样本预测最容易出问题的地方，是没有把任务定义清楚就直接开始调用。先写一个配置文件，把“预测什么、多长、多细”明确规定下来：
```python
spec.py
SPEC = {
"time_col": "ds", # 时间列名
"target": "orders", # 目标列名
"freq": "D", # 频率：D=日，W=周，h=小时
"horizon": 14, # 每次预测未来多少步
"context": 180, # 喂给模型的历史步数
"quantiles": [0.1, 0.5, 0.9], # 要哪些分位数
}
```
关于 context 的选择，有一条经验规则：至少覆盖 3~5 个业务周期。日粒度数据里，周周期是 7 天，年周期是 365 天，取 180 天能吃到周内规律和部分季节趋势；如果你有强年度季节性（比如羽绒服销量），上下文要拉长到一年以上，或者改在周/月粒度上做。
horizon 则要看你的决策周期：排班按周排，就预测 7~14 天；备货按月的，用周粒度预测 8~12 周通常比日粒度预测 30~90 天更稳。
第 2 步：数据接入与清洗
时序模型对输入形态比想象中敏感：缺一天、多一条重复、时区差 8 小时，都可能让结果明显偏离预期。这一步的目标是把数据整理成“等间隔、无重复、时间戳干净”的一条序列。
如果数据在 PostgreSQL 里：
```sql
-- 按天聚合出订单量，注意时区要显式指定
SELECT date_trunc('day', created_at AT TIME ZONE 'Asia/Shanghai') AS ds,
COUNT(*) AS orders
FROM orders
WHERE created_at >= now() - interval '400 days'
GROUP BY 1
ORDER BY 1;
```
导成 CSV 后读入并做规整：
```python
ingest.py
import pandas as pd
def load_series(path: str, time_col: str, target: str, freq: str) -> pd.DataFrame:
df = pd.read_csv(path, parse_dates=[time_col])
1) 去重：同一时间戳保留最后一条
df = df.sort_values(time_col).drop_duplicates(time_col, keep="last")
2) 补全时间轴：缺失的日期补成 NaN，而不是直接跳过
full_idx = pd.date_range(df[time_col].min(), df[time_col].max(), freq=freq)
df = (df.set_index(time_col)[[target]]
.reindex(full_idx)
.rename_axis(time_col)
.reset_index())
3) 标记缺失，交给后续策略处理
df["is_missing"] = df[target].isna()
return df
def fill_gaps(df: pd.DataFrame, target: str, max_gap: int = 3) -> pd.DataFrame:
"""短缺口用线性插值，长缺口保持 NaN 并记录，避免污染训练上下文。"""
s = df[target]
只在连续缺失不超过 max_gap 的位置插值
filled = s.interpolate(method="linear", limit=max_gap, limit_area="inside")
df[target] = filled
df["still_missing"] = df[target].isna()
return df
```
为什么不用“前向填充”一把梭？因为对销量这类序列，前向填充会把断货当天的 0 复制成一段虚假数据，模型会把这当成真实规律。连续缺失超过 3 天的位置，更稳的做法是把这一段从上下文里剔除，或者退回到上一个可用窗口。
洗完做一次小体检，输出几个数字心里有数：
```python
df = fill_gaps(load_series("orders.csv", "ds", "orders", "D"), "orders")
print("总天数:", len(df))
print("缺失天数:", int(df["still_missing"].sum()))
print("零值天数占比:", round((df["orders"] == 0).mean(), 3))
print("日期是否等间隔:", df["ds"].diff().dropna().nunique() == 1)
```
最后一行如果输出 False，说明时间轴还有问题，别往下走。
第 3 步：把序列切成模型能吃的窗口
零样本模型的输入通常是“一段数值序列，以及可选的时间戳和频率”。不同厂商的字段命名不一样，但结构大同小异。这里写一个构造函数，把 DataFrame 转成列表：
```python
windows.py
import pandas as pd
def make_input(df: pd.DataFrame, target: str, context: int, freq: str) -> dict:
tail = df.tail(context)
if tail[target].isna().any():
raise ValueError("AI 词典：上下文窗口">上下文窗口内仍有缺失值，请先处理或缩短窗口")
return {
"timestamps": [t.strftime("%Y-%m-%d") for t in tail["ds"]],
"values": [float(v) for v in tail[target]],
"freq": freq,
}
def rolling_windows(df: pd.DataFrame, target: str, context: int,
horizon: int, step: int = 1):
"""滚动切窗：每个窗口给出 (输入历史, 未来真实值)。"""
n = len(df)
for end in range(context, n - horizon + 1, step):
hist = df.iloc[end - context:end]
future = df.iloc[end:end + horizon]
yield hist, future
```
两个细节：
一是别把目标序列里的“未来信息”带进去。如果你的数据里有“当日促销标记”“当日总访问量”这类字段，它们只有在当天结束后才能知道，做多步预测时不能当作已知协变量喂进去，否则回测会虚高。可以用的是“已知未来”的变量，比如节假日、已经排好的促销日历。
二是量级差异。有些模型的接口对输入量级不敏感，有些则在你传入百万级数值时表现变差。稳妥做法是保存一个缩放因子，预测完再乘回来：
```python
scale = df["orders"].abs().max() or 1.0
values_scaled = [v / scale for v in values]
... 预测完成后： [p * scale for p in preds]
```
第 4 步：封装 API 调用（超时 / 重试 / 缓存）
不要在主流程里裸写 requests.post。加三样东西：超时、带退避的重试、按输入哈希做本地缓存。回测会重复调用几十上百次，缓存能直接省掉一半时间和费用。
```python
client.py
import hashlib, json, os, time
import requests
BASE = os.environ["TS_API_BASE"].rstrip("/")
PATH = os.environ["TS_API_PATH"]
MODEL = os.environ["TS_MODEL_ID"]
KEY = os.environ["TS_API_KEY"]
CACHE_DIR = ".cache_ts"
os.makedirs(CACHE_DIR, exist_ok=True)
def _cache_key(payload: dict) -> str:
raw = json.dumps(payload, sort_keys=True, ensure_ascii=False)
return hashlib.sha256(raw.encode()).hexdigest()
def forecast(values, horizon, freq="D", quantiles=(0.5,),
max_retries: int = 4) -> dict:
payload = {
"model": MODEL,
"input": {"values": values, "freq": freq},
"horizon": horizon,
"quantiles": list(quantiles),
字段名以官方文档为准，必要时在下面加 units / timestamps 等
}
key = _cache_key(payload)
cache_file = os.path.join(CACHE_DIR, f"{key}.json")
if os.path.exists(cache_file):
with open(cache_file, encoding="utf-8") as f:
return json.load(f)
last_err = None
for attempt in range(max_retries):
try:
resp = requests.post(
f"{BASE}{PATH}",
headers={
"Authorization": f"Bearer {KEY}",
"Content-Type": "application/json",
},
json=payload,
timeout=60,
)
if resp.status_code == 429 or resp.status_code >= 500:
raise RuntimeError(f"可重试状态码 {resp.status_code}: {resp.text[:200]}")
resp.raise_for_status()
data = resp.json()
with open(cache_file, "w", encoding="utf-8") as f:
json.dump(data, f, ensure_ascii=False)
return data
except Exception as e: # noqa: BLE001
last_err = e
time.sleep(min(2 ** attempt, 30)) # 指数退避
raise RuntimeError(f"预测接口连续失败: {last_err}")
```
几点说明：
429（限流）和5xx（服务端问题）值得重试，4xx里的参数错误重试多少次都没用，所以上面只在特定状态码时走重试路径。- 缓存键包含完整请求体，换
horizon或换context会自动生成新条目，不会串结果。 - 如果你的环境限制出网，这里换成官方提供的 SDK 或本地推理接口，其余代码不用改。
再包一层结果解析，把不同返回结构统一成 {0.1: [...], 0.5: [...], 0.9: [...]}：
```python
def parse_forecast(resp: dict, quantiles) -> dict:
返回结构以官方文档为准；常见形式是 resp["outputs"] 或 resp["predictions"]
body = resp.get("outputs") or resp.get("predictions") or resp
out = {}
for q in quantiles:
node = body.get(str(q)) or body.get(f"q{q}") or body.get("median")
out[q] = [float(x) for x in node]
return out
```
这个 parse_forecast 是最需要按文档调整的地方，建议先用一次真实调用把原始返回打印出来看一眼，再写死字段路径。
第 5 步：跑通第一次零样本预测
现在把前面几块拼起来，做一次单点预测：
```python
run_once.py
import json
from spec import SPEC
from ingest import load_series, fill_gaps
from windows import make_input
from client import forecast, parse_forecast
df = load_series("orders.csv", SPEC["time_col"], SPEC["target"], SPEC["freq"])
df = fill_gaps(df, SPEC["target"])
inp = make_input(df, SPEC["target"], SPEC["context"], SPEC["freq"])
scale = max(abs(v) for v in inp["values"]) or 1.0
payload_values = [v / scale for v in inp["values"]]
resp = forecast(payload_values, SPEC["horizon"], SPEC["freq"], SPEC["quantiles"])
preds = parse_forecast(resp, SPEC["quantiles"])
for q, seq in preds.items():
print(f"P{int(q*100)}:", [round(v * scale, 1) for v in seq])
```
如果只拿到一条中位数曲线，说明接口默认只回点预测，需要在请求里显式要分位数（字段名以官方文档为准）。对业务来说，区间往往比点值更有用：P90 用来备货，P10 用来排最低人力，比一个孤零零的数字好决策得多。
拿到结果先做一次“肉眼验证”：把历史最后 60 天和预测的 14 天画在同一张图上，看看曲线有没有出现明显不合理的跳变、负值（销量不该为负）、或者一条笔直的水平线（那通常是解析出错，没取到真值）。
```python
import matplotlib.pyplot as plt
hist = df.tail(60)
p50 = [v * scale for v in preds[0.5]]
p10 = [v * scale for v in preds[0.1]]
p90 = [v * scale for v in preds[0.9]]
future_idx = pd.date_range(df["ds"].iloc[-1] + pd.Timedelta(days=1), periods=len(p50), freq="D")
plt.plot(hist["ds"], hist["orders"], label="history")
plt.plot(future_idx, p50, label="P50")
plt.fill_between(future_idx, p10, p90, alpha=0.2, label="P10-P90")
plt.legend(); plt.xticks(rotation=45); plt.tight_layout()
plt.savefig("forecast_once.png", dpi=130)
```
第 6 步：滚动回测，和基线同台比较
单次预测看着像那么回事，不代表模型可用。真正决定要不要上线的，是滚动回测：把历史切成很多个“当时”，每次只用该时点之前的数据做预测，再和之后的真实值比。
同时必须带上基线。没有基线的预测评估没有意义——很多业务序列用“上周同一天”就能解释掉大部分波动，模型如果打不过季节朴素法，就没有使用价值。
```python
backtest.py
import numpy as np
import pandas as pd
from spec import SPEC
from windows import rolling_windows, make_input
from client import forecast, parse_forecast
def naive(hist, h):
return [hist[-1]] * h
def seasonal_naive(hist, h, m=7):
if len(hist) < m:
return naive(hist, h)
return [hist[-m + (i % m)] for i in range(h)]
def model_forecast(hist_values, h, freq, quantiles):
scale = max(abs(v) for v in hist_values) or 1.0
resp = forecast([v / scale for v in hist_values], h, freq, quantiles)
preds = parse_forecast(resp, quantiles)
return {q: [v * scale for v in seq] for q, seq in preds.items()}
def run_backtest(df, target, context, horizon, freq, quantiles, step=1):
rows = []
for hist, future in rolling_windows(df, target, context, horizon, step):
h_vals = hist[target].tolist()
y_true = future[target].tolist()
origin = hist["ds"].iloc[-1]
m = model_forecast(h_vals, horizon, freq, quantiles)
rows.append({
"origin": origin, "y_true": y_true,
"model": m.get(0.5), "p10": m.get(0.1), "p90": m.get(0.9),
"naive": naive(h_vals, horizon),
"snaive": seasonal_naive(h_vals, horizon, m=7),
"train_tail": h_vals[-28:],
})
return rows
```
step=1 是最严格的逐点回测，样本量最大但也最费调用。数据量大时用 step=7（每周滑动一次），评估结论通常不会变，调用量降到七分之一。
注意 train_tail 这个字段：算 MASE 需要它做分母，不能省。
第 7 步：算指标，并把结果落到业务侧
指标不要只算 MAE。至少四个维度：
```python
metrics.py
import numpy as np
def mae(y, p): return float(np.mean(np.abs(np.array(y) - np.array(p))))
def rmse(y, p): return float(np.sqrt(np.mean((np.array(y) - np.array(p)) ** 2)))
def smape(y, p):
y, p = np.array(y, float), np.array(p, float)
denom = (np.abs(y) + np.abs(p)) / 2
denom = np.where(denom == 0, 1e-8, denom)
return float(np.mean(np.abs(y - p) / denom) * 100)
def mase(y, p, train_tail, m=7):
y, p = np.array(y, float), np.array(p, float)
scale = np.mean(np.abs(train_tail[m:] - train_tail[:-m]))
if scale == 0:
return float("nan")
return float(np.mean(np.abs(y - p)) / scale)
def coverage(y, lo, hi):
y = np.array(y, float)
return float(np.mean((y >= np.array(lo)) & (y <= np.array(hi))))
```
MASE 值得重点解释：它用“季节朴素法在训练段上的平均误差”做分母。MASE < 1 表示你的模型比季节朴素法好，MASE > 1 表示不如直接用上周同一天。这个指标比 MAPE 稳，因为 MAPE 在真实值接近 0 时会大幅失真。
汇总所有回测窗口：
```python
def summarize(rows, m=7):
out = {}
for name in ["model", "naive", "snaive"]:
ys, ps, mases, smapes = [], [], [], []
for r in rows:
y, p = r["y_true"], r[name]
ys += y; ps += p
mases.append(mase(y, p, r["train_tail"], m))
smapes.append(smape(y, p))
out[name] = {
"MAE": round(mae(ys, ps), 2),
"RMSE": round(rmse(ys, ps), 2),
"MASE": round(float(np.nanmean(mases)), 3),
"sMAPE": round(float(np.mean(smapes)), 2),
}
cov = np.mean([coverage(r["y_true"], r["p10"], r["p90"]) for r in rows])
out["P10-P90 覆盖率"] = round(float(cov), 3)
return out
```
结果大致会呈现这样的形态（数字仅为示意，实际以你的数据为准）：模型 MASE 在 0.7~0.9 之间，季节朴素法固定在 1.0 附近；P10-P90 的名义覆盖率是 80%，实测落在 0.75~0.85 之间说明区间校准得还不错，如果只有 0.5，说明区间偏窄，做备货时要手动放宽。
落地上，把每个窗口的结果写回一张宽表，供 BI 或看板直接读：
```python
import pandas as pd
flat = []
for r in rows:
for i, y in enumerate(r["y_true"]):
flat.append({
"origin": r["origin"], "step": i + 1, "y_true": y,
"y_pred": r["model"][i], "p10": r["p10"][i], "p90": r["p90"][i],
})
pd.DataFrame(flat).to_parquet("backtest_result.parquet", index=False)
```
有了这张表，你可以按 step 分组看“预测第 1 天”和“预测第 14 天”的误差差多少——这个衰减曲线比单一总分有用，它直接告诉你：14 天预测里，从第几天开始不值得信。
常见坑与排错
预测出来是一条直线。 十有八九是解析出错，parse_forecast 取到了默认值或把 horizon 当成了序列。先把原始响应 print(json.dumps(resp, ensure_ascii=False)[:500]) 打出来，确认字段路径再写代码。
回测分数好得不像话。 检查三处：上下文窗口有没有用到未来数据；协变量里有没有“当天结束后才知道”的字段；时间轴是不是在 reindex 时被未来数据填过。这三点里任意一条踩中，分数都会虚高。
每次调用结果略有不同。 部分接口在采样上有随机性。如果官方支持固定随机种子或温度参数">温度参数，就在请求里固定住；不支持的话，回测结论要看整体分布，别纠结单次差异。
限流。 回测跑几十个窗口很容易触发限流。上面的退避重试能缓解，但更根本的办法是加缓存、把 step 调大、或者把请求串行化并加一个小的固定间隔。
MASE 是 nan。 说明 train_tail 里季节差分全为 0（比如连续四周完全相同的值）。这种序列本身就没什么可预测的，直接看 MAE 更实在。
时区与夏令时。 跨时区数据一定要在聚合前显式指定时区，否则会出现某天 23 小时或 25 小时的记录，导致日粒度序列出现异常低值或异常高值。用 UTC 存储、展示层再转本地时区，是最省心的做法。
累计值当成增量喂进去。 数据库里的“累计注册用户数”是单调上升的曲线，直接喂给模型，预测出来也是一条上升线，但业务要的是“每日新增”。先把累计值差分再预测。
下一步建议
跑通上面这条链路后，按投入产出比排序，建议依次做这几件事：
第一，扩大回测覆盖面。 现在只测了一条序列。把同一套代码套到 20~50 条同类序列上（不同门店、不同 SKU、不同渠道），看 MASE 的分布而不是平均值。如果有一半序列打不过季节朴素法，那说明这类序列不适合零样本，得考虑微调或换方法。
第二，做粒度对比实验。 同一份数据，分别用日粒度预测 14 天、周粒度预测 2 周，再折算回日总量比较误差。很多业务场景里周粒度反而更准，因为日粒度的噪声被平滑掉了。
第三，把区间用起来。 点预测只能回答“大概多少”，区间才能支撑决策。试着按 P90 备货、按 P50 排常规人力、按 P10 排最低人力，跑一个月看缺货率和人力浪费率的变化，这比看 MAE 更有说服力。
第四，建立监控。 上线后每周自动跑一次滚动回测，把 MASE 和覆盖率画成趋势图。指标突然恶化通常对应业务结构性变化（大促、改价、上新），这时候需要人工介入，而不是让模型继续输出结果。
第五，评估成本与替代方案。 零样本的优势是启动快、不用维护训练管线；劣势是按调用量计费、对超长上下文的支持受接口限制。当序列数量上去以后，算一下“API 调用成本”和“自己训一个小模型的人力成本”哪个更划算，这个账要定期重算。具体价格以官方页面为准。
最后提醒一句：零样本不等于零配置。真正决定效果的往往不是模型本身，而是你在数据清洗、窗口构造、基线选择这三步上花的功夫。把这三步做扎实，换成任何一家时序模型，这套流程都能直接复用。