Compressor/models/generalized/features_pressure.py

#!/usr/bin/env python3
"""Unit-relative (generalized) pressure features for cross-unit PKS models.

Absolute oil-pressure levels differ between units, so a shared model trained on
raw values treats one unit's normal level as another unit's anomaly.  These
features express each window relative to its own window and its own unit's
normal baseline (median/MAD), so the same model can be shared across units.

Baselines must be computed from training-split normal windows only.
"""
from __future__ import annotations

from collections import defaultdict
from statistics import median

MIN_BASELINE_SAMPLES = 24

# Absolute oil-pressure features dropped from the generalized feature set.
ABSOLUTE_FEATURES = ("oil_mean", "oil_min", "oil_max", "oil_range", "oil_slope")

# Fixed feature order for the shared model.
FEATURE_NAMES = (
    "oil_last_relative_to_window",
    "oil_range_rel",
    "oil_slope_rel",
    "oil_min_dev",
    "oil_max_dev",
    "oil_low_ratio_mean",
    "oil_low_low_ratio_mean",
    "running_coverage_mean",
    "running_coverage_min",
    "oil_rel_cycle_mean",
    "oil_rel_cycle_min",
    "oil_rel_cycle_last",
    "oil_rel_recent_mean",
    "oil_rel_recent_last",
    "oil_slope_rel_cycle",
    "oil_decline_hours",
    "oil_drawdown_cycle",
    "oil_range_rel_cycle",
    "oil_mean_z",
    "oil_min_z",
    "oil_last_z",
    "oil_slope_z",
)

# Features computed in build_datasets from the run cycle and passed through as-is.
CYCLE_RELATIVE_FEATURES = (
    "oil_rel_cycle_mean",
    "oil_rel_cycle_min",
    "oil_rel_cycle_last",
    "oil_rel_recent_mean",
    "oil_rel_recent_last",
    "oil_slope_rel_cycle",
    "oil_decline_hours",
    "oil_drawdown_cycle",
    "oil_range_rel_cycle",
)


def feature_names() -> list[str]:
    return list(FEATURE_NAMES)


def _safe_ratio(numerator, denominator):
    if numerator is None or denominator is None or denominator == 0:
        return None
    return numerator / denominator


def _last_hour_oil(sample: dict):
    hours = sample.get("source_hours") or []
    if not hours:
        return None
    return hours[-1].get("oil_avg")


def compute_unit_baselines(samples: list[dict]) -> dict[str, dict]:
    """Compute per-unit normal oil-pressure median/MAD from training normal windows.

    Raises if a unit has fewer than MIN_BASELINE_SAMPLES training normal windows
    or a degenerate MAD, because the normalization would be unstable.
    """
    by_unit = defaultdict(list)
    for sample in samples:
        if sample["label"] != 0:
            continue
        value = sample["features"].get("oil_mean")
        if value is not None:
            by_unit[sample["unit"]].append(value)
    if not by_unit:
        raise RuntimeError("训练集没有正常窗口，无法计算机组基准")
    baselines = {}
    for unit, values in sorted(by_unit.items()):
        if len(values) < MIN_BASELINE_SAMPLES:
            raise RuntimeError(
                f"{unit} 训练正常窗口仅 {len(values)} 个，少于 {MIN_BASELINE_SAMPLES} 个，无法计算稳定基准"
            )
        center = float(median(values))
        mad = float(median([abs(value - center) for value in values]))
        if mad <= 0:
            raise RuntimeError(f"{unit} 正常油压 MAD=0，无法归一化")
        baselines[unit] = {"median": center, "mad": mad, "normal_samples": len(values)}
    return baselines


def generalized_features(sample: dict, baselines: dict[str, dict]) -> dict:
    features = sample["features"]
    unit = sample["unit"]
    if unit not in baselines:
        raise KeyError(f"缺少 {unit} 的机组基准")
    center = baselines[unit]["median"]
    mad = baselines[unit]["mad"]
    oil_mean = features.get("oil_mean")
    oil_min = features.get("oil_min")
    oil_max = features.get("oil_max")
    oil_range = features.get("oil_range")
    oil_slope = features.get("oil_slope")
    last_oil = _last_hour_oil(sample)
    result = {
        "oil_last_relative_to_window": features.get("oil_last_relative_to_window"),
        "oil_range_rel": _safe_ratio(oil_range, oil_mean),
        "oil_slope_rel": _safe_ratio(oil_slope, oil_mean),
        "oil_min_dev": _safe_ratio(oil_min - oil_mean, oil_mean) if oil_min is not None and oil_mean is not None else None,
        "oil_max_dev": _safe_ratio(oil_max - oil_mean, oil_mean) if oil_max is not None and oil_mean is not None else None,
        "oil_low_ratio_mean": features.get("oil_low_ratio_mean"),
        "oil_low_low_ratio_mean": features.get("oil_low_low_ratio_mean"),
        "running_coverage_mean": features.get("running_coverage_mean"),
        "running_coverage_min": features.get("running_coverage_min"),
        "oil_mean_z": (oil_mean - center) / mad if oil_mean is not None else None,
        "oil_min_z": (oil_min - center) / mad if oil_min is not None else None,
        "oil_last_z": (last_oil - center) / mad if last_oil is not None else None,
        "oil_slope_z": oil_slope / mad if oil_slope is not None else None,
    }
    for name in CYCLE_RELATIVE_FEATURES:
        result[name] = features.get(name)
    return result


def feature_matrix(samples: list[dict], baselines: dict[str, dict]) -> list[dict]:
    return [generalized_features(sample, baselines) for sample in samples]