Polars中支持DST感知的多日时间滞后实现方案咨询

预期处理结果：

┌───────────────────────────────┬───────────┬───────────┐
│ timestamp ┆ value ┆ lag_value │
│ --- ┆ --- ┆ --- │
│ datetime[ms, Europe/Brussels] ┆ f64 ┆ f64 │
╞═══════════════════════════════╪═══════════╪═══════════╡
│ 2017-03-27 00:00:00 CEST ┆ 2.358506 ┆ -0.453049 │
│ 2017-03-27 00:30:00 CEST ┆ -1.235676 ┆ 1.696162 │
│ 2017-03-27 01:00:00 CEST ┆ -0.430255 ┆ 0.10527 │
│ 2017-03-27 01:30:00 CEST ┆ -1.460279 ┆ 0.93969 │
│ 2017-03-27 02:00:00 CEST ┆ -0.918418 ┆ 1.158872 │
│ 2017-03-27 02:30:00 CEST ┆ -0.933531 ┆ -0.158087 │
│ 2017-03-27 03:00:00 CEST ┆ -0.421031 ┆ 1.158872 │
│ 2017-03-27 03:30:00 CEST ┆ -0.800223 ┆ -0.158087 │
...

尝试过的方法及问题：
- 通用`shift`方法无法针对DST导致的时间不连续场景实现精准滞后
- 直接对本地时间使用`offset_by`会触发Polars计算错误（非存在时间报错）：
```python
from datetime import datetime
df = pl.DataFrame({"localdatetime": [datetime(2017, 3, 27, 2)]}).with_columns(pl.col("localdatetime").dt.replace_time_zone("Europe/Brussels"))

df = df.with_columns(
    pl.col("localdatetime").dt.offset_by("-1d").alias("localdatetime_YTD")
)

>>> polars.exceptions.ComputeError: datetime '2017-03-26 02:00:00' is non-existent in time zone 'Europe/Brussels'. You may be able to use `non_existent='null'` to return `null` in this case.

• 常规向前/向后填充仅适用于单值缺失，无法处理DST导致的整段时间块缺失，需按策略整体偏移匹配。

解决方案 #

核心思路 #

通过UTC时间作为中间层规避DST转换的歧义：先将本地时间转为UTC完成滞后计算，再映射回本地时间；同时构建完整的本地时间序列，确保每个时间间隔都有记录，最后按策略填充缺失值并匹配滞后数据。

Polars实现代码 #

import polars as pl
from datetime import timedelta

def add_lag_with_dst(df: pl.DataFrame, timestamp_col: str, value_col: str, lag_days: int = 1, interval_minutes: int = 30):
    # 1. 将本地时间转换为UTC，避免DST转换的歧义
    df_utc = df.with_columns(
        pl.col(timestamp_col).dt.convert_time_zone("UTC").alias("utc_timestamp")
    )
    
    # 2. 去重：同一本地时间多条记录取第一条（按UTC时间排序后保留最早的）
    df_utc_unique = df_utc.sort("utc_timestamp").unique(subset=[timestamp_col], keep="first")
    
    # 3. 构建完整的本地时间序列：覆盖数据起止时间+滞后天数，按指定间隔生成
    time_zone = df_utc_unique[timestamp_col].min().dt.time_zone()
    min_local = df_utc_unique[timestamp_col].min()
    max_local = df_utc_unique[timestamp_col].max() + timedelta(days=lag_days)
    
    full_local_seq = pl.DataFrame({
        timestamp_col: pl.date_range(
            start=min_local,
            end=max_local,
            interval=f"{interval_minutes}m",
            time_zone=time_zone
        )
    })
    
    # 4. 关联原始数据，对缺失值执行向后填充
    full_df = full_local_seq.join(df_utc_unique, on=timestamp_col, how="left").sort(timestamp_col)
    full_df = full_df.with_columns(
        pl.col(value_col).fill_null(strategy="backward")
    )
    
    # 5. 计算滞后：通过UTC时间偏移n天，避免本地时间的DST错误
    full_df = full_df.with_columns(
        pl.col("utc_timestamp").dt.offset_by(f"-{lag_days}d").alias("lag_utc_timestamp")
    )
    
    # 6. 关联滞后值：用偏移后的UTC时间匹配原始数据中的对应值
    lag_mapping = full_df.select(
        pl.col("utc_timestamp").alias("match_utc"),
        pl.col(value_col).alias(f"lag_{value_col}")
    )
    
    result = full_df.join(lag_mapping, left_on="lag_utc_timestamp", right_on="match_utc", how="left")
    
    # 7. 清理冗余列，保留目标字段
    result = result.select([timestamp_col, value_col, f"lag_{value_col}"])
    
    return result

# 示例使用
if __name__ == "__main__":
    # 模拟示例数据
    sample_rows = [
        ("2017-03-26 00:00:00 CET", -0.453049),
        ("2017-03-26 00:30:00 CET", 1.696162),
        ("2017-03-26 01:00:00 CET", 0.10527),
        ("2017-03-26 01:30:00 CET", 0.93969),
        ("2017-03-26 03:00:00 CEST", 1.158872),
        ("2017-03-26 03:30:00 CEST", -0.158087),
        ("2017-03-27 00:00:00 CEST", 2.358506),
        ("2017-03-27 00:30:00 CEST", -1.235676),
        ("2017-03-27 01:00:00 CEST", -0.430255),
        ("2017-03-27 01:30:00 CEST", -1.460279),
        ("2017-03-27 02:00:00 CEST", -0.918418),
        ("2017-03-27 02:30:00 CEST", -0.933531),
        ("2017-03-27 03:00:00 CEST", -0.421031),
        ("2017-03-27 03:30:00 CEST", -0.800223),
    ]
    
    df = pl.DataFrame(sample_rows, schema=["timestamp", "value"]).with_columns(
        pl.col("timestamp").str.to_datetime(time_zone="Europe/Brussels")
    )
    
    # 生成滞后1天、30分钟间隔的结果
    output_df = add_lag_with_dst(df, "timestamp", "value", lag_days=1, interval_minutes=30)
    print(output_df)

关键步骤说明 #

1. UTC转换：本地时间转UTC后，时间偏移计算不会受DST影响，避免非存在时间的报错。
2. 去重逻辑：按UTC时间排序后保留同一本地时间的第一条记录，符合"记录过多取第一条"的要求。
3. 完整序列构建：生成覆盖所有需要的时间间隔的序列，确保没有缺失的时间点。
4. 向后填充：对缺失的记录值使用后续最近的有效值填充，满足"记录缺失时向后填充"的策略。
5. 滞后匹配：通过UTC时间偏移后再匹配，确保滞后值对应正确的时间点。

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