Compressor/backend/indicator_features/trend.py

#!/usr/bin/env python3
"""Long-term indicator-feature deviation vs the scanner's anomaly segments.

For each cylinder/side it reduces the feature CSV to daily medians, scores them
as robust z against the point's own baseline (median/MAD over its valid days),
and marks days that fall inside a rule anomaly segment from cache/pks_scanner/
evolution (pressure 阶段段 / 振动异常段).  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

UNIT_BATCH = {"7": 30, "8": 31, "9": 32}
TREND_FEATURES = ("p_discharge", "pressure_ratio", "n_comp", "efficiency")


def parse_dt(value):
    return datetime.strptime(value[:19], "%Y-%m-%d %H:%M:%S")


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


def load_segments(evolution_dir, unit):
    """Rule anomaly segments for a unit from the two stage CSVs."""
    segments = []
    for name, kind in (("润滑油压力小时阶段段.csv", "压力"), ("振动异常段.csv", "振动")):
        path = Path(evolution_dir) / name
        if not path.is_file():
            continue
        for row in csv.DictReader(open(path, encoding="utf-8-sig")):
            if row["机组"] != f"{unit}号机":
                continue
            segments.append({
                "kind": kind, "level": row.get("压力阶段") or row.get("振动阶段"),
                "start": parse_dt(row["开始小时"]), "end": parse_dt(row["结束小时"]),
            })
    return segments


def mark_segments(day: str, segments):
    moment = datetime.strptime(day, "%Y-%m-%d")
    return [s["kind"] + "/" + str(s["level"]) for s in segments if s["start"] <= moment <= s["end"]]


def main():
    parser = argparse.ArgumentParser(description="功图特征长期偏离检测（对齐 evolution 异常段）")
    parser.add_argument("csv_path")
    parser.add_argument("--evolution-dir", default=str(Path(__file__).resolve().parents[2] / "cache" / "pks_scanner" / "evolution"))
    parser.add_argument("--unit", required=True, choices=("7", "8", "9"))
    args = parser.parse_args()

    rows = load_features(args.csv_path)
    segments = load_segments(args.evolution_dir, args.unit)
    by_series = defaultdict(list)
    for row in rows:
        by_series[(row["cylinder"], row["side"])].append(row)

    print(f"文件 {args.csv_path}  行 {len(rows)}  evolution段 {len(segments)}")
    for (cylinder, side), series in sorted(by_series.items()):
        days = sorted({r["day"] for r in series})
        print(f"\n== {cylinder}缸{side}  ({len(days)} 天)")
        for feature in TREND_FEATURES:
            daily = {}
            for day in days:
                values = [r[feature] for r in series if r["day"] == day and r[feature] is not None]
                if values:
                    daily[day] = float(np.median(values))
            if len(daily) < 5:
                continue
            values = np.array([daily[d] for d in days if d in daily], dtype=float)
            center = float(np.median(values))
            mad = float(np.median(np.abs(values - center)))
            scale = mad * 1.4826 or float(np.std(values))
            if scale <= 1e-9:
                continue
            print(f"   {feature:<16} 基线 {center:.3f}±{scale:.3f}")
            for day in days:
                if day not in daily:
                    continue
                z = (daily[day] - center) / scale
                flags = mark_segments(day, segments)
                if abs(z) >= 2.0 or flags:
                    tag = "  ".join(flags)
                    print(f"      {day}  {daily[day]:>8.3f}  z={z:+.2f}  {tag}")


if __name__ == "__main__":
    main()