按24小时制逐小时时段拆分事件 计算跨天场景车道开放时长(SQL/Python)

基础规则说明 #

原事件表结构：

• lane：车道标识，记录ON/OFF事件所属车道
• date：事件发生的日期
• time：事件发生的时间
• event：事件状态，仅取值'ON'（车道开放）、'OFF'（车道关闭）

输出要求：
按24小时制拆分小时统计粒度，time_slot取值范围0-23，其中time_slot=0对应每日00:00:00-00:59:59，其余时段依次类推。最终输出lane、date、time_slot、working_time四个字段，working_time为对应车道在对应日期对应时段内的开放分钟数，支持小数格式。

核心解决思路 #

跨小时、跨天事件的统计核心是先配对连续开放区间，再逐小时切片算重叠时长，这套逻辑SQL和Python完全通用，不会出现两端计算结果不一致的问题：

1. 先把每条记录的日期和时间拼接为完整时间戳，按车道分组、时间升序排列，将每条ON事件和它之后紧邻的第一条OFF事件配对，得到每一段连续开放的起止时间对
2. 对每一段开放时间对，枚举所有被这段时间覆盖的整小时时间片，每个时间片的范围是整点到下一个整点
3. 逐片计算开放区间和时间片的重叠时长，转换为分钟单位
4. 按车道、日期、时段分组求和，得到最终统计结果

SQL实现（支持MySQL 8.0+、PostgreSQL等带窗口函数和递归CTE的数据库） #

WITH event_with_ts AS (
    SELECT
        lane,
        event,
        -- 拼接为完整时间戳，字段类型不同时可调整拼接方式
        CAST(CONCAT(date, ' ', time) AS DATETIME) AS event_ts
    FROM event_record
),
open_interval AS (
    SELECT
        lane,
        event_ts AS open_ts,
        -- 取同车道下一条事件的时间作为当前段结束时间
        LEAD(event_ts) OVER (PARTITION BY lane ORDER BY event_ts) AS close_ts,
        LEAD(event) OVER (PARTITION BY lane ORDER BY event_ts) AS next_event_type
    FROM event_with_ts
),
valid_open AS (
    -- 过滤出合法的ON-OFF配对段
    SELECT lane, open_ts, close_ts
    FROM open_interval
    WHERE event = 'ON' AND next_event_type = 'OFF'
),
-- 递归生成每个开放段覆盖的所有小时整点起点
slot_recurse AS (
    SELECT
        lane,
        open_ts,
        close_ts,
        DATE_FORMAT(open_ts, '%Y-%m-%d %H:00:00') AS slot_start
    FROM valid_open
    UNION ALL
    SELECT
        lane,
        open_ts,
        close_ts,
        DATE_ADD(slot_start, INTERVAL 1 HOUR)
    FROM slot_recurse
    WHERE DATE_ADD(slot_start, INTERVAL 1 HOUR) < close_ts
)
-- 计算每片重叠时长，聚合输出
SELECT
    lane,
    DATE(slot_start) AS date,
    HOUR(slot_start) AS time_slot,
    SUM(
        TIMESTAMPDIFF(
            SECOND,
            GREATEST(open_ts, slot_start),
            LEAST(close_ts, DATE_ADD(slot_start, INTERVAL 1 HOUR))
        ) / 60
    ) AS working_time
FROM slot_recurse
GROUP BY lane, DATE(slot_start), HOUR(slot_start)
ORDER BY lane, date, time_slot;

Python实现（基于pandas） #

import pandas as pd
from datetime import timedelta

# 读取数据，替换为实际数据源读取逻辑
df = pd.read_csv("event_record.csv")
# 拼接完整时间戳
df["event_ts"] = pd.to_datetime(df["date"].astype(str) + " " + df["time"].astype(str))
df = df.sort_values(["lane", "event_ts"]).reset_index(drop=True)

# 配对ON-OFF得到连续开放段
df["close_ts"] = df.groupby("lane")["event_ts"].shift(-1)
df["next_event"] = df.groupby("lane")["event"].shift(-1)
open_segments = df[
    (df["event"] == "ON") & (df["next_event"] == "OFF")
][["lane", "event_ts", "close_ts"]].rename(columns={"event_ts": "open_ts"})

result_list = []
for _, seg in open_segments.iterrows():
    lane = seg["lane"]
    open_ts = seg["open_ts"]
    close_ts = seg["close_ts"]
    # 从开放时间所在的第一个整点开始遍历
    current_slot = open_ts.floor("h")
    while current_slot < close_ts:
        slot_end = current_slot + timedelta(hours=1)
        # 计算重叠时长
        overlap_start = max(open_ts, current_slot)
        overlap_end = min(close_ts, slot_end)
        work_min = (overlap_end - overlap_start).total_seconds() / 60
        result_list.append({
            "lane": lane,
            "date": current_slot.date(),
            "time_slot": current_slot.hour,
            "working_time": work_min
        })
        current_slot = slot_end

# 聚合得到最终结果
final_result = pd.DataFrame(result_list).groupby(
    ["lane", "date", "time_slot"], as_index=False
)["working_time"].sum()

边界处理提示 #

• 如果统计周期内存在最后一条事件为ON、无对应OFF的情况，需要按业务规则补充关闭时间（比如统计周期的截止时间），否则会漏掉这段未关闭的开放时长
• 若date、time字段本身是数据库的日期、时间类型，不需要字符串拼接，直接相加即可得到完整时间戳
• 跨多日的长开放段会自动拆分到对应日期的对应时段，不需要额外写跨天判断逻辑

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