Python实现时间区间记录计数及15分钟时段状态标记方案咨询

两种方案在Python环境下均可实现，其中指定时段计数的方案（方案二）实现难度更低，逻辑直白、代码量少、运行效率更高；生成96列状态宽表的方案（方案一）实现稍复杂，更适合有明细留存需求的场景。

核心判断逻辑非常简单：某条人员记录落在目标统计时段内的充要条件是：
人员进入流程时间 < 统计时段结束时间，且人员离开流程时间 > 统计时段开始时间
这个判断逻辑完全匹配给出的样例规则：比如18:00-18:15的时段，IdNum=389的记录Exitdate为18:12，满足10:03 < 18:15且18:12 > 18:00，因此会被统计在内，最终计数为2，和预期一致。

具体实现步骤：

• 预处理原始表：将BeginDate、Exitdate两列从字符串转为pandas的datetime类型，避免时间比较错误。
• 定义接收三个参数的函数：原始数据表、统计时段开始时间、统计时段结束时间，入参时间统一转为datetime格式。
• 用布尔索引筛选符合上述时间重叠条件的记录，返回符合条件的IdNum数量即可。

参考实现代码：

import pandas as pd
from datetime import datetime

def count_active_personnel(df, slot_start: datetime, slot_end: datetime):
    # 首次运行时自动转换时间列格式
    if not pd.api.types.is_datetime64_any_dtype(df["BeginDate"]):
        df["BeginDate"] = pd.to_datetime(df["BeginDate"])
    if not pd.api.types.is_datetime64_any_dtype(df["Exitdate"]):
        df["Exitdate"] = pd.to_datetime(df["Exitdate"])
    
    # 筛选时段内处于流程中的记录
    active_filter = (df["BeginDate"] < slot_end) & (df["Exitdate"] > slot_start)
    # 如果IdNum无重复可以直接用len(df[active_filter])，有重复则用nunique()去重
    return df[active_filter]["IdNum"].nunique()

# 测试样例：统计2022-06-13 18:00-18:15的在流程人数
if __name__ == "__main__":
    df = pd.DataFrame([
        {"IdNum":123, "BeginDate":"2022-06-13 09:03", "Exitdate":"2022-06-13 22:12"},
        {"IdNum":633, "BeginDate":"2022-06-13 08:15", "Exitdate":"2022-06-13 13:09"},
        {"IdNum":389, "BeginDate":"2022-06-13 10:03", "Exitdate":"2022-06-13 18:12"},
        {"IdNum":665, "BeginDate":"2022-06-13 08:30", "Exitdate":"2022-06-13 10:12"},
    ])
    res = count_active_personnel(
        df,
        slot_start=datetime(2022,6,13,18,0),
        slot_end=datetime(2022,6,13,18,15)
    )
    print(res) # 输出2，符合预期

如果需要统计全天96个时段的人数，只需要从0点开始每15分钟生成一个时段，循环调用这个函数即可，不需要额外生成冗余列，数据量大的时候内存占用优势非常明显。

如果需要留存每个IdNum在各个时段的状态明细做后续分析，可以选择这个方案，实现步骤如下：

• 同样先将BeginDate、Exitdate列转为datetime类型。
• 按日期生成对应日期下的96个15分钟间隔时段，记录每个时段的开始时间、结束时间、列名（格式如18:00 - 18:15）。注意如果数据表包含多天数据，需要按日期分组分别生成对应时段，避免跨天记录判断错误。
• 遍历每个时段，逐列判断每行记录是否满足时间重叠条件，赋值为True/False。
• 将所有生成的时段列拼接到原始表后返回。

参考核心实现代码：

def generate_status_widetable(df):
    df = df.copy()
    # 预处理时间列
    df["BeginDate"] = pd.to_datetime(df["BeginDate"])
    df["Exitdate"] = pd.to_datetime(df["Exitdate"])
    
    # 此处以单天数据为例，多天数据请先按日期分组处理
    base_date = df["BeginDate"].dt.date.iloc[0]
    # 生成96个15分钟时段
    time_slots = []
    for minute_offset in range(0, 24*60, 15):
        slot_start = datetime.combine(base_date, datetime.min.time()) + pd.Timedelta(minutes=minute_offset)
        slot_end = slot_start + pd.Timedelta(minutes=15)
        col_name = f"{slot_start.strftime('%H:%M')} - {slot_end.strftime('%H:%M')}"
        time_slots.append( (col_name, slot_start, slot_end) )
    
    # 逐列生成状态值
    for col_name, s, e in time_slots:
        df[col_name] = (df["BeginDate"] < e) & (df["Exitdate"] > s)
    return df

这个方案的缺点是当数据量较大时，新增96列会带来明显的内存占用，且如果存在跨天数据，处理逻辑会更繁琐，更容易出现边界判断错误。

• 没有特殊明细留存需求的话优先选方案二，实现快、逻辑稳、性能好，不管是查单个时段还是批量统计全时段都能满足。
• 只有当后续需要基于单IdNum的时段状态做交叉分析时，再考虑用方案一拼宽表。

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