计算两个事件序列每日重叠时长的Python实现方法

本文介绍如何使用python高效计算两组时间事件（如df_event_a和df_event_b）在每一天内的最大并发重叠时长（单位：秒），确保每日结果不超过86400秒，并自动处理事件内部重叠及跨日边界问题。

要准确计算两组事件（A与B）在每个自然日内同时发生的总时长，核心在于：对每一天，求所有A类事件与所有B类事件在该日内的时间交集并集长度——即只要某时刻至少有一个A事件和一个B事件同时活跃，该时刻即计入重叠；且同日内所有此类时刻的累计时长即为当日重叠时长（上限为24小时 = 86400秒）。

✅ 关键逻辑解析

两段区间 [a_start, a_end] 和 [b_start, b_end] 的重叠时长公式为：

overlap = max(min(a_end, b_end) - max(a_start, b_start), pd.Timedelta(0))

若结果为负数（无重叠），则取0。.total_seconds() 可将其转为浮点型秒数。

但注意：单靠两两配对计算所有A×B组合再求和会严重高估（因未去重，同一时间段被多次计数），且无法处理“多事件叠加”场景。正确做法是：将每日重叠问题转化为时间轴上的区间合并问题。

立即学习“Python免费学习笔记（深入）”；

✅ 推荐实现步骤（基于pandas + interval arithmetic）

import pandas as pd import numpy as np  def compute_daily_overlap_seconds(df_a, df_b, freq='D'):     """     计算df_a与df_b在每个自然日内的并发重叠时长（秒）      Parameters:     df_a, df_b: DataFrame with 'start_ts' and 'end_ts' (datetime64[ns])     freq: pd.Grouper frequency, default 'D' for daily      Returns:     Series indexed by date, values = overlap seconds (0 ≤ x ≤ 86400)     """     # Step 1: 生成所有A-B两两交集区间（仅保留非空交集）     df_a = df_a.copy()     df_b = df_b.copy()     df_a['key'] = 1     df_b['key'] = 1     merged = df_a.merge(df_b, on='key', suffixes=('_a', '_b')).drop('key', axis=1)      # 计算交集端点     merged['overlap_start'] = merged[['start_ts_a', 'start_ts_b']].max(axis=1)     merged['overlap_end'] = merged[['end_ts_a', 'end_ts_b']].min(axis=1)     merged = merged[merged['overlap_start'] < merged['overlap_end']].copy()      # Step 2: 按天切分每个交集区间 → 拆分为「日粒度子区间」     intervals = []     for _, row in merged.iterrows():         start, end = row['overlap_start'], row['overlap_end']         # 生成覆盖该交集的所有自然日日期范围         day_start = start.floor('D')         day_end = end.ceil('D') - pd.Timedelta(seconds=1)  # 向前取整到秒级日末         for day in pd.date_range(day_start, day_end, freq='D'):             day_lower = max(start, day)             day_upper = min(end, day + pd.Timedelta(days=1))             if day_lower < day_upper:                 intervals.append({                     'date': day.date(),                     'start': day_lower,                     'end': day_upper                 })      if not intervals:         return pd.Series([], dtype='float64').rename_axis('date')      # Step 3: 按日期聚合，对每个日期的所有子区间执行「区间合并」     df_intv = pd.DataFrame(intervals)     result = {}     for date, group in df_intv.groupby('date'):         # 排序后合并重叠/邻接区间         sorted_group = group.sort_values('start')         merged_ranges = []         for _, r in sorted_group.iterrows():             if not merged_ranges:                 merged_ranges.append([r['start'], r['end']])             else:                 last = merged_ranges[-1]                 if r['start'] <= last[1]:  # 可合并（重叠或紧邻）                     last[1] = max(last[1], r['end'])                 else:                     merged_ranges.append([r['start'], r['end']])         # 累加合并后各区间长度（秒）         total_sec = sum((end - start).total_seconds() for start, end in merged_ranges)         result[date] = min(total_sec, 86400.0)  # 强制封顶24小时      return pd.Series(result).sort_index()  # ✅ 使用示例 df_a = pd.DataFrame({     'start_ts': pd.to_datetime(['2022-01-01 00:00:00', '2022-01-01 09:00:00']),     'end_ts': pd.to_datetime(['2022-01-01 10:00:00', '2022-01-01 12:00:00']) }) df_b = pd.DataFrame({     'start_ts': pd.to_datetime(['2022-01-01 08:00:00', '2022-01-01 11:00:00']),     'end_ts': pd.to_datetime(['2022-01-01 11:30:00', '2022-01-01 13:00:00']) })  daily_overlap = compute_daily_overlap_seconds(df_a, df_b) print(daily_overlap) # 输出示例：2022-01-01    7200.0  → 即 2 小时（09:00–11:00 与 11:00–11:30 共计 2.5h？实际合并后为 09:00–11:30 = 2.5h = 9000s —— 需根据输入校验）

⚠️ 注意事项

• 性能提示：若事件量大（如每表超千行），merge 会产生 O(n×m) 组合，建议先按日期粗筛（如 start_ts.dt.date join）再精确计算；
• 时区敏感：确保所有 datetime 列已统一时区（推荐转为 UTC 或本地时区并 .dt.tz_localize(None) 显式声明）；
• 边界处理：本方案严格遵循「自然日」（00:00:00–23:59:59.999999），不跨日累加；
• 精度保障：使用 pd.Timedelta 运算，避免浮点截断误差；
• 空结果安全：自动返回空 Series，调用方无需额外判空。

该方法兼顾准确性、可读性与工程鲁棒性，适用于监控告警重叠分析、资源争用评估、用户行为并发建模等典型场景。