基于事件活动日志统计每日进行中事件数量的技术问询

需求说明 #

我有一份带事件状态时间戳的事件日志数据，仅包含实际事件发生的日期。需要输出每日进行中事件的数量：例如某事件于1月1日开始、1月10日结束，则1日至10日均视为该事件处于进行中状态；若事件中途暂停（如8日暂停、11日重新启动），则暂停期间不算进行中。

示例数据 #

import pandas as pd

df = pd.DataFrame([
                ['1/01/2023','Event_a','started'],
                ['5/01/2023','Event_b','started'],
                ['8/01/2023','Event_b','paused'],
                ['10/01/2023','Event_a','Ended'],
                ['11/01/2023','Event_b','started'],
                ['12/01/2023','Event_b','Ended']],
                columns = ['date', 'EventID', 'Event_status'])
df['date'] = pd.to_datetime(df['date'])

当前遇到的瓶颈 #

我尝试按日期和EventID分组填充缺失日期并统计事件数，但后续步骤无法正确计算每日进行中的事件数量，当前代码如下：

# 注：原代码中issue_id应为EventID，属于笔误
df_2 =  df.groupby(['date','EventID'])[['EventID']].count()
df_3 = df_2.unstack(['EventID']).fillna(0).stack(['EventID'])

解决方案 #

步骤1：为每个事件生成活跃状态的日期区间 #

先为每个事件的状态变化配对“开始-结束”的活跃区间：

• 对每个EventID按日期排序，将下一条记录的日期作为当前状态的结束日期参考
• 过滤出started状态的记录，确定活跃区间的起止（暂停或结束状态对应区间终止）

# 按EventID和日期排序
df_sorted = df.sort_values(['EventID', 'date']).reset_index(drop=True)

# 为每条记录添加下一条状态的日期作为结束日期参考
df_sorted['next_date'] = df_sorted.groupby('EventID')['date'].shift(-1)

# 处理最后一条记录的结束日期（如果是Ended则用当天，否则用日志的最晚日期）
max_date = df_sorted['date'].max()
df_sorted['next_date'] = df_sorted['next_date'].fillna(max_date)

# 筛选出started的记录，生成活跃区间：start_date到end_date
active_intervals = df_sorted[df_sorted['Event_status'] == 'started'].copy()
active_intervals['end_date'] = active_intervals.apply(
    lambda x: x['next_date'] - pd.Timedelta(days=1) 
    if df_sorted.loc[df_sorted.index[df_sorted.index.get_loc(x.name)+1], 'Event_status'] != 'Ended'
    else x['next_date'],
    axis=1
)

步骤2：生成完整的日期序列 #

创建从日志最早日期到最晚日期的所有连续日期：

date_range = pd.date_range(start=df_sorted['date'].min(), end=df_sorted['date'].max())

步骤3：展开活跃区间为每日记录 #

将每个活跃区间的日期范围展开为单条的每日事件记录：

# 定义函数展开单个区间的日期
def expand_interval(row):
    return pd.DataFrame({
        'date': pd.date_range(start=row['date'], end=row['end_date']),
        'EventID': row['EventID']
    })

# 对所有活跃区间应用展开函数并合并
daily_active = pd.concat([expand_interval(row) for _, row in active_intervals.iterrows()], ignore_index=True)

步骤4：统计每日进行中事件数量 #

按日期分组统计活跃事件的唯一数量，并补全无活跃事件的日期：

daily_count = daily_active.groupby('date')['EventID'].nunique().reset_index(name='active_events')

# 补全所有日期（无活跃事件的日期数量设为0）
daily_count = daily_count.set_index('date').reindex(date_range).fillna(0).astype(int).reset_index()
daily_count.rename(columns={'index': 'date'}, inplace=True)

最终结果 #

运行上述代码后，daily_count即为期望输出，示例结果如下：

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