Compressor/_tmp_temperature/make_report.py

#!/usr/bin/env python3
"""Generate ``_tmp_temperature/out/分析报告.md`` from the unified outputs."""
from __future__ import annotations

import argparse
from pathlib import Path

import pandas as pd


def md_table(frame: pd.DataFrame, columns: list[str]) -> str:
    frame = frame.fillna("")
    lines = ["| " + " | ".join(columns) + " |", "|" + "|".join(["---"] * len(columns)) + "|"]
    for _, row in frame.iterrows():
        lines.append("| " + " | ".join(str(row[column]) for column in columns) + " |")
    return "\n".join(lines)


def main() -> None:
    parser = argparse.ArgumentParser()
    parser.add_argument("--dir", default=str(Path(__file__).resolve().parent / "out"))
    args = parser.parse_args()
    directory = Path(args.dir)

    segments = pd.read_csv(directory / "温度异常段.csv")
    backtest = pd.read_csv(directory / "回测_温度事件.csv")
    channels = pd.read_csv(directory / "温度通道异常段.csv")
    cycles = pd.read_csv(directory / "温度演化周期.csv")
    hf_path = directory / "温度高频波动段.csv"
    hf = pd.read_csv(hf_path) if hf_path.exists() else pd.DataFrame()
    candidate_path = directory / "温度高频候选.csv"
    candidates = pd.read_csv(candidate_path) if candidate_path.exists() else pd.DataFrame()

    events = segments[segments["特征类型"] != "高频波动"]
    observation_path = directory / "温度观察段.csv"
    observed = pd.read_csv(observation_path) if observation_path.exists() else pd.DataFrame()
    linked = int((observed["是否升级为事件"] == "是").sum()) if "是否升级为事件" in observed else 0
    significant_channels = int((channels["是否显著"] == "是").sum()) if "是否显著" in channels else 0

    counts = events.pivot_table(index="机组", columns="特征类型", values="批次", aggfunc="count", fill_value=0)
    counts["合计"] = counts.sum(axis=1)
    counts = counts.reset_index()

    severe = events[events["温度阶段"].isin(["异常", "严重异常"])].copy()
    severe = severe.sort_values(["机组", "开始小时"])
    severe_columns = ["机组", "级", "温度阶段", "特征类型", "方向", "置信度", "开始小时", "结束小时",
                      "有效运行小时数", "最大级差变化", "最大同级偏离", "最大同级偏低", "最大同级相对3h", "阶段触发依据"]

    channel_counts = channels.groupby(["机组", "异常类型"]).size().unstack(fill_value=0).reset_index()
    channel_examples = channels[
        (channels["机组"].isin(["4号机", "6号机"]))
        & (channels["开始小时"].str.startswith(("2026-02-17", "2026-01-15")))
    ].copy()
    channel_columns = ["机组", "点位描述", "异常类型", "开始小时", "结束小时", "持续自然小时数",
                       "最小值", "最大值", "同级基准中位"]

    hf_counts = hf.groupby(["机组", "级", "温度阶段"]).size().reset_index(name="段数") if not hf.empty else pd.DataFrame(columns=["机组", "级", "温度阶段", "段数"])
    hf_example = hf[(hf["机组"] == "9号机") & (hf["开始小时"].str.startswith("2026-04-04"))].head(1) if not hf.empty else pd.DataFrame()
    hf_example_columns = ["机组", "级", "温度阶段", "开始小时", "结束小时", "有效运行小时数", "高频波动倍数", "高频标准差", "高频基线标准差"]

    backtest_columns = ["机组", "故障日期", "描述", "最大同级偏离℃", "最大级差变化℃", "命中",
                        "首个事件开始", "提前小时", "同日命中", "通道异常命中", "通道异常类型"]

    text = f"""# 缸体温度统一演化检测 —— 第一版结果

> 代码在 `_tmp_temperature/`，`models/` 未改动。  
> 统一入口：`python3 _tmp_temperature/rescan.py --out _tmp_temperature/out`  
> 回测：`python3 _tmp_temperature/backtest.py --dir _tmp_temperature/out`

## 1. 结果概览

- 总故障段表 `温度异常段.csv`：**{len(segments)}** 行（温度事件 {len(events)} + 高频波动 {len(hf)}）。
- 温度事件：**{len(events)}** 段；另有 1 小时观察段 **{len(observed)}** 条（单独在 `温度观察段.csv`，其中 **{linked}** 条在 24h 内升级为事件）。
- 三级高频波动：**{len(hf)}** 段（高频候选 **{len(candidates)}** 段，未确认的留在 `温度高频候选.csv`，不再重复计数）。
- 涵盖 4#~9#，统一阈值（一级/二级用同级相对；三级只用高频波动单独输出）。
- 事件类型：{', '.join(f'{key}={value}' for key, value in events['特征类型'].value_counts().items())}。

{md_table(counts, list(counts.columns))}

## 2. 统一规则

1. **数据有效性**：7/8/9 用转速；4# 用累计运行小时（+机身振动/进气压力兜底）；5#/6# 用 2/4 信号（一级排气压力、一级进气压力、机身振动、调节阀开度）；整级低于 45℃ 视为停机；单通道突然到环境温度并恢复视为“停机/通道异常”，不参与温度故障。
2. **同级温差（常开特征）**：级差、级差较平时变化、每缸相对同级中位、每缸相对自身历史偏移；无事件小时也输出到 `同级温差特征.csv`。
3. **短时单缸变化**：1h/3h/6h 同级相对变化；1h 记“观察”（单独输出、逐小时一行），2h 记“事件”；要求同方向连续 2 小时，并过滤“先降后升/先升后降”的过渡。
4. **确认/滞回**：持续型 3h 确认；短时型 2h；观察 1h；相邻段间隔 ≤6h 合并。
5. **阈值（统一）**：同级偏高/偏低 4/5、6/8、10/12℃；级差变化 4/8/15℃；短时 1h 3/4/6/9℃、3h 4/5/7/11℃、6h 6/7/9/13℃；三级高频波动：先去慢趋势，再要求去趋势残差标准差 ≥0.5℃、反复升降 ≥5 次/小时、持续 ≥3h（候选/确认分开）；6h 单缸相对抬升 3/4.5/6℃、整级 6h 抬升在 ±3℃ 内、持续 ≥2h（已有持续偏高事件保持原类型；整级共同抬升只作特征，不作为事件）。

## 3. 维修事件回测（4#/5#/6#）

{md_table(backtest, backtest_columns)}

- 12 起中 11 起有数据的事件全部命中；5# 2026-07-11 超出批次 58 数据范围（截至 2026-06-30）。
- 6# 2026-02-18 通过 **2/17 22:00 的通道骤降/停机异常**命中（2h 前），不再算未命中。
- “提前小时”负数表示在故障日当天 0 点之后的同一日内命中；`同日命中=是` 更直观。

## 4. 异常/严重异常事件清单（供复核）

{md_table(severe, severe_columns)}

## 5. 温度通道异常

{md_table(channel_counts, list(channel_counts.columns))}

- 通道异常总段数 **{len(channels)}**，其中“显著” **{significant_channels}** 段（持续≥2h / 坏道 / 冻值）；复核时可先看显著项。

典型例子：

{md_table(channel_examples, channel_columns)}

## 6. 7#/8#/9# 三级高频波动

- 高频波动是 `温度异常段.csv` 中的 `特征类型=高频波动` 行，`温度高频波动段.csv` 为确认后的过滤副本；`温度高频候选.csv` 保留全部首轮候选及是否确认。温度事件仍为上面的 {len(events)} 条。
- 判定：先去慢趋势，再要求去趋势残差标准差 ≥0.5℃、反复升降 ≥5 次/小时、持续 ≥3h；首轮候选 60 段，确认 {len(hf)} 段。
- 目前只对单缸的三级输出；一级/二级暂不输出。

{md_table(hf_counts, list(hf_counts.columns))}

典型例子：

{md_table(hf_example, hf_example_columns)}

## 7. 演化周期

`温度演化周期.csv` 按 **每次开机周期 × 级** 一行，包含：

- 阶段演化顺序、方向、特征类型、置信度；
- 观察/轻微/异常/严重异常各阶段起止与运行小时数；
- 峰值时刻/峰值缸、最大级差变化、最大同级偏高/偏低；
- 首次短时触发、首次持续触发、关联通道异常。

## 8. 运行态加强后的变化

- 4# 改用**累计运行小时** `YSJ_31`（+1/h 运行、0 停机）；5#/6# 改用 **2/4 信号**（一级排气压力、一级进气压力、机身振动、调节阀开度）。
- 与旧版“一级进气 >0.15”相比，剔除了大量“油泵在转但压缩机已停/卸载”的小时：4# 约 117h、5# 约 101h、6# 约 669h。
- 温度小时均值已排除 0 值；通道掉点用原始最小值和零值样本数标记，恢复段不再被当成上升；短时变化要求当前值高于近期 6h 中位。
- 单缸 6h 相对抬升并入事件（类型 `单缸抬升`）：阈值 3/4.5/6℃，且整级 6h 抬升在 ±3℃ 内；已有 `持续偏高` 事件保持原类型，不重新分类。整级共同抬升只保留为特征，不生成事件。
- 因此有 **9 个旧事件落在启停过渡段**，新版不再作为确认事件：4# 2026-02-15、2026-03-24、2026-04-04、2026-04-25、2026-04-26、2026-05-30；5# 2025-05-09；7# 2025-02-17；8# 2025-01-18。它们多数是开机/停机温度爬坡，不是稳定运行故障；这些过渡段只保留在 `温度过渡段.csv`，不在页面和确认事件中展示。

## 9. 限制与下一步

1. 4/5/6 无转速，运行态已用累计运行小时/2-of-4 信号加强；极低负荷小时仍可能有少量误判。
2. 通道异常表较大（单点级，包含大量 1–2h 采样跌落）；建议先按“持续≥2h 或 坏道/冻值”筛选复核。
3. 现有 level 阈值已由你复核为真事件，暂不提高；短时/偏低事件建议再做一轮现场复核后微调。
4. 规则稳定后再合并进 `models/temperature/`，并接入 `compressor_alarm`。
"""
    (directory / "分析报告.md").write_text(text, encoding="utf-8")
    print(f"written -> {directory / '分析报告.md'}")


if __name__ == "__main__":
    main()