End-to-End Forecasting with TimesFM 2.5: Backtesting, Covariates, Anomaly Detection, and Scalable Colab Deployment
In this tutorial, we construct a sophisticated end-to-end time-series forecasting workflow with TimesFM 2.5. We start by configuring the runtime, putting in the required dependencies, detecting out there {hardware}, and producing a sensible multi-store retail dataset with pattern, seasonality, pricing, promotions, holidays, temperature results, and random variation. We then load and compile the TimesFM 2.5 mannequin, study its forecast configuration, and use it for zero-shot level and probabilistic forecasting. As we progress, we consider forecast high quality with metrics akin to MAE, RMSE, sMAPE, MASE, pinball loss, and prediction-interval protection, whereas additionally testing batched inference, rolling-origin backtesting, context-length sensitivity, covariate integration by way of XReg, anomaly detection, long-horizon forecasting, throughput tuning, and enter robustness. By working by way of these phases, we develop a sensible understanding of how we configure, validate, benchmark, and deploy TimesFM for practical forecasting duties.
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])
attempt:
import timesfm
besides 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("excessive")
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"gadget : {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)
attempt:
import jax
from sklearn import preprocessing
HAS_XREG_DEPS = True
besides Exception as e:
HAS_XREG_DEPS = False
print(f"[warn] XReg deps lacking ({e}); part 10 will likely be skipped.")
N_DAYS = 1200
N_STORES = 6
REGIONS = ["north", "north", "south", "south", "coast", "coast"]
dates = pd.date_range("2021-01-01", durations=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.regular(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, record(holiday_doy)).astype(int)
rows = []
for s in vary(N_STORES):
degree = 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)
section = np.random.uniform(0, 2 * np.pi)
base_price = np.random.uniform(9.0, 13.0)
worth = base_price + np.random.regular(0, 0.25, N_DAYS)
promo = (np.random.rand(N_DAYS) < 0.09).astype(int)
worth = worth - 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 + section)
gross sales = (degree
+ slope * t
+ weekly
+ yearly
- elasticity * (worth - base_price)
+ promo_lift * promo
+ 55 * is_holiday
+ temp_beta * (temp - 18)
+ np.random.regular(0, 14, N_DAYS))
gross sales = np.clip(gross sales, 5, None)
rows.append(pd.DataFrame({
"date": dates,
"retailer": f"store_{s}",
"area": REGIONS[s],
"gross sales": gross sales.astype(np.float32),
"worth": worth.astype(np.float32),
"promo": promo.astype(np.int32),
"vacation": 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"].distinctive())
print(f"nDataset: {df.form[0]:,} rows | {len(STORES)} shops | "
f"{dates[0].date()} → {dates[-1].date()}")
print(df.head(3).to_string(index=False))
huge = df.pivot(index="date", columns="retailer", values="gross sales")
SEASON = 7
HORIZON = 56
print("nLoading google/timesfm-2.5-200m-pytorch ...")
t0 = time.time()
mannequin = 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,
)
mannequin.compile(timesfm.ForecastConfig(**BASE_CFG))
print("compiled:", {okay: v for okay, v in BASE_CFG.objects() if okay in
("max_context", "max_horizon", "per_core_batch_size")})
def recompile(**overrides):
cfg = {**BASE_CFG, **overrides}
mannequin.compile(timesfm.ForecastConfig(**cfg))
return cfg
We configure the Google Colab surroundings, set up TimesFM and its supporting libraries, detect the out there CPU or GPU, and initialize reproducible random seeds. We generate a sensible multi-store retail dataset containing traits, weekly and yearly seasonality, pricing results, promotions, holidays, temperature variations, and random demand noise. We then load the TimesFM 2.5 mannequin, outline its baseline forecast configuration, compile it, and create a reusable operate for altering mannequin settings in later experiments.
IDX_MEAN, IDX_Q10, IDX_Q50, IDX_Q90 = 0, 1, 5, 9
QUANTILES = [0.1, 0.2, 0.3, 0.4, 0.5, 0.6, 0.7, 0.8, 0.9]
target_store = STORES[0]
sequence = huge[target_store].values.astype(np.float32)
prepare, precise = sequence[:-HORIZON], sequence[-HORIZON:]
level, quant = mannequin.forecast(horizon=HORIZON, inputs=[train.copy()])
print(f"npoint {level.form} # (n_series, horizon)")
print(f"quantile {quant.form} # (n_series, horizon, 10)")
fig, ax = plt.subplots(figsize=(14, 5))
hist_n = 180
ax.plot(dates[-HORIZON - hist_n:-HORIZON], prepare[-hist_n:], shade="#334155",
lw=1.2, label="historical past")
ax.plot(dates[-HORIZON:], precise, shade="#0f172a", lw=1.6, label="precise")
ax.plot(dates[-HORIZON:], level[0], shade="#ea580c", lw=2, label="TimesFM median")
for lo, hello, a in [(1, 9, .12), (2, 8, .16), (3, 7, .20), (4, 6, .24)]:
ax.fill_between(dates[-HORIZON:], quant[0, :, lo], quant[0, :, hi],
shade="#ea580c", alpha=a, lw=0)
ax.axvline(dates[-HORIZON], shade="#94a3b8", ls="--", lw=1)
ax.set_title(f"TimesFM 2.5 zero-shot — {target_store}, {HORIZON}-day horizon "
f"(fan = q10…q90)")
ax.legend(loc="higher left")
ax.xaxis.set_major_formatter(mdates.DateFormatter("%b %Y"))
plt.tight_layout()
plt.present()
print("n--- output anatomy ---")
print("index 0 = MEAN (not q0!). indices 1..9 = q10..q90. index 5 = median.")
print("point_forecast is actually quantile[..., 5]:",
np.allclose(level, quant[..., IDX_Q50]))
print("monotone quantiles (fix_quantile_crossing):",
bool(np.all(np.diff(quant[0, :, 1:], axis=-1) >= -1e-4)))
row = pd.DataFrame({
"index": vary(10),
"that means": ["mean"] + [f"q{int(q*100)}" for q in QUANTILES],
"day+1": quant[0, 0].spherical(1),
f"day+{HORIZON}": quant[0, -1].spherical(1),
})
print(row.to_string(index=False))
print("Interval width grows with horizon — day+1 q10..q90 span "
f"{quant[0,0,9]-quant[0,0,1]:.1f}, day+{HORIZON} span "
f"{quant[0,-1,9]-quant[0,-1,1]:.1f}")
def seasonal_naive(historical past, horizon, season=SEASON):
"""Repeat the final full season ahead — the baseline you could beat."""
reps = int(np.ceil(horizon / season))
return np.tile(historical past[-season:], reps)[:horizon]
def pinball(precise, q, quantiles=QUANTILES):
"""Mean pinball (quantile) loss over q10..q90 — the probabilistic metric."""
out = []
for i, tau in enumerate(quantiles, begin=1):
e = precise - q[:, i]
out.append(np.imply(np.most(tau * e, (tau - 1) * e)))
return float(np.imply(out))
def consider(precise, pred, historical past, q=None, season=SEASON):
precise, pred = np.asarray(precise, float), np.asarray(pred, float)
err = precise - pred
scale = np.imply(np.abs(historical past[season:] - historical past[:-season])) + 1e-9
m = {
"MAE": float(np.imply(np.abs(err))),
"RMSE": float(np.sqrt(np.imply(err ** 2))),
"MAPE%": float(
np.imply(np.abs(err / np.most(np.abs(precise), 1e-9))) * 100
),
"sMAPE%": float(
np.imply(
2 * np.abs(err) /
(np.abs(precise) + np.abs(pred) + 1e-9)
) * 100
),
"MASE": float(np.imply(np.abs(err)) / scale),
}
if q will not be None:
m["pinball"] = pinball(precise, q)
m["cov80%"] = float(
np.imply(
(precise >= q[:, IDX_Q10]) &
(precise <= q[:, IDX_Q90])
) * 100
)
return m
base_pred = seasonal_naive(prepare, HORIZON)
cmp = pd.DataFrame({
"TimesFM 2.5": consider(precise, level[0], prepare, quant[0]),
"seasonal naive": consider(precise, base_pred, prepare),
"final worth": consider(
precise,
np.repeat(prepare[-1], HORIZON),
prepare
),
}).T
print("n--- single-series accuracy ---")
print(cmp.spherical(3).to_string())
print("MASE < 1 means you beat seasonal naive. cov80% ought to sit close to 80.")
inputs = [
wide[s].values[:-HORIZON].astype(np.float32).copy()
for s in STORES
]
t0 = time.time()
b_point, b_quant = mannequin.forecast(
horizon=HORIZON,
inputs=record(inputs)
)
elapsed = time.time() - t0
print(f"n{len(STORES)} sequence forecast in {elapsed:.2f}s "
f"({elapsed/len(STORES)*1000:.0f} ms/sequence)")
per_series = {}
for i, s in enumerate(STORES):
hist = huge[s].values[:-HORIZON]
act = huge[s].values[-HORIZON:]
per_series[s] = consider(
act,
b_point[i],
hist,
b_quant[i]
)
tbl = pd.DataFrame(per_series).T
print(tbl.spherical(3).to_string())
print("imply MASE:", spherical(tbl["MASE"].imply(), 3),
"| imply 80% protection:", spherical(tbl["cov80%"].imply(), 1))
fig, axes = plt.subplots(2, 3, figsize=(16, 7), sharex=True)
for ax, s, i in zip(
axes.ravel(),
STORES,
vary(len(STORES))
):
hist = huge[s].values[:-HORIZON]
ax.plot(
dates[-HORIZON-90:-HORIZON],
hist[-90:],
shade="#475569",
lw=1
)
ax.plot(
dates[-HORIZON:],
huge[s].values[-HORIZON:],
shade="#0f172a",
lw=1.4
)
ax.plot(
dates[-HORIZON:],
b_point[i],
shade="#ea580c",
lw=1.8
)
ax.fill_between(
dates[-HORIZON:],
b_quant[i, :, IDX_Q10],
b_quant[i, :, IDX_Q90],
shade="#ea580c",
alpha=.18,
lw=0
)
ax.set_title(
f"{s} MASE={per_series[s]['MASE']:.2f}",
fontsize=10
)
ax.tick_params(labelsize=8)
fig.suptitle(
"Batched zero-shot forecasts "
"(orange = median, band = 80% PI)"
)
plt.tight_layout()
plt.present()
We generate our first zero-shot forecast and visualize the historic information, precise observations, median predictions, and probabilistic quantile bands in a fan chart. We examine the construction of the purpose and quantile outputs, outline analysis features for MAE, RMSE, MAPE, sMAPE, MASE, pinball loss, and prediction-interval protection, and examine TimesFM towards easy forecasting baselines. We additionally forecast all retailer sequence in a single batch, calculate store-level efficiency metrics, and plot the ensuing forecasts and uncertainty intervals throughout the entire retail dataset.
N_FOLDS = 3 if FAST_MODE else 6
STEP = HORIZON
print(f"n--- rolling-origin backtest: {N_FOLDS} folds x {HORIZON}d x "
f"{len(STORES)} sequence ---")
fold_rows = []
for okay in vary(N_FOLDS):
finish = N_DAYS - okay * STEP
minimize = finish - HORIZON
fold_inputs, fold_actuals, fold_hists = [], [], []
for s in STORES:
v = huge[s].values.astype(np.float32)
fold_inputs.append(v[:cut].copy())
fold_actuals.append(v[cut:end])
fold_hists.append(v[:cut])
fp, fq = mannequin.forecast(
horizon=HORIZON,
inputs=record(fold_inputs)
)
for i, s in enumerate(STORES):
m_tfm = consider(
fold_actuals[i],
fp[i],
fold_hists[i],
fq[i]
)
m_snv = consider(
fold_actuals[i],
seasonal_naive(fold_hists[i], HORIZON),
fold_hists[i]
)
fold_rows.append({
"fold": okay,
"retailer": s,
"mannequin": "TimesFM",
**m_tfm
})
fold_rows.append({
"fold": okay,
"retailer": s,
"mannequin": "snaive",
**m_snv
})
bt = pd.DataFrame(fold_rows)
abstract = bt.groupby("mannequin")[
["MAE", "RMSE", "sMAPE%", "MASE"]
].imply()
print(abstract.spherical(3).to_string())
carry = (
1 -
abstract.loc["TimesFM", "MASE"] /
abstract.loc["snaive", "MASE"]
)
print(
f"TimesFM reduces MASE by "
f"{carry*100:.1f}% vs seasonal naive, zero-shot."
)
fig, ax = plt.subplots(figsize=(10, 4))
piv = bt.pivot_table(
index="fold",
columns="mannequin",
values="MASE"
)
piv.plot(
variety="bar",
ax=ax,
shade=["#ea580c", "#94a3b8"],
width=.8
)
ax.set_ylabel("MASE (decrease is healthier)")
ax.axhline(1.0, shade="okay", ls=":", lw=1)
ax.set_title("Rolling-origin backtest — MASE per fold")
plt.tight_layout()
plt.present()
ctx_grid = (
[128, 512]
if FAST_MODE
else [64, 128, 256, 512, 1024]
)
print(f"n--- context ablation {ctx_grid} ---")
abl = []
for ctx in ctx_grid:
recompile(max_context=ctx)
ins = [
wide[s].values[:-HORIZON][-ctx:]
.astype(np.float32)
.copy()
for s in STORES
]
t0 = time.time()
p, q = mannequin.forecast(
horizon=HORIZON,
inputs=record(ins)
)
dt = time.time() - t0
ms = [
evaluate(
wide[s].values[-HORIZON:],
p[i],
huge[s].values[:-HORIZON],
q[i]
)
for i, s in enumerate(STORES)
]
abl.append({
"context": ctx,
"MASE": np.imply([m["MASE"] for m in ms]),
"sMAPE%": np.imply([m["sMAPE%"] for m in ms]),
"cov80%": np.imply([m["cov80%"] for m in ms]),
"sec": spherical(dt, 2)
})
abl = pd.DataFrame(abl)
print(abl.spherical(3).to_string(index=False))
fig, ax1 = plt.subplots(figsize=(9, 4))
ax1.plot(
abl["context"],
abl["MASE"],
"o-",
shade="#ea580c"
)
ax1.set_xscale("log", base=2)
ax1.set_xlabel("context size (days)")
ax1.set_ylabel("MASE", shade="#ea580c")
ax2 = ax1.twinx()
ax2.bar(
abl["context"],
abl["sec"],
width=abl["context"] * 0.35,
alpha=.25,
shade="#0369a1"
)
ax2.set_ylabel("wall-clock (s)", shade="#0369a1")
ax1.set_title("Accuracy vs. context size vs. price")
plt.tight_layout()
plt.present()
recompile()
We carry out rolling-origin backtesting throughout a number of historic cutoffs to judge TimesFM underneath a number of practical forecasting durations as a substitute of counting on one holdout window. We examine the mannequin with a seasonal-naive baseline, summarize the typical error metrics, calculate the development in MASE, and visualize efficiency throughout backtest folds. We then run a context-length ablation research to find out how completely different quantities of historic information affect forecast accuracy, interval protection, and inference time.
if HAS_XREG_DEPS:
CTX_X, H_X = 512, 56
recompile(
max_context=CTX_X,
max_horizon=128,
return_backcast=True
)
x_inputs = []
dyn_num = {"worth": [], "temp": []}
dyn_cat = {"promo": [], "vacation": [], "dow": []}
stat_cat = {"area": [], "retailer": []}
for s in STORES:
g = df[df["store"] == s].reset_index(drop=True)
n = len(g)
ctx_lo, ctx_hi = n - H_X - CTX_X, n - H_X
x_inputs.append(
g["sales"]
.values[ctx_lo:ctx_hi]
.astype(np.float32)
)
for identify, col in [
("price", "price"),
("temp", "temp")
]:
dyn_num[name].append(
g[col]
.values[ctx_lo:n]
.astype(np.float32)
)
for identify, col in [
("promo", "promo"),
("holiday", "holiday"),
("dow", "dow")
]:
dyn_cat[name].append(
g[col]
.values[ctx_lo:n]
.astype(np.int32)
.tolist()
)
stat_cat["region"].append(g["region"].iloc[0])
stat_cat["store"].append(s)
x_actuals = [
wide[s].values[-H_X:]
for s in STORES
]
x_hists = [
wide[s].values[:-H_X]
for s in STORES
]
print(
f"n--- XReg: inputs len={len(x_inputs[0])}, "
f"covariate len={len(dyn_num['price'][0])} "
f"=> inferred horizon "
f"{len(dyn_num['price'][0]) - len(x_inputs[0])} ---"
)
xreg_results = {}
for mode in ["xreg + timesfm", "timesfm + xreg"]:
t0 = time.time()
p_list, q_list = mannequin.forecast_with_covariates(
inputs=[a.copy() for a in x_inputs],
dynamic_numerical_covariates=dyn_num,
dynamic_categorical_covariates=dyn_cat,
static_categorical_covariates=stat_cat,
xreg_mode=mode,
normalize_xreg_target_per_input=True,
ridge=1.0,
force_on_cpu=False,
)
dt = time.time() - t0
ms = [
evaluate(
x_actuals[i],
np.asarray(p_list[i]),
x_hists[i],
np.asarray(q_list[i])
)
for i in vary(len(STORES))
]
xreg_results[mode] = (
np.asarray(p_list),
np.asarray(q_list),
ms,
dt
)
print(
f"{mode:>16}: "
f"MASE={np.imply([m['MASE'] for m in ms]):.3f} "
f"sMAPE={np.imply([m['sMAPE%'] for m in ms]):.2f}% "
f"({dt:.1f}s)"
)
recompile(
max_context=CTX_X,
max_horizon=128,
return_backcast=False
)
p_uni, q_uni = mannequin.forecast(
horizon=H_X,
inputs=[a.copy() for a in x_inputs]
)
ms_uni = [
evaluate(
x_actuals[i],
p_uni[i],
x_hists[i],
q_uni[i]
)
for i in vary(len(STORES))
]
print(
f"{'univariate':>16}: "
f"MASE={np.imply([m['MASE'] for m in ms_uni]):.3f} "
f"sMAPE={np.imply([m['sMAPE%'] for m in ms_uni]):.2f}%"
)
comp = pd.DataFrame({
"univariate": pd.DataFrame(ms_uni).imply(),
"xreg + timesfm": pd.DataFrame(
xreg_results["xreg + timesfm"][2]
).imply(),
"timesfm + xreg": pd.DataFrame(
xreg_results["timesfm + xreg"][2]
).imply(),
}).T
print("n", comp.spherical(3).to_string())
best_mode = min(
xreg_results,
key=lambda m: np.imply(
[x["MASE"] for x in xreg_results[m][2]]
)
)
p_best = xreg_results[best_mode][0]
fig, axes = plt.subplots(
2,
2,
figsize=(15, 7),
sharex=True
)
for ax, i in zip(axes.ravel(), vary(4)):
ax.plot(
dates[-H_X - 60:-H_X],
x_hists[i][-60:],
shade="#475569",
lw=1
)
ax.plot(
dates[-H_X:],
x_actuals[i],
shade="#0f172a",
lw=1.5,
label="precise"
)
ax.plot(
dates[-H_X:],
p_uni[i],
shade="#94a3b8",
lw=1.6,
ls="--",
label="univariate"
)
ax.plot(
dates[-H_X:],
p_best[i],
shade="#ea580c",
lw=1.8,
label=f"{best_mode}"
)
pr = df[
df["store"] == STORES[i]
]["promo"].values[-H_X:]
for j in np.the place(pr == 1)[0]:
d = dates[-H_X + j]
ax.axvspan(
d - pd.Timedelta(hours=12),
d + pd.Timedelta(hours=12),
shade="#16a34a",
alpha=.25,
lw=0
)
ax.set_title(STORES[i], fontsize=10)
ax.tick_params(labelsize=8)
axes[0, 0].legend(fontsize=8)
fig.suptitle(
"Covariate forecasting "
"(inexperienced = promotion days recognized upfront)"
)
plt.tight_layout()
plt.present()
else:
print(
"n[skip] part 10 — "
"set up scikit-learn + jax to run XReg."
)
recompile()
We incorporate numerical, categorical, and static covariates into the forecasting workflow by way of TimesFM’s XReg performance. We examine the xreg + timesfm and timesfm + xreg fusion modes towards a univariate forecast whereas utilizing worth, temperature, promotion, vacation, weekday, area, and retailer info. We consider the competing approaches, establish the best-performing covariate mode, and visualize how its predictions reply to recognized future promotion durations.
sig = huge[STORES[1]].values.astype(np.float32).copy()
truth_idx = [
N_DAYS - 210,
N_DAYS - 128,
N_DAYS - 61,
N_DAYS - 20
]
sig[truth_idx[0]] *= 1.9
sig[truth_idx[1]] *= 0.35
sig[truth_idx[2]:truth_idx[2] + 5] *= 1.5
sig[truth_idx[3]] *= 0.45
DET_WIN, DET_START = 14, N_DAYS - 280
flags = []
for begin in vary(
DET_START,
N_DAYS - DET_WIN + 1,
DET_WIN
):
ctx = sig[max(0, start - 512):start]
p, q = mannequin.forecast(
horizon=DET_WIN,
inputs=[ctx.copy()]
)
act = sig[start:start + DET_WIN]
lo90 = q[0, :, IDX_Q10]
hi90 = q[0, :, IDX_Q90]
lo_ext = (
q[0, :, IDX_Q10] -
1.5 * (
q[0, :, IDX_Q50] -
q[0, :, IDX_Q10]
)
)
hi_ext = (
q[0, :, IDX_Q90] +
1.5 * (
q[0, :, IDX_Q90] -
q[0, :, IDX_Q50]
)
)
for j in vary(DET_WIN):
sev = (
"CRITICAL"
if (
act[j] < lo_ext[j] or
act[j] > hi_ext[j]
)
else "WARNING"
if (
act[j] < lo90[j] or
act[j] > hi90[j]
)
else "okay"
)
flags.append({
"i": begin + j,
"date": dates[start + j],
"precise": act[j],
"lo": lo90[j],
"hello": hi90[j],
"sev": sev,
"z": (
act[j] -
q[0, j, IDX_Q50]
) / max(
(hi90[j] - lo90[j]) / 2.563,
1e-6
)
})
fl = pd.DataFrame(flags)
alerts = fl[fl.sev != "ok"]
print(
f"n--- anomaly detection: "
f"{len(alerts)} flags in {len(fl)} factors "
f"({len(alerts)/len(fl)*100:.1f}% — "
f"anticipate ~20% by development) ---"
)
print(
alerts[
alerts.sev == "CRITICAL"
][
["date", "actual", "lo", "hi", "z", "sev"]
]
.spherical(2)
.to_string(index=False)
)
print(
"injected anomalies at:",
[str(dates[i].date()) for i in truth_idx]
)
fig, ax = plt.subplots(figsize=(14, 4.5))
ax.plot(
fl["date"],
fl["actual"],
shade="#0f172a",
lw=1.1,
label="noticed"
)
ax.fill_between(
fl["date"],
fl["lo"],
fl["hi"],
shade="#0369a1",
alpha=.15,
lw=0,
label="80% predictive interval"
)
w = fl[fl.sev == "WARNING"]
c = fl[fl.sev == "CRITICAL"]
ax.scatter(
w["date"],
w["actual"],
s=28,
shade="#f59e0b",
zorder=3,
label="warning"
)
ax.scatter(
c["date"],
c["actual"],
s=60,
shade="#dc2626",
zorder=4,
marker="X",
label="crucial"
)
ax.set_title(
"Anomaly detection from TimesFM predictive intervals"
)
ax.legend(loc="higher left", fontsize=8)
plt.tight_layout()
plt.present()
LONG_H = 256
long_train = (
huge[STORES[2]]
.values[:-LONG_H]
.astype(np.float32)
)
long_act = huge[STORES[2]].values[-LONG_H:]
recompile(max_horizon=256)
p_direct, q_direct = mannequin.forecast(
horizon=LONG_H,
inputs=[long_train.copy()]
)
recompile(max_horizon=128)
CHUNK = 64
ctx_roll = long_train.copy()
rec = []
for _ in vary(LONG_H // CHUNK):
p_c, _ = mannequin.forecast(
horizon=CHUNK,
inputs=[ctx_roll[-1024:].copy()]
)
rec.append(p_c[0])
ctx_roll = np.concatenate([
ctx_roll,
p_c[0]
])
p_rec = np.concatenate(rec)
print(f"n--- lengthy horizon ({LONG_H} days) ---")
print(
pd.DataFrame({
"direct": consider(
long_act,
p_direct[0],
long_train,
q_direct[0]
),
f"recursive x{CHUNK}": consider(
long_act,
p_rec,
long_train
),
})
.T
.spherical(3)
.to_string()
)
fig, ax = plt.subplots(figsize=(14, 4.5))
ax.plot(
dates[-LONG_H-120:-LONG_H],
long_train[-120:],
shade="#475569",
lw=1
)
ax.plot(
dates[-LONG_H:],
long_act,
shade="#0f172a",
lw=1.4,
label="precise"
)
ax.plot(
dates[-LONG_H:],
p_direct[0],
shade="#ea580c",
lw=1.8,
label="direct"
)
ax.plot(
dates[-LONG_H:],
p_rec,
shade="#7c3aed",
lw=1.5,
ls="--",
label=f"recursive ({CHUNK}-day chunks)"
)
ax.fill_between(
dates[-LONG_H:],
q_direct[0, :, IDX_Q10],
q_direct[0, :, IDX_Q90],
shade="#ea580c",
alpha=.15,
lw=0
)
ax.legend()
ax.set_title(
"Direct vs recursive long-horizon forecasting"
)
plt.tight_layout()
plt.present()
recompile()
We create an anomaly-detection workflow by evaluating noticed values towards TimesFM’s predictive quantile intervals and assigning warning or crucial severity ranges to uncommon observations. We inject spikes, outages, and sustained shifts right into a retail sequence, calculate interval-based anomaly scores, and visualize the detected occasions alongside the anticipated forecasting vary. We additionally examine direct and recursive long-horizon forecasting to look at how every technique impacts accuracy, uncertainty, and error accumulation over an prolonged prediction interval.
many = [
wide[STORES[i % len(STORES)]]
.values[:-HORIZON]
.astype(np.float32)
.copy()
for i in vary(48)
]
print("n--- throughput (48 sequence, 1024 context) ---")
bench = []
for bs in (
[8, 32]
if FAST_MODE
else [1, 4, 16, 32, 48]
):
recompile(per_core_batch_size=bs)
mannequin.forecast(
horizon=32,
inputs=[many[0].copy()]
)
t0 = time.time()
mannequin.forecast(
horizon=HORIZON,
inputs=record(many)
)
dt = time.time() - t0
bench.append({
"per_core_batch_size": bs,
"sec": spherical(dt, 2),
"sequence/sec": spherical(len(many) / dt, 1)
})
bench = pd.DataFrame(bench)
print(bench.to_string(index=False))
print(
"Tip: additionally attempt force_flip_invariance=False "
"for a ~2x speedup when your sequence are "
"strictly optimistic — it removes the second "
"ahead cross."
)
recompile()
print("n--- robustness checks ---")
soiled = (
huge[STORES[0]]
.values[:-HORIZON]
.astype(np.float32)
.copy()
)
soiled[:20] = np.nan
soiled[400:415] = np.nan
p_nan, _ = mannequin.forecast(
horizon=HORIZON,
inputs=[dirty.copy()]
)
print(
f"(a) NaN enter -> finite output: "
f"{np.isfinite(p_nan).all()}, "
f"MASE "
f"{consider(precise, p_nan[0], prepare)['MASE']:.3f} "
f"(clear: "
f"{consider(precise, level[0], prepare)['MASE']:.3f})"
)
def trim_trailing_nans(a):
a = np.asarray(a, dtype=np.float32)
return (
a[:len(a) - np.argmax(~np.isnan(a[::-1]))]
if np.isnan(a[-1])
else a
)
for L in [8, 32, 128]:
ps, _ = mannequin.forecast(
horizon=14,
inputs=[train[-L:].copy()]
)
print(
f"(b) context={L:>4} -> 14d MASE "
f"{consider("
f"sequence[-HORIZON:-HORIZON+14], "
f"ps[0], "
f"prepare"
f")['MASE']:.3f}"
)
decay = np.clip(
np.linspace(400, 8, 400) +
np.random.regular(0, 6, 400),
0,
None
).astype(np.float32)
recompile(infer_is_positive=True)
p_clip, _ = mannequin.forecast(
horizon=64,
inputs=[decay.copy()]
)
recompile(infer_is_positive=False)
p_free, _ = mannequin.forecast(
horizon=64,
inputs=[decay.copy()]
)
print(
f"(c) decaying sequence: "
f"infer_is_positive=True "
f"min={p_clip.min():+.1f} "
f"(floored at 0) | "
f"False min={p_free.min():+.1f}"
)
recompile()
probe = [
train.copy(),
train.copy(),
train.copy()
]
earlier than = len(probe)
mannequin.forecast(
horizon=16,
inputs=probe
)
print(
f"(d) record size earlier than={earlier than} "
f"after={len(probe)} "
f"<-- cross record(inputs)!"
)
r1, _ = mannequin.forecast(
horizon=32,
inputs=[train.copy()]
)
r2, _ = mannequin.forecast(
horizon=32,
inputs=[train.copy()]
)
print(
f"(e) deterministic: "
f"{np.allclose(r1, r2)}"
)
fc_dates = pd.date_range(
dates[-1] + pd.Timedelta(days=1),
durations=HORIZON,
freq="D"
)
final_in = [
wide[s]
.values
.astype(np.float32)
.copy()
for s in STORES
]
fp, fq = mannequin.forecast(
horizon=HORIZON,
inputs=record(final_in)
)
out = []
for i, s in enumerate(STORES):
out.append(pd.DataFrame({
"date": fc_dates,
"retailer": s,
"forecast": fp[i],
"q10": fq[i, :, IDX_Q10],
"q25": fq[i, :, 3],
"q50": fq[i, :, IDX_Q50],
"q75": fq[i, :, 7],
"q90": fq[i, :, IDX_Q90],
"imply": fq[i, :, IDX_MEAN],
}))
out = pd.concat(out, ignore_index=True)
out.to_csv(
"timesfm_forecasts.csv",
index=False
)
with open("timesfm_backtest.json", "w") as f:
json.dump({
"backtest": abstract.spherical(4).to_dict(),
"context_ablation": abl.spherical(4).to_dict(
orient="data"
),
"throughput": bench.to_dict(
orient="data"
)
}, f, indent=2)
print(
"nWrote timesfm_forecasts.csv "
"and timesfm_backtest.json"
)
print(
out.head(4)
.spherical(1)
.to_string(index=False)
)
print("""
============================ USE IT ON YOUR OWN CSV ==========================
df = pd.read_csv("mydata.csv", parse_dates=["date"]).sort_values("date")
vals = df["value"].astype("float32").to_numpy() # 1-D, evenly spaced!
mannequin = timesfm.TimesFM_2p5_200M_torch.from_pretrained(
"google/timesfm-2.5-200m-pytorch")
mannequin.compile(timesfm.ForecastConfig(
max_context=1024, max_horizon=128, normalize_inputs=True,
per_core_batch_size=32, use_continuous_quantile_head=True,
fix_quantile_crossing=True, infer_is_positive=True))
level, quant = mannequin.forecast(horizon=30, inputs=[vals])
decrease, median, higher = quant[0,:,1], level[0], quant[0,:,9]
CHECKLIST
[ ] common spacing, one row per interval, gaps as NaN (not dropped rows)
[ ] trim trailing NaNs your self
[ ] max_context a number of of 32; max_horizon a number of of 128
[ ] max_context + max_horizon <= 16384; horizon <= 1024 w/ quantile head
[ ] quantile index 0 = MEAN, 1..9 = q10..q90, 5 = median
[ ] infer_is_positive=False for something that may go damaging
[ ] return_backcast=True just for forecast_with_covariates()
[ ] dynamic covariates should cowl context + horizon
[ ] cross record(inputs) — forecast() mutates the record it's given
[ ] backtest with rolling origins and examine towards seasonal naive
=============================================================================""")
We benchmark forecasting throughput throughout completely different per-core batch sizes and measure what number of time sequence we course of per second. We take a look at mannequin robustness with main and inside lacking values, very quick histories, positive-value clipping, mutable enter lists, and deterministic repeated inference. We lastly generate future forecasts for each retailer, export the outcomes and experiment summaries to CSV and JSON information, and present a reusable template for making use of TimesFM 2.5 to our personal time-series dataset.
In conclusion, we accomplished a complete TimesFM 2.5 forecasting pipeline that extends properly past producing a primary prediction. We used probabilistic quantiles to measure uncertainty, examine the mannequin towards seasonal-naive and last-value baselines, and apply rolling-origin backtesting to acquire a extra dependable view of real-world efficiency. We additionally explored how context size, batch dimension, exogenous covariates, compilation flags, and direct or recursive forecasting methods affect accuracy and computational price. Through anomaly detection and robustness checks, we examined how the mannequin behaves with lacking values, quick histories, positive-value constraints, mutable inputs, and lengthy forecast horizons. Finally, we exported forecast, backtest, ablation, and throughput outcomes into reusable information and offered a template for making use of the workflow to our personal evenly spaced time-series information. We completed with a structured and production-oriented basis for adapting TimesFM 2.5 to demand forecasting, operational planning, anomaly monitoring, and different large-scale forecasting purposes.
Check out the Full Codes here. Also, be at liberty to comply with us on Twitter and don’t neglect to affix our 150k+ML SubReddit and Subscribe to our Newsletter. Wait! are you on telegram? now you can join us on telegram as well.
Need to companion with us for selling your GitHub Repo OR Hugging Face Page OR Product Release OR Webinar and so forth.? Connect with us
The publish End-to-End Forecasting with TimesFM 2.5: Backtesting, Covariates, Anomaly Detection, and Scalable Colab Deployment appeared first on MarkTechPost.
