2. 机电监测数据质量检查与缺失段标记

任务对象与数据约定

本节处理一段来自某型船舶主机与发电机组的多通道时序数据，共 6 个通道：主机缸套冷却水出口温度 T_cw、主机滑油压力 P_lo、主机转速 N_main、发电机排烟温度 T_exh、发电机电流 I_gen、舵机液压油压力 P_hyd。原始数据以各传感器本地时间戳记录，采样标称周期 1 s，但存在两类典型问题：

• 时钟漂移：各采集模块独立计时，长时间运行后时间戳相对参考时钟缓慢偏移，导致同一物理时刻的通道值错位。
• 传感器掉线：通信中断期间无数据写入，表现为时间戳跳变或数值冻结（重复采样）。

清洗目标：统一到 1 s 等间隔时间轴，剔除重复采样与冻结值，标记缺失段（含掉线区间与漂移校正后产生的空洞），输出清洗数据集与质量报告。时间参照约定：以主机采集模块时钟为参考时钟 t_ref，其余通道时间戳按线性漂移模型校正到 t_ref。

处理链路与判读逻辑

下图给出本节的核心处理链：原始多通道数据先做时间戳漂移估计与校正，再重采样到统一网格，随后识别重复采样/冻结值并置为缺失，最后做缺失段合并与标记，输出数据集与质量报告。图中同时标出各环节的关键判据。

关键实现：漂移估计与缺失标记

下面给出可直接运行的 Python 实现（依赖 pandas、numpy）。核心思路：用各通道与参考时钟的同步对齐点拟合线性漂移，把时间戳映射到 t_ref；在 1 s 网格上重采样；用“连续相同值计数”识别冻结；最后把连续 NaN 合并成缺失区间并生成报告。

import numpy as np
import pandas as pd

# ---------- 1. 构造教学示例数据（含漂移与掉线） ----------
rng = np.random.default_rng(7)
t_ref = pd.date_range("2024-06-01 08:00:00", periods=600, freq="1s")

def make_channel(name, drift_ppm, drop_slice, freeze_slice, base, noise):
    # 本地时钟相对参考时钟的线性漂移：t_local = t_ref*(1+ppm*1e-6) + offset
    t_local = t_ref + pd.to_timedelta(np.arange(len(t_ref)) * drift_ppm * 1e-6, unit="s")
    vals = base + rng.normal(0, noise, len(t_ref)).cumsum() * 0.01
    df = pd.DataFrame({"t_local": t_local, "value": vals, "ch": name})
    df = df.drop(df.index[drop_slice])          # 模拟掉线：整段无记录
    df.loc[df.index[freeze_slice], "value"] = df.loc[df.index[freeze_slice][0], "value"]  # 冻结
    return df

raw = pd.concat([
    make_channel("T_cw",   +30, slice(120, 150), slice(300, 312), 78.0, 0.05),
    make_channel("P_lo",   -20, slice(400, 415), slice(200, 208), 0.42, 0.002),
    make_channel("N_main", +10, slice(500, 505), slice(100, 106), 620.0, 0.8),
], ignore_index=True)

# ---------- 2. 漂移校正：以 t_ref 为参考，线性拟合 ----------
def correct_drift(df, ref_index):
    # 用序号近似参考时刻（示例中参考通道完整），拟合 t_local = a*t_ref + b
    x = ref_index.astype("int64").to_numpy() / 1e9
    y = df["t_local"].astype("int64").to_numpy() / 1e9
    a, b = np.polyfit(x, y, 1)
    df = df.copy()
    df["t_corr"] = pd.to_datetime((y - b) / a, unit="s")
    return df, a, b

# ---------- 3. 重采样到 1 s 网格 + 冻结识别 ----------
GRID = pd.date_range(t_ref[0], t_ref[-1], freq="1s")
FREEZE_N = 5

def resample_and_flag(df):
    s = df.set_index("t_corr")["value"].sort_index()
    s = s[~s.index.duplicated(keep="first")]          # 同时间戳重复保留首条
    g = s.reindex(GRID, method="nearest", tolerance=pd.Timedelta("2s"))
    # 冻结：连续相同值计数
    same = g.diff().fillna(1) == 0
    run = same.groupby((~same).cumsum()).cumsum()
    g[run >= FREEZE_N - 1] = np.nan
    return g

cleaned, report = {}, []
for ch, df in raw.groupby("ch"):
    dfc, a, b = correct_drift(df, pd.Series(range(len(t_ref))))
    g = resample_and_flag(dfc)
    cleaned[ch] = g
    # 缺失段合并
    miss = g.isna()
    seg = miss.ne(miss.shift()).cumsum()[miss]
    for _, idx in g[miss].groupby(seg).groups.items():
        report.append({"ch": ch, "start": idx[0], "end": idx[-1],
                       "dur_s": len(idx), "drift_a": round(a, 6)})

clean_df = pd.DataFrame(cleaned)
rep_df = pd.DataFrame(report)
print(clean_df.head())
print(rep_df[rep_df.dur_s >= 3].to_string(index=False))

结果判读与质量门槛

运行后 clean_df 为统一 1 s 网格上的清洗数据集，缺失位置为 NaN；rep_df 中每行是一段缺失区间。判读要点：

若某通道有效率低于门槛，应先回到采集侧检查通信与供电，而不是用插值强行补齐——插值会掩盖掉线，使后续基线模型把异常段当作正常样本学习。