Compressor/_tmp_temperature/unified.py

"""Unified cylinder-temperature evolution detector.

One pass over the hourly wide table produces:

* always-on same-stage difference features;
* short-term (1h/3h/6h) same-stage relative rise/fall;
* 7#/8#/9# 三级 high-frequency fluctuation (separate table);
* independent channel anomalies (shutdown / sudden drop / frozen / dead);
* confirmed hourly levels, event segments and one evolution-cycle row per
  (run cycle x stage).

Pressure and vibration are intentionally *not* touched here.
"""
from __future__ import annotations

from collections import defaultdict, deque
from datetime import datetime, timedelta
from statistics import median
import math

import numpy as np

from .algorithm import (
    _assign_cycles,
    _baseline,
    _global_offsets,
    _mad,
    _median,
    _snapshot,
    channel_qc,
    clean_rows,
    linear_slope,
    output_sort_key,
    write_csv,
)
from .config import (
    TEMP_BASELINE_HOURS,
    TEMP_MIN_HISTORY_HOURS,
    TEMP_MIN_STEADY,
    TEMP_SENSOR_MAX,
    TEMP_SENSOR_MIN,
    UnitConfig,
)

SEV_LABELS = {0: "正常", 1: "观察", 2: "轻微", 3: "异常", 4: "严重异常"}
LEVELS = ("观察", "轻微", "异常", "严重异常")
EVENT_LEVELS = ("轻微", "异常", "严重异常")
DIRECTIONS = ("偏高", "偏低", "混合")


def _fmt(value, digits=2):
    return "" if value is None else round(float(value), digits)


def _part_and_point(label, config: UnitConfig) -> tuple[str, str]:
    """Derive the fault-table/3D part label and PKS point from a cylinder label."""
    text = str(label or "").strip()
    if not text:
        return "", ""
    point = next((column for column, name in config.labels.items() if name == text), "")
    part = text
    for suffix in ("排气温度", "温度"):
        if part.endswith(suffix):
            part = part[: -len(suffix)]
            break
    return part, point


def _valid(value):
    return value is not None and TEMP_SENSOR_MIN <= value <= TEMP_SENSOR_MAX and value > 0


def _value_at(history: deque, target: datetime, tolerance_hours: float = 0.6):
    for moment, value in reversed(history):
        delta = abs((moment - target).total_seconds()) / 3600.0
        if delta <= tolerance_hours:
            return value
        if moment < target - timedelta(hours=tolerance_hours + 0.1):
            break
    return None


def _sev_level(value, mild, abnormal, severe) -> int:
    if value is None:
        return 0
    if value >= severe:
        return 4
    if value >= abnormal:
        return 3
    if value >= mild:
        return 2
    return 0


def _sev_rate(value, observe, mild, abnormal, severe) -> int:
    if value is None:
        return 0
    if value >= severe:
        return 4
    if value >= abnormal:
        return 3
    if value >= mild:
        return 2
    if value >= observe:
        return 1
    return 0


def _max_candidate(candidates: list[tuple[int, str, str]]):
    """Pick the highest severity, preferring level over short on ties."""
    candidates = [item for item in candidates if item[0] > 0]
    if not candidates:
        return 0, "", ""
    return max(candidates, key=lambda item: (item[0], 1 if item[1] == "level" else 0))


# ---------------------------------------------------------------------------
# channel anomalies
# ---------------------------------------------------------------------------
def mark_channel_anomalies(rows: list[dict], config: UnitConfig, thresholds: dict) -> list[dict]:
    ambient = thresholds.get("channel_ambient", 45.0)
    drop_delta = thresholds.get("channel_drop_delta", 8.0)
    min_drop = thresholds.get("channel_min_drop", 20.0)
    zero_min = int(thresholds.get("channel_zero_min", 5))
    recover_hours = int(thresholds.get("channel_recover_hours", 3))
    frozen_hours = int(thresholds.get("channel_frozen_hours", 12))
    frozen_eps = thresholds.get("channel_frozen_eps", 0.1)
    peer_variation = thresholds.get("channel_peer_variation", 3.0)
    anomalies: list[dict] = []

    stage_max = []
    for row in rows:
        values = {}
        for stage, columns in config.stages.items():
            nums = [row.get(f"{column}__avg") for column in columns if _valid(row.get(f"{column}__avg"))]
            values[stage] = max(nums) if nums else None
        stage_max.append(values)

    for stage, columns in config.stages.items():
        for column in columns:
            series = [
                (index, row["hour_start"], row.get(f"{column}__avg"))
                for index, row in enumerate(rows)
                if row.get(f"{column}__avg") is not None
                and TEMP_SENSOR_MIN <= row.get(f"{column}__avg") <= TEMP_SENSOR_MAX
            ]
            for position, (index, moment, value) in enumerate(series):
                peak = stage_max[index].get(stage)
                if peak is None or peak < TEMP_MIN_STEADY:
                    continue
                previous = [item[2] for item in series[max(0, position - 6):position]
                            if moment - item[1] <= timedelta(hours=6) and item[2] > 0]
                following = [item[2] for item in series[position + 1:position + 7]
                             if item[1] - moment <= timedelta(hours=6) and item[2] > 0]
                reference_values = previous + following
                reference = median(reference_values) if len(reference_values) >= 3 else None

                next_values = [item[2] for item in series[position + 1:position + 1 + recover_hours]
                               if item[1] - moment <= timedelta(hours=recover_hours)]
                recovers = any(value2 >= max(60.0, (reference or 60.0) - 10) for value2 in next_values)
                minraw = rows[index].get(f"{column}__minraw")
                zeros = rows[index].get(f"{column}__zeros") or 0

                if minraw is not None and minraw <= ambient and zeros >= zero_min and peak >= 60:
                    kind = "停机/通道异常" if recovers else "通道坏道"
                    rows[index][f"{column}__dropout"] = True
                    anomalies.append(_anomaly_record(config, rows[index], column, kind, minraw, reference))
                    continue
                if minraw is not None and reference is not None and minraw < reference - min_drop and recovers and peak >= 60:
                    rows[index][f"{column}__dropout"] = True
                    anomalies.append(_anomaly_record(config, rows[index], column, "通道骤降", minraw, reference))
                    continue

                if value <= ambient and reference is not None and value < reference - 20 and peak >= 60:
                    kind = "停机/通道异常" if recovers else "通道坏道"
                    rows[index][f"{column}__dropout"] = True
                    anomalies.append(_anomaly_record(config, rows[index], column, kind, value, reference))
                elif value > 0 and reference is not None and value < reference - drop_delta:
                    if recovers:
                        rows[index][f"{column}__dropout"] = True
                        anomalies.append(_anomaly_record(config, rows[index], column, "通道骤降", value, reference))
                    # else: persistent relative low -> keep as process low

            # Rows whose hourly average is entirely empty (all samples 0) are
            # not in ``series``; still record them from minraw / zero count.
            for index, row in enumerate(rows):
                if row.get(f"{column}__dropout"):
                    continue
                minraw = row.get(f"{column}__minraw")
                zeros = row.get(f"{column}__zeros") or 0
                peak = stage_max[index].get(stage)
                if minraw is not None and minraw <= ambient and zeros >= zero_min and peak and peak >= 60:
                    row[f"{column}__dropout"] = True
                    anomalies.append(_anomaly_record(config, row, column, "停机/通道异常", minraw, None))

            # frozen channel: identical for >= N hours while peers vary
            run_start = 0
            for position in range(1, len(series) + 1):
                if position == len(series) or abs(series[position][2] - series[position - 1][2]) > frozen_eps:
                    run_length = position - run_start
                    if run_length >= frozen_hours:
                        peer_values = []
                        for item in series[run_start:position]:
                            index = item[0]
                            for peer in columns:
                                if peer == column:
                                    continue
                                peer_value = rows[index].get(f"{peer}__avg")
                                if _valid(peer_value):
                                    peer_values.append(peer_value)
                        if peer_values and max(peer_values) - min(peer_values) >= peer_variation:
                            for item in series[run_start:position]:
                                index = item[0]
                                rows[index][f"{column}__dropout"] = True
                                anomalies.append(_anomaly_record(
                                    config, rows[index], column, "通道冻值", item[2], median(peer_values)
                                ))
                    run_start = position
    return anomalies


def _anomaly_record(config: UnitConfig, row: dict, column: str, kind: str, value, peer) -> dict:
    meta = config.point_meta.get(column, {})
    return {
        "机组": config.name,
        "批次": row["import_batch_id"],
        "点位": column,
        "点位描述": meta.get("label", column),
        "级": meta.get("stage", ""),
        "缸": meta.get("cylinder", ""),
        "异常类型": kind,
        "小时": row["hour_start"],
        "值": value,
        "同级基准": peer,
    }


def build_channel_segments(anomalies: list[dict], thresholds: dict) -> list[dict]:
    group: dict[tuple, list[dict]] = defaultdict(list)
    for item in anomalies:
        group[(item["机组"], item["批次"], item["点位"])].append(item)

    segments = []
    for (unit, batch, column), items in group.items():
        items.sort(key=lambda item: item["小时"])
        current = []
        for item in items:
            if not current or item["小时"] - current[-1]["小时"] <= timedelta(hours=6):
                current.append(item)
            else:
                segments.append(_channel_segment(current, unit, batch, column))
                current = [item]
        if current:
            segments.append(_channel_segment(current, unit, batch, column))
    return sorted(segments, key=output_sort_key)


def _channel_segment(items: list[dict], unit: str, batch: int, column: str) -> dict:
    priority = {"停机/通道异常": 4, "通道坏道": 3, "通道冻值": 2, "通道骤降": 1}
    kind = max({item["异常类型"] for item in items}, key=lambda value: priority.get(value, 0))
    first, last = items[0], items[-1]
    values = [item["值"] for item in items if item["值"] is not None]
    peers = [item["同级基准"] for item in items if item["同级基准"] is not None]
    duration = int((last["小时"] - first["小时"]).total_seconds() / 3600) + 1
    significant = (
        duration >= 2
        or kind in ("通道坏道", "通道冻值")
        or (kind == "停机/通道异常" and values and min(values) <= 0)
    )
    return {
        "机组": unit,
        "批次": batch,
        "点位": column,
        "点位描述": first["点位描述"],
        "级": first["级"],
        "缸": first["缸"],
        "异常类型": kind,
        "开始小时": first["小时"].strftime("%Y-%m-%d %H:%M:%S"),
        "结束小时": last["小时"].strftime("%Y-%m-%d %H:%M:%S"),
        "持续自然小时数": duration,
        "有效小时数": len(items),
        "是否显著": "是" if significant else "否",
        "最小值": _fmt(min(values)) if values else "",
        "最大值": _fmt(max(values)) if values else "",
        "中位数": _fmt(median(values)) if values else "",
        "同级基准中位": _fmt(median(peers)) if peers else "",
        "原因": {
            "停机/通道异常": "单通道突然接近环境温度并恢复，按停机/采集异常处理，不参与温度故障",
            "通道骤降": "单通道短时骤降并恢复，按通道异常处理，不参与同级温差",
            "通道坏道": "单通道长期接近环境温度，按坏道处理",
            "通道冻值": "单通道长期不变而同级其他缸在变化，按冻值处理",
        }.get(kind, ""),
        "_dt": first["小时"],
    }


# ---------------------------------------------------------------------------
# unified features
# ---------------------------------------------------------------------------
def compute_features(rows: list[dict], config: UnitConfig, bad_channels: set[str],
                     thresholds: dict) -> list[dict]:
    valid = list(rows)
    global_spread, global_peer = _global_offsets(valid, config)

    history: dict[str, deque] = {column: deque() for column in config.columns}
    history_peer: dict[str, deque] = {column: deque() for column in config.columns}
    history_std: dict[str, deque] = {column: deque() for column in config.columns}
    history_spread: dict[str, deque] = {stage: deque() for stage in config.stages}
    last_dt: dict[str, datetime | None] = {column: None for column in config.columns}
    rise_count: dict[str, int] = {column: 0 for column in config.columns}

    features: list[dict] = []
    for row in valid:
        dt: datetime = row["hour_start"]
        window_start = dt - timedelta(hours=TEMP_BASELINE_HOURS)
        for values in (list(history.values()) + list(history_peer.values())
                       + list(history_std.values()) + list(history_spread.values())):
            while values and values[0][0] < window_start:
                values.popleft()

        snapshots = {}
        stage_medians = {}
        for stage, columns in config.stages.items():
            usable = [column for column in columns if column not in bad_channels]
            snapshot = _snapshot(row, usable)
            if not snapshot:
                continue
            snapshots[stage] = snapshot
            stage_medians[stage] = median(snapshot[0].values())

        for stage, (values, stage_median, spread, peer) in snapshots.items():
            spread_values = [item[1] for item in history_spread[stage]]
            spread_base = _median(spread_values) if len(spread_values) >= TEMP_MIN_HISTORY_HOURS else global_spread.get(stage)
            spread_delta = spread - spread_base if spread is not None and spread_base is not None else None

            per_column: dict[str, dict] = {}
            for column, value in values.items():
                own = [item[1] for item in history[column]]
                tod = [value for moment, value in history[column] if moment.hour == dt.hour]
                if len(tod) >= 3:
                    base = median(tod)
                    scale = max(1.4826 * _mad(tod, base), abs(base) * 0.005, 0.5)
                else:
                    base, scale = _baseline(own)
                delta = value - base if base is not None else None
                z = delta / scale if delta is not None and scale else None

                peer_values = [item[1] for item in history_peer[column]]
                peer_base = _median(peer_values) if len(peer_values) >= TEMP_MIN_HISTORY_HOURS else global_peer.get(column, 0.0)
                peer_dev = peer[column]
                peer_anom = peer_dev - peer_base if peer_base is not None else None

                change = {}
                for hours in (1, 3, 6):
                    past = _value_at(history[column], dt - timedelta(hours=hours))
                    change[hours] = value - past if past is not None else None

                slope = linear_slope(own[-5:] + [value]) if len(own) >= 2 else None
                if delta is not None and delta > 1.0:
                    rise_count[column] = rise_count[column] + 1 if last_dt[column] and dt - last_dt[column] == timedelta(hours=1) else 1
                else:
                    rise_count[column] = 0

                std_value = row.get(f"{column}__std")
                std_history = [item[1] for item in history_std[column] if item[1] is not None]
                std_base = _median(std_history) if len(std_history) >= TEMP_MIN_HISTORY_HOURS else None
                hf_ratio = (std_value / max(std_base, thresholds["hf_base_floor"])) if (std_value is not None and std_base is not None) else None
                hf_z = None
                if std_value is not None and std_base is not None and len(std_history) >= TEMP_MIN_HISTORY_HOURS:
                    log_values = [math.log(value + 0.01) for value in std_history]
                    log_center = median(log_values)
                    log_scale = max(1.4826 * median([abs(value - log_center) for value in log_values]), 0.1)
                    hf_z = (math.log(std_value + 0.01) - log_center) / log_scale

                recent_values = own[-6:]
                recent_median = median(recent_values) if recent_values else None
                rise_values = [
                    item[1] for item in history[column]
                    if item[0] >= dt - timedelta(hours=thresholds["rise_baseline_hours"])
                ]
                rise_base = median(rise_values) if len(rise_values) >= thresholds["rise_min_history_hours"] else None
                rise_value = value - rise_base if rise_base is not None else None

                per_column[column] = {
                    "value": value, "base": base, "delta": delta, "z": z,
                    "peer_dev": peer_dev, "peer_anom": peer_anom,
                    "d1": change[1], "d3": change[3], "d6": change[6],
                    "slope": slope, "rise": rise_count[column],
                    "std": std_value, "std_base": std_base, "hf_ratio": hf_ratio, "hf_z": hf_z,
                    "recent_median": recent_median, "rise_base": rise_base, "rise_value": rise_value,
                }

            median_rate = {}
            for hours in (1, 3, 6):
                rate_values = [item[f"d{hours}"] for item in per_column.values() if item[f"d{hours}"] is not None]
                median_rate[hours] = median(rate_values) if rate_values else None

            for column, item in per_column.items():
                for hours in (1, 3, 6):
                    peer_rate = item[f"d{hours}"] - median_rate[hours] if item[f"d{hours}"] is not None and median_rate[hours] is not None else None
                    item[f"peer_d{hours}"] = peer_rate

            # ---- grading ----
            deltas = [item["delta"] for item in per_column.values() if item["delta"] is not None]
            zs = [item["z"] for item in per_column.values() if item["z"] is not None]
            peer_high = max((item["peer_anom"] for item in per_column.values()
                             if item["peer_anom"] is not None), default=None)
            min_peer = min((item["peer_anom"] for item in per_column.values()
                            if item["peer_anom"] is not None), default=0.0)
            peer_low = abs(min(0.0, min_peer))
            rate_max = {hours: max((item[f"peer_d{hours}"] for item in per_column.values()
                                    if item[f"peer_d{hours}"] is not None
                                    and item["recent_median"] is not None
                                    and item["value"] > item["recent_median"]), default=None)
                        for hours in (1, 3, 6)}
            rate_min = {hours: min((item[f"peer_d{hours}"] for item in per_column.values()
                                    if item[f"peer_d{hours}"] is not None
                                    and item["recent_median"] is not None
                                    and item["value"] < item["recent_median"]), default=None)
                        for hours in (1, 3, 6)}

            # A drop followed by a rebound (or a rise followed by a fall) is a
            # load/transition/channel artifact, not a one-sided short-term fault.
            rebound = (
                (rate_max[1] is not None and rate_min[3] is not None and rate_max[1] >= 4 and rate_min[3] <= -4)
                or (rate_min[1] is not None and rate_max[3] is not None and rate_min[1] <= -4 and rate_max[3] >= 4)
                or (rate_max[1] is not None and rate_min[6] is not None and rate_max[1] >= 4 and rate_min[6] <= -4)
                or (rate_min[1] is not None and rate_max[6] is not None and rate_min[1] <= -4 and rate_max[6] >= 4)
            )
            if rebound:
                rate_max = {hours: None for hours in (1, 3, 6)}
                rate_min = {hours: None for hours in (1, 3, 6)}

            peak_peer_column = max(
                (column for column in per_column if per_column[column]["peer_anom"] is not None),
                key=lambda column: per_column[column]["peer_anom"], default=None)
            low_peer_column = min(
                (column for column in per_column if per_column[column]["peer_anom"] is not None),
                key=lambda column: per_column[column]["peer_anom"], default=None)
            spread_high_column = max(values, key=values.get) if values else None
            spread_low_column = min(values, key=values.get) if values else None

            hf_columns = [column for column in per_column if per_column[column]["hf_ratio"] is not None]
            hf_column = max(hf_columns, key=lambda column: per_column[column]["hf_ratio"], default=None)
            hf_ratio = per_column[hf_column]["hf_ratio"] if hf_column else None
            hf_std = per_column[hf_column]["std"] if hf_column else None
            hf_base = per_column[hf_column]["std_base"] if hf_column else None
            hf_zs = [per_column[column]["hf_z"] for column in per_column if per_column[column]["hf_z"] is not None]
            hf_z = max(hf_zs) if hf_zs else None
            hf_sev = 0
            if hf_std is not None and hf_std >= thresholds["hf_min_std"]:
                hf_sev = max(
                    _sev_level(hf_ratio, thresholds["hf_ratio_mild"], thresholds["hf_ratio_abnormal"], thresholds["hf_ratio_severe"]),
                    _sev_level(hf_z, thresholds["hf_z_mild"], thresholds["hf_z_abnormal"], thresholds["hf_z_severe"]),
                )
            hf_reason = ""
            if hf_sev:
                label = config.labels.get(hf_column, "") if hf_column else ""
                if hf_ratio is not None and hf_std is not None and hf_base is not None:
                    hf_reason = f"{label}高频波动×{hf_ratio:.1f}（基线{hf_base:.2f}，实测{hf_std:.2f}）"
                elif hf_z is not None:
                    hf_reason = f"{label}高频波动z={hf_z:.1f}"

            rise_values = [item["rise_value"] for item in per_column.values() if item["rise_value"] is not None]
            stage_rise = median(rise_values) if rise_values else None
            relative_rises = {
                column: item["rise_value"] - stage_rise
                for column, item in per_column.items()
                if item["rise_value"] is not None and stage_rise is not None
            }
            if len(values) >= 2:
                rise_single_column = max(relative_rises, key=relative_rises.get, default=None)
                rise_single = relative_rises[rise_single_column] if rise_single_column else None
                rise_stage_value = stage_rise
            else:
                rise_single_column = max(
                    (column for column in per_column if per_column[column]["rise_value"] is not None),
                    key=lambda column: per_column[column]["rise_value"], default=None)
                rise_single = per_column[rise_single_column]["rise_value"] if rise_single_column else None
                rise_stage_value = rise_single

            previous_values = []
            for column in per_column:
                previous = _value_at(history[column], dt - timedelta(hours=1))
                if previous is not None:
                    previous_values.append(previous)
            previous_stage_median = median(previous_values) if previous_values else None
            stage_rising = previous_stage_median is not None and stage_median > previous_stage_median
            rise_flat = stage_rise is not None and abs(stage_rise) <= thresholds["rise_stage_flat_limit"]

            # Only a single-cylinder rise *relative to its stage peers* is a
            # fault candidate.  A whole-stage rise is the normal daily load
            # ramp; it is kept as a feature, not an event.
            if len(values) >= 2:
                rise_single_sev = _sev_level(
                    rise_single, thresholds["rise_single_mild"], thresholds["rise_single_abnormal"], thresholds["rise_single_severe"]
                ) if (stage_rising and rise_flat) else 0
                rise_sev = rise_single_sev
                if rise_sev:
                    rise_type = "单缸抬升"
                    rise_label = config.labels.get(rise_single_column, "") if rise_single_column else ""
                    rise_reason = f"6h整级抬升{rise_stage_value:+.1f}℃；{rise_label}相对同级{rise_single:+.1f}℃"
                else:
                    rise_type = ""
                    rise_label = ""
                    rise_reason = ""
            else:
                rise_sev = 0
                rise_type = ""
                rise_label = ""
                rise_reason = ""

            def peer_reason(column, anomaly, sign):
                if column is None or anomaly is None:
                    return ""
                item = per_column[column]
                dev = item["peer_dev"]
                base = dev - item["peer_anom"]
                label = config.labels.get(column, column)
                return f"{label}较同级中位{dev:+.1f}℃（平时{base:+.1f}℃），较平时{sign}{anomaly:.1f}℃"

            def spread_reason():
                if spread is None or spread_delta is None:
                    return ""
                high = config.labels.get(spread_high_column, "")
                low = config.labels.get(spread_low_column, "")
                base = f"{spread_base:.1f}" if spread_base is not None else "-"
                return f"级差{spread:.1f}℃（{high}-{low}，平时{base}℃）较平时+{spread_delta:.1f}℃"

            if len(values) >= 2:
                high_candidates = [
                    (_sev_level(peer_high, thresholds["mild_peer"], thresholds["abnormal_peer"], thresholds["severe_peer"]), "level", peer_reason(peak_peer_column, peer_high, "+")),
                    (_sev_level(spread_delta, thresholds["mild_spread"], thresholds["abnormal_spread"], thresholds["severe_spread"]), "level", spread_reason()),
                ]
                low_candidates = [
                    (_sev_level(peer_low, thresholds["mild_peer_low"], thresholds["abnormal_peer_low"], thresholds["severe_peer_low"]), "level", peer_reason(low_peer_column, peer_low, "-")),
                ]
                for hours in (1, 3, 6):
                    if rate_max[hours] is not None:
                        high_candidates.append((_sev_rate(rate_max[hours], thresholds[f"observe_rate{hours}"],
                                                          thresholds[f"mild_rate{hours}"], thresholds[f"abnormal_rate{hours}"],
                                                          thresholds[f"severe_rate{hours}"]), "short",
                                                f"{hours}h同级升{rate_max[hours]:.1f}℃"))
                    if rate_min[hours] is not None:
                        low_candidates.append((_sev_rate(-rate_min[hours], thresholds[f"observe_rate{hours}"],
                                                         thresholds[f"mild_rate{hours}"], thresholds[f"abnormal_rate{hours}"],
                                                         thresholds[f"severe_rate{hours}"]), "short",
                                               f"{hours}h同级降{-rate_min[hours]:.1f}℃"))
                high_sev, high_source, high_reason = _max_candidate(high_candidates)
                low_sev, low_source, low_reason = _max_candidate(low_candidates)
            else:
                # Single-cylinder stage: no same-stage peers.  The old
                # cross-stage residual feature is intentionally removed; only
                # high-frequency fluctuation is kept for the tertiary stage.
                high_sev = low_sev = 0
                high_source = low_source = ""
                high_reason = low_reason = ""

            if high_sev and low_sev:
                base_direction, base_severity, base_source = "混合", max(high_sev, low_sev), "混合"
                base_reason = "；".join(item for item in (high_reason, low_reason) if item)
            elif high_sev:
                base_direction, base_severity, base_source, base_reason = "偏高", high_sev, high_source, high_reason
            elif low_sev:
                base_direction, base_severity, base_source, base_reason = "偏低", low_sev, low_source, low_reason
            else:
                base_direction, base_severity, base_source, base_reason = "", 0, "", ""

            direction, severity, source, reason = base_direction, base_severity, base_source, base_reason
            if base_severity == 0 and rise_sev > 0:
                direction, severity, source, reason = "偏高", rise_sev, "rise", rise_reason

            if severity == 0:
                feature_type = ""
            elif source == "hf":
                feature_type = "高频波动"
            elif source == "rise":
                feature_type = rise_type
            elif direction == "混合":
                feature_type = "混合"
            elif source == "short":
                feature_type = "短时突升" if direction == "偏高" else "短时突降"
            else:
                feature_type = "持续偏高" if direction == "偏高" else "持续偏低"

            confidence = "高" if source == "level" else "中"

            strongest = max((column for column in per_column if per_column[column]["delta"] is not None),
                            key=lambda column: per_column[column]["delta"], default=None)
            high_values = [config.limits.get(column, {}).get("high") for column in values]
            high_values = [value for value in high_values if value is not None]
            high_high_values = [config.limits.get(column, {}).get("high_high") for column in values]
            high_high_values = [value for value in high_high_values if value is not None]

            features.append({
                "机组": config.name,
                "批次": row["import_batch_id"],
                "周期编号": row["_cycle_id"],
                "小时": dt.strftime("%Y-%m-%d %H:%M:%S"),
                "级": stage,
                "有效缸数": len(values),
                "运行覆盖率": round(row.get("coverage") or 0.0, 4),
                "最高温度": _fmt(max(values.values())),
                "最低温度": _fmt(min(values.values())),
                "最高温度缸": config.labels.get(max(values, key=values.get), ""),
                "级差": _fmt(spread),
                "级差基线": _fmt(spread_base),
                "级差变化": _fmt(spread_delta),
                "最大自身偏离": _fmt(max(deltas) if deltas else None),
                "最大自身偏离缸": config.labels.get(strongest, "") if strongest else "",
                "最大同级偏离": _fmt(peer_high),
                "最大同级偏离缸": config.labels.get(peak_peer_column, "") if peak_peer_column else "",
                "最大同级偏低": _fmt(peer_low if peer_low else None),
                "最大同级偏低缸": config.labels.get(low_peer_column, "") if low_peer_column else "",
                "最大稳健z": _fmt(max(zs) if zs else None),
                "高频波动倍数": _fmt(hf_ratio),
                "高频波动z": _fmt(hf_z),
                "高频标准差": _fmt(hf_std),
                "高频基线标准差": _fmt(hf_base),
                "高频波动缸": config.labels.get(hf_column, "") if hf_column else "",
                "6小时整级抬升": _fmt(rise_stage_value),
                "6小时单缸抬升": _fmt(rise_single),
                "6小时单缸相对抬升": _fmt(relative_rises.get(rise_single_column)) if rise_single_column else "",
                "6小时基线": _fmt(per_column[rise_single_column]["rise_base"]) if rise_single_column else "",
                "抬升类型": rise_type,
                "抬升缸": rise_label,
                "最强缸": config.labels.get(strongest, "") if strongest else "",
                "最高报警阈值": _fmt(min(high_values)) if high_values else "",
                "高高报警阈值": _fmt(min(high_high_values)) if high_high_values else "",
                "超告警缸": "、".join(
                    config.labels.get(column, column) for column in values
                    if (config.limits.get(column, {}).get("high") is not None
                        and values[column] >= config.limits[column]["high"])
                ),
                "超高高缸": "、".join(
                    config.labels.get(column, column) for column in values
                    if (config.limits.get(column, {}).get("high_high") is not None
                        and values[column] >= config.limits[column]["high_high"])
                ),
                "近6小时斜率": _fmt(per_column[strongest]["slope"], 4) if strongest else "",
                "连续上升小时数": per_column[strongest]["rise"] if strongest else 0,
                "同级相对1h": _fmt(rate_max[1] if rate_max[1] is not None and (rate_max[1] or 0) >= (abs(rate_min[1] or 0)) else rate_min[1]),
                "同级相对3h": _fmt(rate_max[3] if rate_max[3] is not None and (rate_max[3] or 0) >= (abs(rate_min[3] or 0)) else rate_min[3]),
                "同级相对6h": _fmt(rate_max[6] if rate_max[6] is not None and (rate_max[6] or 0) >= (abs(rate_min[6] or 0)) else rate_min[6]),
                "各缸温度": "；".join(f"{config.labels.get(column, column)}={value:.2f}" for column, value in values.items()),
                "各缸自身偏离": "；".join(f"{config.labels.get(column, column)}={per_column[column]['delta']:+.2f}"
                                          for column in values if per_column[column]["delta"] is not None),
                "各缸同级偏离": "；".join(f"{config.labels.get(column, column)}={per_column[column]['peer_dev']:+.2f}"
                                          for column in values),
                "方向": direction,
                "特征类型": feature_type,
                "来源": source,
                "置信度": confidence,
                "原始等级": SEV_LABELS[severity],
                "判定依据": reason,
                "温度等级": "",
                "是否事件": "",
                "首次短时触发": "",
                "首次持续触发": "",
                "关联通道异常": "",
                "_raw_sev": severity,
                "_direction": direction,
                "_source": source,
                "_hf_sev": hf_sev,
                "_hf_ratio": hf_ratio,
                "_hf_z": hf_z,
                "_hf_std": hf_std,
                "_hf_base": hf_base,
                "_hf_column": hf_column,
                "_rise_sev": rise_sev,
                "_rise_type": rise_type,
                "_rise_stage": rise_stage_value,
                "_rise_single": rise_single,
                "_type": feature_type,
                "_confidence": confidence,
                "_dt": dt,
                "_strongest": strongest,
            })

            # append histories after all computations for this stage
            for column, value in values.items():
                history[column].append((dt, value))
                history_peer[column].append((dt, peer[column]))
                history_std[column].append((dt, per_column[column]["std"]))
                last_dt[column] = dt
            if spread is not None:
                history_spread[stage].append((dt, spread))

    return features


# ---------------------------------------------------------------------------
# confirmation / segments / cycles
# ---------------------------------------------------------------------------
def confirm_levels(features: list[dict], thresholds: dict) -> None:
    confirm_hours = int(thresholds.get("confirm_hours", 3))
    event_hours = int(thresholds.get("event_hours", 2))
    hf_persist = int(thresholds.get("hf_persist_hours", 3))
    rise_persist = int(thresholds.get("rise_persist_hours", 2))
    grouped: dict[tuple, list[dict]] = defaultdict(list)
    for row in features:
        grouped[(row["机组"], row["批次"], row["级"])].append(row)

    for rows in grouped.values():
        rows.sort(key=lambda row: row["_dt"])
        index = 0
        while index < len(rows):
            if rows[index]["_raw_sev"] == 0:
                rows[index]["温度等级"] = "正常"
                rows[index]["是否事件"] = "否"
                index += 1
                continue
            end = index
            while end + 1 < len(rows) and rows[end + 1]["_raw_sev"] > 0:
                if rows[end + 1]["_dt"] - rows[end]["_dt"] != timedelta(hours=1):
                    break
                end += 1
            run = rows[index:end + 1]
            run_length = len(run)

            level_rows = [row for row in run if row["_source"] == "level" and row["_raw_sev"] > 0]
            short_rows = [row for row in run if row["_source"] == "short" and row["_raw_sev"] > 0]
            hf_rows = [row for row in run if row["_source"] == "hf" and row["_raw_sev"] > 0]
            rise_rows = [row for row in run if row["_source"] == "rise" and row["_raw_sev"] > 0]
            max_level = max((row["_raw_sev"] for row in level_rows), default=0)
            max_short = max((row["_raw_sev"] for row in short_rows), default=0)
            strong_short = [row for row in short_rows if row["_raw_sev"] >= 2]
            short_streak = best_short_streak = 0
            previous_direction = None
            for row in strong_short:
                if row["_direction"] == previous_direction and previous_direction in ("偏高", "偏低"):
                    short_streak += 1
                else:
                    short_streak = 1
                previous_direction = row["_direction"]
                best_short_streak = max(best_short_streak, short_streak)
            max_hf = max((row["_raw_sev"] for row in hf_rows), default=0)
            strong_hf = [row for row in hf_rows if row["_raw_sev"] >= 2]
            max_rise = max((row["_raw_sev"] for row in rise_rows), default=0)
            strong_rise = [row for row in rise_rows if row["_raw_sev"] >= 2]

            final = 0
            if max_level:
                if run_length >= confirm_hours:
                    final = max(final, max_level)
                elif run_length >= event_hours:
                    if max_level >= 3:
                        final = max(final, 3)
                    elif max_level >= 2:
                        final = max(final, 2)
                elif max_level >= 2:
                    final = max(final, 1)
            if max_short >= 2 and best_short_streak >= 2:
                final = max(final, max_short)
            if max_hf >= 2 and len(strong_hf) >= hf_persist:
                final = max(final, max_hf)
            if max_rise >= 2 and len(strong_rise) >= rise_persist:
                final = max(final, max_rise)
            if final == 0:
                final = 1

            peak = max(run, key=lambda row: (row["_raw_sev"], 1 if row["_source"] == "level" else 0))
            active_directions = {row["_direction"] for row in run if row["_direction"] and row["_raw_sev"] >= 2}
            if "混合" in active_directions or len(active_directions) > 1:
                direction = "混合"
            else:
                direction = peak["_direction"] or ""

            first_short = next((row["小时"] for row in run
                                if row["_source"] == "short" and row["_raw_sev"] >= 2), "")
            first_level = next((row["小时"] for row in run
                                if row["_source"] in ("level", "rise") and row["_raw_sev"] >= 2), "")
            for row in run:
                row["温度等级"] = SEV_LABELS[final]
                row["是否事件"] = "是" if final >= 2 else "否"
                row["方向"] = direction
                row["首次短时触发"] = first_short
                row["首次持续触发"] = first_level
                if final >= 2 and direction == "混合":
                    row["特征类型"] = "混合"
                elif final >= 2 and row["_type"]:
                    row["特征类型"] = row["_type"]
            index = end + 1


def build_observations(features: list[dict], segments: list[dict], config: UnitConfig) -> list[dict]:
    """One row per 1h observation hour, linked to the next event within 24h."""
    events: dict[tuple, list[dict]] = defaultdict(list)
    for segment in segments:
        if segment["是否事件"] == "是":
            events[(segment["机组"], segment["级"])].append(segment)
    rows = []
    for row in features:
        if row["温度等级"] != "观察":
            continue
        candidates = []
        for segment in events.get((row["机组"], row["级"]), []):
            delta = (segment["_dt"] - row["_dt"]).total_seconds() / 3600
            if 0 <= delta <= 24:
                candidates.append(segment)
        event = min(candidates, key=lambda item: item["_dt"]) if candidates else None
        rows.append({
            "机组": row["机组"], "批次": row["批次"], "周期编号": row["周期编号"],
            "小时": row["小时"], "级": row["级"],
            "方向": row["方向"], "特征类型": row["特征类型"], "置信度": row["置信度"],
            "最高温度": row["最高温度"], "级差": row["级差"], "级差变化": row["级差变化"],
            "最大同级偏离": row["最大同级偏离"], "最大同级偏低": row["最大同级偏低"],
            "最大同级相对1h": row["同级相对1h"], "最大同级相对3h": row["同级相对3h"],
            "最大同级相对6h": row["同级相对6h"],
            "判定依据": row["判定依据"],
            "关联周期编号": event["周期编号"] if event else "",
            "关联事件开始": event["开始小时"] if event else "",
            "关联事件等级": event["温度阶段"] if event else "",
            "是否升级为事件": "是" if event else "否",
            "_dt": row["_dt"],
        })
    return sorted(rows, key=output_sort_key)


def build_segments(features: list[dict], config: UnitConfig, channel_segments: list[dict],
                   thresholds: dict, min_hours: int = 1) -> list[dict]:
    channel_by_key: dict[tuple, set[str]] = defaultdict(set)
    for segment in channel_segments:
        channel_by_key[(segment["机组"], segment["级"])].add(segment["异常类型"])

    grouped: dict[tuple, list[dict]] = defaultdict(list)
    for row in features:
        if row["温度等级"] in EVENT_LEVELS:
            grouped[(row["机组"], row["批次"], row["级"], row["温度等级"])].append(row)

    segments = []
    for (unit, batch, stage, level), rows in grouped.items():
        rows.sort(key=lambda row: row["_dt"])
        current = []
        for row in rows:
            if not current or row["_dt"] - current[-1]["_dt"] <= timedelta(hours=thresholds.get("merge_gap_hours", 6)):
                current.append(row)
            else:
                segments.append(_build_segment(current, unit, batch, stage, level, channel_by_key, config))
                current = [row]
        if current:
            segments.append(_build_segment(current, unit, batch, stage, level, channel_by_key, config))

    segments = [segment for segment in segments if segment["有效运行小时数"] >= min_hours]
    return sorted(segments, key=output_sort_key)


def _build_segment(rows: list[dict], unit: str, batch: int, stage: str, level: str,
                   channel_by_key: dict, config: UnitConfig) -> dict:
    first, last = rows[0], rows[-1]
    peak = max(rows, key=lambda row: (row["_raw_sev"], 1 if row["_source"] == "level" else 0))
    peak_label = peak.get("最强缸") or peak.get("最高温度缸") or ""
    spread_rows = [row for row in rows if row["级差"] not in ("", None)]
    spread_peak = max(spread_rows, key=lambda row: row["级差"]) if spread_rows else None
    peer_rows = [row for row in rows if row["最大同级偏离"] not in ("", None)]
    peer_peak = max(peer_rows, key=lambda row: row["最大同级偏离"]) if peer_rows else None
    low_rows = [row for row in rows if row["最大同级偏低"] not in ("", None)]
    low_peak = max(low_rows, key=lambda row: row["最大同级偏低"]) if low_rows else None
    short_times = [row["小时"] for row in rows if row["_source"] == "short" and row["_raw_sev"] >= 2]
    level_times = [row["小时"] for row in rows if row["_source"] == "level" and row["_raw_sev"] >= 2]
    directions = {row["方向"] for row in rows if row["方向"]}
    direction = "混合" if "混合" in directions or len(directions) > 1 else (next(iter(directions), ""))
    return {
        "机组": unit,
        "批次": batch,
        "周期编号": first["周期编号"],
        "级": stage,
        "温度阶段": level,
        "阶段部位": _part_and_point(peak_label, config)[0],
        "阶段点位": _part_and_point(peak_label, config)[1],
        "是否事件": "是" if level in EVENT_LEVELS else "否",
        "是否形成确认等级": "是" if level in ("异常", "严重异常") else "否",
        "方向": direction or peak["_direction"],
        "特征类型": "混合" if direction == "混合" else peak["_type"],
        "置信度": peak["_confidence"],
        "开始小时": first["小时"],
        "结束小时": last["小时"],
        "持续自然小时数": int((last["_dt"] - first["_dt"]).total_seconds() / 3600) + 1,
        "有效运行小时数": len(rows),
        "峰值时刻": peak["小时"],
        "峰值缸": peak_label,
        "最高温度": max((row["最高温度"] for row in rows if row["最高温度"] not in ("", None)), default=""),
        "最大自身偏离": peak["最大自身偏离"],
        "最大级差": spread_peak["级差"] if spread_peak else "",
        "最大级差基线": spread_peak["级差基线"] if spread_peak else "",
        "最大级差变化": spread_peak["级差变化"] if spread_peak else "",
        "最大同级偏离": peer_peak["最大同级偏离"] if peer_peak else "",
        "最大同级偏低": low_peak["最大同级偏低"] if low_peak else "",
        "最大稳健z": peak["最大稳健z"],
        "最大同级相对1h": max((row["同级相对1h"] for row in rows if row["同级相对1h"] not in ("", None)), default=""),
        "最大同级相对3h": max((row["同级相对3h"] for row in rows if row["同级相对3h"] not in ("", None)), default=""),
        "最大同级相对6h": max((row["同级相对6h"] for row in rows if row["同级相对6h"] not in ("", None)), default=""),
        "首次短时触发": min(short_times) if short_times else "",
        "首次持续触发": min(level_times) if level_times else "",
        "阶段最高报警阈值": min((row["最高报警阈值"] for row in rows if row["最高报警阈值"] not in ("", None)), default=""),
        "阶段高高报警阈值": min((row["高高报警阈值"] for row in rows if row["高高报警阈值"] not in ("", None)), default=""),
        "阶段高报警小时数": sum(1 for row in rows if row.get("超告警缸") not in ("", None)),
        "阶段高高报警小时数": sum(1 for row in rows if row.get("超高高缸") not in ("", None)),
        "关联通道异常": "、".join(sorted(channel_by_key.get((unit, stage), set()))),
        "阶段触发依据": "；".join(
            list(dict.fromkeys(
                row["判定依据"] for row in rows if row["判定依据"]
            ))
        [:3]),
        "_dt": first["_dt"],
    }


def build_hf_segments(features: list[dict], config: UnitConfig, thresholds: dict) -> list[dict]:
    """High-frequency fluctuation events, tertiary stage only."""
    hf_persist = int(thresholds.get("hf_persist_hours", 3))
    rise_persist = int(thresholds.get("rise_persist_hours", 2))
    grouped: dict[tuple, list[dict]] = defaultdict(list)
    for row in features:
        if row["级"] != "三级" or row.get("_hf_sev", 0) <= 0:
            continue
        grouped[(row["机组"], row["批次"], row["级"])].append(row)

    raw_segments = []
    for (unit, batch, stage), rows in grouped.items():
        rows.sort(key=lambda row: row["_dt"])
        current = []
        for row in rows:
            if not current or row["_dt"] - current[-1]["_dt"] == timedelta(hours=1):
                current.append(row)
            else:
                if len(current) >= hf_persist:
                    raw_segments.append(_build_hf_segment(current, unit, batch, stage, config))
                current = [row]
        if len(current) >= hf_persist:
            raw_segments.append(_build_hf_segment(current, unit, batch, stage, config))

    raw_segments.sort(key=lambda row: (row["机组"], row["批次"], row["_level"], row["_dt"]))
    merged = []
    for segment in raw_segments:
        if (merged and merged[-1]["机组"] == segment["机组"] and merged[-1]["批次"] == segment["批次"]
                and merged[-1]["_level"] == segment["_level"]
                and segment["_dt"] - merged[-1]["_dt"] <= timedelta(hours=6)):
            merged[-1]["结束小时"] = segment["结束小时"]
            merged[-1]["有效运行小时数"] += segment["有效运行小时数"]
            merged[-1]["持续自然小时数"] += segment["持续自然小时数"]
            merged[-1]["最高温度"] = max(merged[-1]["最高温度"], segment["最高温度"])
            merged[-1]["高频波动倍数"] = max(merged[-1]["高频波动倍数"], segment["高频波动倍数"])
            merged[-1]["高频波动z"] = max(merged[-1]["高频波动z"], segment["高频波动z"])
            merged[-1]["高频标准差"] = max(merged[-1]["高频标准差"], segment["高频标准差"])
            merged[-1]["阶段触发依据"] = segment["阶段触发依据"]
        else:
            merged.append(segment)
    return sorted(merged, key=output_sort_key)


def _build_hf_segment(rows: list[dict], unit: str, batch: int, stage: str, config: UnitConfig) -> dict:
    peak = max(rows, key=lambda row: (row["_hf_sev"], row.get("_hf_ratio") or 0))
    level = peak["_hf_sev"]
    column = peak.get("_hf_column") or ""
    label = config.labels.get(column, "")
    ratio = peak.get("_hf_ratio") or 0
    z_value = peak.get("_hf_z")
    std_value = peak.get("_hf_std") or 0
    base_value = peak.get("_hf_base") or 0
    reason = f"{label}高频波动×{ratio:.1f}（基线{base_value:.2f}，实测{std_value:.2f}）"
    if z_value is not None:
        reason += f"，z={z_value:.1f}"
    return {
        "机组": unit,
        "批次": batch,
        "周期编号": rows[0]["周期编号"],
        "级": stage,
        "温度阶段": SEV_LABELS[level],
        "阶段部位": _part_and_point(label, config)[0],
        "阶段点位": _part_and_point(label, config)[1],
        "是否事件": "是",
        "是否形成确认等级": "是" if SEV_LABELS[level] in ("异常", "严重异常") else "否",
        "方向": "波动",
        "特征类型": "高频波动",
        "置信度": "中",
        "开始小时": rows[0]["小时"],
        "结束小时": rows[-1]["小时"],
        "持续自然小时数": int((rows[-1]["_dt"] - rows[0]["_dt"]).total_seconds() / 3600) + 1,
        "有效运行小时数": len(rows),
        "峰值时刻": peak["小时"],
        "峰值缸": label,
        "最高温度": max((row["最高温度"] for row in rows if row["最高温度"] not in ("", None)), default=""),
        "高频波动倍数": round(ratio, 2),
        "高频波动z": round(z_value, 2) if z_value is not None else "",
        "高频标准差": round(std_value, 2),
        "高频基线标准差": round(base_value, 2),
        "高频波动缸": label,
        "阶段触发依据": reason,
        "_dt": rows[0]["_dt"],
        "_level": level,
    }


def summarize_cycles(features: list[dict], segments: list[dict], config: UnitConfig) -> list[dict]:
    grouped: dict[tuple, list[dict]] = defaultdict(list)
    for row in features:
        grouped[(row["机组"], row["批次"], row["周期编号"], row["级"])].append(row)

    result = []
    for (unit, batch, cycle_id, stage), rows in grouped.items():
        rows.sort(key=lambda row: row["_dt"])
        stage_segments = [segment for segment in segments
                          if segment["周期编号"] == cycle_id and segment["级"] == stage]
        sequence = []
        previous = None
        for row in rows:
            label = row["温度等级"]
            if label != previous:
                sequence.append(label)
                previous = label
        if len(sequence) > 1 and sequence[-1] == "正常":
            sequence[-1] = "恢复"
        peak = max(rows, key=lambda row: (row["_raw_sev"], 1 if row["_source"] == "level" else 0))
        summary = {
            "机组": unit,
            "批次": batch,
            "周期编号": cycle_id,
            "级": stage,
            "周期开始小时": rows[0]["小时"],
            "周期结束小时": rows[-1]["小时"],
            "有效运行小时数": len(rows),
            "阶段演化顺序": "→".join(sequence) if sequence else "正常",
            "方向": peak["方向"] if peak["_raw_sev"] else "",
            "特征类型": peak["_type"] if peak["_raw_sev"] else "",
            "置信度": peak["置信度"] if peak["_raw_sev"] else "",
            "峰值时刻": peak["小时"] if peak["_raw_sev"] else "",
            "峰值缸": peak["最强缸"] if peak["_raw_sev"] else "",
            "首次短时触发": min((row["首次短时触发"] for row in rows if row["首次短时触发"]), default=""),
            "首次持续触发": min((row["首次持续触发"] for row in rows if row["首次持续触发"]), default=""),
            "关联通道异常": "、".join(sorted({segment["关联通道异常"] for segment in stage_segments if segment["关联通道异常"]})),
        }
        for level in LEVELS:
            selected = [segment for segment in stage_segments if segment["温度阶段"] == level]
            summary[f"{level}阶段开始小时"] = selected[0]["开始小时"] if selected else ""
            summary[f"{level}阶段结束小时"] = selected[-1]["结束小时"] if selected else ""
            summary[f"{level}阶段运行小时数"] = sum(segment["有效运行小时数"] for segment in selected)
            summary[f"是否达到{level}"] = "是" if selected else "否"
        summary["周期最高温度"] = max((row["最高温度"] for row in rows if row["最高温度"] not in ("", None)), default="")
        summary["周期最大级差"] = max((row["级差"] for row in rows if row["级差"] not in ("", None)), default="")
        summary["周期最大级差变化"] = max((row["级差变化"] for row in rows if row["级差变化"] not in ("", None)), default="")
        summary["周期最大同级偏离"] = max((row["最大同级偏离"] for row in rows if row["最大同级偏离"] not in ("", None)), default="")
        summary["周期最大同级偏低"] = max((row["最大同级偏低"] for row in rows if row["最大同级偏低"] not in ("", None)), default="")
        result.append(summary)
    return result


# ---------------------------------------------------------------------------
# outputs
# ---------------------------------------------------------------------------
FEATURE_FIELDS = [
    "机组", "批次", "周期编号", "小时", "级", "有效缸数", "运行覆盖率",
    "最高温度", "最低温度", "最高温度缸", "最高报警阈值", "高高报警阈值", "超告警缸", "超高高缸",
    "级差", "级差基线", "级差变化",
    "最大自身偏离", "最大自身偏离缸", "最大同级偏离", "最大同级偏离缸",
    "最大同级偏低", "最大同级偏低缸", "最大稳健z",
    "高频波动倍数", "高频波动z", "高频标准差", "高频基线标准差", "高频波动缸",
    "6小时整级抬升", "6小时单缸抬升", "6小时单缸相对抬升", "6小时基线", "抬升类型", "抬升缸", "最强缸",
    "近6小时斜率", "连续上升小时数", "同级相对1h", "同级相对3h", "同级相对6h",
    "各缸温度", "各缸自身偏离", "各缸同级偏离",
    "方向", "特征类型", "来源", "置信度",
    "温度等级", "原始等级", "是否事件", "首次短时触发", "首次持续触发", "关联通道异常", "判定依据",
]

CANONICAL_FIELDS = [
    "机组", "批次", "周期编号", "级", "温度阶段", "阶段部位", "阶段点位",
    "开始小时", "结束小时", "持续自然小时数", "有效运行小时数",
    "峰值时刻", "峰值缸", "最高温度",
    "最大自身偏离", "最大级差", "最大级差基线", "最大级差变化",
    "最大同级偏离", "最大同级偏低", "最大稳健z",
    "最大同级相对1h", "最大同级相对3h", "最大同级相对6h",
    "首次短时触发", "首次持续触发",
    "阶段最高报警阈值", "阶段高高报警阈值",
    "阶段高报警小时数", "阶段高高报警小时数",
    "方向", "特征类型", "置信度",
    "高频波动倍数", "高频波动z", "高频标准差", "高频基线标准差", "高频波动缸",
    "去趋势残差标准差", "每小时反复升降次数",
    "是否形成确认等级", "关联通道异常", "阶段触发依据",
]

SEGMENT_FIELDS = CANONICAL_FIELDS
HF_FIELDS = CANONICAL_FIELDS
HF_CANDIDATE_FIELDS = CANONICAL_FIELDS + ["是否确认"]

CYCLE_FIELDS = [
    "机组", "批次", "周期编号", "级", "周期开始小时", "周期结束小时", "有效运行小时数",
    "阶段演化顺序", "方向", "特征类型", "置信度", "峰值时刻", "峰值缸",
    "首次短时触发", "首次持续触发", "关联通道异常",
    "观察阶段开始小时", "观察阶段结束小时", "观察阶段运行小时数", "是否达到观察",
    "轻微阶段开始小时", "轻微阶段结束小时", "轻微阶段运行小时数", "是否达到轻微",
    "异常阶段开始小时", "异常阶段结束小时", "异常阶段运行小时数", "是否达到异常",
    "严重异常阶段开始小时", "严重异常阶段结束小时", "严重异常阶段运行小时数", "是否达到严重异常",
    "周期最高温度", "周期最大级差", "周期最大级差变化", "周期最大同级偏离", "周期最大同级偏低",
]

CHANNEL_FIELDS = [
    "机组", "批次", "点位", "点位描述", "级", "缸", "异常类型",
    "开始小时", "结束小时", "持续自然小时数", "有效小时数", "是否显著",
    "最小值", "最大值", "中位数", "同级基准中位", "原因",
]

QC_FIELDS = [
    "机组", "点位", "点位描述", "运行小时数", "最小值", "最大值", "中位数",
    "不同值个数", "最长不变小时", "零值比例", "越界比例", "判定", "原因",
]

SAME_STAGE_FIELDS = [
    "机组", "批次", "周期编号", "小时", "级", "有效缸数",
    "级差", "级差基线", "级差变化", "最大同级偏离", "最大同级偏低",
    "各缸温度", "各缸同级偏离",
]

SHORT_FIELDS = [
    "机组", "批次", "周期编号", "小时", "级",
    "同级相对1h", "同级相对3h", "同级相对6h", "方向", "最强缸",
]

OBSERVATION_FIELDS = [
    "机组", "批次", "周期编号", "小时", "级", "方向", "特征类型", "置信度",
    "最高温度", "级差", "级差变化", "最大同级偏离", "最大同级偏低",
    "最大同级相对1h", "最大同级相对3h", "最大同级相对6h",
    "判定依据", "关联周期编号", "关联事件开始", "关联事件等级", "是否升级为事件",
]



def write_results(output_dir, qc, features, segments, cycles, channel_segments, observations,
                  hf_segments, hf_candidates=None):
    output_dir.mkdir(parents=True, exist_ok=True)
    candidates = hf_candidates if hf_candidates is not None else hf_segments
    canonical = clean_rows(segments) + clean_rows(hf_segments)
    canonical.sort(key=output_sort_key)
    canonical_hf = [row for row in canonical if row.get("特征类型") == "高频波动"]
    write_csv(output_dir / "温度小时特征.csv", clean_rows(features), FEATURE_FIELDS, {field: field for field in FEATURE_FIELDS})
    write_csv(output_dir / "温度异常段.csv", canonical, CANONICAL_FIELDS, {field: field for field in CANONICAL_FIELDS})
    write_csv(output_dir / "温度观察段.csv", clean_rows(observations), OBSERVATION_FIELDS, {field: field for field in OBSERVATION_FIELDS})
    write_csv(output_dir / "温度高频波动段.csv", canonical_hf, HF_FIELDS, {field: field for field in HF_FIELDS})
    write_csv(output_dir / "温度高频候选.csv", clean_rows(candidates), HF_CANDIDATE_FIELDS, {field: field for field in HF_CANDIDATE_FIELDS})
    write_csv(output_dir / "温度演化周期.csv", clean_rows(cycles), CYCLE_FIELDS, {field: field for field in CYCLE_FIELDS})
    write_csv(output_dir / "温度通道异常段.csv", clean_rows(channel_segments), CHANNEL_FIELDS, {field: field for field in CHANNEL_FIELDS})
    write_csv(output_dir / "温度通道质控.csv", clean_rows(qc), QC_FIELDS, {field: field for field in QC_FIELDS})

    same_stage = [{**row} for row in features]
    write_csv(output_dir / "同级温差特征.csv", clean_rows(same_stage), SAME_STAGE_FIELDS, {field: field for field in SAME_STAGE_FIELDS})
    short = [{**row} for row in features]
    write_csv(output_dir / "温度短时特征.csv", clean_rows(short), SHORT_FIELDS, {field: field for field in SHORT_FIELDS})


def run(rows: list[dict], config: UnitConfig, thresholds: dict | None = None, min_hours: int = 1):
    merged = {**config.thresholds, **(thresholds or {})}
    valid = [row for row in rows if row.get("valid")]
    bad_channels, qc = channel_qc(valid, config)
    anomalies = mark_channel_anomalies(valid, config, merged)
    channel_segments = build_channel_segments(anomalies, merged)
    valid = _assign_cycles(valid, config)
    features = compute_features(valid, config, bad_channels, merged)
    confirm_levels(features, merged)
    segments = build_segments(features, config, channel_segments, merged, min_hours)
    observations = build_observations(features, segments, config)
    hf_segments = build_hf_segments(features, config, merged)
    cycles = summarize_cycles(features, segments, config)
    return qc, features, segments, cycles, channel_segments, observations, hf_segments


__all__ = ["run", "write_results"]