基于Pandas实现带max_time限制的Sample and Hold空值填充

问题场景 #

给定如下结构的DataFrame：

需要针对每个ID执行Sample and Hold操作：对Value为空的行，用同一ID下最近的非空Value填充，但要求该非空值的时间t_other满足 t - max_time ≤ t_other ≤ t（t为当前行时间）；超出时间窗口的空值保留NaN。

例如max_time=10时，ID=1、Time=20的空值可被Time=10的1.0填充，但ID=1、Time=30的空值因与最近非空时间差为20>10，需保留NaN。

原有代码的问题 #

原方案先使用ffill填充所有空值，再通过相邻行时间差判断是否还原为NaN，逻辑存在缺陷：

def fill_nulls(group):
    group['Value'] = group['Value'].ffill()
    time_diff = group['Time'].diff()
    group['Value'][time_diff > 10] = np.nan
    return group

filled_df = df.groupby('ID').apply(fill_nulls)

该方法中，time_diff计算的是当前行与前一行的时间差，而非与原始非空值行的时间差。例如ID=1的Time=20行是填充后的结果，Time=30与20的时间差为10≤max_time，导致错误保留填充值，但实际原始非空值的时间是10，与30的时间差为20>10，应还原为NaN。

正确解决方案 #

方法1：分组后追踪原始非空值时间 #

通过记录每个行对应的最近原始非空值的时间，直接计算与当前时间的差来判断是否填充：

import pandas as pd
import numpy as np

# 构造示例数据
df = pd.DataFrame({
    'ID': [1,2,1,3,1,2,4],
    'Time': [10,10,20,20,30,30,30],
    'Value': [1.0,2.0,np.nan,4.0,np.nan,np.nan,np.nan]
})

max_time = 10

def sample_and_hold(group):
    # 确保组内按时间升序排列
    group = group.sort_values('Time').reset_index(drop=True)
    # 标记非空值行，并用ffill传递最近的非空时间和值
    non_null_mask = group['Value'].notna()
    group['last_valid_time'] = group['Time'].where(non_null_mask).ffill()
    group['last_valid_value'] = group['Value'].ffill()
    
    # 计算时间差，仅填充符合窗口限制的空值
    time_diff = group['Time'] - group['last_valid_time']
    group['Value'] = np.where(
        (group['Value'].isna()) & (time_diff <= max_time),
        group['last_valid_value'],
        group['Value']
    )
    
    # 清理临时列
    return group.drop(['last_valid_time', 'last_valid_value'], axis=1)

# 分组处理并恢复原索引顺序
filled_df = df.groupby('ID', group_keys=False).apply(sample_and_hold)
filled_df = filled_df.loc[df.index]
print(filled_df)

方法2：使用merge_asof高效匹配（适合大数据集） #

利用merge_asof向量化匹配最近的非空值行，效率更高：

import pandas as pd
import numpy as np

df = pd.DataFrame({
    'ID': [1,2,1,3,1,2,4],
    'Time': [10,10,20,20,30,30,30],
    'Value': [1.0,2.0,np.nan,4.0,np.nan,np.nan,np.nan]
})

max_time = 10

# 分离非空值作为参考表并排序
ref_df = df[df['Value'].notna()].sort_values(['ID', 'Time'])
# 待填充表排序
df_sorted = df.sort_values(['ID', 'Time'])

# 按ID匹配最近的非空值行（时间≤当前行）
merged = pd.merge_asof(
    df_sorted,
    ref_df[['ID', 'Time', 'Value']],
    on='Time',
    by='ID',
    direction='backward'
)

# 计算时间差并填充符合条件的空值
merged['time_diff'] = merged['Time'] - merged['Time_y']
merged['Value'] = np.where(
    (merged['Value_x'].isna()) & (merged['time_diff'] <= max_time),
    merged['Value_y'],
    merged['Value_x']
)

# 整理结果并恢复原索引
filled_df = merged[['ID', 'Time', 'Value']].loc[df.index]
print(filled_df)

验证结果 #

两种方法执行后，输出结果均符合预期：

内容的提问来源于stack exchange，提问作者k-c