如何基于自定义可变时长时间范围对Pandas DataFrame分箱?
根据Timestamp匹配测试归属的Pandas解决方案
看起来你需要把每条测试结果记录,根据它的Timestamp归属到对应的测试批次里——也就是新增一列标记该记录属于哪个预先定义好时间范围的测试对吧?我来分享几个实用的Pandas实现方法,你可以根据自己的场景选择:
先明确前提:定义测试时间范围
首先你得把每个测试对应的时间区间整理好,比如用字典或者单独的DataFrame来存储:
# 示例:用字典存储测试名称和对应的时间区间(起始时间,结束时间) test_time_ranges = { "测试A": ("2017-10-14 00:00:00", "2017-10-14 00:30:00"), "测试B": ("2017-10-14 00:30:00", "2017-10-14 01:00:00"), "测试C": ("2017-10-14 01:00:00", "2017-10-14 01:30:00") }
另外,先确保你的主DataFrame里的Timestamp列是datetime类型,如果还没转的话先执行:
df['Timestamp'] = pd.to_datetime(df['Timestamp'])
方法1:用pd.cut快速分箱(适合连续无重叠区间)
如果你的测试时间是连续且没有重叠的,pd.cut是最高效的方式:
- 先把测试的时间区间转换成有序的时间戳数组,同时准备好对应的测试名称标签:
# 提取所有区间的起始时间,加上最后一个区间的结束时间 time_bins = [] test_labels = [] for test_name, (start, end) in test_time_ranges.items(): time_bins.append(pd.to_datetime(start)) test_labels.append(test_name) # 添加上最后一个区间的结束边界 time_bins.append(pd.to_datetime(list(test_time_ranges.values())[-1][1]))
- 调用
pd.cut新增归属列:
df['所属测试'] = pd.cut( df['Timestamp'], bins=time_bins, labels=test_labels, include_lowest=True # 确保第一个区间包含左边界,避免第一条记录被排除 )
提示:如果你的区间是左闭右开的格式,
include_lowest=True会帮你处理第一个时间点的匹配问题。
方法2:用merge_asof匹配(适合复杂场景)
如果你的测试时间范围存在单独的DataFrame里,或者需要处理不同设备对应不同测试的情况,merge_asof会更灵活:
首先假设你有一个测试信息的DataFrame:
test_df = pd.DataFrame({ '测试名称': ['测试A', '测试B', '测试C'], '开始时间': pd.to_datetime(['2017-10-14 00:00:00', '2017-10-14 00:30:00', '2017-10-14 01:00:00']), '结束时间': pd.to_datetime(['2017-10-14 00:30:00', '2017-10-14 01:00:00', '2017-10-14 01:30:00']) })
然后执行以下步骤:
# 对两个DataFrame按时间排序(merge_asof要求必须排序) df_sorted = df.sort_values('Timestamp') test_sorted = test_df.sort_values('开始时间') # 用merge_asof匹配最近的测试开始时间,再过滤掉超出测试结束时间的记录 merged = pd.merge_asof(df_sorted, test_sorted, left_on='Timestamp', right_on='开始时间') merged = merged[merged['Timestamp'] <= merged['结束时间']] # 把匹配结果合并回原DataFrame(保留所有原始行,未匹配的会显示NaN) df = df.merge(merged[['DeviceID', 'Timestamp', '测试名称']], on=['DeviceID', 'Timestamp'], how='left')
方法3:用apply自定义匹配(灵活但大数据量慎用)
如果你的时间区间规则比较特殊(比如允许重叠区间,需要自定义优先级),可以用apply写个简单的函数:
def match_test(timestamp): # 遍历所有测试区间,找到匹配的测试名称 for test_name, (start, end) in test_time_ranges.items(): start_dt = pd.to_datetime(start) end_dt = pd.to_datetime(end) if start_dt <= timestamp <= end_dt: return test_name # 处理不在任何区间的记录 return '未匹配测试' # 新增所属测试列 df['所属测试'] = df['Timestamp'].apply(match_test)
注意:这个方法虽然直观,但如果你的DataFrame有几十万甚至上百万行,效率会比前两种低很多,所以大数据量场景优先选前两种。
额外提醒
- 确保所有时间都是
datetime类型,否则会出现匹配错误; - 如果存在时间区间重叠的情况,
pd.cut会报错,这时候用apply或者merge_asof更合适; - 对于未匹配到任何测试的行,可以设置默认值(比如上面的'未匹配测试'),避免出现
NaN。
内容的提问来源于stack exchange,提问作者ze'ev
相关产品推荐
相关产品推荐

