如何基于readPeriod列限制填充daily_cons列的空行？

需求说明 #

现有包含多客户每日数据的DataFrame，部分行的cust_id和read_dt有值，但readPeriod和daily_cons为null。需实现：当readPeriod>1时，将当前行的daily_cons值回填至当前行及之前X行（X = readPeriod - 1）；仅处理readPeriod>1的已知值，其余空值保留为NaN。

背景：daily_cons由计量表起始/结束读数计算得出，正常读数对应1天，因通信问题，部分读数会覆盖多天，需将该值分配至对应周期内的所有日期。

示例数据 #

初始数据 #

starting_data = {
    'cust_id': [1, 1, 1, 1, 1, 1, 2, 2, 2, 2, 2, 2],
    'read_dt': ['2023-08-01', '2023-08-02', '2023-08-03', '2023-08-04', '2023-08-05', '2023-08-06', 
                '2023-08-01', '2023-08-02', '2023-08-03', '2023-08-04', '2023-08-05', '2023-08-06'],
    'readPeriod': [1, None, None, 3, None, 1, 1, 1, None, None, None, 4],
    'daily_cons': [10, None, None, 15, None, 24, 13, 16, None, None, None, 14]
}

期望结果 #

result_data = {
    'cust_id': [1, 1, 1, 1, 1, 1, 2, 2, 2, 2, 2, 2],
    'read_dt': ['2023-08-01', '2023-08-02', '2023-08-03', '2023-08-04', '2023-08-05', '2023-08-06', 
                '2023-08-01', '2023-08-02', '2023-08-03', '2023-08-04', '2023-08-05', '2023-08-06'],
    'readPeriod': [1, None, None, 3, None, 1, 1, 1, None, None, None, 4],
    'daily_cons': [10, 15, 15, 15, None, 24, 13, 16, 14, 14, 14, 14]
}

尝试过的方法及问题 #

方法1：分组后bfill #

condition = (dc['readPeriod'] > 1)
dc['daily_cons'] = dc.groupby('cust_id')['daily_cons'].apply(lambda x: x.bfill() if condition.any() else x)

问题：无差别填充所有后续空值，未限制回填范围，不符合需求。

方法2：累计和标记填充 #

m1 = dc['readPeriod'].notna()

m2 = (dc.loc[::-1, 'readPeriod'].fillna(-1)
        .groupby(dc['id']).cumsum()
        .ge(1)
     )

dc['daily_cons'] = dc['daily_cons'].bfill().where(m1|m2)

问题：误填充了不需要的行（如read_dt为2023-02-09和2023-02-10的行，应保持NaN），错误结果如下：

解决方案 #

核心思路：按客户分组，对每个分组内readPeriod>1的行，精准计算回填区间，仅填充对应范围的空值。

代码实现 #

import pandas as pd

# 构造初始DataFrame（修正原数据语法错误）
starting_data = {
    'cust_id': [1, 1, 1, 1, 1, 1, 2, 2, 2, 2, 2, 2],
    'read_dt': ['2023-08-01', '2023-08-02', '2023-08-03', '2023-08-04', '2023-08-05', '2023-08-06', 
                '2023-08-01', '2023-08-02', '2023-08-03', '2023-08-04', '2023-08-05', '2023-08-06'],
    'readPeriod': [1, None, None, 3, None, 1, 1, 1, None, None, None, 4],
    'daily_cons': [10, None, None, 15, None, 24, 13, 16, None, None, None, 14]
}

dc = pd.DataFrame(starting_data)
dc['read_dt'] = pd.to_datetime(dc['read_dt'])

def backfill_group(group):
    # 按日期排序，确保时间顺序正确
    group = group.sort_values('read_dt').reset_index(drop=True)
    # 遍历所有readPeriod>1的行
    for idx, row in group[group['readPeriod'] > 1].iterrows():
        # 计算回填起始索引：当前索引 - (readPeriod-1)，最小为0
        start_idx = max(0, idx - int(row['readPeriod']) + 1)
        # 填充区间内的daily_cons
        group.loc[start_idx:idx, 'daily_cons'] = row['daily_cons']
    return group

# 按客户分组处理，重置索引
dc_filled = dc.groupby('cust_id').apply(backfill_group).reset_index(drop=True)
print(dc_filled)

代码说明 #

1. 分组排序：按cust_id分组后，先对每个组按read_dt排序，确保时间顺序正确，避免因数据乱序导致回填错误。
2. 精准区间计算：对每个readPeriod>1的行，计算需要回填的起始位置（当前行索引减去readPeriod-1，且不小于0），确保只填充指定的X天范围。
3. 定向填充：将计算出的区间内的daily_cons设置为当前行的值，仅覆盖目标范围，其余空值保持NaN。

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