如何用Pandas按组补全小时级缺失日期并处理indicator字段

问题背景 #

原始DataFrame如下：

import pandas as pd

data = [['A', '2022-09-01 10:00:00', False, 2], ['A', '2022-09-01 12:00:00', True, 3], ['A', '2022-09-01 14:00:00', False, 1], 
        ['B', '2022-09-01 13:00:00', False, 1], ['B', '2022-09-01 16:00:00', True, 4], ['B', '2022-09-01 18:00:00', False, 3]]
df = pd.DataFrame(data=data, columns=['group', 'date', 'indicator', 'value'])

需求说明：

• 按group分组，补全每组内缺失的小时级日期
• 补全行的value沿用前一行的现有值
• 若前一行（原始存在的行）的indicator为True，则后续补全行的indicator需设为False（value保持不变）

期望输出的DataFrame：

data = [['A', '2022-09-01 10:00:00', False, 2], ['A', '2022-09-01 11:00:00', False, 2], ['A', '2022-09-01 12:00:00', True, 3], ['A', '2022-09-01 13:00:00', False, 3], ['A', '2022-09-01 14:00:00', False, 1], 
        ['B', '2022-09-01 13:00:00', False, 1], ['B', '2022-09-01 14:00:00', False, 1], ['B', '2022-09-01 15:00:00', False, 1], ['B', '2022-09-01 16:00:00', True, 4], ['B', '2022-09-01 17:00:00', False, 4], ['B', '2022-09-01 18:00:00', False, 3]]
df_desired = pd.DataFrame(data=data, columns=['group', 'date', 'indicator', 'value'])

实现步骤与代码 #

1. 预处理日期列 #

将date列转换为datetime类型，确保时间序列处理的正确性：

df['date'] = pd.to_datetime(df['date'])

2. 按组生成完整小时时间序列 #

对每个group，生成从该组最早时间到最晚时间的所有小时时间点，构建完整的时间框架：

def create_full_time_range(group):
    start = group['date'].min()
    end = group['date'].max()
    # 生成每小时的时间序列
    full_dates = pd.date_range(start=start, end=end, freq='H')
    return pd.DataFrame({'date': full_dates, 'group': group['group'].iloc[0]})

# 按group分组应用函数，合并结果
full_time_df = df.groupby('group').apply(create_full_time_range).reset_index(drop=True)

3. 合并原始数据与完整时间序列 #

将原始数据合并到完整时间框架中，保留原始行的字段值，缺失行留空以便后续填充：

merged_df = pd.merge(full_time_df, df, on=['group', 'date'], how='left')

4. 填充value字段 #

使用前向填充（ffill）补全value列，确保补全行沿用前一行的value值：

merged_df['value'] = merged_df.groupby('group')['value'].ffill()

5. 处理indicator字段 #

• 标记原始行（indicator非空的行）
• 前向填充indicator的原始值，用于判断前一行的状态
• 对补全行：如果前一个原始行的indicator为True，则将当前行的indicator设为False，否则保留前向填充值

# 标记原始行
merged_df['is_original'] = merged_df['indicator'].notna()
# 前向填充原始indicator值，用于判断前置状态
merged_df['prev_indicator'] = merged_df.groupby('group')['indicator'].ffill()

# 处理补全行的indicator逻辑
merged_df['indicator'] = merged_df.apply(
    lambda row: False if (not row['is_original'] and row['prev_indicator']) else row['indicator'],
    axis=1
)

# 填充剩余的indicator缺失值（兼容边界场景）
merged_df['indicator'] = merged_df.groupby('group')['indicator'].ffill()
# 清理临时列
merged_df.drop(['is_original', 'prev_indicator'], axis=1, inplace=True)

完整代码 #

import pandas as pd

# 原始数据
data = [['A', '2022-09-01 10:00:00', False, 2], ['A', '2022-09-01 12:00:00', True, 3], ['A', '2022-09-01 14:00:00', False, 1], 
        ['B', '2022-09-01 13:00:00', False, 1], ['B', '2022-09-01 16:00:00', True, 4], ['B', '2022-09-01 18:00:00', False, 3]]
df = pd.DataFrame(data=data, columns=['group', 'date', 'indicator', 'value'])

# 1. 转换日期类型
df['date'] = pd.to_datetime(df['date'])

# 2. 生成完整时间序列
def create_full_time_range(group):
    start = group['date'].min()
    end = group['date'].max()
    full_dates = pd.date_range(start=start, end=end, freq='H')
    return pd.DataFrame({'date': full_dates, 'group': group['group'].iloc[0]})

full_time_df = df.groupby('group').apply(create_full_time_range).reset_index(drop=True)

# 3. 合并数据
merged_df = pd.merge(full_time_df, df, on=['group', 'date'], how='left')

# 4. 填充value
merged_df['value'] = merged_df.groupby('group')['value'].ffill()

# 5. 处理indicator
merged_df['is_original'] = merged_df['indicator'].notna()
merged_df['prev_indicator'] = merged_df.groupby('group')['indicator'].ffill()

merged_df['indicator'] = merged_df.apply(
    lambda row: False if (not row['is_original'] and row['prev_indicator']) else row['indicator'],
    axis=1
)

merged_df['indicator'] = merged_df.groupby('group')['indicator'].ffill()
merged_df.drop(['is_original', 'prev_indicator'], axis=1, inplace=True)

# 查看结果
print(merged_df)

执行上述代码后，merged_df将与期望的df_desired完全一致。

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