功图特征升级：Otsu平台检测、相位对齐指示功/效率、三种归一化对照与通道QC

backend/indicator_features/README.md

@@ -15,7 +15,13 @@ python indicator_features/extract_from_db.py \
 ```
 
 - `--per-day N`：每个测点每天最多取 N 个文件（均匀取样），控制拉取量。
-- 输出每周期一行的 CSV：吸/排气平台、压比、相位占比、斜率、拐角、指示功等。
+- 输出每周期一行的 CSV：吸/排气平台、压比、相位占比、斜率、拐角、指示功、效率、相位剖面等。
+
+对照分析（同级缸间 / 同缸双腔 / 本缸自身基线，含通道 QC）：
+
+```bash
+python indicator_features/analyze.py cache/indicator/7_1stage_valve_202605.csv
+```
 
 ## 特征（当前）
 

backend/indicator_features/analyze.py

@@ -0,0 +1,173 @@
+#!/usr/bin/env python3
+"""Compare indicator features with three normalizations (read-only).
+
+  * peer     : within the same stage and side at the same timestamp;
+  * chamber  : cover side vs shaft side of the same cylinder;
+  * self     : the cylinder own trailing baseline (median/MAD) -> z-score.
+
+Absolute levels depend on the process, so only relative quantities are scored.
+"""
+from __future__ import annotations
+
+import argparse
+import csv
+from collections import defaultdict
+
+import numpy as np
+
+DEAD_DISCHARGE = 0.20
+PEER_FEATURES = ("p_discharge", "pressure_ratio", "n_comp", "discharge_frac", "efficiency", "indicated_work")
+PROFILE = tuple(f"prof_{i:02d}" for i in range(12))
+
+
+def to_float(value):
+    try:
+        result = float(value)
+    except (TypeError, ValueError):
+        return None
+    return result if np.isfinite(result) else None
+
+
+def load(path: str) -> list[dict]:
+    rows = list(csv.DictReader(open(path, encoding="utf-8-sig")))
+    for row in rows:
+        for key in PEER_FEATURES + PROFILE + ("p_suction", "p_span"):
+            row[key] = to_float(row.get(key))
+    return rows
+
+
+def healthy(row: dict, exclude: set | None = None) -> bool:
+    if exclude and (row["cylinder"], row["side"]) in exclude:
+        return False
+    pd = row.get("p_discharge")
+    ps = row.get("p_suction")
+    span = row.get("p_span")
+    return pd is not None and pd >= DEAD_DISCHARGE and ps and ps > 0.05 and span and span >= 0.3
+
+
+def series_medians(rows: list[dict], feature: str) -> dict:
+    by_series = defaultdict(list)
+    for row in rows:
+        value = row.get(feature)
+        if value is not None:
+            by_series[(row["cylinder"], row["side"])].append(value)
+    return {key: float(np.median(values)) for key, values in by_series.items() if values}
+
+
+def flag_channels(rows: list[dict]) -> dict:
+    """Flag weak/dead channels relative to the stage-wide expected discharge level."""
+    medians = series_medians(rows, "p_discharge")
+    reference = np.median([v for v in medians.values() if v > DEAD_DISCHARGE]) if medians else 0.0
+    flags = {}
+    for key, value in medians.items():
+        if value < DEAD_DISCHARGE:
+            flags[key] = "dead"
+        elif reference > 0 and value < 0.6 * reference:
+            flags[key] = "weak"
+        elif reference > 0 and value > 1.5 * reference:
+            flags[key] = "high"
+    return flags
+
+
+def peer_deviations(rows: list[dict], exclude: set) -> dict:
+    by_key = defaultdict(dict)
+    for row in rows:
+        if healthy(row, exclude):
+            by_key[(row["stage"], row["side"], row["sample_time"])][row["cylinder"]] = row
+    result = defaultdict(list)
+    for (stage, side, _time), members in by_key.items():
+        if len(members) < 2:
+            continue
+        for feature in PEER_FEATURES:
+            values = [m[feature] for m in members.values() if m[feature] is not None]
+            if len(values) < 2:
+                continue
+            med = float(np.median(values))
+            if abs(med) < 1e-9:
+                continue
+            for cylinder, member in members.items():
+                if member[feature] is not None:
+                    result[(cylinder, side, feature)].append((member[feature] - med) / med)
+    return {key: float(np.mean(values)) for key, values in result.items()}
+
+
+def chamber_asymmetry(rows: list[dict], exclude: set) -> dict:
+    by_key = defaultdict(dict)
+    for row in rows:
+        by_key[(row["cylinder"], row["sample_time"])][row["side"]] = row
+    result = defaultdict(list)
+    for (cylinder, _time), pair in by_key.items():
+        if "盖侧" not in pair or "轴侧" not in pair:
+            continue
+        cover, shaft = pair["盖侧"], pair["轴侧"]
+        if not (healthy(cover, exclude) and healthy(shaft, exclude)):
+            continue
+        for feature in ("p_discharge", "pressure_ratio", "n_comp"):
+            if cover[feature] is not None and shaft[feature] is not None and abs(shaft[feature]) > 1e-9:
+                result[(cylinder, feature)].append((cover[feature] - shaft[feature]) / abs(shaft[feature]))
+        profile_cover = np.array([cover[name] for name in PROFILE], dtype=float)
+        profile_shaft = np.array([shaft[name] for name in PROFILE], dtype=float)
+        if np.all(np.isfinite(profile_cover)) and np.all(np.isfinite(profile_shaft)):
+            result[(cylinder, "profile_l2")].append(float(np.linalg.norm(profile_cover - profile_shaft)))
+    return {key: float(np.mean(values)) for key, values in result.items()}
+
+
+def self_baseline(rows: list[dict], exclude: set) -> dict:
+    by_series = defaultdict(list)
+    for row in rows:
+        if healthy(row, exclude):
+            by_series[(row["cylinder"], row["side"])].append(row)
+    result = {}
+    for (cylinder, side), series in by_series.items():
+        series.sort(key=lambda r: r["sample_time"])
+        times = [r["sample_time"] for r in series]
+        for feature in ("p_discharge", "pressure_ratio", "n_comp"):
+            values = np.array([r[feature] if r[feature] is not None else np.nan for r in series], dtype=float)
+            valid = values[np.isfinite(values)]
+            if valid.size < 5:
+                continue
+            center = float(np.median(valid))
+            mad = float(np.median(np.abs(valid - center)))
+            scale = mad * 1.4826 or float(np.std(valid))
+            if scale <= 1e-9:
+                continue
+            z = (values - center) / scale
+            result[(cylinder, side, feature)] = {
+                "center": center, "scale": scale,
+                "first_half_z": float(np.nanmean(z[: len(z) // 2])),
+                "second_half_z": float(np.nanmean(z[len(z) // 2:])),
+                "max_abs_z": float(np.nanmax(np.abs(z))) if np.isfinite(z).any() else None,
+                "last_time": times[-1],
+            }
+    return result
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description="功图特征三种归一化对照")
+    parser.add_argument("csv_path")
+    args = parser.parse_args()
+    rows = load(args.csv_path)
+    flags = flag_channels(rows)
+    exclude = set(flags)
+    units = sorted({(r["cylinder"], r["side"]) for r in rows})
+    print(f"总行 {len(rows)}；测点 {len(units)}")
+    print(f"通道标记(排除对照)：{flags if flags else '无'}")
+
+    print("\n=== 同级缸间偏差（同侧、同刻，相对中位数）===")
+    for (cylinder, side, feature), value in sorted(peer_deviations(rows, exclude).items()):
+        print(f"  {cylinder}缸{side:<3} {feature:<16} {value * 100:+.2f}%")
+
+    print("\n=== 同缸双腔差（盖 - 轴）===")
+    for (cylinder, feature), value in sorted(chamber_asymmetry(rows, exclude).items()):
+        if feature == "profile_l2":
+            print(f"  {cylinder}缸 profile L2 {value:.3f}")
+        else:
+            print(f"  {cylinder}缸 {feature:<16} {value * 100:+.2f}%")
+
+    print("\n=== 本缸自身基线 z（前段 vs 后段，沿时间趋势）===")
+    for (cylinder, side, feature), stat in sorted(self_baseline(rows, exclude).items()):
+        print(f"  {cylinder}缸{side:<3} {feature:<16} 前段z={stat['first_half_z']:+.2f} 后段z={stat['second_half_z']:+.2f} max|z|={stat['max_abs_z']:.2f}")
+
+
+if __name__ == "__main__":
+    main()

backend/indicator_features/indicator_features.py

@@ -2,15 +2,17 @@
 """Phase-level features from one compressor indicator diagram cycle.
 
 One stored cycle (wave_sample_one) is the pressure of a single cylinder chamber
-over one crankshaft revolution (~60 ms at ~996 RPM).  Absolute pressure levels
-depend on the process, so these features are meant to be compared:
-  * against the same cylinder's own recent baseline (trend), and
-  * against the peer cylinders of the same compression stage at the same time,
-  * and between the cover/shaft chambers of the same cylinder.
+over one crankshaft revolution (~60 ms at ~996 RPM).
+
+Design goals:
+  * robust two-plateau (suction P_s / discharge P_d) detection via Otsu split;
+  * physical phase alignment (search the crank offset that maximises the PV loop
+    area) so indicated work and a pressure-conversion efficiency are meaningful;
+  * a normalised phase profile for peer / double-chamber / self-baseline review.
 
 Compression staging (confirmed with the site):
   一级(1st) = 2/4/6 缸,  二级(2nd) = 3/5 缸,  三级(3rd) = 1 缸.
-No model is trained here; the goal is to expose interpretable features.
+Absolute levels depend on the process, so features are meant to be compared.
 """
 from __future__ import annotations
 
@@ -21,6 +23,8 @@ CYLINDER_STAGE = {"一": "三级", "二": "一级", "三": "二级", "四": "一
 PISTON_STROKE_MM = 148.0
 CONNECTING_ROD_LENGTH_MM = 460.0
 CLEARANCE_VOLUME_L_BY_BORE = {490.0: 1.51, 390.0: 0.74, 360.0: 0.62}
+PROFILE_BINS = 12
+K_ISENTROPIC = 1.30
 
 
 def smooth(values: np.ndarray, window: int) -> np.ndarray:
@@ -31,12 +35,37 @@ def smooth(values: np.ndarray, window: int) -> np.ndarray:
     return np.convolve(values, kernel, mode="same")
 
 
+def otsu_threshold(values: np.ndarray, bins: int = 96) -> float:
+    hist, edges = np.histogram(values, bins=bins)
+    centers = (edges[:-1] + edges[1:]) / 2.0
+    total = float(values.size)
+    if total <= 0:
+        return float(np.mean(values))
+    sum_total = float(np.sum(centers * hist))
+    sum_before = 0.0
+    weight_before = 0.0
+    best_variance = -1.0
+    threshold = float(centers[0])
+    for index in range(len(hist)):
+        weight_before += float(hist[index])
+        if weight_before == 0.0:
+            continue
+        weight_after = total - weight_before
+        if weight_after == 0.0:
+            break
+        sum_before += centers[index] * hist[index]
+        mean_before = sum_before / weight_before
+        mean_after = (sum_total - sum_before) / weight_after
+        variance = weight_before * weight_after * (mean_before - mean_after) ** 2
+        if variance > best_variance:
+            best_variance = variance
+            threshold = float(centers[index])
+    return threshold
+
+
 def volume_from_angle(count: int, bore_mm: float) -> np.ndarray:
-    """Chamber volume (L) over one revolution, from crank-slider geometry."""
     crank_radius = PISTON_STROKE_MM / 2.0
-    clearance = CLEARANCE_VOLUME_L_BY_BORE.get(bore_mm)
-    if clearance is None:
-        clearance = 1.0
+    clearance = CLEARANCE_VOLUME_L_BY_BORE.get(bore_mm, 1.0)
     theta = np.linspace(0.0, 2.0 * np.pi, count, endpoint=False)
     travel = (
         crank_radius * (1.0 - np.cos(theta))
@@ -47,50 +76,85 @@ def volume_from_angle(count: int, bore_mm: float) -> np.ndarray:
     return clearance + piston_area * travel / 1_000_000.0
 
 
-def _plateau_levels(p: np.ndarray) -> tuple[float, float]:
-    """Suction (low) and discharge (high) plateau levels via robust quantiles."""
-    low = np.quantile(p, 0.12)
-    high = np.quantile(p, 0.88)
-    near_low = p[p <= low + 0.15 * (high - low)]
-    near_high = p[p >= high - 0.15 * (high - low)]
-    suction = float(np.median(near_low)) if near_low.size else float(np.min(p))
-    discharge = float(np.median(near_high)) if near_high.size else float(np.max(p))
-    return suction, discharge
+def _plateau_levels(p: np.ndarray) -> tuple[float, float, float]:
+    threshold = otsu_threshold(p)
+    low = p[p <= threshold]
+    high = p[p > threshold]
+    suction = float(np.median(low)) if low.size else float(np.min(p))
+    discharge = float(np.median(high)) if high.size else float(np.max(p))
+    return suction, discharge, threshold
+
+
+def _align_offset(p: np.ndarray, dvolume: np.ndarray) -> tuple[int, float]:
+    """Offset (in samples) that maximises the indicated work -sum(p dV)."""
+    count = len(p)
+    cross = np.real(np.fft.ifft(np.conj(np.fft.fft(p)) * np.fft.fft(dvolume)))
+    work = -cross
+    offset = int(np.argmax(work))
+    return offset, float(work[offset])
+
+
+def _polytropic_index(volume: np.ndarray, pressure: np.ndarray, low: float, high: float, rising: bool) -> float | None:
+    dp = np.diff(pressure, prepend=pressure[0])
+    band = (pressure > low + 0.15 * (high - low)) & (pressure < low + 0.85 * (high - low))
+    direction = dp > 0 if rising else dp < 0
+    mask = band & direction
+    v = volume[mask]
+    p = pressure[mask]
+    valid = (v > 0) & (p > 0)
+    if np.count_nonzero(valid) < 8:
+        return None
+    slope, _ = np.polyfit(np.log(v[valid]), np.log(p[valid]), 1)
+    return float(-slope)
 
 
 def cycle_features(signal: np.ndarray, cylinder: str, smooth_window: int = 21) -> dict:
     signal = np.asarray(signal, dtype=float)
     signal = signal[np.isfinite(signal)]
     count = len(signal)
-    if count < 50:
+    if count < 60:
         return {}
     p = smooth(signal, smooth_window)
     p = np.where(np.isfinite(p), p, signal)
-    suction, discharge = _plateau_levels(p)
+    suction, discharge, threshold = _plateau_levels(p)
     span = discharge - suction
-    threshold = suction + 0.5 * span
-    high = p >= threshold
+    volume = volume_from_angle(count, CYLINDER_BORE_MM.get(cylinder, 490.0))
+    dvolume = np.diff(volume, append=volume[0])
+
+    offset, indicated_work = _align_offset(p, dvolume)
+    aligned_volume = np.roll(volume, -offset)
+    aligned_dvolume = np.roll(dvolume, -offset)
+
+    # Isentropic ideal work (MPa*L = kJ) and pressure-conversion efficiency.
+    v_suction = float(np.max(aligned_volume))
+    ratio = (discharge / suction) if suction > 0 else None
+    if ratio and ratio > 1:
+        ideal = (K_ISENTROPIC / (K_ISENTROPIC - 1.0)) * suction * v_suction * (
+            ratio ** ((K_ISENTROPIC - 1.0) / K_ISENTROPIC) - 1.0
+        )
+    else:
+        ideal = None
+    efficiency = (indicated_work / ideal) if ideal and ideal > 1e-9 else None
+
+    # Phase boundaries by pressure crossing (aligned).
+    high_mask = p >= threshold
+    suction_frac = float(np.mean(~high_mask))
+    discharge_frac = float(np.mean(high_mask))
     dp = np.diff(p)
     rising = dp[dp > 0]
     falling = dp[dp < 0]
-    # Compression/expansion line steepness (per sample).
     compression_slope = float(np.quantile(rising, 0.9)) if rising.size else 0.0
     expansion_slope = float(-np.quantile(falling, 0.1)) if falling.size else 0.0
-    # Plateau occupancy: fraction of the cycle spent near high/low plateau.
-    near = 0.08 * span + 1e-9
-    suction_frac = float(np.mean(np.abs(p - suction) <= near))
-    discharge_frac = float(np.mean(np.abs(p - discharge) <= near))
-    # Indicated work over the revolution (net area) and mean effective pressure.
-    volume = volume_from_angle(count, CYLINDER_BORE_MM.get(cylinder, 490.0))
-    dvolume = np.diff(volume, append=volume[0])
-    work = float(-np.sum(p * dvolume))  # net work done on the gas (positive)
-    sweep = float(np.max(volume) - np.min(volume))
-    imep = work / sweep if sweep else 0.0
-    # Corner sharpness: mean |curvature| of pressure where it crosses the mid band.
     d2 = np.diff(p, 2)
     band = np.abs(p[1:-1] - threshold) <= 0.1 * span
     corner = float(np.mean(np.abs(d2[band]))) if band.any() else 0.0
-    return {
+
+    # Normalised phase profile (0..1) on the aligned cycle.
+    idx = np.linspace(0, count - 1, PROFILE_BINS).round().astype(int)
+    prof = (p[idx] - suction) / span if span else np.zeros(PROFILE_BINS)
+    profile = {f"prof_{i:02d}": float(prof[i]) for i in range(PROFILE_BINS)}
+
+    features = {
         "n_samples": count,
         "p_min": float(np.min(p)),
         "p_max": float(np.max(p)),
@@ -98,13 +162,18 @@ def cycle_features(signal: np.ndarray, cylinder: str, smooth_window: int = 21) -
         "p_suction": suction,
         "p_discharge": discharge,
         "p_span": span,
-        "pressure_ratio": (discharge / suction) if suction else None,
-        "pressure_rise": span,
+        "pressure_ratio": ratio,
         "suction_frac": suction_frac,
         "discharge_frac": discharge_frac,
         "compression_slope": compression_slope,
         "expansion_slope": expansion_slope,
-        "indicated_work": work,
-        "imep": imep,
         "corner_sharpness": corner,
+        "n_comp": _polytropic_index(aligned_volume, p, suction, discharge, rising=True),
+        "n_exp": _polytropic_index(aligned_volume, p, suction, discharge, rising=False),
+        "indicated_work": indicated_work,
+        "ideal_work_isentropic": ideal,
+        "efficiency": efficiency,
+        "phase_offset": offset,
     }
+    features.update(profile)
+    return features

backend/indicator_features/观察记录.md

@@ -1,57 +1,47 @@
 # 功图找气阀/活塞故障 —— 观察记录
 
-> 只读分析，样本为 `wave_file` + `wave_sample_one`（首周期）。故障锚点来自 `7,8,9号压缩机故障汇总.xlsx`。
+> 只读分析。样本为 `wave_file` + `wave_sample_one`（首周期，DB 实测约 1545 点）。故障锚点来自 `7,8,9号压缩机故障汇总.xlsx`；异常段只认 `cache/pks_scanner/evolution/`。
 
 ## 1. 功图数据方法（确认）
 
-- 采样：25.6 kHz、键相 8 脉冲/转、≈996 RPM → 单周期≈60 ms、≈1540 点。
-- `wave_sample_one` 现在每个文件就是**一个首周期**（DB 实测 1545 点）。
-- `wave_file.cycle_start/cycle_end` 已废弃，不使用。
-- 只认 `cache/pks_scanner/evolution/` 作为异常段来源。
+- 采样：25.6 kHz、键相 8 脉冲/转、≈996 RPM → 单周期≈60 ms、≈1540 点；15 分钟一个文件。
+- `wave_sample_one` 现在每文件就是**一个首周期**；`wave_file.cycle_start/cycle_end` 已废弃。
+- 级配：**一级=2/4/6、二级=3/5、三级=1**；压力逐级升高（一级 Pd≈1.4~2.2、二级≈3.2~4.4、三级≈5.0~5.1 MPa）。
 
-## 2. 数据覆盖（按月，rpm>0 运行文件）
+## 2. 已完成改进（对应任务 1/2/3）
 
-| 测点 | 运行覆盖 |
-|---|---|
-| 8 号机 全部测点 | 2025-03~05、2026-03~06 |
-| 7 号机 一级(2/4/6)、四级… | 2025-04~07、2025-11、2026-02~06（2025-08/10 无） |
-| 9 号机 一级(2缸) | 2025-03、06、09、10、2026-01~06 |
-| 7 号机 一/六缸、9 号机 一/五/六缸 | 2026-01 起才有 |
+1. **稳健平台检测**：用 Otsu 双峰分割替代分位数，稳定求 `P_s`（吸气平台）、`P_d`（排气平台）、`pressure_ratio`。
+2. **物理相位对齐**：在整转内搜索使 PV 环面积最大的曲柄偏移，得到可用的 `indicated_work`、等熵理想功与 `efficiency`、多变指数 `n_comp`。
+3. **归一化相位形状/对照**（`analyze.py`）：
+   - 同级缸间偏差（同侧、同刻相对中位数）；
+   - 同缸双腔差（盖侧 − 轴侧，含归一化相位剖面 L2 距离）；
+   - 本缸自身近期基线 z（沿时间趋势）。
+   - 通道 QC：按“相对同级期望排气压力”自动标记 dead/weak/high 并排除。
 
-**含义**：多数 2025 年气阀/活塞事件没有功图数据；可用的候选事件：
+## 3. 7 号机一级 2026-05（故障前两周，任务 4 的最佳样本）
 
-- **7 号机 2026.5.21 气阀**：2026-05-06~05-20 连续密集运行（最优，故障前两周）
-- **8 号机 2026.3.29 / 4.1 气阀**：运行仅到 2026-03-20，随后停机
-- **9 号机 2026.1.18 活塞（三级=1 缸）**：1 缸数据从 2026-01-30 起（故障后才运行）
+- 通道标记：**六缸盖侧 weak、四缸盖侧 dead**（已排除对照）。
+- 同级（轴侧）偏差稳定：二/四/六 p_discharge 偏差 −0.5% / +1.8% / −6.6%，pressure_ratio −6.3% / +1.9% / +3.7%。
+- 本缸自身趋势：前段 z 与后段 z 基本在 ±0.7 内，**无持续漂移**。
+- 结论：**故障前无功图趋势**，与该事件“进气阀组油压管突发崩开”的性质一致（突发而非渐进）。
 
-## 3. 已验证的观察
+## 4. 事件可用性（任务 4）
 
-### 3.1 7 号机 一级 2026-05 故障前两周
-特征按天均值基本平稳，无显著趋势：
-
-| 缸·侧 | p_discharge 走势 | 压比 前段→末段 |
+| 类型 | 事件 | 故障前功图 |
 |---|---|---|
-| 二·盖 | 1.79→1.91→1.85 | 2.93→2.99 (+2.1%) |
-| 二·轴 | 1.69→1.89→1.81 | 2.69→2.69 (−0.2%) |
-| 四·轴 | 1.77→1.92→1.82 | 2.91→3.01 (+3.5%) |
-| 六·轴 | 1.62→1.76→1.68 | 3.05→3.10 (+1.5%) |
-
-- 该事件是“进气阀组油压管崩开”，属**突发**，**故障前期无趋势符合预期**。
-- 相邻时刻功图几乎重合；跨天变化在 ±3% 内 → 与“压力变化微弱”的观察一致。
-
-### 3.2 问题
-- **坏通道**：8 号机六缸盖侧 Pd≈0.02；7 号机六缸盖侧 Pd≈0.8、四缸盖侧 NaN。做缸间对照前必须剔除。
-- **指示功/IMEP 不可用**：`∮P dV` 依赖周期起始与活塞相位的对齐，当前未锚定，出现负值/爆量。
-
-## 4. 下一步（找特征）
-
-1. **稳健平台检测**：用压力直方图双峰/众数定 P_s、P_d，替代分位数，稳住压比。
-2. **物理相位锚定**：用键相/触发把周期对齐到活塞相位，使 `∮P dV`、压力转换效率可用。
-3. **归一化形状特征**：每周期按相位重采样 P(θ) 并做“同级缸间 / 同缸双腔 / 本缸近期基线”对照。
-4. **只挑渐进型故障**：现有最佳是 7 号机 2026-05（突发），需再找带“故障前持续运行”的渐进事件验证。
-5. 坏通道自动剔除 + 每测点近期基线。
-
-## 5. 结论（当前）
-
-- 抽取链路可用，平台类特征稳定（同侧同级可比、双腔可比）。
-- 但在目前可用的故障前数据上**尚未找到明显趋势**；需先完善相位/形状特征，并找合适的渐进型事件。
+| 气阀 | 7 号机 2026.5.21 | ✅ 2026-05-06~20 连续运行（但属突发） |
+| 气阀 | 8 号机 2026.3.29 / 4.1 | ❌ 8 号机 2026-03-20 后停机 |
+| 气阀 | 8 号机 2025.2.17 | ❌ 8 号机 2025-03 才有数据 |
+| 气阀 | 7 号机 2025.8.8 / 2025.10.7 | ❌ 无 2025-08 / 2025-10 数据 |
+| 活塞 | 8 号机 2025.12.6 | ❌ 故障前已停机 |
+| 活塞 | 9 号机 2026.1.18（三级=1缸） | ❌ 1 缸数据 2026-01-30 才有（故障后） |
+| 活塞 | 7 号机 2025.1.10 | ❌ 早于数据范围 |
+
+**结论：当前功图数据里没有一个“故障前持续运行 + 渐进退化”的气阀/活塞事件**，因此暂无法证明或否定功图趋势。唯一可用事件是突发型。
+
+## 5. 下一步
+
+1. 等新数据积累（2026-06 后有 4/6 号机与 7/8/9 号机的密集数据），找渐进型事件；
+2. 校准等熵指数 k（当前 efficiency≈1.2，偏高），把“压力转换效率”做成可解释指标；
+3. 用 `n_comp`、相位剖面 L2、双腔不对称做长期基线，做“偏离检测”而非分类；
+4. 与 `cache/pks_scanner/evolution/` 的压力/振动异常段做时间对齐，多证据互验。