Compressor/backend/indicator_features/metric_trace.py

#!/usr/bin/env python3
"""Trace one (point, angle bin, metric) over the whole available timeline.

Prints the daily medians, splits them into data segments (day gaps), fits a
linear trend per segment and overall, and optionally saves a figure.  This is
for checking whether a candidate feature has a slow pre-event evolution.
"""
from __future__ import annotations

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

import numpy as np

BACKEND = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(BACKEND))

import matplotlib
matplotlib.use("Agg")
import matplotlib.pyplot as plt  # noqa: E402

matplotlib.rcParams["font.sans-serif"] = ["Arial Unicode MS", "Heiti TC", "Hiragino Sans GB", "DejaVu Sans"]
matplotlib.rcParams["axes.unicode_minus"] = False

import cn_names as CN  # noqa: E402
from event_bins import load  # noqa: E402

EVENT_DATES = ["2025-08-08", "2025-10-07", "2026-05-21"]


def robust_scale(values):
    if len(values) == 0:
        return 0.0
    values = np.asarray(values)
    mad = float(np.median(np.abs(values - np.median(values)))) * 1.4826
    return mad if mad > 1e-12 else float(np.std(values))


def main() -> None:
    parser = argparse.ArgumentParser(description="单指标全时间轴演化轨迹")
    parser.add_argument("--csv", required=True)
    parser.add_argument("--unit", required=True)
    parser.add_argument("--cylinder", required=True)
    parser.add_argument("--side", required=True)
    parser.add_argument("--bin-start", type=float, required=True)
    parser.add_argument("--metric", required=True)
    parser.add_argument("--gap", type=int, default=10, help="数据段切分的天数间隔")
    parser.add_argument("--event", action="append", default=[], help="事件日期（画竖线）")
    parser.add_argument("--out", default="")
    args = parser.parse_args()

    rows = load(args.csv)
    sel = [r for r in rows if r["unit"] == args.unit and r["cylinder"] == args.cylinder
           and r["side"] == args.side and abs(r["bin_start"] - args.bin_start) < 1e-6
           and r[args.metric] is not None]
    if not sel:
        print("没有匹配数据")
        return
    sel.sort(key=lambda r: r["time"])
    by_day = defaultdict(list)
    for r in sel:
        by_day[r["time"].date()].append(r[args.metric])
    days = sorted(by_day)
    daily = {d: float(np.median(by_day[d])) for d in days}
    print(f"{args.unit}号机 {args.cylinder}缸{args.side} 箱{args.bin_start:g}° {args.metric}"
          f"  周期 {len(sel)}  日 {len(days)}  {days[0]}~{days[-1]}")

    segments = []
    for d in days:
        if segments and (d - segments[-1][-1]).days <= args.gap:
            segments[-1].append(d)
        else:
            segments.append([d])
    all_scale = robust_scale(list(daily.values()))
    for seg in segments:
        y = np.asarray([daily[d] for d in seg])
        span_days = (seg[-1] - seg[0]).days
        line = f"  段 {seg[0]}~{seg[-1]} ({len(seg)}天有数据/{span_days}天)  中位 {np.median(y):.5g}  范围 {y.min():.5g}~{y.max():.5g}"
        if len(seg) >= 4 and span_days > 0:
            x = np.asarray([(d - seg[0]).days for d in seg], dtype=float)
            slope = float(np.polyfit(x, y, 1)[0])
            drift = slope * span_days
            line += f"  趋势 {drift:+.4g} ({drift / all_scale:+.1f}全局MAD)"
        print(line)
    if len(days) >= 6 and (days[-1] - days[0]).days > 0:
        x = np.asarray([(d - days[0]).days for d in days], dtype=float)
        y = np.asarray([daily[d] for d in days])
        # only fit inside the last contiguous-ish run toward the event
        slope = float(np.polyfit(x, y, 1)[0])
        print(f"  全程线性趋势 {slope * (days[-1] - days[0]).days:+.4g} / {all_scale:+.4g} = {(slope * (days[-1] - days[0]).days) / all_scale:+.1f} 全局MAD")

    print("  逐日: " + ", ".join(f"{d}:{daily[d]:.4g}" for d in days))

    if args.out:
        fig, ax = plt.subplots(figsize=(11, 3.6))
        xs = [datetime.combine(d, datetime.min.time()) for d in days]
        ys = [daily[d] for d in days]
        ax.plot(xs, ys, "o-", color="crimson", ms=3, lw=1)
        for ev in (args.event or EVENT_DATES):
            ax.axvline(datetime.fromisoformat(ev), color="k", ls="--", lw=0.8)
            ax.text(datetime.fromisoformat(ev), max(ys), ev, fontsize=7, rotation=90, va="top")
        ax.set_title(f"{args.unit}号机 {args.cylinder}缸{args.side} 箱{args.bin_start:g}° {CN.metric_col(args.metric)} 日度演化")
        ax.grid(alpha=0.25)
        fig.tight_layout()
        Path(args.out).parent.mkdir(parents=True, exist_ok=True)
        fig.savefig(args.out, dpi=110)
        print("图:", args.out)


if __name__ == "__main__":
    main()