本文介绍如何使用 python 计算两个事件数据集（df_event_a 和 df_event_b）在每一天内发生的**跨集合时间重叠总时长**，结果以秒为单位，且单日最大值不超过 86400 秒（24 小时）。
要准确计算 A 类事件与 B 类事件在每日粒度上的并发持续时间，核心在于：对每一对可能跨日相交的 (A, B) 事件，先求其时间交集区间，再将该交集按自然日切分，最后汇总每天的有效重叠秒数（注意：同一天内多个重叠片段需合并去重，避免重复计时；且单日总和不可超过 86400 秒）。
✅ 关键逻辑说明
两个时间区间 [a_start, a_end] 和 [b_start, b_end] 的重叠时长公式为：
overlap = max(min(a_end, b_end) - max(a_start, b_start), pd.Timedelta(0))
若结果为负或零，则无重叠；否则 .total_seconds() 即得重叠秒数。
macOS UI 自动化技能包 —— 用 Python 控制你的 Mac
下载
让 macOS 自动化像说话一样简单 —— 用代码控制你的 Mac，无需学习 AppleScript 系统要求 macOS 14+ (Sonoma) Python 3.9+ 需要开启：辅助功能 + 屏幕录制权限
但直接两两计算所有组合再按天 sum() 会高估——因为：
- 同一天内多个 A/B 事件可能产生多段重叠，存在时间覆盖；
- 重叠区域本身可能横跨多日（如 A: 2022-01-01 22:00 → 2022-01-02 03:00，B: 2022-01-01 23:00 → 2022-01-02 04:00），需拆分为 2022-01-01 和 2022-01-02 两天分别计算。
✅ 推荐实现步骤（使用 pandas + numpy）
import pandas as pd
import numpy as np
def compute_daily_overlap_seconds(df_a, df_b, freq='D'):
"""
计算 df_a 与 df_b 在每一天的并发总时长（秒），自动去重、截断至 86400s/天
要求：df_a / df_b 包含列 'start_ts', 'end_ts'，类型为 datetime64[ns]
"""
# 确保时间列是 datetime
for df in [df_a, df_b]:
df['start_ts'] = pd.to_datetime(df['start_ts'])
df['end_ts'] = pd.to_datetime(df['end_ts'])
# 生成所有 (a, b) 组合（笛卡尔积）
df_a = df_a.assign(_key=1)
df_b = df_b.assign(_key=1)
joined = df_a.merge(df_b, on='_key', suffixes=('_a', '_b')).drop(columns='_key')
# 计算每对事件的全局重叠区间
joined['overlap_start'] = joined[['start_ts_a', 'start_ts_b']].max(axis=1)
joined['overlap_end'] = joined[['end_ts_a', 'end_ts_b']].min(axis=1)
joined['raw_overlap'] = (joined['overlap_end'] - joined['overlap_start']).clip(lower=pd.Timedelta(0))
# 过滤掉无重叠的行
valid = joined[joined['raw_overlap'] > pd.Timedelta(0)].copy()
if valid.empty:
# 返回空日期索引的 Series
return pd.Series([], dtype='float64')
# 拆分跨日重叠：对每个重叠区间，生成其覆盖的每一天的 [day_start, day_end]
records = []
for _, row in valid.iterrows():
start, end = row['overlap_start'], row['overlap_end']
# 生成该重叠区间覆盖的所有自然日
days = pd.date_range(start.date(), end.date(), freq='D')
for day in days:
day_start = max(start, pd.Timestamp(day.date()))
day_end = min(end, pd.Timestamp(day.date()) + pd.Timedelta('1D'))
duration_sec = (day_end - day_start).total_seconds()
if duration_sec > 0:
records.append({'date': day.date(), 'seconds': duration_sec})
# 转为 DataFrame 并按日聚合（合并同日所有片段）
if not records:
return pd.Series([], dtype='float64')
daily_df = pd.DataFrame(records)
daily_total = daily_df.groupby('date')['seconds'].sum().round().astype(int)
# 强制单日上限为 86400 秒（24 小时）
daily_total = daily_total.clip(upper=86400)
# 补全缺失日期（可选）：若需连续日期索引，可用 resample
# daily_total = daily_total.resample('D').sum().fillna(0).astype(int)
return daily_total
# 示例用法：
# result = compute_daily_overlap_seconds(df_event_a, df_event_b)
# print(result)⚠️ 注意事项
- 性能提示：若事件量大（如各超万条），笛卡尔积可能内存爆炸。此时建议改用区间树（如 intervaltree 库）或分天预过滤（先按日期分桶，再在同一天内做交叉）。
- 时区敏感：确保所有 datetime 列已统一时区（推荐转为 UTC 或本地 tz-aware），否则跨日计算易出错。
- 边界处理：本实现采用左闭右开语义（[start, end)），符合 pandas 时间差惯例；若业务需包含终点，请在 end_ts 上加 1ns 再计算。
- 输出解释：返回 pd.Series，索引为 datetime.date，值为当日 A 与 B 同时活跃的去重总秒数，严格 ≤ 86400。
通过该方法，你可精准量化系统中两类事件（如服务请求 vs 资源维护、用户在线 vs 推送发送）的真实并发压力，为容量规划与 SLA 分析提供可靠依据。
