Compressor/backend/indicator_features/finalize_pseudo_online.py

#!/usr/bin/env python3
"""Merge final pseudo-online event replays and unlabelled controls."""
from __future__ import annotations

import csv
from pathlib import Path

ROOT=Path("cache/indicator")
OUT=ROOT/"伪在线检测"
EVENT_DIRS=[ROOT/"伪在线检测_7号2025final",ROOT/"伪在线检测_7号2026final",
            ROOT/"伪在线检测_4号final",ROOT/"伪在线检测_6号final"]
HEALTH=ROOT/"伪在线健康段v2"


def load(path):
    with path.open(encoding="utf-8-sig",newline="") as handle:return list(csv.DictReader(handle))


def write(path,rows,fields=None):
    fields=list(dict.fromkeys(key for row in rows for key in row)) if rows else list(fields or [])
    with path.open("w",encoding="utf-8-sig",newline="") as handle:
        writer=csv.DictWriter(handle,fieldnames=fields);writer.writeheader();writer.writerows(rows)


def main():
    OUT.mkdir(parents=True,exist_ok=True)
    details=[];bands=[];validations=[]
    for folder in EVENT_DIRS:
        details+=load(folder/"伪在线_窗口检测明细.csv")
        bands+=load(folder/"伪在线_角度带告警.csv")
        validations+=load(folder/"伪在线_事件验证.csv")
    write(OUT/"伪在线_窗口检测明细.csv",details)
    write(OUT/"伪在线_角度带告警.csv",bands)
    write(OUT/"伪在线_事件验证.csv",validations)

    health=[]
    for unit in ("7","8","9"):
        path=HEALTH/f"伪在线_未标注对照触发明细_{unit}号.csv"
        if path.exists():health+=load(path)
    write(OUT/"伪在线_未标注对照触发明细.csv",health,["机组","样本标签"])
    control_summary=[]
    for unit in sorted({row["机组"] for row in health}):
        rows=[row for row in health if row["机组"]==unit]
        point_days={(row["测点"],row["日期"]) for row in rows}
        candidate={(row["测点"],row["日期"]) for row in rows if row["等级"] in ("候选","持续预警","未分类异常")}
        persistent={(row["测点"],row["日期"]) for row in rows if row["等级"]=="持续预警"}
        control_summary.append({"机组":unit,"标签":"未标注对照，非确认健康","日期数":len({row["日期"] for row in rows}),
            "测点数":len({row["测点"] for row in rows}),"测点日数":len(point_days),
            "候选测点日数":len(candidate),"候选测点日率":round(len(candidate)/max(1,len(point_days)),4),
            "持续预警测点日数":len(persistent),"持续预警测点日率":round(len(persistent)/max(1,len(point_days)),4),
            "说明":"7日运行片段；8号片段距维修17天，不保证健康。用于粗略候选压力测试，不是确认误报率。"})
    write(OUT/"伪在线_未标注对照统计.csv",control_summary,["机组","标签"])

    common_by_angle={}
    grouped={}
    for row in details:
        if row["等级"] not in ("候选","持续预警"):continue
        angle=int(row["角度箱"]);grouped.setdefault(angle,[]).append(row)
    for angle,rows in grouped.items():
        events=sorted({row["事件"] for row in rows})
        if len(events)<2:continue
        common_by_angle[angle]={"角度范围":f"{angle}-{angle+10}°","事件数":len(events),"事件":";".join(events),
            "测点":";".join(sorted({row["测点"] for row in rows})),
            "候选窗":sum(row["等级"]=="候选" for row in rows),"持续预警窗":sum(row["等级"]=="持续预警" for row in rows),
            "历史签名匹配窗":sum(row.get("历史事件签名匹配")=="1" for row in rows),
            "未分类异常窗":sum(row["等级"]=="未分类异常" for row in rows),
            "最大快RMS":round(max(float(row["综合RMS"]) for row in rows),3),
            "最大慢RMS":round(max(float(row["慢通道综合RMS"]) for row in rows),3)}
    common=sorted(common_by_angle.values(),key=lambda row:(-row["事件数"],-row["持续预警窗"],-row["候选窗"]))
    write(OUT/"伪在线_跨事件共性.csv",common,["角度范围","事件数"])

    lines=["# 伪在线双通道故障检测验证", "", "## 回放设置", "",
        "- 快通道：长历史基线，每测点每天粗采最多3周期；24小时检测窗口使用全部周期并分4个6小时子窗。",
        "- 慢通道：本次运行段最早3个有效日期建立固定基线；后续有效日检测。",
        "- 两路均使用周期平台归一化、测点自身稳健Z和单机公共变化扣除。",
        "- 历史签名只用于标记相似性；未匹配异常仍保留为未分类异常，不静默丢弃。",
        "- 全周期大范围异常单列为通道质控，不计入局部气阀候选。", ""]
    for event in sorted({row["事件"] for row in validations}):
        rows=[row for row in validations if row["事件"]==event]
        rows.sort(key=lambda row:(-int(row["持续预警窗口数"]),-float(row["最大综合RMS"])))
        lines += [f"## {event}","",f"- 验证候选角度带：{len(rows)}。"]
        for i,row in enumerate(rows[:12],1):
            lines.append(f"{i}. {row['测点']} {row['角度带']}：首次候选{row['首次候选日期']}，提前{row['提前日数']}天，持续预警{row['持续预警窗口数']}窗，快通道RMS {float(row['最大综合RMS']):.1f}。")
        lines.append("")
    lines += ["## 目标特征验证", "",
        "- 7号2026一缸盖侧240~270°：5月17日左右相邻箱形成候选，5月19日前后达到持续条件，较5月21日事件提前约2~4天。",
        "- 7号2026二缸盖侧140~150°：慢通道能在事件前触发，但长历史健康对照触发率高，当前只能视为慢漂移候选，尚未证明故障特异性。",
        "- 6号三缸轴侧160~180°：快通道在事件前约1~3天触发。",
        "- 4号出现多测点候选；候选分散，需降低误报后再判断哪些与故障相关。",
        "- 7号2025二缸盖侧全周期异常被标为通道质控，不进入局部故障签名；事件附近缺测，不能证明预测了8月或10月故障。", "",
        "## 未标注运行段压力测试", ""]
    for row in control_summary:
        lines.append(f"- {row['机组']}号：{row['日期数']}日、{row['测点数']}测点；候选测点日率{row['候选测点日率']:.1%}，持续预警测点日率{row['持续预警测点日率']:.1%}。这是未标注片段触发率，不是真实误报率。")
    lines += ["", "## 判定", "",
        "当前实现仍不能上线。健康压力测试中7/8/9号触发率偏高，且慢通道会捕捉季节、负荷和运行段状态转换。检测器可以发现事件前的局部变化，但还没有足够证据判断这些变化是否具有故障特异性。",
        "正式阈值冻结前，需要更长、确认无维修/故障的负样本；建立事件级而不是10度箱级的误报控制；对传感器异常和跨运行段工况切换设置独立状态分类。", "",
        "## 推荐下一版规则", "",
        "1. 保留快慢双通道，慢通道只在每个新运行段开始时重置，不能用全历史模板。",
        "2. 先做整通道质控：近零幅、异常覆盖超过半周期、异常平台量程，剔除后再做局部检测。",
        "3. 对候选角度带要求至少两个特征族、连续时间支持和局部公共变化残差；历史签名匹配只加分，不作为唯一条件。",
        "4. 误报按事件/机组/运行段计数，不能把重叠10度箱和相邻24小时窗口当独立误报。",
        "5. 目前未标注对照每台仅7日，且8号段距维修17日；需更多独立健康运行段后再估计生产误报率。"]
    (OUT/"伪在线检测结论.md").write_text("\n".join(lines)+"\n",encoding="utf-8")
    print("merged",len(details),len(bands),len(validations),len(health),len(common))


if __name__=="__main__":main()