TimesFM 2.5 端到端预测教程：回测、协变量与异常检测
End-to-End Forecasting with TimesFM 2.5: Backtesting, Covariates, Anomaly Detection, and Scalable Colab Deployment
In this tutorial, we build an advanced end-to-end time-series forecasting workflow with TimesFM 2.5. We begin by configuring the runtime, installing the required dependencies, detecting available hardware, and generating a realistic multi-store retail dataset with trend, seasonality, pricing, promotions, holidays, temperature effects, and random variation. We then load and compile the TimesFM 2.5 model, examine its forecast configuration, and use it for zero-shot point and probabilistic forecasting. As we progress, we evaluate forecast quality with metrics such as MAE, RMSE, sMAPE, MASE, pinball loss, and prediction-interval coverage, while also testing batched inference, rolling-origin backtesting, context-length sensitivity, covariate integration through XReg, anomaly detection, long-horizon forecasting, throughput tuning, and input robustness. By working through these stages, we develop a practical understanding of how we configure, validate, benchmark, and deploy TimesFM for realistic forecasting tasks.
FAST_MODE = False
SEED = 7
import subprocess, sys, os, time, json, math, warnings
warnings.filterwarnings("ignore")
def _pip(*args):
subprocess.check_call([sys.executable, "-m", "pip", "install", "-q", *args])
try:
import timesfm
except ImportError:
print("Installing timesfm[torch] ... (~1-2 min)")
_pip("timesfm[torch]")
import timesfm
import numpy as np
import pandas as pd
import torch
import matplotlib.pyplot as plt
import matplotlib.dates as mdates
np.random.seed(SEED)
torch.manual_seed(SEED)
torch.set_float32_matmul_precision("high")
DEVICE = "cuda" if torch.cuda.is_available() else "cpu"
print("=" * 78)
print(f"timesfm : {getattr(timesfm, '__version__', 'n/a')}")
print(f"torch : {torch.__version__}")
print(f"device : {DEVICE}")
if DEVICE == "cuda":
print(f"gpu : {torch.cuda.get_device_name(0)} "
f"({torch.cuda.get_device_properties(0).total_memory/1e9:.1f} GB)")
print("=" * 78)
try:
import jax
from sklearn import preprocessing
HAS_XREG_DEPS = True
except Exception as e:
HAS_XREG_DEPS = False
print(f"[warn] XReg deps missing ({e}); section 10 will be skipped.")
N_DAYS = 1200
N_STORES = 6
REGIONS = ["north", "north", "south", "south", "coast", "coast"]
dates = pd.date_range("2021-01-01", periods=N_DAYS, freq="D")
t = np.arange(N_DAYS)
dow = dates.dayofweek.values
doy = dates.dayofyear.values
temp_base = 18 + 12 * np.sin(2 * np.pi * (doy - 105) / 365.25)
temp = temp_base + np.cumsum(np.random.normal(0, 0.6, N_DAYS)) * 0.15
temp = temp - np.linspace(0, temp[-1] - temp_base[-1], N_DAYS)
holiday_doy = {1, 2, 45, 100, 120, 185, 240, 300, 358, 359, 360, 361, 362, 363, 364, 365}
is_holiday = np.isin(doy, list(holiday_doy)).astype(int)
rows = []
for s in range(N_STORES):
level = 180 + 60 * s
slope = np.random.uniform(0.02, 0.09)
week_amp = np.random.uniform(15, 35)
year_amp = np.random.uniform(20, 45)
elasticity = np.random.uniform(18, 32)
promo_lift = np.random.uniform(35, 70)
temp_beta = np.random.uniform(0.8, 2.2)
phase = np.random.uniform(0, 2 * np.pi)
base_price = np.random.uniform(9.0, 13.0)
price = base_price + np.random.normal(0, 0.25, N_DAYS)
promo = (np.random.rand(N_DAYS) < 0.09).astype(int)
price = price - promo * np.random.uniform(1.2, 2.2)
weekly = week_amp * np.array([0.9, 0.7, 0.7, 0.85, 1.25, 1.8, 1.5])[dow]
yearly = year_amp * np.sin(2 * np.pi * doy / 365.25 + phase)
sales = (level
+ slope * t
+ weekly
+ yearly
- elasticity * (price - base_price)
+ promo_lift * promo
+ 55 * is_holiday
+ temp_beta * (temp - 18)
+ np.random.normal(0, 14, N_DAYS))
sales = np.clip(sales, 5, None)
rows.append(pd.DataFrame({
"date": dates,
"store": f"store_{s}",
"region": REGIONS[s],
"sales": sales.astype(np.float32),
"price": price.astype(np.float32),
"promo": promo.astype(np.int32),
"holiday": is_holiday.astype(np.int32),
"dow": dow.astype(np.int32),
"temp": temp.astype(np.float32),
}))
df = pd.concat(rows, ignore_index=True)
STORES = sorted(df["store"].unique())
print(f"\nDataset: {df.shape[0]:,} rows | {len(STORES)} stores | "
f"{dates[0].date()} → {dates[-1].date()}")
print(df.head(3).to_string(index=False))
wide = df.pivot(index="date", columns="store", values="sales")
SEASON = 7
HORIZON = 56
print("\nLoading google/timesfm-2.5-200m-pytorch ...")
t0 = time.time()
model = timesfm.TimesFM_2p5_200M_torch.from_pretrained(
"google/timesfm-2.5-200m-pytorch"
)
print(f"loaded in {time.time() - t0:.1f}s")
BASE_CFG = dict(
max_context=1024,
max_horizon=256,
normalize_inputs=True,
per_core_batch_size=16,
use_continuous_quantile_head=True,
force_flip_invariance=True,
infer_is_positive=True,
fix_quantile_crossing=True,
return_backcast=False,
)
model.compile(timesfm.ForecastConfig(**BASE_CFG))
print("compiled:", {k: v for k, v in BASE_CFG.items() if k in
("max_context", "max_horizon", "per_core_batch_size")})
def recompile(**overrides):
cfg = {**BASE_CFG, **overrides}
model.compile(timesfm.ForecastConfig(**cfg))
return cfgWe configure the Google Colab environment, install TimesFM and its supporting libraries, detect the available CPU or GPU, and initialize reproducible random seeds. We generate a realistic multi-store retail dataset containing trends, weekly and yearly seasonality, pricing effects, promotions, holidays, temperature variations, and random demand noise. We then load the TimesFM 2.5 model, define its baseline forecast configuration, compile it, and create a reusable function for changing model settings in later experiments.
更进一步：量化金融体系
看懂新闻只是起点——沿量化金融路径，把它变成能交付的工程能力