Привет, Хабр!
Сегодня рассмотрим тему обработки временных рядов с помощью Polars.
Почему groupby_dynamic() лучше resample() из Pandas
Начну с того, что в Pandas для агрегации временных рядов принято использовать метод resample(). Он удобен и привычен, но имеет свои ограничения по производительности и гибкости. Polars, в свою очередь, имеет метод groupby_dynamic(), который позволяет группировать данные по динамическим временным интервалам.
Рассмотрим, как можно сгруппировать данные с часовыми метками по дневным интервалам:
import polars as pl
from datetime import datetime, timedelta
# Генерируем данные: неделя записей с часовым интервалом
dates = [datetime(2025, 1, 1) + timedelta(hours=i) for i in range(24 * 7)]
values = [i % 5 + 1 for i in range(24 * 7)]
df = pl.DataFrame({"timestamp": dates, "value": values})
# Группируем данные по дням и агрегируем: суммируем и считаем среднее
resampled = df.groupby_dynamic("timestamp", every="1d").agg([
pl.col("value").sum().alias("daily_sum"),
pl.col("value").mean().alias("daily_mean")
])
print(resampled)С помощью groupby_dynamic() определяем временной интервал (every=»1d») и сразу же агрегируем данные. Метод компилируется в код на Rust, что даёт прирост производительности по сравнению с Pandas. Если сравнить с Pandas, то код будет выглядеть примерно так:
import pandas as pd
df_pandas = pd.DataFrame({"timestamp": dates, "value": values})
df_pandas.set_index("timestamp", inplace=True)
daily = df_pandas.resample("D").agg({"value": ["sum", "mean"]})
print(daily)Rolling Windows: rolling_mean() и rolling_std() без overhead
Поговорим о скользящих окнах. Все мы знаем, что расчет скользящих средних и стандартного отклонения — стандартная операция при анализе временных рядов.
Polars предлагает функции rolling_mean() и rolling_std(), которые оптимизированы до предела. Они реализованы на Rust
Пример вычисления скользящего среднего и стандартного отклонения:
df = pl.DataFrame({
"timestamp": dates,
"value": values
}).with_columns([
pl.col("value").rolling_mean(window_size=24, min_periods=1).alias("rolling_mean"),
pl.col("value").rolling_std(window_size=24, min_periods=1).alias("rolling_std")
])
print(df.head(30))Рассчитываем 24-часовое скользящее среднее и стандартное отклонение. Параметр min_periods=1 позволяет начать вычисления уже с первого значения, а оптимизированная реализация гарантирует, что даже при миллионах записей задержек практически не будет.
Также все это можно комбинировать с другими методами Polars.
Интерполяция пропущенных значений: Polars.interpolate()
Работая с временными рядами, мы часто сталкиваемся с пробелами в данных. От сбоя датчиков до ошибок при сборе информации — пропуски неизбежны. Хорошая новость: Polars предлагает метод interpolate() для быстрого заполнения пропущенных значений.
Создадим DataFrame с пропущенными значениями и применим линейную интерполяцию:
df_missing = pl.DataFrame({
"timestamp": dates,
"value": [None if i % 10 == 0 else i % 5 + 1 for i in range(len(dates))]
})
# Применяем линейную интерполяцию для заполнения пропусков
df_interpolated = df_missing.with_columns([
pl.col("value").interpolate(method="linear").alias("value_interpolated")
])
print(df_interpolated.head(30))interpolate() позволяет указать метод интерполяции (в данном случае — «linear»).
Оптимизация анализа с помощью LazyFrame
Polars поддерживает ленивые вычисления через объект LazyFrame. Можно писать длинные цепочки преобразований, а Polars сам оптимизирует план выполнения и выполняет только необходимые вычисления в самый последний момент.
Пример оптимизированной обработки данных:
# Преобразуем DataFrame в LazyFrame
lazy_df = df.lazy()
# Строим цепочку операций: фильтрация, группировка по дням, агрегирование
result = lazy_df.filter(pl.col("value") > 2)\
.groupby_dynamic("timestamp", every="1d")\
.agg([
pl.col("value").mean().alias("daily_mean")
])\
.collect() # Выполняем вычисления
print(result)Суть в том, что до вызова .collect() никаких вычислений не происходит. Это позволяет оптимизировать запрос, минимизировать избыточные проходы по данным и сократить время выполнения.
Пример обработкт временных рядов в магазине
Представим, магазина собирает данные о продажах с интервалом в час. Данные собираются постоянно, но иногда происходят сбои, и в ряде случаев значения пропадают. Задача — собрать данные, заполнить пропуски, агрегировать продажи по дням и вычислить скользящую недельную среднюю. Всё это — на Polars, и всё это должно работать быстро.
Сначала сгенерируем данные с часовыми записями за 30 дней, включая случайные пропуски:
import random
random.seed(42)
# Создаем список дат: 30 дней, 24 записи в день
dates_shop = [datetime(2021, 6, 1) + timedelta(hours=i) for i in range(30 * 24)]
# Генерируем данные по продажам: случайные числа, пропуски ~10%
sales = [random.randint(50, 200) if random.random() > 0.1 else None for _ in range(len(dates_shop))]
df_shop = pl.DataFrame({
"timestamp": dates_shop,
"sales": sales
})Заполним пробелы в данных, используя линейную интерполяцию:
df_shop = df_shop.with_columns([
pl.col("sales").interpolate(method="linear").alias("sales_interpolated")
])Теперь агрегируем данные по дням, чтобы получить суммарные продажи и среднее значение за день:
daily_sales = df_shop.groupby_dynamic("timestamp", every="1d").agg([
pl.col("sales_interpolated").sum().alias("daily_total_sales"),
pl.col("sales_interpolated").mean().alias("daily_avg_sales")
])Чтобы отследить тренды и сезонные колебания, посчитаем скользящую среднюю продаж за последние 7 дней:
daily_sales = daily_sales.with_columns([
pl.col("daily_total_sales").rolling_mean(window_size=7, min_periods=1).alias("weekly_sales_avg")
])
print(daily_sales)В результате получаем DataFrame, где каждая запись содержит дневные итоги и рассчитанное значение скользящего среднего.
В завершение напоминаю об открытых уроках, которые пройдут в Otus в марте:
25 марта: «Метрики и Prometheus».
Узнать подробнее27 марта: «PostgreSQL на стероидах: большие данные, высокие нагрузки и масштабирование без боли».
Узнать подробнее
Больше актуальных навыков по аналитике данных вы можете получить в рамках практических онлайн-курсов от экспертов отрасли.