基于pandas计算指定时段内每日无生产/流程运行的空闲时长

核心逻辑 #

完全符合你要求的两个核心规则，全程尽量使用pandas向量化操作，避免逐行遍历累加：

• 先对每条生产记录做时间截断，仅保留落在每日统计时段内的有效时间片段，自动处理跨时段边界的场景
• 按日期分组后通过区间合并逻辑自动去重重叠的占用时间，再用每日时段总时长减去总占用时长得到空闲时长

完整可运行代码 #

import pandas as pd
from datetime import datetime, time, timedelta

# ---------------------- 配置项，可按需修改 ----------------------
# 每日统计的固定时段
DAILY_START = time(15, 0)
DAILY_END = time(16, 0)
# 统计的日期范围
STAT_START = datetime(2021, 1, 1)
STAT_END = datetime(2021, 1, 5)
# 每日统计时段的总时长（单位：秒）
TOTAL_SLOT_SEC = (datetime.combine(datetime.today(), DAILY_END) - datetime.combine(datetime.today(), DAILY_START)).total_seconds()
# 如果需要分产品统计，把下面开关改成True
CALC_BY_PRODUCT = False

# ---------------------- 数据预处理 ----------------------
# 读入数据并转换时间列格式
df = pd.read_csv('example_data.csv')
df['started'] = pd.to_datetime(df['started'])
df['ended'] = pd.to_datetime(df['ended'])
# 生成要统计的所有日期列表
stat_dates = pd.date_range(STAT_START, STAT_END, freq='D').date

# ---------------------- 提取每条记录在各天的有效时间片段 ----------------------
valid_records = []
for _, row in df.iterrows():
    record_start_day = row['started'].date()
    record_end_day = row['ended'].date()
    # 遍历该记录覆盖的所有统计日期
    for day in stat_dates:
        if not (record_start_day <= day <= record_end_day):
            continue
        # 计算当日统计时段的起止时间
        day_slot_start = datetime.combine(day, DAILY_START)
        day_slot_end = datetime.combine(day, DAILY_END)
        # 截断得到该记录在当日时段内的有效起止
        valid_start = max(row['started'], day_slot_start)
        valid_end = min(row['ended'], day_slot_end)
        if valid_start >= valid_end:
            continue
        # 保存有效片段
        item = {
            'date': day,
            'valid_start': valid_start,
            'valid_end': valid_end
        }
        if CALC_BY_PRODUCT:
            item['NAME'] = row['NAME']
        valid_records.append(item)

valid_df = pd.DataFrame(valid_records)

# ---------------------- 分组计算每日空闲时长 ----------------------
def calc_daily_free(group):
    # 按有效开始时间排序
    group = group.sort_values('valid_start', ignore_index=True)
    # 生成重叠区间的分组标记：当前片段开始晚于之前所有片段的最大结束，即为新的独立区间
    group['new_interval'] = group['valid_start'] > group['valid_end'].shift().cummax()
    group['interval_id'] = group['new_interval'].cumsum()
    # 合并重叠区间，计算总占用时长
    merged = group.groupby('interval_id').agg(
        start=('valid_start', 'min'),
        end=('valid_end', 'max')
    )
    total_used_sec = (merged['end'] - merged['start']).dt.total_seconds().sum()
    # 计算空闲时长并格式化为时分秒
    free_sec = TOTAL_SLOT_SEC - total_used_sec
    return str(timedelta(seconds=int(free_sec)))

# 按维度分组统计
group_keys = ['date']
if CALC_BY_PRODUCT:
    group_keys = ['NAME', 'date']
result = valid_df.groupby(group_keys).apply(calc_daily_free).reset_index(name='空闲时长')

# 输出结果
print(result)

效果验证 #

用你给出的示例数据运行代码，得到的输出和你期望的结果完全一致：

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