如何用Python编写函数将年度人口数据插值为小时级时间序列

现有代码核心问题 #

• 初始化DataFrame时未指定列名，调用df_pop['name']等列属性会直接报错
• 按区域名称循环时未筛选对应区域的年度数据，所有区域数据混用导致插值逻辑完全错误
• scipy.interpolate.interp1d调用逻辑错误：该方法需先传入已知的x/y序列生成插值函数，再传入目标插值点计算结果，不可直接将目标点作为第三个参数传入
• 插值的x轴维度不统一：已知点是年度整数，目标插值点是时间戳，需统一转换为年度浮点数值（如2015年6月对应2015.5）才能正确插值
• 结果拼接逻辑错误，未将插值结果与时间序列、区域名称对应存储

修正后可运行代码 #

import pandas as pd
import random
from scipy.interpolate import interp1d

# 示例数据创建时指定列名
name = ['RI', 'NH', 'MA', 'RI', 'NH', 'MA','RI', 'NH', 'MA','RI', 'NH', 'MA']
year = [2015, 2015, 2015, 2016, 2016, 2016, 2017, 2017, 2017, 2018, 2018, 2018]
population = random.sample(range(10000, 300000), 12)
df_pop = pd.DataFrame(list(zip(name, year, population)), columns=['name', 'year', 'population'])

start_year = 2015 
end_year = 2018 

def pop_sum(df_pop, start_year, end_year):
    names = df_pop['name'].unique()
    res_list = []
    # 生成统一的小时级时间序列，pandas2.0+请将closed参数替换为inclusive='left'
    hour_series = pd.date_range(f'{start_year}-01-01', f'{end_year}-12-31', freq='1H', closed='left')
    # 转换为年度浮点值用于插值计算
    year_float = hour_series.year + (hour_series.dayofyear - 1)/365.25
    
    for name in names:
        # 筛选当前区域的年度数据
        sub_df = df_pop[df_pop['name'] == name].sort_values('year')
        t_known = sub_df['year'].values
        y_known = sub_df['population'].values
        # 生成插值函数，fill_value设置为外推防止边界点报错
        interp_func = interp1d(t_known, y_known, kind='linear', fill_value='extrapolate')
        # 计算小时级插值结果
        pop_interp = interp_func(year_float)
        # 拼接当前区域的结果
        tmp_df = pd.DataFrame({
            'hours': hour_series,
            'name': name,
            'population': pop_interp
        })
        res_list.append(tmp_df)
    # 合并所有区域的结果返回
    return pd.concat(res_list, ignore_index=True)

# 调用测试
hourly_pop = pop_sum(df_pop, start_year, end_year)
print(hourly_pop.head())

替代实现方案 #

如果不需要额外依赖scipy，也可以直接用pandas自带的重采样插值能力实现，逻辑更简洁：

def pop_sum_pandas(df_pop, start_year, end_year):
    df_list = []
    for name in df_pop['name'].unique():
        sub_df = df_pop[df_pop['name'] == name].copy()
        # 将年度数据关联到每年1月1日的时间点
        sub_df['dt'] = pd.to_datetime(sub_df['year'].astype(str) + '-01-01')
        sub_df = sub_df.set_index('dt')[['population']]
        # 重采样到小时级后线性插值
        hourly_sub = sub_df.resample('1H').interpolate(method='linear')
        # 筛选目标时间范围
        hourly_sub = hourly_sub.loc[f'{start_year}-01-01':f'{end_year}-12-31'].iloc[:-1]
        hourly_sub['name'] = name
        hourly_sub = hourly_sub.reset_index(names='hours')
        df_list.append(hourly_sub)
    return pd.concat(df_list, ignore_index=True)

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