本文介绍如何基于多条带起止时间的记录，为每个 ID 计算其所有关联 code 的共同重叠时间段（即所有 code 的时间区间交集），并精准过滤出覆盖全部预定义 code 的完整重叠结果。
本文介绍如何基于多条带起止时间的记录，为每个 id 计算其所有关联 code 的**共同重叠时间段**（即所有 code 的时间区间交集），并精准过滤出覆盖全部预定义 code 的完整重叠结果。
在实际数据分析中（如医疗事件共现分析、设备并发运行检测、用户行为交叉验证等），我们常需判断：对于某实体（如 id），其所有关联子项（如 code）是否存在同时生效的时间段？这本质上是求多个时间区间的交集，且要求该交集必须覆盖该 ID 下全部预设 code —— 而非任意两两重叠。
以下以示例数据为基础，分步实现这一目标：
✅ 步骤 1：准备与类型转换
确保时间列为 datetime64 类型，这是后续区间运算的基础：
import pandas as pd
# 原始数据
data_group = {'id': ['0111','0123'], 'code': [['1', '2', '3'],['1','2']]}
df_group = pd.DataFrame(data_group)
data = {
'codice': ['1', '2', '3', '1','1','1'],
'id': ['0111', '0111', '0111', '0111','0123','0123'],
'data1': ['2025-02-03 02:16:00', '2025-02-03 02:18:00', '2025-02-03 02:17:00', '2025-02-03 12:02:00','2025-02-03 12:02:00','2025-02-03 12:02:00'],
'data2': ['2025-02-03 02:44:00', '2025-02-03 02:44:00', '2025-02-03 02:39:00', '2025-02-03 12:05:00','2025-02-03 12:06:00','2025-02-03 12:04:00']
}
df = pd.DataFrame(data)
# 关键：转为 datetime
df[['data1', 'data2']] = df[['data1', 'data2']].apply(pd.to_datetime)✅ 步骤 2：合并重叠区间（按 ID 分组）
核心思想：对同一 id 下所有区间按起始时间排序后，使用“滑动窗口”逻辑合并重叠部分，并标记连续重叠组（n）。这里利用 data1 > data2.shift() 判断是否断开，再通过 cumsum() 生成组号：
merged = (
df.sort_values(['id', 'data1', 'data2'])
.assign(n=lambda x: x['data1'].gt(x['data2'].shift()).groupby(x['id']).cumsum())
.groupby(['id', 'n'], as_index=False)
.agg({
'data1': 'max', # 同一组内最晚开始 → 实际交集起点
'data2': 'min', # 同一组内最早结束 → 实际交集终点
'codice': set # 记录该组覆盖的 code 集合
})
)? 原理说明：
data1.max()和data2.min()在已合并的连续重叠组中，恰好对应该组所有原始区间的最大交集（即所有区间共同覆盖的时间窗）。这是区间交集的经典计算方式。
✅ 步骤 3：筛选「全覆盖」结果
仅保留那些 codice 集合完全匹配 df_group 中该 ID 对应 code 列表的结果（注意：set(['1','2','3']) == set(['3','1','2'])）：
# 构建 id → code_set 映射
expected_codes = df_group.set_index('id')['code'].apply(set)
result = merged.loc[
merged['id'].map(expected_codes).eq(merged['codice'])
]输出示例：
id n data1 data2 codice
0 0111 0 2025-02-03 02:18:00 2025-02-03 02:39:00 {1, 2, 3}✅ 这正是题目所求：id='0111' 的 codes ['1','2','3'] 共同存在的唯一重叠区间为 [02:18:00, 02:39:00]。
✅ 可选：格式化为字典输出
如需对接下游 API 或 JSON 序列化，可快速转为嵌套字典：
output_dict = (
result.set_index('id')[['data1', 'data2']]
.astype(str) # 转字符串便于序列化
.T.to_dict()
)
# {'0111': {'data1': '2025-02-03 02:18:00', 'data2': '2025-02-03 02:39:00'}}⚠️ 注意事项
-
时间精度：若存在毫秒级差异，建议统一
round()或floor()处理，避免因浮点误差导致min/max偏移； -
空交集处理：当无全局重叠时，
result为空 DataFrame，建议增加.empty检查； -
边界语义：本解法默认
[start, end]为闭区间；若业务为左闭右开（如data2表示“失效时刻”），需将data2统一减去pd.Timedelta(1)再计算； -
性能优化：大数据量时，可先用
df.groupby('id').filter(lambda g: len(g) >= len(df_group.loc[df_group['id']==g.name, 'code'].iloc[0]))预过滤 ID。
通过以上方法，你不仅能准确识别多维时间区间的严格交集，还能确保结果满足业务定义的「全要素覆盖」约束——这是构建可信时序分析流水线的关键能力。
