3. 水下机器人故障日志的缺失值与异常值清洗

清洗对象与判定口径

航次故障日志来自控制台、传感器与任务日志三路汇合，清洗前必须先固定“什么算脏”。本例采用三类判定：缺失（关键字段为空或占位符）、重复（同一事件被多路重复上报）、越界（数值超出物理量程）。判定口径一旦确定，后续插补与剔除才有可复现的依据。

以水下机器人常见字段为例，物理量程按设备手册设定：航深 0–300 m、航速 0–2.0 m/s、电池电压 44–58 V、推进器转速 0–3000 rpm、舱温 -5–45 ℃。任何超出该范围的读数视为传感器异常或单位错位，不能直接参与分类。

缺失字段的识别与插补

缺失分两种：结构性缺失（该航次本就没有该传感器，如某次任务未装侧扫声呐）与随机缺失（采集丢包、写入中断）。结构性缺失不能插补，应保留为空并标记“不适用”；随机缺失才适合插补。判定方法是看该字段在同一航次内的缺失是否连续成段且与设备配置表对应。

对随机缺失的数值字段，采用“短缺口线性插值、长缺口同航次中位数”的策略：连续缺失 ≤ 3 个采样点用前后有效值线性插值；超过 3 个点则用该航次同类字段的中位数填充，并写入 imputed_flag 标记，便于后续分类时区分真实值与插补值。

import pandas as pd
import numpy as np

# 物理量程（下限, 上限）
RANGE = {
    "depth_m":      (0.0, 300.0),
    "speed_ms":     (0.0, 2.0),
    "batt_v":       (44.0, 58.0),
    "thruster_rpm": (0.0, 3000.0),
    "cabin_temp_c": (-5.0, 45.0),
}
KEY_FIELDS = ["event_id", "timestamp", "fault_label"] + list(RANGE)

def load_log(path):
    df = pd.read_csv(path, parse_dates=["timestamp"])
    # 占位符统一转为 NaN
    df = df.replace({"": np.nan, "NA": np.nan, "null": np.nan, "--": np.nan})
    return df

def mark_missing(df):
    df["missing_flag"] = df[KEY_FIELDS].isna().any(axis=1)
    return df

def impute_numeric(df, max_gap=3):
    df = df.sort_values("timestamp").copy()
    df["imputed_flag"] = False
    for col in RANGE:
        s = df[col]
        # 连续缺失段长度
        grp = (s.notna() != s.notna().shift()).cumsum()
        gap_len = s.isna().groupby(grp).transform("sum")
        short = s.isna() & (gap_len <= max_gap)
        long_ = s.isna() & (gap_len > max_gap)
        # 短缺口：线性插值
        df.loc[short, col] = s.interpolate(method="linear")[short]
        # 长缺口：同航次中位数
        df.loc[long_, col] = s.median()
        df.loc[short | long_, "imputed_flag"] = True
    return df

重复记录去重与越界值剔除

重复记录来自多路日志对同一事件的重复上报。去重键取 (event_id, timestamp, fault_label)，保留来源优先级最高的一条（控制台 > 传感器 > 任务日志），并记录被合并的条数。越界值不插补，直接置空并标记 out_of_range_flag，因为越界往往意味着单位错位或传感器故障，插补会掩盖真实问题。

SOURCE_PRIORITY = {"console": 0, "sensor": 1, "mission": 2}

def dedup(df):
    df = df.copy()
    df["src_rank"] = df["source"].map(SOURCE_PRIORITY).fillna(9)
    before = len(df)
    df = (df.sort_values("src_rank")
            .drop_duplicates(subset=["event_id", "timestamp", "fault_label"], keep="first"))
    print(f"去重：{before} -> {len(df)}，合并 {before - len(df)} 条")
    return df.drop(columns="src_rank")

def flag_out_of_range(df):
    df = df.copy()
    df["out_of_range_flag"] = False
    for col, (lo, hi) in RANGE.items():
        bad = df[col].notna() & ((df[col] < lo) | (df[col] > hi))
        df.loc[bad, col] = np.nan
        df.loc[bad, "out_of_range_flag"] = True
    return df

def clean_pipeline(path):
    df = load_log(path)
    df = mark_missing(df)
    df = dedup(df)
    df = flag_out_of_range(df)
    df = impute_numeric(df, max_gap=3)
    return df

清洗前后标签分布一致性验证

清洗最容易犯的错误是“洗掉了某一类故障”。验证方法是比较清洗前后各故障类别的占比：若某类别占比偏差超过 1.0 个百分点，说明该类别的记录被过度剔除或插补，需要回查判定口径。同时检查总样本保留率，低于 90% 说明剔除过猛。

def label_dist(df, label_col="fault_label"):
    return df[label_col].value_counts(normalize=True).sort_index()

def verify_consistency(raw_df, clean_df, tol=0.01, min_keep=0.90):
    before = label_dist(raw_df)
    after = label_dist(clean_df)
    idx = before.index.union(after.index)
    before, after = before.reindex(idx, fill_value=0), after.reindex(idx, fill_value=0)
    diff = (after - before).abs()
    keep_rate = len(clean_df) / len(raw_df)
    report = pd.DataFrame({
        "清洗前占比": before.round(4),
        "清洗后占比": after.round(4),
        "偏差": diff.round(4),
        "通过": diff <= tol,
    })
    ok = bool((diff <= tol).all() and keep_rate >= min_keep)
    print(report)
    print(f"保留率 {keep_rate:.1%}，整体判定：{'通过' if ok else '不通过'}")
    return report, ok

raw = load_log("voyage_fault_log.csv")
clean = clean_pipeline("voyage_fault_log.csv")
verify_consistency(raw, clean)

通过条件：各类别占比偏差 ≤ 1.0%，总保留率 ≥ 90%，且 imputed_flag 与 out_of_range_flag 均已写入输出表。满足这三条，即可将干净数据集交给后续的文本标准化与标签映射环节。