!pip install -q chronos-forecasting
import numpy as np, pandas as pd, torch, matplotlib.pyplot as plt
from chronos import BaseChronosPipeline
model = BaseChronosPipeline.from_pretrained("amazon/chronos-2", device_map="cpu")
print("ready")1 Спрогнозуйте власну метрику за допомогою foundation model
School of Digital Business 5.0 — Часові ряди в епоху AI.
Чотири кроки: оберіть ряд → бектест → прочитайте результат → ухваліть рішення. Запускайте клітинки згори донизу (Shift + Enter). Під час першого запуску модель завантажується — близько 500 МБ, хвилина-дві.
Лише публічні або анонімізовані дані. Усе працює всередині вашої сесії Colab; нічого не надсилається стороннім сервісам прогнозування.
1.1 0. Встановлення
1.2 1. Оберіть ряд
Варіант A — один із десяти продуктів з лекції: вкажіть PRODUCT нижче.
Варіант B — власний CSV з двома колонками, дата й значення, один рядок на день (або на тиждень). Вкажіть PRODUCT = None, запустіть клітинку й завантажте файл, коли з’явиться запит.
PRODUCT = "ChatGPT" # ChatGPT, Duolingo, Spotify, Netflix, Tinder, Strava, TikTok, Notion, Threads, Black Friday
if PRODUCT:
url = ("https://raw.githubusercontent.com/Aranaur/aranaur.rbind.io/main/"
"lectures/genesis/sdb/data/pageviews.csv")
data = pd.read_csv(url, parse_dates=["ds"])
y = data[data["unique_id"] == PRODUCT].set_index("ds")["y"].dropna()
else:
from google.colab import files
uploaded = files.upload()
raw = pd.read_csv(next(iter(uploaded)))
raw.columns = ["ds", "y"] + list(raw.columns[2:])
y = raw.assign(ds=pd.to_datetime(raw["ds"])).set_index("ds")["y"].sort_index().dropna()
y = y[y.index >= y.index.max() - pd.Timedelta(days=3 * 365)] # the last three years are plenty
print(f"{len(y)} точок, з {y.index.min():%Y-%m-%d} по {y.index.max():%Y-%m-%d}")
y.plot(figsize=(11, 3), title="Ваш ряд"); plt.show()1.3 2. Бектест: модель проти «як тиждень тому»
Шість вікон по 28 кроків наприкінці історії. У кожному вікні обидва прогнозисти бачать лише те, що було до нього, — рівно те, що ви мали б того дня.
SEASON — довжина повторюваного патерну: 7 для щоденних даних з тижневим ритмом, 52 для щотижневих з річним.
H, WINDOWS, SEASON = 28, 6, 7
rows = []
for k in range(WINDOWS):
end = len(y) - (WINDOWS - k) * H
history, actual = y.iloc[:end], y.iloc[end:end + H].to_numpy()
q, _ = model.predict_quantiles([torch.tensor(history.to_numpy(), dtype=torch.float32)],
prediction_length=H, quantile_levels=[0.1, 0.5, 0.9])
q = np.asarray(q[0]).reshape(H, 3)
rule = history.to_numpy()[-SEASON:][np.arange(H) % SEASON]
rows.append(pd.DataFrame({"ds": y.index[end:end + H], "actual": actual, "rule": rule,
"P10": q[:, 0], "P50": q[:, 1], "P90": q[:, 2], "window": k + 1}))
bt = pd.concat(rows)
past = y.iloc[:len(y) - WINDOWS * H].to_numpy()
scale = np.mean(np.abs(past[SEASON:] - past[:-SEASON])) # typical miss of the rule, before the test
mase = lambda f: np.mean(np.abs(bt["actual"] - bt[f])) / scale
wins = sum(np.mean(np.abs(g.actual - g.P50)) < np.mean(np.abs(g.actual - g.rule)) for _, g in bt.groupby("window"))
print("MASE як тиждень тому: {rule:.2f} Chronos-2: {model:.2f} (менше — краще)".format(rule=mase("rule"), model=mase("P50")))
print("Chronos-2 краща у {k} з {n} вікон".format(k=wins, n=WINDOWS))
print("Частка фактичних значень усередині P10-P90: {share:.0%} (близько 80%, якщо відкалібровано)".format(share=np.mean((bt.actual >= bt.P10) & (bt.actual <= bt.P90))))
fig, ax = plt.subplots(figsize=(12, 3.5))
ax.plot(y.index[-(WINDOWS * H + 60):], y.iloc[-(WINDOWS * H + 60):], color="black", lw=1, label="факт")
for _, g in bt.groupby("window"):
ax.fill_between(g.ds, g.P10, g.P90, color="#e64173", alpha=0.2)
ax.plot(g.ds, g.P50, color="#e64173", lw=1.5)
ax.plot(g.ds, g.rule, "--", color="#181485", lw=1)
ax.set_title("Бектест: Chronos-2 (рожевий, смуга P10-P90) і «як тиждень тому» (пунктир)")
plt.show()1.4 3. Прогноз, яким ви б користувалися
Якщо бектест каже, що модель того варта, — ось наступні 28 кроків із діапазоном.
q, _ = model.predict_quantiles([torch.tensor(y.to_numpy(), dtype=torch.float32)], prediction_length=H,
quantile_levels=[0.1, 0.5, 0.9])
q = np.asarray(q[0]).reshape(H, 3)
step = y.index[-1] - y.index[-2]
future = pd.date_range(y.index[-1] + step, periods=H, freq=step)
forecast = pd.DataFrame({"P10": q[:, 0], "P50": q[:, 1], "P90": q[:, 2]}, index=future)
fig, ax = plt.subplots(figsize=(12, 3.5))
ax.plot(y.index[-120:], y.iloc[-120:], color="black", lw=1)
ax.fill_between(future, forecast.P10, forecast.P90, color="#e64173", alpha=0.2, label="P10-P90")
ax.plot(future, forecast.P50, color="#e64173", lw=2, label="P50")
ax.legend(); plt.show()
forecast.round(0).head(10)1.5 4. Ухваліть рішення — запишіть його
Дві-три фрази для вашої команди:
- Чи обіграє модель правило? Наскільки (MASE) і в скількох із шести вікон?
- Чи можна довіряти діапазону? Частка всередині P10–P90 близька до 80% чи далека від цього?
- Що б ви спланували за цим прогнозом — і що змусило б вас змінити план?
- Чого модель не знає про наступні 28 днів — кампанія, запуск, свято?