Коротко про библиотеку TSFresh
Привет, Хабр!
Сегодня в коротком формате познакомимся с библиотекой TSFresh.
TSFresh берет на себя две основные задачи:
Извлечение признаков: функция
extract_features()генерирует огромный набор статистик по заданным временным рядам. Внутри неё используются так называемыеFeatureCalculators— функции, рассчитывающие конкретные признаки. Например, автокорреляция, энтропия, число нулевых пересечений.Отбор признаков: функция
select_features()выполняет статистические тесты для отбора действительно релевантных признаков, что помогает избежать переобучения и уменьшить размерность.
Основная фича в том, что TSFresh позволяет работать с данными в так называемом long format — где каждая строка соответствует одному измерению, а идентификатор временного ряда задаётся через column_id.
Взглянем на простой пример извлечения признаков из временного ряда:
import pandas as pd
from tsfresh import extract_features, select_features
from tsfresh.utilities.dataframe_functions import impute
# Создаём тестовый DataFrame с данными временного ряда
df = pd.DataFrame({
'id': [1]*100,
'time': range(100),
'value': [i + (i % 10)*5 for i in range(100)]
})
# Извлекаем признаки
extracted_features = extract_features(df, column_id="id", column_sort="time")
# Обрабатываем NaN значения, если они есть
impute(extracted_features)
print("Извлечённые признаки:")
print(extracted_features.head())Извлечённые признаки:
value__abs_energy value__agg_autocorrelation__lag_1 value__mean ... value__number_peaks
id
1 1234.56 0.987 45.67 ... 3За пару строчек кода получили DataFrame с сотнями признаков, рассчитанных для нашего ряда.
Форматы данных и настройка извлечения признаков
Для адекватной работы TSFresh данные должны быть в длинном формате. Что это значит? Каждый временной ряд должен иметь уникальный идентификатор column_id, временную метку column_sort и значение column_value. Если данные представлены в широком формате (где каждая колонка — отдельный признак), их необходимо преобразовать в длинный формат.
Пример данных в длинном формате:
# Пример DataFrame в длинном формате
df_long = pd.DataFrame({
'id': [1, 1, 1, 2, 2, 2],
'time': [1, 2, 3, 1, 2, 3],
'value': [10, 15, 14, 20, 25, 24]
})При вызове extract_features() можно задавать параметры:
column_id— идентификатор временного ряда.column_sort— временная метка, по которой производится сортировка.column_kind— если у вас несколько типов значений в одном наборе данных.column_value— колонка с самими значениями.
Также TSFresh имеет набор встроенных параметров, которые по умолчанию рассчитывают более 60 категорий признаков. Однако, зачастую требуется настроить набор признаков под конкретную задачу. Для этого используется объект EfficientFCParameters или можно создать собственный набор с помощью KindToFCParameters.
Пример настройки параметров:
from tsfresh.feature_extraction.settings import EfficientFCParameters
# Используем набор эффективных параметров для быстрого извлечения признаков
extraction_settings = EfficientFCParameters()
# Извлечение признаков с заданными настройками
features = extract_features(
df,
column_id="id",
column_sort="time",
default_fc_parameters=extraction_settings
)
impute(features)
print("Настроенные признаки:")
print(features.head())А если нужно оставить только релевантные признаки (например, автокорреляции и энтропии), можно создать кастомный словарь с нужными параметрами:
custom_fc_parameters = {
"value": {
"autocorrelation": [{"lag": 1}],
"approximate_entropy": [{"m": 2, "r": 0.3}],
# Добавляем другие необходимые функции...
}
}
features_custom = extract_features(
df,
column_id="id",
column_sort="time",
default_fc_parameters=custom_fc_parameters
)
impute(features_custom)
print("Кастомные признаки:")
print(features_custom.head())Кастомные признаки:
value__autocorrelation value__approximate_entropy
id
1 0.950000 0.450000Интеграция с scikit-learn и пайплайны
Благодаря классам TSFreshFeatureExtractor и TSFreshRelevantFeatureExtractor можно встроить автоматическое извлечение признаков в стандартный ML‑пайплайн.
Пример: сначала извлекаем признаки, затем стандартизируем их и подаем на классификатор:
from sklearn.pipeline import Pipeline
from sklearn.ensemble import RandomForestClassifier
from sklearn.preprocessing import StandardScaler
from tsfresh.transformers import TSFreshFeatureExtractor
# Предположим, что у нас есть DataFrame с временными рядами в длинном формате
# и метками для классификации
df_ts = pd.DataFrame({
'id': [1]*50 + [2]*50,
'time': list(range(50))*2,
'value': [i + (i % 5) for i in range(50)] + [i - (i % 3) for i in range(50)]
})
y = [0, 1] # Метки классов для двух временных рядов
# Создаём пайплайн
pipeline = Pipeline([
('tsfresh', TSFreshFeatureExtractor(column_id="id", column_sort="time")),
('scaler', StandardScaler()),
('clf', RandomForestClassifier(n_estimators=100, random_state=42))
])
# Применяем fit_transform() для обучения
pipeline.fit(df_ts, y)
# Получаем предсказания
predictions = pipeline.predict(df_ts)
print("Предсказания классификатора:")
print(predictions)Предсказания классификатора:
[0 1]Пайплайн извлекает признаки для двух временных рядов (id 1 и 2), масштабирует их и обучает классификатор. В данном примере модель предсказывает класс 0 для первого ряда и 1 для второго.
Фильтрация признаков и кастомные фичи
Механизм фильтрации признаков
Допустим, вы вытащили из временного ряда 500 признаков. Из них полезных — от силы 10. Остальные шумят, переобучают, тормозят, и вообще мешают жить. Именно поэтому TSFresh сразу предлагает нам воспользоваться встроенной функцией select_features() — она отбирает только те признаки, которые статистически значимы для целевой переменной.
Работают разные тесты:
f_classif— для классификации с непрерывными фичами.chi2— для категориальных.mutual_info_classif— если хочется что‑то нестандартное.Kendall,Spearman— для ранговой корреляции.
Но в основном всё делается за вас — автоматически.
from tsfresh import extract_features, select_features
from tsfresh.utilities.dataframe_functions import impute
import pandas as pd
import numpy as np
# Создаём игрушечный датасет с двумя временными рядами
df = pd.DataFrame({
'id': [1]*50 + [2]*50,
'time': list(range(50))*2,
'value': np.concatenate([
np.sin(np.linspace(0, 10, 50)) + np.random.normal(0, 0.1, 50),
np.random.normal(0, 1, 50)
])
})
# Целевая переменная (например, к какому классу относится временной ряд)
y = pd.Series([1, 0], index=[1, 2]) # 1 - синус, 0 - шум
# Извлекаем фичи
features = extract_features(df, column_id="id", column_sort="time")
impute(features)
# Фильтруем признаки по значимости
filtered_features = select_features(features, y)
print("Отобранные признаки:")
print(filtered_features.head())select_features() вернёт только те признаки, которые реально статистически связаны с y.
Создание кастомных признаков
Иногда TSFresh не знает про вашу предметную область. А вы знаете. Тогда можно (и нужно) писать свою фичу. TSFresh поддерживает кастомные функции через два декоратора:
@set_property(...)— указывает мета‑инфу про фичу.@aggregate_feature(...)— регистрирует саму фичу.
Представим, вы хотите фичу, которая считает количество резких пиков — мест, где значение превышает соседей и в два раза стандартное отклонение. Это уже было выше. А теперь — как это встроить в extract_features() в полноценной задаче:
# 1. Определяем кастомную фичу
from tsfresh.feature_extraction.feature_calculators import set_property, aggregate_feature
@set_property("fctype", "simple")
@aggregate_feature("value")
def count_peaks_above_2std(x):
"""Количество пиков, превышающих 2 стандартных отклонения."""
if len(x) < 3:
return 0
std = np.std(x)
threshold = 2 * std
peaks = [x[i] for i in range(1, len(x)-1)
if x[i] > x[i-1] and x[i] > x[i+1] and x[i] > threshold]
return len(peaks)
# 2. Добавляем её в словарь кастомных фичей
from tsfresh.feature_extraction import extract_features
from tsfresh.utilities.dataframe_functions import impute
# Кастомный конфиг, где добавлена наша функция
from tsfresh.feature_extraction.settings import ComprehensiveFCParameters
custom_fc_parameters = {
"value": {
"count_peaks_above_2std": None # функция без параметров
}
}
# 3. Применяем на датафрейме
features = extract_features(
df,
column_id="id",
column_sort="time",
default_fc_parameters=custom_fc_parameters
)
impute(features)
print("Кастомная фича на данных:")
print(features.head())Можно использовать и свою кастомную фичу, и любые встроенные. Кастомные функции могут иметь параметры, типы, валидаторы (например, применимость только к float, или только если ряд длиннее 10).
Работа с большими данными, rolling-окнами и параллелизмом
При работе с большими объемами данных, скажем, временными рядами для 100k пользователей, важно правильно настроить параллелизм и управление памятью. TSFresh позволяет это сделать с помощью трёх параметров:
n_jobs
позволяет распределить вычисления на несколько ядер процессора. Если у вас, например, 8 ядер, установкаn_jobs=8может значительно сократить время обработки. Однако не стоит забывать, что слишком большое число ядер может привести к избыточной нагрузке, особенно если параллельные процессы конкурируют за память.disable_progressbar
лишний вывод прогресс‑бара может замедлять работу. Установка этого параметра в True убирает визуальный индикатор, освобождая ресурсы для вычислений.chunksize
помогает оптимизировать использование оперативной памяти. Если данные очень большие, разумно указать, например,chunksize=1000или меньше, чтобы TSFresh обрабатывал порции данных по частям, не загружая всю выборку сразу в память.
Пример настроек:
# Извлечение признаков с оптимизированными параметрами для больших данных:
features = extract_features(
df_large, # Ваш большой DataFrame
column_id="id",
column_sort="time",
n_jobs=4, # Распараллеливание на 4 ядрах
disable_progressbar=True, # Выключение прогресс-бара
chunksize=1000 # Обработка порциями по 1000 строк
)
impute(features) # Не забываем обработать NaN значения
print("Оптимизированные признаки:")
print(features.head())Rolling‑окна — тоже хороший инструмент, когда задача предсказания или анализа зависит от локальных характеристик временного ряда. TSFresh имеет функцию roll_time_series(), которая позволяет создать новые скользящие подвыборки данных. Например, если есть данные по 200 точкам времени на каждого пользователя, можно применить окно с максимальным сдвигом в 50 единиц времени, чтобы получить агрегированные характеристики для каждого интервала.
Рассмотрим пример:
import pandas as pd
import numpy as np
from tsfresh.utilities.dataframe_functions import roll_time_series
from tsfresh import extract_features
from tsfresh.utilities.dataframe_functions import impute
# Допустим, у нас есть большой DataFrame с временными рядами
df_large = pd.DataFrame({
'id': [i for i in range(1, 101) for _ in range(200)],
'time': list(range(200)) * 100,
'value': np.random.randn(200 * 100)
})
# Применяем rolling окна с максимальным сдвигом в 50 единиц времени.
rolled_df = roll_time_series(
df_large,
column_id="id",
column_sort="time",
max_timeshift=50
)
# Теперь извлекаем признаки для каждого окна с использованием оптимизированных настроек
features_rolled = extract_features(
rolled_df,
column_id="id",
column_sort="time",
n_jobs=4, # Распараллеливание вычислений
disable_progressbar=True,
chunksize=1000 # Обработка порциями для экономии RAM
)
impute(features_rolled)
print("Признаки после применения rolling-окон:")
print(features_rolled.head())Функция roll_time_series() генерирует скользящие окна для каждого временного ряда, что позволяет анализировать локальные изменения в данных.
Иногда хочется объединить извлечение признаков, их масштабирование и обучение модели в одном цепном процессе. Используя класс TSFreshFeatureExtractor вместе с стандартными компонентами scikit‑learn, можно собрать пайплайн для задач forecasting. Плюсом можно применить rolling‑окна для создания лаговых признаков.
import pandas as pd
import numpy as np
from sklearn.pipeline import Pipeline
from sklearn.linear_model import LinearRegression
from sklearn.preprocessing import StandardScaler
from tsfresh.transformers import TSFreshFeatureExtractor
from tsfresh.utilities.dataframe_functions import roll_time_series
# Генерируем данные для forecasting: 100 временных рядов, 150 точек каждый
df_forecast = pd.DataFrame({
'id': [i for i in range(1, 101) for _ in range(150)],
'time': list(range(150)) * 100,
'value': np.random.randn(150 * 100)
})
# Применяем roll_time_series для создания лаговых признаков (окно с максимальным сдвигом в 20)
rolled_forecast = roll_time_series(
df_forecast,
column_id="id",
column_sort="time",
max_timeshift=20
)
# Создаем пайплайн для прогнозирования
forecast_pipeline = Pipeline([
('tsfresh', TSFreshFeatureExtractor(
column_id="id",
column_sort="time",
n_jobs=4, # Распараллеливание вычислений
disable_progressbar=True,
chunksize=500 # Обработка порциями для больших данных
)),
('scaler', StandardScaler()),
('regressor', LinearRegression())
])
# Предположим, что таргет – одно значение на временной ряд
y_forecast = np.random.randn(rolled_forecast['id'].nunique())
# Обучаем пайплайн и получаем прогнозы
forecast_pipeline.fit(rolled_forecast, y_forecast)
preds = forecast_pipeline.predict(rolled_forecast)
print("Прогнозы модели (первые 10 значений):")
print(preds[:10])Генерируем данные для 100 временных рядов. Применяем rolling‑окна, создавая лаговые признаки с максимальным сдвигом в 20. Собираем пайплайн, который сначала извлекает признаки через TSFresh, затем масштабирует их и обучает линейную регрессию. Используем параметры n_jobs, disable_progressbar и chunksize для оптимизации работы с большими данными.
Все актуальные методы и инструменты DS и ML можно освоить на онлайн‑курсах OTUS: в каталоге можно посмотреть список всех программ, а в календаре — записаться на открытые уроки.
