如何在Polars的group_by_dynamic中应用前向填充(ffill)？

问题场景 #

使用Polars的group_by_dynamic方法做数据下采样时，无数据的时间段会被直接过滤，但业务需要保留这些时段并以前一行有效值填充（类似Pandas中reindex+ffill的效果）。尝试过调用fill_nan()、fill_null()、forward_fill()均未成功，核心原因是缺失的时段根本没有出现在结果集中，填充方法无法作用于不存在的行。

现有Polars处理代码 #

import polars as pl

def resampling_pl(df: pl.DataFrame, timeframe: int) -> pl.DataFrame:
    df = df.with_columns(pl.col("Date") + " " + pl.col("Time"))
    df = df.with_columns(
        pl.col("Date").
            str.to_datetime(
                format="%Y-%m-%d %H:%M:%S%.f", 
                strict=False
            )
        .alias("Open time")
    )

    df = df.select("Open time", "Price")
    
    data_rs = (df.group_by_dynamic(
        index_column="Open time", every=str(timeframe)+"s", closed="both")
            .agg(
                pl.col("Price").first().alias("Open"),
                pl.col("Price").max().alias("High"),
                pl.col("Price").min().alias("Low"),
                pl.col("Price").last().alias("Close")
            )
        )

    return data_rs

预期效果的Pandas参考实现 #

### 此方法通过Pandas实现了预期的前向填充效果
def ffill_(data, timeframe):
    df = data.copy()
    df.index = df["Open time"]
    df.drop(["Open time"], axis=1, inplace=True)

    start_date = df.index.min()
    end_date = df.index.max()

    new_index = pd.date_range(start=start_date, end=end_date, freq=str(timeframe)+'S')
    df = df.reindex(new_index)
    df = df.fillna(method='ffill')
    del data

    return df


### 调用示例
file_path = "nq.csv"
df = pl.read_csv(file_path)
data1 = resampling_pl(df, 30).to_pandas()
data1.index = data1["Open time"]
data1 = ffill_(data1, 30)

比如示例中，2023-03-03 16:49:30的时段原本缺失，需要填充为与16:49:15行相同的值。

Polars中的解决方案 #

核心思路是先生成完整的目标时间序列，再将下采样结果与该序列左连接，最后进行前向填充：

1. 从原始数据中提取时间范围，生成包含所有目标时段的时间序列
2. 将该时间序列转为DataFrame，与group_by_dynamic的结果左连接，保留所有时段
3. 对连接后的空值列执行前向填充

修改后的resampling_pl函数如下：

import polars as pl

def resampling_pl(df: pl.DataFrame, timeframe: int) -> pl.DataFrame:
    df = df.with_columns(pl.col("Date") + " " + pl.col("Time"))
    df = df.with_columns(
        pl.col("Date").
            str.to_datetime(
                format="%Y-%m-%d %H:%M:%S%.f", 
                strict=False
            )
        .alias("Open time")
    )

    df = df.select("Open time", "Price")
    
    # 生成完整的目标时间序列
    time_range = df.select(
        pl.col("Open time").min().alias("start"),
        pl.col("Open time").max().alias("end")
    ).row(0)
    
    full_time_index = pl.date_range(
        start=time_range[0],
        end=time_range[1],
        interval=f"{timeframe}s",
        name="Open time"
    ).to_frame()
    
    # 执行下采样
    data_rs = (df.group_by_dynamic(
        index_column="Open time", every=f"{timeframe}s", closed="both")
            .agg(
                pl.col("Price").first().alias("Open"),
                pl.col("Price").max().alias("High"),
                pl.col("Price").min().alias("Low"),
                pl.col("Price").last().alias("Close")
            )
        )
    
    # 左连接完整时间序列并前向填充
    result = full_time_index.join(data_rs, on="Open time", how="left").forward_fill()
    
    return result

关键说明 #

• pl.date_range用于生成连续的时间刻度，确保所有目标时段都被包含
• 左连接（how="left"）保证缺失数据的时段保留在结果中，对应列值为null
• forward_fill()对空值进行前向填充，实现与Pandas一致的效果

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