TSFresh – извлечение признаков из временных рядов

Что такое TSFresh и зачем она нужна

TSFresh (Time Series Feature extraction based on scalable hypothesis tests) представляет собой мощную Python-библиотеку, специально созданную для автоматического извлечения большого количества признаков из временных рядов. Разработанная командой исследователей из Германии, эта библиотека решает одну из самых трудоемких задач в области анализа данных - преобразование сырых временных данных в информативные характеристики, пригодные для машинного обучения.

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

Архитектура и принципы работы TSFresh

TSFresh построена на принципе статистически обоснованного отбора признаков. Библиотека использует двухэтапный подход: сначала извлекает максимально возможное количество признаков из временных рядов, а затем применяет статистические тесты для определения их значимости относительно целевой переменной.

Внутренняя архитектура библиотеки включает несколько ключевых компонентов:

• Модуль извлечения признаков с более чем 60 различными функциями
• Систему статистической фильтрации на основе p-значений
• Оптимизированные алгоритмы для работы с большими объемами данных
• Интеграцию с популярными библиотеками машинного обучения

Преимущества использования TSFresh

TSFresh предлагает ряд существенных преимуществ для работы с временными рядами:

Автоматизация процесса feature engineering - библиотека самостоятельно генерирует сотни признаков, что экономит значительное время разработчика.

Статистическая обоснованность - все отобранные признаки проходят строгую статистическую проверку на значимость.

Масштабируемость - поддержка параллельной обработки позволяет эффективно работать с большими датасетами.

Гибкость конфигурации - возможность настройки параметров извлечения признаков под конкретную задачу.

Широкий спектр функций - от простых статистических мер до сложных показателей нелинейности и хаотичности.

Совместимость с экосистемой Python - полная интеграция с pandas, scikit-learn, NumPy и другими популярными библиотеками.

Установка и настройка окружения

Для работы с TSFresh требуется установка нескольких компонентов. Основная установка выполняется через pip:

pip install tsfresh

Для полноценной работы рекомендуется также установить дополнительные зависимости:

pip install scikit-learn pandas numpy matplotlib seaborn

При работе с большими данными может потребоваться установка дополнительных пакетов для оптимизации производительности:

pip install dask[complete] numba

Базовый импорт необходимых модулей:

from tsfresh import extract_features, select_features
from tsfresh.utilities.dataframe_functions import impute
from tsfresh.feature_extraction import ComprehensiveFCParameters, EfficientFCParameters
from tsfresh.feature_selection.relevance import calculate_relevance_table
import pandas as pd
import numpy as np

Структура данных и требования к формату

TSFresh работает с данными в специальном формате, называемом "длинной таблицей" (long format). Этот формат требует определенной структуры DataFrame:

import pandas as pd

# Пример корректной структуры данных
df = pd.DataFrame({
    'id': [1, 1, 1, 1, 2, 2, 2, 2, 3, 3, 3, 3],
    'time': [1, 2, 3, 4, 1, 2, 3, 4, 1, 2, 3, 4],
    'value': [10, 12, 11, 13, 20, 22, 21, 23, 30, 32, 31, 33],
    'sensor_type': ['A', 'A', 'A', 'A', 'B', 'B', 'B', 'B', 'C', 'C', 'C', 'C']
})

Ключевые столбцы:

• id - уникальный идентификатор временного ряда
• time - временная метка или порядковый номер
• value - значение измерения
• sensor_type (опционально) - тип сенсора или канала данных

Основные методы извлечения признаков

Базовое извлечение признаков

Основная функция extract_features является центральным элементом библиотеки:

from tsfresh import extract_features

# Извлечение всех доступных признаков
features = extract_features(df, 
                          column_id='id', 
                          column_sort='time',
                          column_value='value')

Использование предустановленных конфигураций

TSFresh предлагает несколько готовых наборов параметров:

from tsfresh.feature_extraction import EfficientFCParameters, ComprehensiveFCParameters, MinimalFCParameters

# Быстрая обработка с основными признаками
features_efficient = extract_features(df, 
                                    column_id='id', 
                                    column_sort='time',
                                    default_fc_parameters=EfficientFCParameters())

# Полный набор признаков
features_comprehensive = extract_features(df, 
                                        column_id='id', 
                                        column_sort='time',
                                        default_fc_parameters=ComprehensiveFCParameters())

Мультиканальные данные

При работе с данными из нескольких источников или сенсоров:

# Извлечение признаков для мультиканальных данных
features_multivariate = extract_features(df, 
                                        column_id='id', 
                                        column_sort='time',
                                        column_value='value',
                                        column_kind='sensor_type')

Фильтрация и отбор значимых признаков

После извлечения признаков необходимо отобрать наиболее значимые для конкретной задачи:

from tsfresh import select_features
from tsfresh.utilities.dataframe_functions import impute

# Создание целевой переменной
y = pd.Series([0, 1, 0], index=[1, 2, 3])

# Заполнение пропусков
features_imputed = impute(features)

# Отбор значимых признаков
features_filtered = select_features(features_imputed, y)

Настройка параметров фильтрации

from tsfresh.feature_selection.relevance import calculate_relevance_table

# Получение таблицы значимости
relevance_table = calculate_relevance_table(features_imputed, y)

# Фильтрация с настройкой уровня значимости
features_filtered = select_features(features_imputed, y, 
                                  fdr_level=0.05,  # Уровень ложных открытий
                                  hypotheses_independent=False)

Оптимизация производительности

TSFresh предлагает несколько способов оптимизации для работы с большими данными:

Параллельная обработка

# Использование всех доступных ядер
features = extract_features(df, 
                          column_id='id', 
                          column_sort='time',
                          n_jobs=-1,  # Все доступные ядра
                          chunksize=10)  # Размер чанка для обработки

Отключение прогресс-бара

# Для автоматизированных скриптов
features = extract_features(df, 
                          column_id='id', 
                          column_sort='time',
                          disable_progressbar=True)

Кэширование результатов

# Сохранение промежуточных результатов
features.to_pickle('extracted_features.pkl')

# Загрузка сохраненных признаков
features_loaded = pd.read_pickle('extracted_features.pkl')

Создание пользовательских функций признаков

TSFresh позволяет создавать собственные функции для извлечения признаков:

from tsfresh.feature_extraction.feature_calculators import set_property

@set_property("fctype", "simple")
def custom_mean_absolute_change(x):
    """Вычисляет среднее абсолютное изменение"""
    return np.mean(np.abs(np.diff(x)))

@set_property("fctype", "combiner")
def custom_ratio_beyond_r_sigma(x, r):
    """Доля значений за пределами r сигм"""
    mean_x = np.mean(x)
    std_x = np.std(x)
    return np.mean(np.abs(x - mean_x) > r * std_x)

# Использование пользовательских функций
from tsfresh.feature_extraction.settings import from_columns
custom_fc_parameters = from_columns([custom_mean_absolute_change, custom_ratio_beyond_r_sigma])

Интеграция с машинным обучением

Классификация

from sklearn.ensemble import RandomForestClassifier
from sklearn.model_selection import train_test_split
from sklearn.metrics import accuracy_score, classification_report

# Разделение данных
X_train, X_test, y_train, y_test = train_test_split(features_filtered, y, test_size=0.3, random_state=42)

# Обучение модели
clf = RandomForestClassifier(n_estimators=100, random_state=42)
clf.fit(X_train, y_train)

# Прогнозирование
y_pred = clf.predict(X_test)
accuracy = accuracy_score(y_test, y_pred)
print(f"Accuracy: {accuracy:.3f}")

Регрессия

from sklearn.ensemble import RandomForestRegressor
from sklearn.metrics import mean_squared_error, r2_score

# Для задач регрессии
y_continuous = pd.Series([1.5, 2.7, 3.2], index=[1, 2, 3])

regressor = RandomForestRegressor(n_estimators=100, random_state=42)
regressor.fit(X_train, y_continuous)

y_pred_reg = regressor.predict(X_test)
mse = mean_squared_error(y_test, y_pred_reg)
r2 = r2_score(y_test, y_pred_reg)

Работа с реальными данными

Пример с данными сенсоров

from tsfresh.examples.robot_execution_failures import download_robot_execution_failures, load_robot_execution_failures

# Загрузка примера данных
download_robot_execution_failures()
timeseries, y = load_robot_execution_failures()

# Полный пайплайн обработки
X = extract_features(timeseries, column_id="id", column_sort="time")
X_imputed = impute(X)
X_filtered = select_features(X_imputed, y)

# Обучение модели
from sklearn.ensemble import RandomForestClassifier
clf = RandomForestClassifier()
clf.fit(X_filtered, y)

# Оценка качества
from sklearn.model_selection import cross_val_score
scores = cross_val_score(clf, X_filtered, y, cv=5)
print(f"Cross-validation accuracy: {scores.mean():.3f} (+/- {scores.std() * 2:.3f})")

Визуализация и анализ результатов

Анализ важности признаков

import matplotlib.pyplot as plt
import seaborn as sns

# Получение важности признаков
feature_importance = pd.DataFrame({
    'feature': X_filtered.columns,
    'importance': clf.feature_importances_
}).sort_values('importance', ascending=False)

# Визуализация топ-20 признаков
plt.figure(figsize=(10, 8))
sns.barplot(data=feature_importance.head(20), x='importance', y='feature')
plt.title('Top 20 Most Important Features')
plt.xlabel('Feature Importance')
plt.tight_layout()
plt.show()

Анализ корреляций

# Матрица корреляций для отобранных признаков
correlation_matrix = X_filtered.corr()

plt.figure(figsize=(12, 10))
sns.heatmap(correlation_matrix, annot=False, cmap='coolwarm', center=0)
plt.title('Feature Correlation Matrix')
plt.tight_layout()
plt.show()

Полная справочная таблица методов и функций TSFresh

Практические рекомендации по использованию

Выбор конфигурации параметров

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

• MinimalFCParameters() - для быстрого прототипирования и небольших данных
• EfficientFCParameters() - для большинства практических задач
• ComprehensiveFCParameters() - для исследовательских задач и максимальной полноты

Оптимизация памяти

# Для больших данных используйте итеративную обработку
def process_large_dataset(df, chunk_size=1000):
    unique_ids = df['id'].unique()
    all_features = []
    
    for i in range(0, len(unique_ids), chunk_size):
        chunk_ids = unique_ids[i:i+chunk_size]
        chunk_df = df[df['id'].isin(chunk_ids)]
        
        features = extract_features(chunk_df, 
                                  column_id='id', 
                                  column_sort='time',
                                  n_jobs=1)  # Ограничиваем параллелизм
        all_features.append(features)
    
    return pd.concat(all_features, ignore_index=True)

Валидация и тестирование

from sklearn.model_selection import TimeSeriesSplit

# Для временных рядов используйте временную кросс-валидацию
tscv = TimeSeriesSplit(n_splits=5)

for train_index, test_index in tscv.split(X_filtered):
    X_train, X_test = X_filtered.iloc[train_index], X_filtered.iloc[test_index]
    y_train, y_test = y.iloc[train_index], y.iloc[test_index]
    
    model = RandomForestClassifier()
    model.fit(X_train, y_train)
    score = model.score(X_test, y_test)
    print(f"Fold accuracy: {score:.3f}")

Интеграция с другими библиотеками

Работа с Dask для больших данных

import dask.dataframe as dd
from tsfresh.utilities.distribution import MultiprocessingDistributor

# Параллельная обработка с Dask
distributor = MultiprocessingDistributor(n_workers=4)
features = extract_features(df, 
                          column_id='id', 
                          column_sort='time',
                          distributor=distributor)

Интеграция с MLflow

import mlflow
import mlflow.sklearn

with mlflow.start_run():
    # Извлечение признаков
    features = extract_features(df, column_id='id', column_sort='time')
    features = impute(features)
    features_filtered = select_features(features, y)
    
    # Обучение модели
    model = RandomForestClassifier()
    model.fit(features_filtered, y)
    
    # Логирование метрик
    mlflow.log_metric("n_features_extracted", len(features.columns))
    mlflow.log_metric("n_features_selected", len(features_filtered.columns))
    mlflow.sklearn.log_model(model, "model")

Обработка ошибок и отладка

Типичные проблемы и их решения

import warnings
warnings.filterwarnings('ignore')

try:
    features = extract_features(df, column_id='id', column_sort='time')
except Exception as e:
    print(f"Ошибка при извлечении признаков: {e}")
    # Проверка формата данных
    print("Проверка структуры данных:")
    print(f"Столбцы: {df.columns.tolist()}")
    print(f"Типы данных: {df.dtypes}")
    print(f"Уникальные ID: {df['id'].nunique()}")
    print(f"Диапазон времени: {df['time'].min()} - {df['time'].max()}")

Логирование процесса

import logging

# Настройка логирования
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

def extract_features_with_logging(df, **kwargs):
    logger.info(f"Начало извлечения признаков для {df['id'].nunique()} временных рядов")
    
    features = extract_features(df, **kwargs)
    
    logger.info(f"Извлечено {len(features.columns)} признаков")
    logger.info(f"Размер результирующей матрицы: {features.shape}")
    
    return features

Примеры применения в различных областях

Анализ финансовых данных

# Пример с финансовыми временными рядами
financial_df = pd.DataFrame({
    'id': np.repeat(range(1, 101), 252),  # 100 акций, 252 торговых дня
    'time': np.tile(range(1, 253), 100),
    'price': np.random.randn(25200).cumsum() + 100,
    'volume': np.random.exponential(1000, 25200)
})

# Извлечение признаков для цены и объема
features = extract_features(financial_df, 
                          column_id='id', 
                          column_sort='time',
                          column_value='price')

Анализ IoT данных

# Пример с данными IoT сенсоров
iot_df = pd.DataFrame({
    'device_id': np.repeat(range(1, 51), 1440),  # 50 устройств, 1440 минут в сутках
    'timestamp': np.tile(range(1, 1441), 50),
    'temperature': 20 + 5 * np.sin(np.tile(np.linspace(0, 2*np.pi, 1440), 50)) + np.random.normal(0, 1, 72000),
    'humidity': 50 + 10 * np.cos(np.tile(np.linspace(0, 2*np.pi, 1440), 50)) + np.random.normal(0, 2, 72000)
})

# Мультивариантное извлечение признаков
features = extract_features(iot_df, 
                          column_id='device_id', 
                          column_sort='timestamp',
                          column_value='temperature')

Заключение

TSFresh представляет собой мощный и универсальный инструмент для работы с временными рядами в задачах машинного обучения. Библиотека значительно упрощает процесс feature engineering, автоматизируя извлечение и отбор признаков на основе статистических методов. Благодаря богатому набору функций, гибкой конфигурации и интеграции с экосистемой Python, TSFresh находит широкое применение в различных областях - от финансовой аналитики до промышленного мониторинга.

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