Compressor/backend/indicator_features/summarize_pseudo_online.py

#!/usr/bin/env python3
"""Summarize merged pseudo-online replay results without database access."""
from __future__ import annotations

import csv
from collections import Counter, defaultdict
from pathlib import Path


ROOT = Path("cache/indicator/伪在线检测")


def read(name):
    with (ROOT / name).open(encoding="utf-8-sig", newline="") as handle:
        return list(csv.DictReader(handle))


def write(name, rows):
    if not rows:
        return
    with (ROOT / name).open("w", encoding="utf-8-sig", newline="") as handle:
        writer = csv.DictWriter(handle, fieldnames=list(rows[0]))
        writer.writeheader(); writer.writerows(rows)


def main():
    detail = read("伪在线_窗口检测明细.csv")
    bands = read("伪在线_角度带告警.csv")
    validation = read("伪在线_事件验证.csv")

    common = []
    by_angle = defaultdict(list)
    for row in detail:
        if row["等级"] in ("候选", "持续预警"):
            by_angle[int(row["角度箱"])].append(row)
    for angle, rows in sorted(by_angle.items()):
        events = sorted({row["事件"] for row in rows})
        if len(events) < 2:
            continue
        common.append({
            "角度范围": f"{angle}-{angle + 10}°", "事件数": len(events),
            "事件": ";".join(events), "测点数": len({row["测点"] for row in rows}),
            "测点": ";".join(sorted({row["测点"] for row in rows})),
            "候选日数": len({(row["事件"], row["日期"]) for row in rows}),
            "持续预警记录数": sum(row["等级"] == "持续预警" for row in rows),
            "最大幅度平均Z": round(max(float(row["幅度平均Z"]) for row in rows), 3),
            "最大综合RMS": round(max(float(row["综合RMS"]) for row in rows), 3),
            "共同特征族": ";".join(sorted({part.split("↑")[0].split("↓")[0]
                                           for row in rows for part in row["特征摘要"].split(";") if part})),
        })
    common.sort(key=lambda row: (-row["事件数"], -row["持续预警记录数"], -row["候选日数"]))
    write("伪在线_跨事件共性.csv", common)

    lines = ["# 伪在线局部故障检测结论", "", "## 回放设置", "",
             "- 基线使用事件前长日期数据，每测点每天粗采最多3个周期。",
             "- 检测窗口按有数据的自然日0~24小时处理窗口内全部有效周期，不抽样。",
             "- 每个24小时窗口拆成4个6小时子窗。",
             "- 使用周期平台归一化、测点自身历史稳健Z、单机公共变化扣除。",
             "- 回放检测不使用维修后数据。", ""]
    for event in sorted({row["事件"] for row in detail}):
        rows = [row for row in detail if row["事件"] == event]
        event_bands = [row for row in bands if row["事件"] == event]
        event_validation = [row for row in validation if row["事件"] == event]
        counts = Counter(row["等级"] for row in rows)
        lines += [f"## {event}", "",
                  f"- 完整日×测点×角度箱记录：{len(rows)}。",
                  f"- 正常{counts['正常']}，观察{counts['观察']}，候选{counts['候选']}，持续预警{counts['持续预警']}。",
                  f"- 角度带状态记录：{len(event_bands)}；曾达到候选的测点角度带：{len(event_validation)}。"]
        ranked = sorted(event_validation, key=lambda row: (-int(row["持续预警窗口数"]),
                                                            -float(row["最大综合RMS"])))
        for index, row in enumerate(ranked[:15], 1):
            sensor = "（疑似整通道异常，排除局部故障）" if row["角度带"] == "0-360°" else ""
            lines.append(f"{index}. {row['测点']} {row['角度带']}：首次候选{row['首次候选日期']}，"
                         f"相对事件提前{row['提前日数']}天，持续预警{row['持续预警窗口数']}窗，"
                         f"最大综合RMS {float(row['最大综合RMS']):.1f}{sensor}。")
        lines.append("")

    lines += ["## 已知目标命中情况", "",
              "- 7号2026一缸盖侧240~270°：5月16日进入观察，5月17日三个箱开始形成候选，5月19日形成持续预警，相对5月21日提前约2~4天。",
              "- 7号2026二缸盖侧140~150°：仅达到观察，没有达到24小时候选；它是数月慢漂移特征，说明必须增加运行段早期基线/慢趋势通道，不能只用长历史24小时幅度阈值。",
              "- 4号四缸轴侧60~70°：6月7~9日为观察，6月10日达到候选；附近70~80°更早形成持续预警，提示角度带边界应允许相邻箱漂移。",
              "- 6号三缸轴侧160~180°：4月19日形成候选，4月21日形成持续预警，相对4月22日提前约1~3天。",
              "- 7号2025二缸盖侧出现0~360°超大异常，是已知传感器卡死污染，不是局部故障命中。", "",
              "## 方法结论", "",
              "1. 长历史粗采样基线 + 24小时全周期检测能够检出事件前局部变化，但单通道会漏掉缓慢、小幅、长期单向漂移。",
              "2. 正式方法必须并行运行快通道和慢通道：快通道使用长历史基线检测24小时突变；慢通道使用本运行段最早日期固定基线检测数日/数周定向漂移。",
              "3. 角度带不能要求每天边界完全一致，应允许相邻10度箱在前后一天移动一个箱，并按20~40度范围追踪。",
              "4. 全周期大范围同时异常应转入传感器/工况分类，不进入局部气阀告警。",
              "5. 当前阈值在4号和6号产生较多候选，尚不能直接上线；必须用更多无故障运行段校准候选日触发率。",
              "6. 跨机器只比较无量纲Z、特征族、持续性和角度带结构，不比较原始压力或原始特征。", "",
              "## 建议正式检测条件", "",
              "- 快通道：24小时内至少两个6小时子窗异常、至少两个特征族、幅度平均不低于2Z、综合RMS不低于3Z。",
              "- 慢通道：相对运行段早期固定基线连续至少3个有效日期同方向，或最近5个有效日期至少3天达到2Z；不要求单日达到快通道阈值。",
              "- 持续预警：快通道最近5个有效窗口至少3个候选，或慢通道持续3天以上；同时要求局部残差显著。",
              "- 传感器/公共工况排除：异常覆盖超过半个周期、多个同级测点同步或绝对压力平台不合理时，降级为数据质量/工况告警。",
              "- 基线有效日期少于7天只输出探索性结果；7~19天为中置信；20天以上才允许高置信。"]
    (ROOT / "伪在线检测结论.md").write_text("\n".join(lines) + "\n", encoding="utf-8")
    print("跨事件共性", len(common), "->", ROOT / "伪在线_跨事件共性.csv")
    print("结论 ->", ROOT / "伪在线检测结论.md")


if __name__ == "__main__":
    main()