基于Pandas高效汇总指定时间区间内DataFrame数据的方法问询

需求说明 #

• 时间区间由DataFrame A提供，包含start_timestamp和end_timestamp列，每行对应一个待汇总的时间区间
• 待汇总数据来自DataFrame B，包含timestamp列及多列metric字段，需对metric执行均值、最大值、最小值等聚合操作
• 输出结果的行数与DataFrame A完全一致，每行对应A中一个区间的汇总结果
• 额外要求：DataFrame A、B可能包含数千行数据，B最多包含100个metric列，必须避免使用for循环实现

可复现示例 #

输入DataFrame A #

# 输入DataFrame A
import pandas as pd

df_a = pd.DataFrame({
    "start_timestamp": ["2022-08-09 00:30", "2022-08-09 01:00", "2022-08-09 01:15"],
    "end_timestamp": ["2022-08-09 03:30", "2022-08-09 04:00", "2022-08-09 08:15"]
})
df_a.loc[:, "start_timestamp"] = pd.to_datetime(df_a["start_timestamp"])
df_a.loc[:, "end_timestamp"] = pd.to_datetime(df_a["end_timestamp"])
print(df_a)

输出表格：

输入DataFrame B #

# 输入DataFrame B
df_b = pd.DataFrame({
    "timestamp":[
        "2022-08-09 01:00", 
        "2022-08-09 02:00", 
        "2022-08-09 03:00", 
        "2022-08-09 04:00", 
        "2022-08-09 05:00", 
        "2022-08-09 06:00", 
        "2022-08-09 07:00", 
        "2022-08-09 08:00", 
    ],
    "metric": [1, 2, 3, 4, 5, 6, 7, 8], 
})
df_b.loc[:, "timestamp"] = pd.to_datetime(df_b["timestamp"])
print(df_b)

输出表格：

预期输出DataFrame #

# 预期输出（循环实现示例）
df_target = df_a.copy()
for i, row in df_target.iterrows():
    condition = (df_b["timestamp"] >= row["start_timestamp"]) & (df_b["timestamp"] <= row["end_timestamp"])
    df_b_sub = df_b.loc[condition, :]
    df_target.loc[i, "metric_mean"] = df_b_sub["metric"].mean()
    df_target.loc[i, "metric_max"] = df_b_sub["metric"].max()
    df_target.loc[i, "metric_min"] = df_b_sub["metric"].min()
print(df_target)

输出表格：

高效矢量化实现方案 #

核心思路 #

利用Pandas的IntervalIndex和矢量化索引匹配替代循环实现区间匹配，再结合分组聚合完成计算，全程无循环，性能远优于逐行遍历。

代码实现 #

import pandas as pd

# 1. 将DataFrame A的时间区间转换为IntervalIndex，closed='both'表示区间包含两端点
intervals = pd.IntervalIndex.from_arrays(
    df_a['start_timestamp'], 
    df_a['end_timestamp'], 
    closed='both'
)

# 2. 为DataFrame B的每个timestamp匹配所属的区间索引（矢量化操作，无循环）
df_b['interval_idx'] = intervals.get_indexer(df_b['timestamp'])

# 3. 按区间索引分组，对metric列执行聚合操作
# 若有多个metric列，将['metric']替换为所有metric列名的列表即可
agg_results = df_b.groupby('interval_idx')['metric'].agg(
    metric_mean='mean',
    metric_max='max',
    metric_min='min'
)

# 4. 将聚合结果与df_a合并，保证行数与df_a完全一致（无匹配的区间会填充NaN）
df_result = df_a.join(agg_results, how='left')

print(df_result)

输出验证 #

执行上述代码后，输出结果与预期完全一致：

多metric列适配 #

如果DataFrame B包含多个metric列（如metric1、metric2...），可批量指定聚合函数：

# 假设B的metric列是['metric1', 'metric2', 'metric3']
agg_funcs = {col: ['mean', 'max', 'min'] for col in ['metric1', 'metric2', 'metric3']}
agg_results = df_b.groupby('interval_idx').agg(agg_funcs)
# 扁平化列名，方便与df_a合并
agg_results.columns = [f'{col}_{func}' for col, func in agg_results.columns]

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