本文介绍如何使用 python 高效计算两个事件数据集（df_event_a 和 df_event_b）在每一天内的最大并发重叠时长（单位：秒），确保每日结果不超过 86400 秒，并自动处理事件内部重叠及跨事件交叉。
要准确计算事件 A 与事件 B 在每个自然日内同时发生的总时长（即二者时间区间交集在该日的并集长度），不能简单对所有 (A, B) 组合暴力求交后累加——因为多个重叠可能覆盖同一时间段，导致重复计数；同时，单日输出必须满足物理约束：最大不超过 24 小时（86400 秒）。
核心思路是：按天切分 → 对每一天分别构造 A 和 B 在该日的有效时间片段 → 求两组区间集合的交集并合并 → 计算总长度。
✅ 正确实现步骤（基于 pandas + interval arithmetic）
import pandas as pd
import numpy as np
from datetime import datetime, timedelta
def compute_daily_overlap_seconds(df_a, df_b, start_col='start_ts', end_col='end_ts'):
"""
计算 df_a 与 df_b 在每个自然日内的最大重叠时长（秒），去重合并后返回 daily_series
"""
# 确保时间列为 datetime
for df in [df_a, df_b]:
df[start_col] = pd.to_datetime(df[start_col])
df[end_col] = pd.to_datetime(df[end_col])
# 获取全局日期范围（按天）
all_days = pd.date_range(
start=min(df_a[start_col].min(), df_b[start_col].min()).date(),
end=max(df_a[end_col].max(), df_b[end_col].max()).date(),
freq='D'
)
results = {}
for day in all_days:
day_start = pd.Timestamp(day)
day_end = day_start + pd.Timedelta('1D') - pd.Timedelta('1ns') # 23:59:59.999999999
# 提取当天在 df_a 中有交集的所有事件片段（裁剪到当日范围内）
a_clipped = []
for _, row in df_a.iterrows():
s, e = row[start_col], row[end_col]
overlap_start = max(s, day_start)
overlap_end = min(e, day_end)
if overlap_start <= overlap_end:
a_clipped.append((overlap_start, overlap_end))
# 同理处理 df_b
b_clipped = []
for _, row in df_b.iterrows():
s, e = row[start_col], row[end_col]
overlap_start = max(s, day_start)
overlap_end = min(e, day_end)
if overlap_start <= overlap_end:
b_clipped.append((overlap_start, overlap_end))
# 计算两组区间集合的交集（即 A∩B 在当日的覆盖时间）
overlap_intervals = []
for a_s, a_e in a_clipped:
for b_s, b_e in b_clipped:
inter_s = max(a_s, b_s)
inter_e = min(a_e, b_e)
if inter_s <= inter_e:
overlap_intervals.append((inter_s, inter_e))
# 合并重叠/相邻区间（关键！避免重复计数）
if not overlap_intervals:
results[day.date()] = 0
continue
# 排序并合并
overlap_intervals.sort()
merged = [overlap_intervals[0]]
for curr in overlap_intervals[1:]:
last = merged[-1]
if curr[0] <= last[1]: # 可合并（含相接）
merged[-1] = (last[0], max(last[1], curr[1]))
else:
merged.append(curr)
total_seconds = sum((e - s).total_seconds() for s, e in merged)
results[day.date()] = min(total_seconds, 86400) # 强制 ≤24h
return pd.Series(results)
# 示例用法：
# df_event_a = pd.DataFrame({
# 'start_ts': ['2022-01-01 00:00:00', '2022-01-01 09:00:00'],
# 'end_ts': ['2022-01-01 10:00:00', '2022-01-01 12:00:00']
# })
# df_event_b = pd.DataFrame({
# 'start_ts': ['2022-01-01 08:00:00', '2022-01-01 11:00:00'],
# 'end_ts': ['2022-01-01 11:30:00', '2022-01-01 13:00:00']
# })
# daily_overlap = compute_daily_overlap_seconds(df_event_a, df_event_b)
# print(daily_overlap)⚠️ 关键注意事项
- 不可直接套用单对区间公式：max(min(A_end,B_end) - max(A_start,B_start), 0).total_seconds() 仅适用于一对事件；多事件场景必须先做区间合并，否则严重高估。
- 时间精度：建议统一使用 pd.Timestamp，避免 datetime.datetime 与 numpy.datetime64 混用引发隐式转换错误。
- 性能优化提示：若数据量大（>10k 行），可用 line sweep 算法或 intervaltree 库替代双重循环；pandas 的 merge_asof 不适用于此问题（非最近匹配，而是全量区间交集）。
- 边界处理：本实现将 [00:00:00, 23:59:59.999...] 视为一日，符合常规日粒度定义；如需 ISO 周或业务日历，可扩展传入 freq 或自定义 day_bounds_func。
✅ 总结
该方案严格满足题设要求：
? 输出为每个自然日的标量重叠秒数；
? 自动合并所有 A-B 交集片段，消除重复覆盖；
? 单日结果上限为 86400 秒（兜底防护）；
? 代码模块化、可读性强，便于集成进 ETL 流程或监控看板。
