Compressor/backend/indicator_features/summary_features.py

#!/usr/bin/env python3
"""Aggregate the per-event change tables into one ranked summary.

Reads one or more ``event_bins.py`` detail CSVs and groups the per-bin and
whole-cycle changes by ``(event, point, bin)`` so several metrics that move in
the same bin are seen as one candidate.  Ranks by the summed |MAD| and prints a
compact metric list per candidate.  This is the table to hand to later
summarisation.
"""
from __future__ import annotations

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

VOLATILE = ("std", "mad", "range", "roughness", "d_abs_mean", "d_abs_max", "d_rms",
            "d2_rms", "zero_cross", "n_extrema", "d2_abs_max", "tv", "mean_cross",
            "mobility", "complexity")


def to_float(value):
    try:
        return float(value)
    except (TypeError, ValueError):
        return None


def main() -> None:
    parser = argparse.ArgumentParser(description="事件候选汇总（跨指标合并）")
    parser.add_argument("--csv", nargs="+", required=True)
    parser.add_argument("--threshold", type=float, default=2.0, help="计入的 |MAD| 下限")
    parser.add_argument("--max-gap", type=int, default=0, help="只保留前后段相隔 <= N 天的（0=不限）")
    parser.add_argument("--max-pre-span", type=int, default=0, help="只保留前段跨度 <= N 天的（0=不限）")
    parser.add_argument("--top", type=int, default=40)
    parser.add_argument("--out", default="")
    args = parser.parse_args()

    groups = defaultdict(list)
    for path in args.csv:
        for row in csv.DictReader(open(path, encoding="utf-8-sig")):
            if row.get("family") not in ("bin", "cycle"):
                continue
            mad = to_float(row.get("mad"))
            if mad is None or abs(mad) < args.threshold:
                continue
            bin_start = row.get("bin_start") or ""
            key = (row["unit"], row["date"], row["kind"], row["family"], row["point"], bin_start)
            groups[key].append({
                "metric": row["metric"], "type": row.get("type", ""), "mad": mad,
                "delta": to_float(row.get("delta")), "rel": to_float(row.get("rel")),
                "pre": to_float(row.get("pre")), "post": to_float(row.get("post")),
                "candidate": row.get("candidate") == "1",
                "gap_days": to_float(row.get("gap_days")),
                "pre_span_days": to_float(row.get("pre_span_days")),
                "pre_end": row.get("pre_end", ""), "post_start": row.get("post_start", ""),
            })

    records = []
    for key, items in groups.items():
        unit, date, kind, family, point, bin_start = key
        gap = items[0]["gap_days"]
        pre_span = items[0].get("pre_span_days")
        if args.max_gap and (gap is None or gap > args.max_gap):
            continue
        if args.max_pre_span and (pre_span is None or pre_span > args.max_pre_span):
            continue
        score = sum(abs(i["mad"]) for i in items)
        strongest = max(items, key=lambda i: abs(i["mad"]))
        if bin_start in ("", None):
            scope = "整周期"
        elif float(bin_start) < 0:
            scope = "整体(全角度)"
        else:
            scope = "局部(单角度箱)"
        records.append({
            "unit": unit, "date": date, "kind": kind, "family": family, "point": point,
            "bin_start": bin_start, "scope": scope,
            "n_metrics": len(items), "score": score,
            "max_mad": abs(strongest["mad"]), "candidate": any(i["candidate"] for i in items),
            "gap_days": gap, "pre_span_days": pre_span,
            "pre_end": items[0]["pre_end"], "post_start": items[0]["post_start"],
            "items": items,
        })
    records.sort(key=lambda r: -r["score"])

    print(f"候选组 {len(records)}（每个 = 事件×测点×角度箱/整周期，跨指标合并；|MAD|>={args.threshold}"
          + (f"；前后相隔<={args.max_gap}天" if args.max_gap else "") + "）")
    print(f"{'#':>3} {'机组':<4} {'事件':<16} {'测点':<9} {'角度箱':<10} {'范围':<10} {'指标数':>5} {'总|MAD|':>8} {'相隔':>4} {'前跨度':>5} 指标（前→后方向）")
    out_rows = []
    for i, r in enumerate(records[:args.top], 1):
        binst = "整周期" if r["bin_start"] == "" else (
            f"{float(r['bin_start']):g}-{float(r['bin_start']) + 10:g}°")
        parts = []
        for it in sorted(r["items"], key=lambda x: -abs(x["mad"])):
            arrow = "↑" if (it["delta"] or 0) > 0 else "↓"
            parts.append(f"{it['metric']}{arrow}{abs(it['mad']):.0f}")
        summary = ", ".join(parts[:8])
        tag = "★" if r["candidate"] else ""
        gap = "" if r["gap_days"] is None else f"{int(r['gap_days'])}"
        pspan = "" if r["pre_span_days"] is None else f"{int(r['pre_span_days'])}"
        print(f"{i:>3} {r['unit']+'号':<4} {r['date'][5:]+' '+r['kind']:<16} {r['point']:<9} {binst:<10} "
              f"{r['scope']:<10} {r['n_metrics']:>5} {r['score']:>8.0f} {gap:>4} {pspan:>5} {summary} {tag}")
        out_rows.append({
            "unit": r["unit"], "date": r["date"], "kind": r["kind"], "family": r["family"],
            "point": r["point"], "bin_start": r["bin_start"], "scope": r["scope"],
            "n_metrics": r["n_metrics"],
            "score": round(r["score"], 2), "max_mad": round(r["max_mad"], 2),
            "gap_days": r["gap_days"], "pre_span_days": r["pre_span_days"],
            "pre_end": r["pre_end"], "post_start": r["post_start"],
            "candidate": int(r["candidate"]), "metrics": ";".join(
                f"{it['metric']}{'up' if (it['delta'] or 0) > 0 else 'dn'}{abs(it['mad']):.1f}"
                for it in sorted(r["items"], key=lambda x: -abs(x["mad"]))),
        })

    if args.out and out_rows:
        out = Path(args.out)
        out.parent.mkdir(parents=True, exist_ok=True)
        with out.open("w", encoding="utf-8-sig", newline="") as handle:
            writer = csv.DictWriter(handle, fieldnames=list(out_rows[0].keys()))
            writer.writeheader()
            writer.writerows(out_rows)
        print(f"\n汇总 -> {out}（{len(out_rows)} 组）")


if __name__ == "__main__":
    main()