基于timestamp列检测Mode列'RISK'值是否连续一小时无变化

核心逻辑 #

• 先对全表按timestamp字段升序排序
• 识别连续的RISK状态块：只要某条记录的Mode是RISK且上一条记录的Mode不是RISK，就标记为新的RISK块起点
• 统计每个连续RISK块的起止时间差，只要差值≥1小时就满足要求

方案1：SQL实现（适配主流关系型数据库） #

下面的代码以MySQL为例，其他数据库仅需调整时间差计算函数：

WITH risk_mark AS (
    -- 标记是否为新RISK块的起点
    SELECT
        `timestamp`,
        `Mode`,
        CASE WHEN LAG(`Mode`, 1, '') OVER (ORDER BY `timestamp`) = 'RISK' THEN 0 ELSE 1 END AS new_block_flag
    FROM 你的表名
),
risk_block AS (
    -- 为每个连续RISK块分配唯一ID
    SELECT
        `timestamp`,
        SUM(new_block_flag) OVER (ORDER BY `timestamp`) AS block_id
    FROM risk_mark
    WHERE `Mode` = 'RISK'
)
-- 筛选持续时间≥1小时的RISK块
SELECT
    block_id AS 连续RISK块ID,
    MIN(`timestamp`) AS 块开始时间,
    MAX(`timestamp`) AS 块结束时间,
    TIMESTAMPDIFF(SECOND, MIN(`timestamp`), MAX(`timestamp`))/3600 AS 持续时长_小时
FROM risk_block
GROUP BY block_id
HAVING TIMESTAMPDIFF(SECOND, MIN(`timestamp`), MAX(`timestamp`)) >= 3600;

时间差函数适配说明：

• PostgreSQL替换为EXTRACT(EPOCH FROM (MAX(timestamp) - MIN(timestamp)))
• SQL Server替换为DATEDIFF(SECOND, MIN(timestamp), MAX(timestamp))

方案2：Python Pandas实现 #

如果是用Python处理内存中的数据表，可使用如下代码：

import pandas as pd

# 提前确保timestamp列是datetime类型
df['timestamp'] = pd.to_datetime(df['timestamp'])
# 按时间升序排序
df = df.sort_values('timestamp').reset_index(drop=True)
# 生成连续RISK块的唯一ID
df['risk_block_id'] = ((df['Mode'] == 'RISK') & (df['Mode'].shift() != 'RISK')).cumsum()
# 计算每个RISK块的持续时长
risk_duration = df[df['Mode'] == 'RISK'].groupby('risk_block_id')['timestamp'].agg(['min', 'max'])
risk_duration['duration_hour'] = (risk_duration['max'] - risk_duration['min']).dt.total_seconds() / 3600
# 输出所有持续≥1小时的RISK块
print(risk_duration[risk_duration['duration_hour'] >= 1])

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