1. 海洋平台报警数据采集与字段梳理

为什么先做字段梳理，而不是直接统计

海洋平台的报警记录通常分散在 DCS、ESD、F&G 三套系统中。三套系统各自记录报警，字段命名、时间基准、优先级编码、状态语义都不一致。如果直接把这些原始记录拼在一起做统计，会出现三类典型问题：同一报警在三个系统里被重复计数、时间戳错位导致抖动报警被误判、优先级编码含义不同导致高优先级报警被低估。因此第一步必须把三套系统的导出记录统一成一张标准化报警表，后续的基线指标计算才有可靠输入。

三套系统的字段差异与映射规则

以常见的导出结构为例，三套系统的关键字段对应关系如下表。映射的核心原则是：保留原始字段值，新增标准字段，便于回溯核对。

标准字段 DCS 来源字段 ESD 来源字段 F&G 来源字段 处理规则
alarm_id TagName + EventTime PointID + TimeStamp DetectorTag + OccurTime 拼接生成唯一键，去重用
source_system 固定 "DCS" 固定 "ESD" 固定 "F&G" 常量赋值
tag TagName PointID DetectorTag 统一大写、去空格
alarm_time EventTime（本地时间） TimeStamp（UTC） OccurTime（本地时间） 统一转 UTC，见下节
priority Pri（1–4，1最高） Severity（A/B/C） AlarmLevel（0–3，3最高） 映射到 ISA 18.2 的 P1–P4
state EventType（ALM/RTN） State（ACT/CLR） Status（ON/OFF） 统一为 ACTIVE/CLEARED
description Message Desc DetectorName 保留原文，不做翻译

优先级映射需要特别小心。DCS 的 Pri=1 是最高，F&G 的 AlarmLevel=3 才是最高，方向相反。建议在映射表中显式写出目标值，而不是靠公式推导，避免方向搞反：

# 优先级映射（示例，需按实际系统配置核对）
PRIORITY_MAP = {
    "DCS": {"1": "P1", "2": "P2", "3": "P3", "4": "P4"},
    "ESD": {"A": "P1", "B": "P2", "C": "P3"},
    "F&G": {"3": "P1", "2": "P2", "1": "P3", "0": "P4"},
}

STATE_MAP = {
    "DCS": {"ALM": "ACTIVE", "RTN": "CLEARED"},
    "ESD": {"ACT": "ACTIVE", "CLR": "CLEARED"},
    "F&G": {"ON": "ACTIVE", "OFF": "CLEARED"},
}

时间对齐：三套系统的时间基准不一致

ESD 导出的是 UTC 时间，DCS 和 F&G 导出的是平台本地时间。如果平台位于东八区，直接把 ESD 的 UTC 时间当作本地时间使用，会产生 8 小时的系统性偏移，导致跨系统关联的报警对不上。处理方式有两种：

  • 统一转 UTC：DCS 和 F&G 的本地时间减去时区偏移（东八区减 8 小时），ESD 保持不变。推荐这种做法，因为 UTC 无歧义。
  • 统一转本地时间：ESD 的 UTC 时间加上时区偏移。适合报表需要按本地班次统计的场景。

无论选哪种，都要在数据集里保留一个 tz_offset_min 字段记录转换时使用的偏移量，便于后续审计。另外要检查时间戳的精度：DCS 通常到秒，ESD 可能到毫秒，统一截断到秒即可，避免同一秒内的多条记录因精度不同被误判为不同事件。

数据质量检查：四类必须拦截的问题

标准化之后不能直接进入统计，需要先跑一遍质量检查。以下四类问题在海洋平台报警数据中最常见，建议逐项检查并记录拦截数量：

  1. 重复记录：同一 tag、同一时间、同一状态出现多次。用 alarm_id 去重，保留第一条。
  2. 时间戳异常：时间为空、时间早于平台投运日期、时间晚于导出时刻。这类记录直接剔除并计数。
  3. 孤立清除：某条 CLEARED 记录找不到对应的 ACTIVE 记录。可能是导出窗口截断导致，标记为 orphan_clear 但保留，供后续判断。
  4. 优先级缺失：priority 映射后为空。说明原始编码不在映射表中，需要回查系统配置，不能默认填 P4。
不要用“删除异常行”一句话带过。每类问题都要记录被拦截的记录数和涉及的 tag 清单，否则后续发现指标异常时无法判断是治理效果还是数据清洗造成的假象。

标准化报警数据集的结构

完成映射、对齐和清洗后，输出一张宽表，字段顺序固定，便于后续案例直接读取。建议至少包含以下列：

alarm_id, source_system, tag, alarm_time_utc, priority, state,
description, tz_offset_min, quality_flag

其中 quality_flag 取值为 OK、ORPHAN_CLEAR、DUP_REMOVED 等，保留清洗痕迹而不是直接丢弃。这样后续做基线统计时，可以按 flag 过滤，也可以对比清洗前后的差异。

源系统导出 DCS TagName / EventTime Pri 1–4（1最高） 本地时间 · 秒级 ESD PointID / TimeStamp Severity A/B/C UTC 时间 · 毫秒级 F&G DetectorTag / OccurTime AlarmLevel 0–3（3最高） 本地时间 · 秒级 字段映射 tag 统一大写去空格 priority → P1–P4 state → ACTIVE/CLEARED 保留原始字段值 时间对齐 DCS/F&G 本地 → UTC ESD 保持 UTC 统一截断到秒 记录 tz_offset_min 质量检查 去重 / 时间戳异常 拦截并计数 重复记录 → 去重 时间戳异常 → 剔除 孤立清除 → 标记保留 优先级缺失 → 回查 标准化报警数据集 alarm_id source_system tag alarm_time_utc priority（P1–P4） state（ACTIVE/CLEARED） description tz_offset_min / quality_flag 交付给后续治理 保留 quality_flag 清洗痕迹 可按 flag 过滤或对比清洗前后 字段顺序固定，便于直接读取 原始字段值可回溯核对

一个可复现的字段梳理脚本骨架

下面这段代码演示三套系统导出记录如何合并成标准化表。示例数据是构造的，用于说明处理逻辑，实际使用时替换为真实导出文件即可。

import pandas as pd

PRIORITY_MAP = {
    "DCS": {"1": "P1", "2": "P2", "3": "P3", "4": "P4"},
    "ESD": {"A": "P1", "B": "P2", "C": "P3"},
    "F&G": {"3": "P1", "2": "P2", "1": "P3", "0": "P4"},
}
STATE_MAP = {
    "DCS": {"ALM": "ACTIVE", "RTN": "CLEARED"},
    "ESD": {"ACT": "ACTIVE", "CLR": "CLEARED"},
    "F&G": {"ON": "ACTIVE", "OFF": "CLEARED"},
}

def normalize(df, system, tz_offset_min):
    out = pd.DataFrame()
    out["source_system"] = [system] * len(df)
    out["tag"] = df["tag"].str.upper().str.strip()
    out["alarm_time_utc"] = pd.to_datetime(df["time"]) - pd.Timedelta(minutes=tz_offset_min)
    out["alarm_time_utc"] = out["alarm_time_utc"].dt.floor("s")
    out["priority"] = df["pri"].astype(str).map(PRIORITY_MAP[system])
    out["state"] = df["state"].map(STATE_MAP[system])
    out["description"] = df["desc"]
    out["tz_offset_min"] = tz_offset_min
    out["alarm_id"] = out["tag"] + "|" + out["alarm_time_utc"].astype(str)
    return out

# 示例：DCS/F&G 为东八区本地时间，ESD 已是 UTC
dcs = pd.DataFrame({"tag": ["P-101A"], "time": ["2024-03-01 08:15:02"],
                    "pri": ["1"], "state": ["ALM"], "desc": ["泵出口压力低"]})
esd = pd.DataFrame({"tag": ["ESD-201"], "time": ["2024-03-01 00:15:05"],
                    "pri": ["A"], "state": ["ACT"], "desc": ["关断阀动作"]})
fg  = pd.DataFrame({"tag": ["GD-301"], "time": ["2024-03-01 08:15:07"],
                    "pri": ["3"], "state": ["ON"], "desc": ["可燃气体高报"]})

frames = [
    normalize(dcs, "DCS", 480),
    normalize(esd, "ESD", 0),
    normalize(fg,  "F&G", 480),
]
std = pd.concat(frames, ignore_index=True)

# 质量检查
std["quality_flag"] = "OK"
dup = std.duplicated(subset="alarm_id", keep="first")
std.loc[dup, "quality_flag"] = "DUP_REMOVED"
std = std[~dup]

bad_time = std["alarm_time_utc"].isna()
std.loc[bad_time, "quality_flag"] = "BAD_TIME"
std = std[~bad_time]

missing_pri = std["priority"].isna()
std.loc[missing_pri, "quality_flag"] = "MISSING_PRIORITY"

print(std[["alarm_id", "source_system", "priority", "state", "quality_flag"]])

运行后应看到三条记录，alarm_time_utc 都落在 2024-03-01 00:15 附近，说明三套系统的时间已经对齐到同一基准。如果 ESD 那条记录的时间明显偏离，说明时区偏移量填错了，需要回查平台所在时区。

验证时间对齐是否正确的快速方法：找一条在 DCS 和 ESD 中都会触发的联锁报警（例如同一工艺扰动同时引起 DCS 压力低报和 ESD 关断），检查两条记录在标准化后的时间差是否在秒级以内。如果差出整小时，就是时区问题。

交付物：字段映射与质量检查记录表

建议在完成本节任务时同步填写下表，作为后续案例的输入依据。表中“拦截数”一列必须来自实际运行结果，不能预估。

检查项 检查方法 拦截数 涉及 tag 示例 处理动作
重复记录 alarm_id 重复计数 待填 待填 保留首条，标记 DUP_REMOVED
时间戳异常 为空 / 超出投运—导出区间 待填 待填 剔除，标记 BAD_TIME
孤立清除 CLEARED 无对应 ACTIVE 待填 待填 保留，标记 ORPHAN_CLEAR
优先级缺失 映射后 priority 为空 待填 待填 回查系统配置，不默认填值

完成以上步骤后，得到的标准化报警数据集即可作为后续基线指标计算的输入。判断本节任务是否完成的标准是：三套系统的记录都已进入同一张表、时间戳统一到 UTC、每条记录都有明确的 priority 和 state、quality_flag 能解释每一条被处理过的记录。

← 上一章 📖 返回目录 下一章 →