Darts: тестируем временные ряды с нуля05.10.2024 10:15

44a8613644599478c2021269af0aeb57.png

Привет, Хабр!

Если вы когда-либо имели дело с временными рядами, то, вероятно, слышали о Darts. А для тех, кто ещё в танке: Darts — это мощный инструмент, который поддерживает мультиварибельные временные ряды и легко интегрируется с PyTorch и TensorFlow.

Зачем же тестировать временные ряды, когда в классическом машинном обучении всё так просто с кросс-валидацией? Временные ряды обладают своей изюминкой: они подвержены временным зависимостям, сезонности, трендам и другим радостям жизни. Так что, если вы хотите, чтобы ваши модели не провалились на тестах, время разобраться с их особенностями!

Основы тестирования временных рядов

Перед тем как приступать к работе с данными, нужно убедиться, что они в правильном формате.

Формат данных

Первым делом нужно проверить временные метки на корректность. Важно, чтобы не было пропусков! С помощью pandas это делается легко: 

import pandas as pd

# Загружаем данные
data = pd.read_csv('your_data.csv', parse_dates=['timestamp'], index_col='timestamp')

# Проверка на пропуски
missing_values = data.isnull().sum()
print(f"Пропуски в данных:\n{missing_values}")

# Проверка на равномерность временных меток
time_diff = data.index.to_series().diff()
print(f"Минимальный интервал между временными метками: {time_diff.min()}")
print(f"Максимальный интервал между временными метками: {time_diff.max()}")

Распределение данных

Важно убедиться, что данные равномерно распределены. Если выйдет так, что обнаружатся большие интервалы без наблюдений, может возникнуть необходимость в ресэмплинге или интерполяции. Помните, что данные должны быть в форме, удобной для анализа.

Перед тем как запустить свою модель, выполним несколько предобработок, которые могут существенно повлиять на точность прогнозов.

Скейлинг: Приведение данных к общему масштабу поможет вашему алгоритму быстрее схватывать закономерности. Используйте StandardScaler или MinMaxScaler из scikit-learn: 

from sklearn.preprocessing import MinMaxScaler

scaler = MinMaxScaler()
data_scaled = scaler.fit_transform(data.values.reshape(-1, 1))

Логарифмирование: Если ваши данные демонстрируют экспоненциальный рост, логарифмирование станет вашим лучшим другом: 

import numpy as np

data_log = np.log(data + 1)  # Добавляем 1, чтобы избежать log(0)

Иммутация пропусков: Обработка пропущенных значений — ещё один важный шаг: 

data_filled = data.interpolate(method='linear')

Метрики для оценки моделей

Когда дело доходит до оценки моделей временных рядов, стандартные метрики могут вас разочаровать. Временные ряды требуют специфических подходов:

RMSE: Это показатель отклонения предсказаний от реальных значений.

from sklearn.metrics import mean_squared_error

rmse = np.sqrt(mean_squared_error(y_true, y_pred))
print(f"RMSE: {rmse}")

MAE: Более устойчивый к выбросам, MAE дает ясное представление о точности модели.

mae = np.mean(np.abs(y_true - y_pred))
print(f"MAE: {mae}")

MASE: Эта метрика помогает сравнивать качество модели с наивным подходом.

mase = np.mean(np.abs(y_true - y_pred)) / np.mean(np.abs(y_true - np.roll(y_true, 1)[1:]))
print(f"MASE: {mase}")

Как протестировать данные и прогнозы с помощью Darts

Итак, начнем с загрузки данных. Допустим, есть данные о продажах, хранящиеся в CSV-файле.

import pandas as pd
from darts import TimeSeries

# Загружаем данные
data = pd.read_csv('sales_data.csv', parse_dates=['date'], index_col='date')

# Создаем временной ряд
series = TimeSeries.from_dataframe(data, 'date', 'sales')
print("Данные загружены и преобразованы в временной ряд:")
print(series)

Перед тем как двигаться дальше, убедимся, что с временным рядом всё в порядке. Проверим на наличие пропусков и аномалий: 

# Проверка на пропуски
if series.isnull().any():
    print("Обнаружены пропуски в данных!")
else:
    print("Пропусков нет!")

# Визуализация данных
series.plot(title='График продаж', xlabel='Дата', ylabel='Количество продаж')

Теперь, когда мы уверены в целостности данных, проведём их предобработку.

Логарифмирование: 

import numpy as np

# Логарифмирование
series_log = TimeSeries.from_dataframe(np.log1p(data.set_index('date')['sales']))
series_log.plot(title='Логарифмированные данные')

Скейлинг: 

from sklearn.preprocessing import MinMaxScaler

scaler = MinMaxScaler()
scaled_data = scaler.fit_transform(data['sales'].values.reshape(-1, 1))
scaled_series = TimeSeries.from_dataframe(pd.DataFrame(scaled_data, index=data['date']))
scaled_series.plot(title='Масштабированные данные')

Теперь всё готово к построению модели. В Darts есть множество моделей, и мы будем использовать N‑BEATS.

from darts.models import NBEATSModel

# Определяем модель
model = NBEATSModel(input_chunk_length=30, output_chunk_length=10, n_epochs=100)

# Обучаем модель
model.fit(series_log)
print("Модель N-BEATS успешно обучена.")

Теперь протестируем модель на различных горизонтах прогнозирования: 

# Прогноз на 10 шагов вперед
forecast_horizon = 10
predictions = model.predict(forecast_horizon)

# Визуализация прогноза
predictions.plot(label='Прогноз N-BEATS', title='Прогнозирование на 10 шагов вперед')
series_log.plot(label='Исторические данные')

После получения прогнозов важно оценить их точность с помощью метрик: 

from darts.utils.statistics import mean_absolute_error, mean_squared_error

# Оценка точности
mae = mean_absolute_error(series_log[-forecast_horizon:], predictions)
rmse = np.sqrt(mean_squared_error(series_log[-forecast_horizon:], predictions))

print(f"MAE: {mae:.4f}")
print(f"RMSE: {rmse:.4f}")

Также стоит визуализировать остатки: 

# Остатки
residuals = series_log[-forecast_horizon:] - predictions
residuals.plot(title='Остатки прогноза')

Если остатки показывают закономерности, это сигнализирует о проблемах в модели.

Как улучшить прогноз

Если N-BEATS не даёт желаемых результатов, можно попробовать другие подходы:

Изменение модели: Попробуйте, например, модель Prophet.

from darts.models import Prophet

# Обучаем модель Prophet
prophet_model = Prophet()
prophet_model.fit(series)

# Прогнозируем
prophet_predictions = prophet_model.predict(forecast_horizon)

# Визуализация
prophet_predictions.plot(label='Прогноз Prophet')
series_log.plot(label='Исторические данные')

Тонкая настройка гиперпараметров: Используйте кросс-валидацию для подбора гиперпараметров.

from darts.models import NBEATSModel
from sklearn.model_selection import GridSearchCV

# Определяем параметры для поиска
param_grid = {
    'input_chunk_length': [10, 30],
    'output_chunk_length': [5, 10],
    'n_epochs': [100, 200],
}

# Настраиваем GridSearchCV
grid_search = GridSearchCV(estimator=NBEATSModel(), param_grid=param_grid, scoring='neg_mean_absolute_error')
grid_search.fit(series_log)

print(f"Лучшие параметры: {grid_search.best_params_}")

Анализ различных горизонтов: Прогнозируйтена разных интервалах и проверяйте точность.

for horizon in [1, 5, 10]:
    pred = model.predict(horizon)
    error = mean_absolute_error(series_log[-horizon:], pred)
    print(f"MAE для горизонта {horizon}: {error:.4f}")

Вот и всё! Darts предоставляет мощные инструменты для анализа временных рядов, так что не бойтесь экспериментировать с разными моделями и гиперпараметрами. Удачи вам и пусть ваши прогнозы всегда будут точными!

Подробнее с библиотекой можно ознакомиться здесь.

А в ближайшие дни пройдут открытые уроки по ML и CV, которые можно посетить бесплатно:

  • 7 октября: «Word2Vec — классика векторных представлений слов для решения задач текстовой обработки». Узнать подробнее

  • 10 октября: «OpenCV: Как Начать Работать с Компьютерным Зрением». Узнать подробнее

© Habrahabr.ru