Разработка признаков временных рядов с помощью Python Itertools

Узнайте, как использовать Python itertools для создания эффективных и масштабируемых признаков временных рядов.

# Введение

Разработка признаков для временных рядов не подчиняется тем же правилам, что и для табличных данных. Наблюдения не являются независимыми, порядок строк не случаен, и наиболее полезными признаками редко оказываются отдельные показания. Вам придется выявлять закономерности во времени, такие как темпы изменений, сравнения с запаздыванием, отклонения от скользящей базовой линии и многое другое.

Создание задержек, скользящих окон и группировка по разрешениям — все это, по своей сути, задачи итерации по упорядоченным последовательностям. Модуль itertools в Python идеально подходит для такой работы. Он не заменяет высокоуровневые абстракции pandas, такие как .rolling(), но предоставляет низкоуровневые строительные блоки для создания именно тех функций, которые вам нужны, с полным контролем над логикой.

В этой статье вы создадите семь категорий признаков временных рядов с помощью инструмента itertools. Вы также примените каждую из них к тестовому набору данных.

Код можно найти на GitHub .

# Создание тестового набора данных

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

import numpy as np import pandas as pd import itertools np.random.seed(42) periods = 168 # одна неделя почасовых показаний index = pd.date_range(start=»2024-03-01″, periods=periods, freq=»h») hours = np.arange(periods) # Температура (°C): суточный цикл + постепенный дрейф + шум temp_base = 3.5 temp_daily = 1.2 * np.sin(2 * np.pi * hours / 24) temp_drift = 0.003 * hours temp_noise = np.random.normal(0, 0.3, periods) temperature = temp_base + temp_daily + temp_drift + temp_noise # Влажность (%): обратная зависимость от температуры + шума humidity = 78 — 2.1 * (temperature — temp_base) + np.random.normal(0, 1.2, periods) # Потребляемая мощность (кВт): пиковые значения в рабочее время, выше в будние дни day_of_week = index.dayofweek business_hours = ((index.hour >= 8) & (index.hour = 5, 0.6, 1.0) power = ( 42.0 + 18.0 * business_hours * weekend_factor + np.random.normal(0, 2.1, periods) ) df = pd.DataFrame({ «temperature_c»: np.round(temperature, 3), «humidity_pct»: np.round(humidity, 2), «power_kw»: np.round(power, 2), }, index=index) df.index.name = «timestamp» print(df.head(8)) print(f»nShape: {df.shape}»)

Выход:

температура_°C влажность_процент мощности_кВт временная метка 2024-03-01 00:00:00 3.649 77.39 40.27 2024-03-01 01:00:00 3.772 76.52 41.33 2024-03-01 02:00:00 4.300 75.25 42.87 2024-03-01 03:00:00 4.814 74.26 40.82 2024-03-01 04:00:00 4.481 75.85 40.27 2024-03-01 05:00:00 4.604 76.09 42.51 2024-03-01 06:00:00 5.192 74.78 42.51 2024-03-01 07:00:00 4.910 76.03 40.94 Форма: (168, 3)

Теперь у нас есть 168 почасовых показаний по трем каналам датчиков. Давайте теперь создадим признаки.

# 1. Генерация признаков задержки с помощью islice

Запаздывание — это наиболее фундаментальная характеристика временного ряда: значение переменной на фиксированном количестве шагов в прошлом . Например, значения, полученные на 1 шаг назад, на 6 шагов назад или на 24 шага назад, могут отражать различные закономерности, такие как краткосрочные колебания, повторяющееся поведение внутри периода и долгосрочные тенденции или сезонность.

Давайте создадим признаки задержки для нашего тестового набора данных с помощью islice:

sensor_readings = df[«temperature_c»].tolist() lag_offsets = [1, 6, 12, 24] lag_features = {} for lag in lag_offsets: lagged = list(itertools.islice(sensor_readings, 0, len(sensor_readings) — lag)) # Добавляем None в начало, чтобы сохранить выравнивание индексов lag_features[f»temp_lag_{lag}h»] = [None] * lag + lagged lag_df = pd.DataFrame(lag_features, index=df.index) lag_df[«temperature_c»] = df[«temperature_c»] print(lag_df.iloc[24:30])

Выход:

temp_lag_1h temp_lag_6h temp_lag_12h temp_lag_24h timestamp 2024-03-02 00:00:00 2.831 2.082 3.609 3.649 2024-03-02 01:00:00 3.409 1.974 2.654 3.772 2024-03-02 02:00:00 3.919 2.960 2.425 4.300 2024-03-02 03:00:00 3.833 2.647 2.528 4.814 2024-03-02 04:00:00 4.542 2.986 2.205 4.481 2024-03-02 05:00:00 4.443 2.831 2.486 4.604 temperature_c timestamp 2024-03-02 00:00:00 3.409 2024-03-02 01:00:00 3.919 2024-03-02 02:00:00 3.833 2024-03-02 03:00:00 4.542 2024-03-02 04:00:00 4.443 2024-03-02 05:00:00 4.659

Функция islice(sensor_readings, 0, len — lag) извлекает последовательность, сдвинутую назад на несколько шагов задержки, без создания копии всего списка. Заполнение None в начале списка обеспечивает выравнивание каждого признака задержки с исходным индексом. Это важно, когда вы позже удаляете значения NaN при обучении модели.

# 2. Создание элементов раздвижных окон с использованием функций островного и накопительного режимов.

Одно значение задержки показывает, что датчик зафиксировал в определенный момент времени в прошлом. Скользящая статистика показывает, что датчик делал в течение определенного временного интервала, что зачастую гораздо полезнее.

readings = df[«temperature_c»].tolist() window_size = 6 # 6-часовое скользящее окно rolling_features = [] for i in range(len(readings)): if i < window_size: rolling_features.append({ "rolling_mean_6h": None, "rolling_std_6h": None, "rolling_min_6h": None, "rolling_max_6h": None, }) continue window = list(itertools.islice(readings, i - window_size, i)) # Используйте accumulate для вычисления скользящей суммы для среднего значения running_sum = list(itertools.accumulate(window)) window_mean = running_sum[-1] / window_size window_mean_sq = sum(x**2 for x in window) / window_size rolling_features.append({ "rolling_mean_6h": round(window_mean, 4), "rolling_std_6h": round((window_mean_sq - window_mean**2) ** 0.5, 4), "rolling_min_6h": round(min(window), 4), "rolling_max_6h": round(max(window), 4), }) roll_df = pd.DataFrame(rolling_features, index=df.index) roll_df["temperature_c"] = df["temperature_c"] print(roll_df.iloc[6:12])

Выход:

rolling_mean_6h rolling_std_6h rolling_min_6h timestamp 2024-03-01 06:00:00 4.2700 0.4256 3.649 2024-03-01 07:00:00 4.5272 0.4386 3.772 2024-03-01 08:00:00 4.7168 0.2929 4.300 2024-03-01 09:00:00 4.7372 0.2662 4.422 2024-03-01 10:00:00 4.6912 0.2728 4.422 2024-03-01 11:00:00 4.6095 0.3769 3.991 rolling_max_6h temperature_c timestamp 2024-03-01 06:00:00 4.814 5.192 2024-03-01 07:00:00 5.192 4.910 2024-03-01 08:00:00 5.192 4.422 2024-03-01 09:00:00 5.192 4.538 2024-03-01 10:00:00 5.192 3.991 2024-03-01 11:00:00 5.192 3.704

Вызов функции accumulate здесь вычисляет текущую сумму окна, поэтому мы получаем итог за один проход — running_sum[-1] — без отдельного вызова sum(). Для больших наборов данных, обрабатываемых в потоковом режиме, эффективно избегать избыточных проходов по одним и тем же данным.

# 3. Создание сезонных интерактивных функций для продукта

Многие временные ряды демонстрируют многослойную сезонность, где взаимодействуют несколько временных циклов — таких как время суток, день недели и более широкие операционные или циклические периоды. Характеристики взаимодействия, объединяющие эти измерения, могут выявлять закономерности, которые отдельные временные компоненты сами по себе могут упустить из виду.

Теперь давайте создадим функции взаимодействия с продуктом:

hours_of_day = list(range(24)) day_types = [«weekday», «weekend»] operational_shifts = [«off_peak», «on_peak»] # on_peak: 08:00–18:00 # Создание полной таблицы соответствия для всех комбинаций season_grid = list(itertools.product(hours_of_day, day_types, operational_shifts)) season_df = pd.DataFrame(season_grid, columns=[«hour», «day_type», «shift»]) # Моделирование ожидаемой базовой температуры для каждой комбинации np.random.seed(14) season_df[«baseline_temp_c»] = np.round( 3.5 + 0.8 * np.sin(2 * np.pi * season_df[«hour»] / 24) + np.where(season_df[«day_type»] == «weekend», 0.3, 0.0) + np.where(season_df[«shift»] == «on_peak», 0.5, 0.0) + np.random.normal(0, 0.1, len(season_df)), 3 ) print(season_df[season_df[«hour»].isin([0, 8, 14, 20])].head(16).to_string(index=False)) print(f»nВсего комбинаций сетки: {len(season_df)}»)

Выход:

час_день_тип_сдвиг_базовая_температура_в_градусах_°C_в_непиковое_время_в_будние_дни_в_пиковое_время_3.655_в_будние_дни_в_пиковое_время_4.008_в_выходные_в_непиковое_время_3.817_в_выходные_в_пиковое_время_4.293_в_будние_дни_в_непиковое_время_4.325_в_будние_дни_в_пиковое_время_4.601_в_выходные_в_непиковое_время_4.446_в_выходные_в_пиковое_время_4.978_в_будние_дни_в_непиковое_время_3.370_в_будние_дни_в_пиковое_время_3.628_в_выходные_в_непиковое_время_3.279_в_выходные_в_пиковое_время_3.959_в_будние_дни_в_непиковое_время_2.726_в_будние_дни_в_пиковое_время_3.256_в_выходные_в_непиковое_время_3.056_в_выходные_в_пиковое_время_3.530_всего_комбинаций сетки: 96

Эта сетка объединяется с вашим основным набором данных в качестве признака baseline_temp_c для каждой строки, присваивая каждому показанию контекстно-зависимое ожидаемое значение. Отклонение от этого базового уровня, temperature_c — baseline_temp_c, затем становится полезным признаком для обнаружения аномалий.

# 4. Извлечение статистики скользящего окна с помощью tee

Иногда необходимо одновременно обрабатывать одну и ту же последовательность с помощью нескольких статистических методов — среднее значение, дисперсия, скорость изменения — без многократной итерации. itertools.tee создает независимые итераторы из одного источника, что как раз и нужно.

def sliding_window_stats(series, window_size): «»»Вычисляет среднее значение, размах и скорость изменения по скользящим окнам с помощью tee.»»» results = [] it = iter(series) window = list(itertools.islice(it, window_size)) if len(window) < window_size: return results results.append({ "window_mean": round(sum(window) / window_size, 4), "window_range": round(max(window) - min(window), 4), "rate_of_change": round(window[-1] - window[0], 4), }) for next_val in it: window = window[1:] + [next_val] # tee создает два независимых итератора по одному и тому же окну iter_a, iter_b = itertools.tee(iter(window)) values_a = list(iter_a) values_b = list(iter_b) mean_val = sum(values_a) / window_size results.append({ "window_mean": round(mean_val, 4), "window_range": round(max(values_b) - min(values_b), 4), "rate_of_change": round(window[-1] - window[0], 4), }) return results power_readings = df["power_kw"].tolist() stats = sliding_window_stats(power_readings, window_size=8) stats_df = pd.DataFrame(stats, index=df.index[7:]) stats_df["power_kw"] = df["power_kw"].iloc[7:].values print(stats_df.iloc[0:8])

Выход:

window_mean window_range rate_of_change power_kw timestamp 2024-03-01 07:00:00 41.4400 2.60 0.67 40.94 2024-03-01 08:00:00 43.7825 18.74 17.68 59.01 2024-03-01 09:00:00 46.1775 20.22 17.62 60.49 2024-03-01 10:00:00 47.9387 20.22 16.14 56.96 2024-03-01 11:00:00 49.9663 20.22 16.77 57.04 2024-03-01 12:00:00 52.2437 19.55 15.98 58.49 2024-03-01 13:00:00 54.3738 19.55 17.04 59.55 2024-03-01 14:00:00 56.6412 19.71 19.71 60.65

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

# 5. Объединение многоразрешенных временных характеристик с помощью цепочки

Полезные характеристики временных рядов часто получаются одновременно из нескольких временных разрешений: исходные почасовые показания, скользящее среднее за 6 часов, скользящее среднее за 24 часа и календарная характеристика, например, час суток. Обычно они находятся в отдельных массивах и требуют объединения в один аккуратный список характеристик. Вот как можно использовать chain для объединения таких характеристик:

humidity = df[«humidity_pct»].tolist() def rolling_means(series, window): means = [] for i in range(len(series)): if i < window: means.append(None) else: w = list(itertools.islice(series, i - window, i)) means.append(round(sum(w) / window, 3)) return means rolling_6h = rolling_means(humidity, 6) rolling_24h = rolling_means(humidity, 24) hour_of_day = df.index.hour.tolist() is_business_hour = [1 if 8 0 else None correlations.append(corr) pairwise_features[feature_name] = correlations corr_df = pd.DataFrame(pairwise_features, index=df.index) print(corr_df.iloc[12:18])

Выход:

corr_temp_humi_12h corr_temp_powe_12h timestamp 2024-03-01 12:00:00 -0.6700 -0.2281 2024-03-01 13:00:00 -0.7208 -0.4960 2024-03-01 14:00:00 -0.7442 -0.6669 2024-03-01 15:00:00 -0.7678 -0.7076 2024-03-01 16:00:00 -0.8116 -0.7265 2024-03-01 17:00:00 -0.8368 -0.7482 corr_humi_powe_12h timestamp 2024-03-01 12:00:00 0.5380 2024-03-01 13:00:00 0.6614 2024-03-01 14:00:00 0.7202 2024-03-01 15:00:00 0.7311 2024-03-01 16:00:00 0.7233 2024-03-01 17:00:00 0.7219

#7. Накопление базовых показателей в процессе эксплуатации.

Значение того или иного параметра может различаться в зависимости от момента его появления в последовательности. Важно его отклонение от изменяющегося базового уровня — скользящего среднего до этого момента времени. Используя инкрементальный подход, такой как накопление, можно эффективно вычислить это скользящее среднее, не сохраняя всю историю.

readings = df[«temperature_c»].tolist() running_sums = list(itertools.accumulate(readings)) running_counts = list(itertools.accumulate([1] * len(readings))) running_means = [ round(s / c, 4) for s, c in zip(running_sums, running_counts) ] # Running max — самая высокая температура, зафиксированная на данный момент, полезна для отслеживания нарушений running_max = list(itertools.accumulate(readings, func=max)) deviation_from_baseline = [ round(r — m, 4) for r, m in zip(readings, running_means) ] baseline_df = pd.DataFrame({ «temperature_c»: readings, «running_mean»: running_means, «running_max»: running_max, «deviation_from_baseline»: deviation_from_baseline, }, index=df.index) print(baseline_df.iloc[20:28])

Выход:

temperature_c running_mean running_max timestamp 2024-03-01 20:00:00 2.960 3.5857 5.192 2024-03-01 21:00:00 2.647 3.5430 5.192 2024-03-01 22:00:00 2.986 3.5188 5.192 2024-03-01 23:00:00 2.831 3.4902 5.192 2024-03-02 00:00:00 3.409 3.4869 5.192 2024-03-02 01:00:00 3.919 3.5035 5.192 2024-03-02 02:00:00 3.833 3.5157 5.192 2024-03-02 03:00:00 4.542 3.5524 5.192 отклонение_от_базовой_линии временная метка 2024-03-01 20:00:00 -0.6257 2024-03-01 21:00:00 -0.8960 2024-03-01 22:00:00 -0.5328 2024-03-01 23:00:00 -0.6592 2024-03-02 00:00:00 -0.0779 2024-03-02 01:00:00 0.4155 2024-03-02 02:00:00 0.3173 2024-03-02 03:00:00 0.9896

# Краткое содержание

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

Вот краткое изложение рассмотренных в этой статье закономерностей:

А поскольку itertools работает на уровне итераторов, все эти шаблоны легко объединяются в потоковые конвейеры. Успешной разработки признаков!

Бала Прия С. — разработчик и технический писатель из Индии. Ей нравится работать на стыке математики, программирования, анализа данных и создания контента. В сферу её интересов и компетенции входят DevOps, анализ данных и обработка естественного языка. Она любит читать, писать, программировать и пить кофе! В настоящее время она работает над изучением и распространением своих знаний среди сообщества разработчиков, создавая учебные пособия, руководства, аналитические статьи и многое другое. Бала также создает увлекательные обзоры ресурсов и обучающие материалы по программированию.

Источник: www.kdnuggets.com