Python实现持续时长条件校验：滚动窗口无重采样标记阀门告警

原代码存在两个核心逻辑硬伤，无法满足需求：

• 滚动求和的阈值无法固定：工业采集的时间间隔不固定，1小时窗口内的原始记录数动态变化，没有一个固定数值能满足“窗口内Valve1全为1、Valve2全为0”的判断要求
• 滚动窗口会漏判边界外的状态影响：pandas的时间滚动窗口只统计时间戳落在窗口范围内的原始记录，完全忽略了窗口起始边界之前的采集点状态——按照规则，这些边界外的采集点状态会通过“沿用最近状态”的逻辑覆盖到窗口内的缺失时段，滚动函数的统计结果完全反映不了这部分状态，极易出现误告警

核心思路：不做重采样、不修改原始Valve值，通过状态分段计算连续符合要求的累计时长，等效实现“缺失时段沿用最近状态”的规则，完全符合所有约束：

• 先对单条原始记录做初步匹配标记，判断当前采集点是否满足Valve1=1且Valve2=0的要求
• 识别连续的相同状态段：只要状态从“符合要求”切换为“不符合”、或反向切换，就开启一个新的状态段
• 计算每个状态段的起始时间，对符合要求的状态段，累计计算从段起始时间到当前点的时长
• 当累计达到1小时时长时，在当前原始记录上标记告警

该逻辑全程不需要新增任何缺失时间点，也不会对原始采集的阀门状态做任何插值、修改，所有计算都是基于原始记录的时间戳和状态完成的。

import pandas as pd

# 1. 按Datetime排序并设为索引（必须步骤，否则连续状态计算会错位）
data = data.set_index('Datetime').sort_index()

# 2. 临时计算单条记录的匹配标记
data['_match'] = (data['Valve1'] == 1) & (data['Valve2'] == 0)

# 3. 识别状态变化点，计算每个连续状态段的起始时间
state_change = data.index.where(data['_match'] != data['_match'].shift())
data['_group_start'] = state_change.ffill()

# 4. 向量化计算累计符合状态的时长：不符合的点时长直接为0
data['_duration'] = (data.index - data['_group_start']) * data['_match']

# 5. 标记告警：累计时长达到1小时就打标
data.loc[data['_duration'] >= pd.Timedelta('1h'), 'alarm'] = 'Yes'

# 6. 清理临时辅助列，保证输出和原始列结构一致
data = data.drop(columns=['_match', '_group_start', '_duration'])

• 全程保留所有原始采集时间点，不新增、不删除记录，没有执行任何重采样操作
• 原始的Valve1、Valve2字段完全保留采集的原始值，没有任何插值、填充生成的数值
• 自动实现缺失时段沿用最近状态的逻辑：连续状态段的时长计算已经把两个采集点之间的无记录时段的状态等效为和最近采集点一致，不需要单独补点
• 自动实现重置计时逻辑：只要出现不符合要求的状态，就会开启新的状态段，之前的累计时长自动清零，重新从符合要求的状态出现开始计数
• 告警触发时机准确：仅当从当前点往前1小时的所有时段（含无记录的缺失时段）的状态全部符合要求时，才会标记告警

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