如何让Polars的group_by_dynamic仅保留完整降采样周期？

原始数据与现有问题 #

1分钟时间序列数据定义 #

from datetime import datetime
import polars as pl

start_dt = datetime(2023,11,30,7,00)
end_dt = datetime(2023,11,30,7,11)

df_1 = pl.DataFrame(
    {
        'date': pl.datetime_range(start_dt, end_dt, interval='1m', eager=True),
        'open': 1,
        'close': 2
    }
)

现有5分钟降采样代码 #

df_5 = (
    df_1
    .group_by_dynamic('date', every='5m')
    .agg(
        pl.col('open').first(),
        pl.col('close').last()
    )
)

问题描述 #

上述代码会生成07:00、07:05、07:10三个5分钟周期，但07:10周期仅包含07:10-07:11的1分钟数据，属于不完整周期。

实际场景中数据可能不连续：比如数据仅包含07:00、07:03、07:05、07:06时，07:00周期（07:00-07:05）是完整的（因为存在07:05数据证明周期已结束），但07:05周期（07:05-07:10）不完整（无足够数据证明周期完成），需要过滤掉这类不完整周期。

优雅实现方法 #

核心思路：判断每个降采样周期是否已完成——即该周期的结束时间（周期起始时间+5分钟）是否小于等于原始数据中的最大时间。

实现代码如下：

# 获取数据中的最大时间戳
max_dt = df_1['date'].max()

df_complete = (
    df_1
    .group_by_dynamic('date', every='5m')
    .agg(
        pl.col('open').first(),
        pl.col('close').last()
    )
    # 计算每个周期的结束时间，过滤掉未完成的周期
    .filter(pl.col('date') + pl.duration(minutes=5) <= max_dt)
)

逻辑说明 #

1. 先获取原始数据的最大时间max_dt，这是判断周期是否完成的基准
2. 对每个动态分组的周期起始时间date，加上5分钟得到周期结束时间
3. 过滤掉结束时间大于max_dt的周期——这类周期还未到结束时间，属于不完整周期

这种方法既处理了数据连续的场景，也适配了数据不连续的情况：比如数据到07:06时，07:00周期的结束时间是07:05，小于等于07:06，会被保留；07:05周期的结束时间是07:10，大于07:06，会被过滤。

内容的提问来源于stack exchange，提问作者HapiDaze