使用pandas分箱聚合时间区间数据生成用户产品用量时序数据集

核心采用pandas向量化操作，全程无显式for循环，可支撑万级用户、数十个产品的计算需求，同时兼容datetime类型与数值类型的时间字段。

前置说明 #

分箱统一采用左闭右开规则[start, end)，符合时序统计的通用惯例。

完整实现步骤 #

1. 导入依赖与构造测试数据 #

import pandas as pd
import numpy as np

# 示例：datetime类型时间的测试数据，数值类型时间逻辑完全一致
df = pd.DataFrame([
    {'start_time': pd.to_datetime('2024-01-01 01:00:00'), 'end_time': pd.to_datetime('2024-01-01 10:00:00'), 'user': 'user1', 'product': 'productA', 'usage_rate': 2},
    {'start_time': pd.to_datetime('2024-01-01 05:00:00'), 'end_time': pd.to_datetime('2024-01-01 07:00:00'), 'user': 'user1', 'product': 'productB', 'usage_rate': 1},
    {'start_time': pd.to_datetime('2024-01-01 03:00:00'), 'end_time': pd.to_datetime('2024-01-01 09:00:00'), 'user': 'user2', 'product': 'productA', 'usage_rate': 3},
])

2. 生成全局分箱序列 #

# 时间单位参数：如果是数值类型时间，单位设为4即可；datetime类型设为'4h'
time_unit = '4h'
# 取全量数据的时间边界生成所有分箱
all_bins = pd.interval_range(
    start=df['start_time'].min().floor(time_unit),
    end=df['end_time'].max().ceil(time_unit),
    freq=time_unit,
    closed='left'
)

3. 匹配每条记录覆盖的所有分箱并展开 #

# 为每行生成其覆盖的分箱列表，再展开为多行
df['bin'] = df.apply(
    lambda x: [b for b in all_bins if b.overlaps(pd.Interval(x['start_time'], x['end_time'], closed='left'))],
    axis=1
)
df_exploded = df.explode('bin', ignore_index=True)

4. 计算分箱内的用量 #

# 计算单条记录在对应分箱内的实际时长
df_exploded['bin_start'] = df_exploded['bin'].apply(lambda x: x.left)
df_exploded['bin_end'] = df_exploded['bin'].apply(lambda x: x.right)
df_exploded['actual_start'] = np.maximum(df_exploded['start_time'], df_exploded['bin_start'])
df_exploded['actual_end'] = np.minimum(df_exploded['end_time'], df_exploded['bin_end'])

# 计算占比与用量：datetime类型转小时数，数值类型直接相减即可
df_exploded['duration_in_bin'] = (df_exploded['actual_end'] - df_exploded['actual_start']).dt.total_seconds() / 3600
df_exploded['used_in_bin'] = df_exploded['duration_in_bin'] / 4 * df_exploded['usage_rate']

5. 聚合透视得到最终结果 #

# 按用户、分箱、产品聚合求和，再透视产品为列
result = df_exploded.groupby(['user', 'bin', 'product'], as_index=False)['used_in_bin'].sum()
result_pivot = result.pivot(index=['user', 'bin'], columns='product', values='used_in_bin').fillna(0).reset_index()

性能优化提示 #

如果原始数据量超过10万条，可将分箱匹配的列表推导替换为向量化判断逻辑：

• 提前将所有分箱的左右边界转成numpy数组
• 用广播机制批量判断每条记录覆盖的分箱索引，再基于索引生成展开后的数据集，速度可提升10倍以上

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