Compressor/backend/indicator_features/lookback_scan.py

#!/usr/bin/env python3
"""Look back over the previous 1-2 valid data segments before each event.

Many repairs have no data immediately before them (machine stopped long before
the repair was recorded).  This groups the days that actually contain an
indicator cycle into segments and inspects the last one or two segments before
the event: peer-relative levels and within-segment drift, plus a comparison
between the two segments to expose a per-cylinder divergence.  Read-only.
"""
from __future__ import annotations

import argparse
import csv
from collections import defaultdict
from datetime import datetime
from pathlib import Path

import numpy as np

EVENTS = [
    ("7", "2025-01-10", "活塞"),
    ("7", "2025-08-08", "气阀"),
    ("7", "2025-10-07", "气阀"),
    ("7", "2026-05-21", "气阀"),
    ("8", "2025-02-17", "气阀"),
    ("8", "2025-12-06", "活塞"),
    ("8", "2026-03-10", "十字头"),
    ("8", "2026-03-29", "气阀"),
    ("8", "2026-04-01", "气阀"),
    ("9", "2026-01-18", "活塞"),
]
CYLINDERS = ("一", "二", "三", "四", "五", "六")
SIDES = ("盖侧", "轴侧")
FEATURES = ("p_discharge", "pressure_ratio", "efficiency")


def load(path):
    rows = []
    for row in csv.DictReader(open(path, encoding="utf-8-sig")):
        row["day"] = row["sample_time"][:10]
        for name in FEATURES:
            try:
                row[name] = float(row[name])
            except (TypeError, ValueError):
                row[name] = None
        rows.append(row)
    return rows


def segments(days, gap=10):
    days = sorted(set(days))
    out = []
    for day in days:
        if out and (datetime.strptime(day, "%Y-%m-%d") - datetime.strptime(out[-1][-1], "%Y-%m-%d")).days <= gap:
            out[-1].append(day)
        else:
            out.append([day])
    return out


def peer_relative(rows, feature):
    by_ts = defaultdict(dict)
    for row in rows:
        if row[feature] is not None:
            by_ts[(row["day"], row["side"])][row["cylinder"]] = row[feature]
    out = defaultdict(dict)
    for (day, side), members in by_ts.items():
        clean = {c: v for c, v in members.items() if v is not None}
        if len(clean) < 2:
            continue
        med = float(np.median(list(clean.values())))
        if abs(med) < 1e-9:
            continue
        for cylinder, value in clean.items():
            out[(cylinder, side)][day] = (value - med) / med
    return out


def seg_summary(rows, seg):
    seg_rows = [r for r in rows if r["day"] in set(seg)]
    peer = {feature: peer_relative(seg_rows, feature) for feature in ("p_discharge", "pressure_ratio")}
    summary = {}
    drift = {}
    for feature, series in peer.items():
        for key, daily in series.items():
            summary[(key[0], key[1], feature)] = (float(np.median(list(daily.values()))), len(daily))
            days = sorted(daily)
            if len(days) >= 8:
                x = np.array([(datetime.strptime(d, "%Y-%m-%d") - datetime.strptime(days[0], "%Y-%m-%d")).days for d in days], dtype=float)
                y = np.array([daily[d] for d in days], dtype=float)
                if x[-1] > 0:
                    slope = float(np.polyfit(x, y, 1)[0])
                    scale = float(np.median(np.abs(y - np.median(y)))) * 1.4826 or float(np.std(y))
                    if scale > 1e-9:
                        drift[(key[0], key[1], feature)] = float(slope * x[-1] / scale)  # total drift in MADs
    return summary, drift


def main():
    parser = argparse.ArgumentParser(description="事件前 1~2 个有效数据段观察")
    parser.add_argument("--data-dir", default=str(Path(__file__).resolve().parents[1] / "cache" / "indicator"))
    parser.add_argument("--segments", type=int, default=2)
    args = parser.parse_args()

    cache = {}
    for unit in ("7", "8", "9"):
        path = Path(args.data_dir) / f"all_{unit}_2025_2026.csv"
        cache[unit] = load(path) if path.is_file() else []

    for unit, date, kind in EVENTS:
        rows = cache[unit]
        all_days = sorted({r["day"] for r in rows})
        segs = [s for s in segments(all_days) if s[-1] < date]
        print(f"\n== {unit}号机 {date} {kind}  事件前有效段 {len(segs)}，取最近 {args.segments}")
        chosen = segs[-args.segments:]
        summaries = []
        for i, seg in enumerate(reversed(chosen), start=1):
            summary, drift = seg_summary(rows, seg)
            summaries.append((seg, summary))
            print(f"   段{i} {seg[0]}~{seg[-1]} ({len(seg)}天)   peer相对 p_discharge | pressure_ratio")
            for cylinder in CYLINDERS:
                cells = []
                for side in SIDES:
                    pd = summary.get((cylinder, side, "p_discharge"))
                    pr = summary.get((cylinder, side, "pressure_ratio"))
                    if pd is None and pr is None:
                        continue
                    pd_s = f"{pd[0]*100:+.0f}%" if pd else "-"
                    pr_s = f"{pr[0]*100:+.0f}%" if pr else "-"
                    cells.append(f"{side}:Pd{pd_s}/R{pr_s}")
                if cells:
                    print(f"      {cylinder}缸  " + "  ".join(cells))
            strong = [(k, v) for k, v in sorted(drift.items(), key=lambda kv: -abs(kv[1])) if abs(v) >= 2.0][:4]
            if strong:
                print("      段内漂移(>=2MAD): " + "; ".join(f"{k[0]}缸{k[1]} {k[2]} {v:+.1f}" for k, v in strong))
        if len(summaries) == 2:
            (seg_prev, sum_prev), (seg_last, sum_last) = summaries[1], summaries[0]
            changes = []
            for key in set(sum_prev) & set(sum_last):
                cylinder, side, feature = key
                delta = sum_last[key][0] - sum_prev[key][0]
                # only peer-relative, ignore tiny absolute
                if abs(delta) >= 0.05:
                    changes.append((abs(delta), f"{cylinder}缸{side} {feature}: {sum_prev[key][0]*100:+.1f}% -> {sum_last[key][0]*100:+.1f}%"))
            changes.sort(reverse=True)
            if changes:
                print("   两段对比（变化>=5%）：")
                for _, text in changes[:6]:
                    print("      " + text)


if __name__ == "__main__":
    main()