Python计算事件重叠时长：跨DataFrame每日交集时长求解

核心思路 #

先对每一类事件按日期拆分并合并当日内的重叠/连续区间，得到该日期下事件的实际覆盖范围；再针对每日的A、B事件覆盖区间计算交集，最后累加交集的总时长（以秒为单位）。

具体实现步骤 #

1. 定义辅助工具函数 #

import pandas as pd
from datetime import datetime, timedelta

def merge_overlapping_intervals(intervals):
    """合并同一组内重叠或连续的时间区间"""
    if not intervals:
        return []
    # 按区间开始时间排序
    sorted_intervals = sorted(intervals, key=lambda x: x[0])
    merged = [sorted_intervals[0]]
    for current_start, current_end in sorted_intervals[1:]:
        last_start, last_end = merged[-1]
        if current_start <= last_end:
            # 重叠或连续则合并区间
            merged[-1] = (last_start, max(last_end, current_end))
        else:
            merged.append((current_start, current_end))
    return merged

def calculate_intersection_duration(intervals_a, intervals_b):
    """计算两个区间列表的交集总时长（单位：秒）"""
    total_seconds = 0
    i = j = 0
    while i < len(intervals_a) and j < len(intervals_b):
        start_a, end_a = intervals_a[i]
        start_b, end_b = intervals_b[j]
        # 确定交集的起止时间
        intersect_start = max(start_a, start_b)
        intersect_end = min(end_a, end_b)
        if intersect_start < intersect_end:
            duration = (intersect_end - intersect_start).total_seconds()
            total_seconds += duration
        # 移动指针：先结束的区间指针后移
        if end_a < end_b:
            i += 1
        else:
            j += 1
    return total_seconds

def process_events_by_day(df):
    """将事件按日期拆分，合并每日重叠区间，返回{日期: 合并后区间列表}"""
    daily_intervals = {}
    for _, row in df.iterrows():
        start_ts = row['start_ts']
        end_ts = row['end_ts']
        current_date = start_ts.date()
        # 拆分跨天事件为单日事件
        while current_date <= end_ts.date():
            day_start = datetime.combine(current_date, datetime.min.time())
            day_end = day_start + timedelta(days=1)
            # 计算事件在当日的实际区间
            event_day_start = max(start_ts, day_start)
            event_day_end = min(end_ts, day_end)
            # 存入当日区间列表
            if current_date not in daily_intervals:
                daily_intervals[current_date] = []
            daily_intervals[current_date].append((event_day_start, event_day_end))
            current_date += timedelta(days=1)
    # 合并每日的重叠区间
    for date in daily_intervals:
        daily_intervals[date] = merge_overlapping_intervals(daily_intervals[date])
    return daily_intervals

2. 处理事件并计算结果 #

# 处理A、B两类事件，得到每日合并后的区间
daily_a = process_events_by_day(df_event_a)
daily_b = process_events_by_day(df_event_b)

# 获取所有涉及的日期
all_dates = set(daily_a.keys()).union(set(daily_b.keys()))

# 计算每日的交集时长
result_list = []
for date in sorted(all_dates):
    intervals_a = daily_a.get(date, [])
    intervals_b = daily_b.get(date, [])
    overlap_sec = calculate_intersection_duration(intervals_a, intervals_b)
    result_list.append({
        'date': date,
        'overlap_seconds': overlap_sec,
        'overlap_hours': overlap_sec / 3600  # 可选：转换为小时单位
    })

# 转换为DataFrame输出最终结果
result_df = pd.DataFrame(result_list)
print(result_df)

代码说明 #

• 合并区间：先对同日期内的同类型事件排序，再合并重叠/连续区间，避免重复计算覆盖范围
• 跨天事件拆分：将跨日期的事件拆分为多个单日事件，确保每日计算仅针对当天的时间范围
• 交集计算：采用双指针法遍历区间列表，高效计算交集总时长，时间复杂度为O(n+m)（n、m分别为当日A、B事件的区间数量）

示例验证 #

针对你给出的示例：

• 事件A的两个区间合并后为2022-01-01 00:00:00 - 2022-01-01 12:00:00（覆盖12小时）
• 假设事件B当日合并区间为2022-01-01 08:00:00 - 2022-01-01 14:00:00（覆盖6小时）
• 交集为2022-01-01 08:00:00 - 2022-01-01 12:00:00，对应时长为4小时（14400秒），运行代码后会输出该日期的对应结果。

内容的提问来源于stack exchange，提问作者LaGabriella