如何计算并比对两列设备上报数据的状态变更次数(Python/SQL)

核心逻辑说明 #

两种上报模式的状态变更次数统计规则统一为：

• 首先将上报数据按时间字段升序排序，保证时序正确
• 逐行对比当前行状态值与前一行状态值，不一致则变更次数+1
• 首条上报记录为初始状态，不计入变更次数

Python实现 #

这里提供基于pandas的实现方案，也可适配原生Python逻辑：

import pandas as pd

def count_status_changes(data, time_col: str = "report_time", status_col: str = "status"):
    """
    统计时序上报数据的状态变更总次数
    :param data: 上报数据，支持DataFrame或嵌套列表格式
    """
    # 转换为DataFrame并按时间排序
    if not isinstance(data, pd.DataFrame):
        df = pd.DataFrame(data, columns=[time_col, status_col])
    else:
        df = data.copy()
    df_sorted = df.sort_values(by=time_col).reset_index(drop=True)
    
    # 数据量小于等于1时无变更
    if len(df_sorted) <= 1:
        return 0
    
    # 统计当前值与前一个值不一致的次数
    return (df_sorted[status_col] != df_sorted[status_col].shift(1)).sum()

# 调用示例
if __name__ == "__main__":
    # 全量上报示例数据
    full_data = [
        ["9:00AM", 1], ["9:02AM", 1], ["9:03AM", 0],
        ["9:04AM", 1], ["9:05AM", 1], ["9:06AM", 0]
    ]
    # 增量上报示例数据
    incr_data = [
        ["9:00AM", 1], ["9:03AM", 0], ["9:04AM", 1], ["9:06AM", 0]
    ]
    
    full_change_cnt = count_status_changes(full_data)
    incr_change_cnt = count_status_changes(incr_data)
    
    print(f"全量上报变更次数：{full_change_cnt}")
    print(f"增量上报变更次数：{incr_change_cnt}")
    print(f"二者是否相等：{full_change_cnt == incr_change_cnt}")

SQL实现 #

基于窗口函数LAG实现，适配支持标准SQL的数据库（MySQL 8.0+/PostgreSQL/Oracle等）：
假设全量上报表名为full_report，增量上报表名为incr_report，两张表均包含report_time（上报时间）、status（状态值）两个字段。

单表变更次数统计SQL #

-- 统计全量上报变更次数
SELECT COUNT(*) AS full_change_count
FROM (
    SELECT
        status,
        LAG(status) OVER(ORDER BY report_time ASC) AS prev_status
    FROM full_report
) t
WHERE prev_status IS NOT NULL AND status != prev_status;

-- 统计增量上报变更次数
SELECT COUNT(*) AS incr_change_count
FROM (
    SELECT
        status,
        LAG(status) OVER(ORDER BY report_time ASC) AS prev_status
    FROM incr_report
) t
WHERE prev_status IS NOT NULL AND status != prev_status;

直接校验是否相等的SQL #

SELECT
    (
        SELECT COUNT(*)
        FROM (
            SELECT LAG(status) OVER(ORDER BY report_time ASC) AS prev_status, status
            FROM full_report
        ) t1
        WHERE prev_status IS NOT NULL AND status != prev_status
    ) = (
        SELECT COUNT(*)
        FROM (
            SELECT LAG(status) OVER(ORDER BY report_time ASC) AS prev_status, status
            FROM incr_report
        ) t2
        WHERE prev_status IS NOT NULL AND status != prev_status
    ) AS is_change_count_equal;

返回结果为1/true代表变更次数相等，0/false代表不相等。

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