# -*- coding: utf-8 -*- """异常检测模型模板化（固定主干 + 配方加载）。 对应 issue #37（父 EPIC #5「③ AI 模型框架 配置化重构」、PRD 5.3 「网络结构策略 / 模板化技术路径」）。 PRD 5.3 的核心诉求 ------------------ 异常检测属于 PRD 5.3「四类模型模板」之一（③ 异常检测），同样采用 「**固定主干 + 可配置超参**」默认模式：同一主干代码不变，切换行业 / 工况只改 *配方（recipe）* —— 一个声明式 JSON 超参包。本模块与 ``quality_forecast``（issue #36）同源，共享「主干工厂 + Recipe + 验收口径」 骨架，但任务语义是无监督异常检测： * **输入**：多维工艺特征时序点（无需标注，无监督）； * **输出**：每个样本的异常分数（越大越异常）+ 二值异常标签（由阈值决定）； * **验收**：检出率 / 误报率 / F1（PRD 5.3 / 第 6 章里程碑：关键异常检出率 ≥ 95%、误报率 ≤ 5%）。 本模块交付什么 -------------- 1. **``AnomalyDetectionModel``**：固定主干的异常检测模型。默认主干是 ``iforest``（隔离森林，PRD 5.3 推荐的无监督异常检测默认结构）；当运行 环境存在 ``sklearn`` 时自动升级为真实实现，否则退化为确定性 stub， 保证边缘 / 离线 / CI 环境可加载与校验——与 issue #34 / #36 的 「numpy/sklearn 可选」策略一致。 2. **``Recipe`` 配方加载器**：声明式 JSON 超参包（``load_recipe`` / ``build_from_recipe``）。配方描述「主干类型 + 超参 + 特征列 + 阈值策略 + 验收口径」，业务侧只 ``build_from_recipe(path)`` 一行即可拿到一个 可训练 / 可推理的异常检测模型——切换模板仅改配方，模型代码零改动。 3. **``Metrics`` 验收口径**：PRD 5.3 / 第 6 章里程碑要求「关键异常检出率 ≥ 95%、误报率 ≤ 5%」。``evaluate`` 直接给出检出率 / 误报率 / 精确率 / 召回率 / F1，便于配置台与 UAT 直接读取。 4. **样例配方（``samples/`` JSON）**：Ti（海绵钛氯化车间炉层杂质预警）+ 树脂两套异常检测超参包样例，验证「同框架加载两套配方均跑通」的验收 口径。 与 issue #34 ``model_recipe`` / #36 ``quality_forecast`` 的关系 -------------------------------------------------------------- 接口风格对齐 #34 的 ``ModelHandle`` / ``ModelRecipe``（``fit`` / ``decision_function`` / ``to_dict``、不可变声明式数据对象），以及 #36 的「主干工厂注册表 + Recipe + 验收口径」骨架。本模块**自包含、不依赖 #34 / #36 未合并分支**，待二者合入后，异常检测主干可平滑注册为 ``register_backbone("iforest", ...)`` 的一个具名主干，配方可映射为一条 ``ModelRecipe``——届时本模块零业务侧改动。 零外部强依赖 ------------ * 主干默认走纯 Python stub（``StubBackbone``）：无 sklearn 时也能加载、 构造、（伪）拟合与打分，保证 CI 可加载与校验； * 存在 ``sklearn`` 时，``iforest`` 主干自动升级为真实 ``IsolationForest`` 实现，其余情况退化为 stub，不影响接口契约与测试。 """ from __future__ import annotations import json import math import os from dataclasses import dataclass, field from typing import Any, Dict, List, Optional, Sequence, Tuple __all__ = [ # 数据对象 "Recipe", "Metrics", "AnomalyDetectionError", # 模型 "AnomalyDetectionModel", "ModelHandle", # 主干工厂 "BACKBONES", "register_backbone", "iforest_backbone", "lof_backbone", "stub_backbone", # 配方 API "load_recipe", "build_from_recipe", "list_sample_recipes", "sample_recipe_path", ] class AnomalyDetectionError(Exception): """异常检测模板化层的统一异常（配方非法 / 主干未注册 / 校验失败）。""" # --------------------------------------------------------------------------- # 配方（Recipe）：声明式超参包，不可变数据对象 # --------------------------------------------------------------------------- #: PRD 5.3 允许的固定主干类型（默认 iforest，PRD 5.3 推荐无监督异常检测默认结构） ALLOWED_BACKBONES = ("iforest", "lof", "stub") #: PRD 5.3 允许的阈值策略：contamination（污染率）分数阈值；sigma（Nσ 法则） ALLOWED_THRESHOLD_POLICIES = ("contamination", "sigma") #: PRD 5.3 / 第 6 章里程碑：关键异常检出率（召回率）验收线 ≥ 95% DEFAULT_RECALL_FLOOR = 0.95 #: PRD 5.3 / 第 6 章里程碑：异常误报率上限 ≤ 5%（即特异性 ≥ 0.95） DEFAULT_FALSE_ALARM_CEIL = 0.05 #: 默认污染率（预期异常比例），对齐 sklearn IsolationForest 默认值 DEFAULT_CONTAMINATION = 0.05 #: 默认 Nσ 法则阈值（3σ 覆盖 ~99.7% 正常区） DEFAULT_SIGMA = 3.0 @dataclass(frozen=True) class Recipe: """异常检测配方（声明式超参包）。 一个 Recipe 描述「用什么固定主干 + 如何从超参构造一个可训练 / 可推理 的异常检测模型 + 用哪些特征列 + 阈值策略 + 验收口径」。它是不可变数据 对象，``to_dict`` / ``from_dict`` 可序列化往返，便于配置台展示与审计。 切换行业 / 工况只改 Recipe，模型代码（``AnomalyDetectionModel``）零改动 ——对齐 PRD 5.3「固定主干 + 可配置超参」默认模式。 """ name: str backbone: str = "iforest" hyperparams: Dict[str, Any] = field(default_factory=dict) feature_columns: Tuple[str, ...] = field(default_factory=tuple) threshold_policy: str = "contamination" contamination: float = DEFAULT_CONTAMINATION sigma: float = DEFAULT_SIGMA recall_floor: float = DEFAULT_RECALL_FLOOR false_alarm_ceil: float = DEFAULT_FALSE_ALARM_CEIL industry: str = "" notes: str = "" def __post_init__(self) -> None: if not self.name: raise AnomalyDetectionError("Recipe 缺少 name") if self.backbone not in ALLOWED_BACKBONES: raise AnomalyDetectionError( f"非法主干类型 {self.backbone!r}，允许：{ALLOWED_BACKBONES}") if self.threshold_policy not in ALLOWED_THRESHOLD_POLICIES: raise AnomalyDetectionError( f"非法阈值策略 {self.threshold_policy!r}，" f"允许：{ALLOWED_THRESHOLD_POLICIES}") if not (0.0 < self.contamination < 1.0): raise AnomalyDetectionError( f"contamination 越界：{self.contamination}（应在 (0,1)）") if self.sigma <= 0: raise AnomalyDetectionError( f"sigma 非法：{self.sigma}（应 > 0）") if not (0.0 <= self.recall_floor <= 1.0): raise AnomalyDetectionError( f"recall_floor 越界：{self.recall_floor}（应在 [0,1]）") if not (0.0 <= self.false_alarm_ceil <= 1.0): raise AnomalyDetectionError( f"false_alarm_ceil 越界：{self.false_alarm_ceil}（应在 [0,1]）") def to_dict(self) -> Dict[str, Any]: return { "name": self.name, "backbone": self.backbone, "hyperparams": dict(self.hyperparams), "feature_columns": list(self.feature_columns), "threshold_policy": self.threshold_policy, "contamination": self.contamination, "sigma": self.sigma, "recall_floor": self.recall_floor, "false_alarm_ceil": self.false_alarm_ceil, "industry": self.industry, "notes": self.notes, } @classmethod def from_dict(cls, data: Dict[str, Any]) -> "Recipe": try: return cls( name=data["name"], backbone=data.get("backbone", "iforest"), hyperparams=dict(data.get("hyperparams", {})), feature_columns=tuple(data.get("feature_columns", [])), threshold_policy=data.get( "threshold_policy", "contamination"), contamination=float( data.get("contamination", DEFAULT_CONTAMINATION)), sigma=float(data.get("sigma", DEFAULT_SIGMA)), recall_floor=float( data.get("recall_floor", DEFAULT_RECALL_FLOOR)), false_alarm_ceil=float( data.get("false_alarm_ceil", DEFAULT_FALSE_ALARM_CEIL)), industry=data.get("industry", ""), notes=data.get("notes", ""), ) except KeyError as exc: # pragma: no cover - 防御性 raise AnomalyDetectionError( f"配方缺少必填字段：{exc}") from exc def load_recipe(path: str) -> Recipe: """从 JSON 文件加载一个异常检测配方。 配方 JSON 结构见 ``Recipe.to_dict``；样例见 ``samples/``。 """ with open(path, "r", encoding="utf-8") as fh: data = json.load(fh) if not isinstance(data, dict): raise AnomalyDetectionError(f"配方根必须是对象：{path}") return Recipe.from_dict(data) # --------------------------------------------------------------------------- # 主干工厂：固定主干网络（iforest / lof / stub） # --------------------------------------------------------------------------- class ModelHandle: """统一模型句柄：fit / decision_function / to_dict，与硬件和具体库无关。 业务代码只持有 ``ModelHandle``，不感知底层是 sklearn 还是 stub。 约定 ``decision_function`` 返回**异常分数**：**越大越异常**（与 sklearn ``score_samples`` 取负号一致），便于阈值策略统一处理。 """ def __init__(self, backbone: str, params: Dict[str, Any], fitted: bool = False, meta: Optional[Dict[str, Any]] = None): self.backbone = backbone self.params = dict(params) self._fitted = fitted self.meta: Dict[str, Any] = dict(meta or {}) @property def fitted(self) -> bool: return self._fitted def fit(self, X: Sequence[Sequence[float]]) -> "ModelHandle": """拟合主干（无监督，仅需 X）。""" X = list(X) if not X: raise AnomalyDetectionError("训练数据为空") self._fit_impl(X) self._fitted = True return self # 子类/工厂填充 def _fit_impl(self, X: Sequence[Sequence[float]]) -> None: raise NotImplementedError def decision_function( self, X: Sequence[Sequence[float]]) -> List[float]: """返回每个样本的异常分数（越大越异常）。""" if not self._fitted: raise AnomalyDetectionError("模型未拟合，无法打分") return [self._score_one(list(row)) for row in X] def _score_one(self, row: Sequence[float]) -> float: raise NotImplementedError def to_dict(self) -> Dict[str, Any]: return { "backbone": self.backbone, "params": dict(self.params), "fitted": self._fitted, "meta": dict(self.meta), } class _StubBackbone(ModelHandle): """确定性 stub 主干：无 sklearn 时的保底实现。 拟合阶段记录每维特征的均值与标准差；打分取各维偏离均值的标准差倍数 之和（马氏距离的简化版），保证可复现、可校验、可对比，便于 CI 与 配置台预览。 """ def __init__(self, params: Dict[str, Any]): super().__init__(backbone="stub", params=params) self._means: List[float] = [] self._stds: List[float] = [] def _fit_impl(self, X) -> None: n_feat = len(X[0]) self._means = [0.0] * n_feat self._stds = [1.0] * n_feat for j in range(n_feat): col = [float(row[j]) for row in X] mean = sum(col) / len(col) var = sum((v - mean) ** 2 for v in col) / len(col) self._means[j] = mean self._stds[j] = math.sqrt(var) or 1.0 self.meta.update({"n_features": n_feat}) def _score_one(self, row) -> float: # 各维偏离均值的标准差倍数之和（≥0，越大越异常） total = 0.0 for j, v in enumerate(row): total += abs(float(v) - self._means[j]) / (self._stds[j] or 1.0) return total class _SklearnIForestBackbone(ModelHandle): """真实隔离森林主干（sklearn IsolationForest）。 仅当运行环境存在 sklearn 时启用；与 stub 接口完全一致。 ``decision_function`` 对 sklearn ``score_samples`` 取负号， 统一为「越大越异常」。 """ def __init__(self, params: Dict[str, Any]): super().__init__(backbone="iforest", params=params) from sklearn.ensemble import IsolationForest # type: ignore self._Clz = IsolationForest self._model: Any = None def _fit_impl(self, X) -> None: kw = { "n_estimators": int(self.params.get("n_estimators", 100)), "max_samples": self.params.get("max_samples", "auto"), "contamination": float( self.params.get("contamination", "auto")), "random_state": int(self.params.get("random_state", 42)), } self._model = self._Clz(**kw) self._model.fit(list(X)) # 记录实际生效的关键超参（max_samples 可能是 'auto'） self.meta.update({"n_estimators": kw["n_estimators"], "random_state": kw["random_state"]}) def _score_one(self, row) -> float: # score_samples 越大越正常，取负号统一为「越大越异常」 return float(-self._model.score_samples([list(row)])[0]) class _SklearnLOFBackbone(ModelHandle): """真实局部离群因子主干（sklearn LocalOutlierFactor）。 PRD 5.3 备选结构；仅当运行环境存在 sklearn 时启用。 novelty=True 以 支持 predict / score_samples 对新样本打分。 """ def __init__(self, params: Dict[str, Any]): super().__init__(backbone="lof", params=params) from sklearn.neighbors import LocalOutlierFactor # type: ignore self._Clz = LocalOutlierFactor self._model: Any = None def _fit_impl(self, X) -> None: kw = { "n_neighbors": int(self.params.get("n_neighbors", 20)), "contamination": float( self.params.get("contamination", "auto")), "novelty": True, } self._model = self._Clz(**kw) self._model.fit(list(X)) self.meta.update({"n_neighbors": kw["n_neighbors"]}) def _score_one(self, row) -> float: return float(-self._model.score_samples([list(row)])[0]) def _has_sklearn() -> bool: try: import sklearn # noqa: F401 return True except Exception: return False def stub_backbone(hyperparams: Dict[str, Any]) -> ModelHandle: """stub 主干工厂（恒可用）。""" return _StubBackbone(hyperparams) def iforest_backbone(hyperparams: Dict[str, Any]) -> ModelHandle: """iforest 主干工厂：有 sklearn 用真实隔离森林，否则退化为 stub。 PRD 5.3 推荐的无监督异常检测默认结构（隔离森林）。 """ if _has_sklearn(): return _SklearnIForestBackbone(hyperparams) # 无 sklearn：退化 stub 但保留声明主干名，便于审计 h = _StubBackbone(hyperparams) h.meta["degraded_from"] = "iforest" return h def lof_backbone(hyperparams: Dict[str, Any]) -> ModelHandle: """lof 主干工厂：有 sklearn 用真实 LOF，否则退化为 stub。""" if _has_sklearn(): return _SklearnLOFBackbone(hyperparams) h = _StubBackbone(hyperparams) h.meta["degraded_from"] = "lof" return h #: 主干注册表：新增结构走 ``register_backbone`` 注册，不动内核 #: （对齐 PRD 5.3「新增结构走插件注册」理念，风格对齐 #34 / #36）。 BACKBONES: Dict[str, Any] = { "iforest": iforest_backbone, "lof": lof_backbone, "stub": stub_backbone, } def register_backbone(name: str, factory: Any) -> None: """注册一个新主干工厂 ``factory(hyperparams) -> ModelHandle``。 允许高级行业模板声明非默认主干（如自研流式异常检测），不动内核——对齐 PRD「新增结构走插件注册而非改内核」。 """ if not callable(factory): raise AnomalyDetectionError("主干工厂必须是可调用对象") BACKBONES[name] = factory def _build_backbone(backbone: str, hyperparams: Dict[str, Any]) -> ModelHandle: factory = BACKBONES.get(backbone) if factory is None: raise AnomalyDetectionError( f"未注册的主干类型：{backbone!r}，已注册：{list(BACKBONES)}") return factory(hyperparams) # --------------------------------------------------------------------------- # 异常检测模型：固定主干 + 配方加载 # --------------------------------------------------------------------------- class AnomalyDetectionModel: """异常检测模型（固定主干 + 配方加载）。 业务侧两种等价入口： 1. 直接构造（显式主干）:: m = AnomalyDetectionModel(backbone="iforest", hyperparams={...}) 2. 配方加载（推荐，切换模板仅改配方）:: m = build_from_recipe( "templates/.../anomaly-detection/recipe.ti.json") """ def __init__(self, backbone: str = "iforest", hyperparams: Optional[Dict[str, Any]] = None, feature_columns: Optional[Sequence[str]] = None, threshold_policy: str = "contamination", contamination: float = DEFAULT_CONTAMINATION, sigma: float = DEFAULT_SIGMA, recall_floor: float = DEFAULT_RECALL_FLOOR, false_alarm_ceil: float = DEFAULT_FALSE_ALARM_CEIL): self.threshold_policy = threshold_policy self.contamination = contamination self.sigma = sigma self.recall_floor = recall_floor self.false_alarm_ceil = false_alarm_ceil self.recipe_meta: Dict[str, Any] = { "backbone": backbone, "hyperparams": dict(hyperparams or {}), "feature_columns": list(feature_columns or []), "threshold_policy": threshold_policy, "contamination": contamination, "sigma": sigma, "recall_floor": recall_floor, "false_alarm_ceil": false_alarm_ceil, } self._handle: ModelHandle = _build_backbone( backbone, hyperparams or {}) self._threshold: Optional[float] = None @classmethod def from_recipe(cls, recipe: Recipe) -> "AnomalyDetectionModel": """从一个 ``Recipe`` 构造模型（推荐入口）。""" m = cls( backbone=recipe.backbone, hyperparams=recipe.hyperparams, feature_columns=recipe.feature_columns, threshold_policy=recipe.threshold_policy, contamination=recipe.contamination, sigma=recipe.sigma, recall_floor=recipe.recall_floor, false_alarm_ceil=recipe.false_alarm_ceil, ) m.recipe_meta["recipe_name"] = recipe.name m.recipe_meta["industry"] = recipe.industry return m # ---- 训练 / 推理 ---- def fit(self, X: Sequence[Sequence[float]]) -> "AnomalyDetectionModel": """拟合主干（无监督）。同时在训练集上确定异常分数阈值。""" X = list(X) self._handle.fit(X) # 用训练分布确定阈值：contamination 取高分位数；sigma 取均值+Nσ scores = self._handle.decision_function(X) self._threshold = self._derive_threshold(scores) return self def _derive_threshold(self, scores: Sequence[float]) -> float: """根据阈值策略从训练分数分布确定异常分数阈值。 - ``contamination``：取高分位数（1 - contamination），高于即判异常； - ``sigma``：取均值 + Nσ（N=3 默认覆盖 ~99.7% 正常区）。 """ scores = sorted(float(s) for s in scores) if not scores: raise AnomalyDetectionError("训练分数为空，无法确定阈值") if self.threshold_policy == "sigma": mean = sum(scores) / len(scores) var = sum((s - mean) ** 2 for s in scores) / len(scores) std = math.sqrt(var) or 1.0 return mean + self.sigma * std # contamination：高分位数（线性插值） k = (1.0 - self.contamination) * (len(scores) - 1) lo = int(math.floor(k)) hi = int(math.ceil(k)) if lo == hi: return scores[lo] frac = k - lo return scores[lo] + (scores[hi] - scores[lo]) * frac def decision_function( self, X: Sequence[Sequence[float]]) -> List[float]: """返回每个样本的异常分数（越大越异常）。""" return self._handle.decision_function(X) def predict(self, X: Sequence[Sequence[float]]) -> List[int]: """返回每个样本的二值异常标签：1=异常，0=正常。 依据 ``fit`` 时确定的阈值（未拟合或阈值未定则报错）。 """ if self._threshold is None: raise AnomalyDetectionError( "阈值未确定：请先 fit，或阈值策略未被应用") scores = self.decision_function(X) return [1 if s > self._threshold else 0 for s in scores] @property def fitted(self) -> bool: return self._handle.fitted @property def threshold(self) -> Optional[float]: return self._threshold # ---- 验收口径 ---- def evaluate(self, X: Sequence[Sequence[float]], y_true: Sequence[int]) -> "Metrics": """评估并返回检出率 / 误报率 / 精确率 / 召回率 / F1 与是否达标。 ``y_true`` 中 1=异常、0=正常。检出率即召回率（PRD 5.3 / 里程碑： ≥ 95%）；误报率即假阳性率（1 - 特异性，里程碑：≤ 5%）。 ``recall >= recall_floor`` 且 ``false_alarm <= false_alarm_ceil`` 即视为达标。 """ y_pred = self.predict(X) return Metrics.compute( y_true=list(y_true), y_pred=y_pred, recall_floor=self.recall_floor, false_alarm_ceil=self.false_alarm_ceil) def to_dict(self) -> Dict[str, Any]: return { "recipe_meta": dict(self.recipe_meta), "handle": self._handle.to_dict(), "threshold": self._threshold, } # --------------------------------------------------------------------------- # 验收：Metrics # --------------------------------------------------------------------------- @dataclass(frozen=True) class Metrics: """异常检测验收结果（PRD 5.3 检出率 / 误报率口径）。""" recall: float # 检出率（TP/TP+FN），里程碑 ≥ 95% precision: float # 精确率（TP/TP+FP） f1: float # F1 false_alarm_rate: float # 误报率（FP/FP+TN），里程碑 ≤ 5% n_anomaly_true: int n_normal_true: int recall_floor: float false_alarm_ceil: float passed: bool def to_dict(self) -> Dict[str, Any]: return { "recall": self.recall, "precision": self.precision, "f1": self.f1, "false_alarm_rate": self.false_alarm_rate, "n_anomaly_true": self.n_anomaly_true, "n_normal_true": self.n_normal_true, "recall_floor": self.recall_floor, "false_alarm_ceil": self.false_alarm_ceil, "passed": self.passed, } @classmethod def compute(cls, y_true: Sequence[int], y_pred: Sequence[int], recall_floor: float = DEFAULT_RECALL_FLOOR, false_alarm_ceil: float = DEFAULT_FALSE_ALARM_CEIL) -> "Metrics": if len(y_true) != len(y_pred): raise AnomalyDetectionError( f"y_true/y_pred 长度不一致：{len(y_true)} != {len(y_pred)}") if not y_true: raise AnomalyDetectionError("评估数据为空") # 统计混淆矩阵四元 tp = fp = fn = tn = 0 for yt, yp in zip(y_true, y_pred): if yt == 1 and yp == 1: tp += 1 elif yt == 0 and yp == 1: fp += 1 elif yt == 1 and yp == 0: fn += 1 else: tn += 1 n_anomaly = tp + fn n_normal = fp + tn recall = tp / n_anomaly if n_anomaly > 0 else 0.0 precision = tp / (tp + fp) if (tp + fp) > 0 else 0.0 f1 = (2 * precision * recall / (precision + recall) if (precision + recall) > 0 else 0.0) far = fp / n_normal if n_normal > 0 else 0.0 passed = recall >= recall_floor and far <= false_alarm_ceil return cls( recall=recall, precision=precision, f1=f1, false_alarm_rate=far, n_anomaly_true=n_anomaly, n_normal_true=n_normal, recall_floor=recall_floor, false_alarm_ceil=false_alarm_ceil, passed=passed, ) # --------------------------------------------------------------------------- # 配方构建入口 + 样例协议 # --------------------------------------------------------------------------- def build_from_recipe(path: str) -> AnomalyDetectionModel: """从 JSON 配方文件加载并构造一个异常检测模型（推荐入口）。 切换模板仅改配方文件，业务代码零改动——对齐 PRD 5.3 验收口径。 """ return AnomalyDetectionModel.from_recipe(load_recipe(path)) def _samples_dir() -> str: return os.path.join(os.path.dirname(os.path.abspath(__file__)), "samples", "anomaly-detection") def list_sample_recipes() -> List[str]: """列出内置样例配方（树脂 + Ti 两套，验证同框架加载多套配方）。""" d = _samples_dir() if not os.path.isdir(d): return [] return sorted(f for f in os.listdir(d) if f.endswith(".json")) def sample_recipe_path(name: str) -> str: """返回样例配方的完整路径。""" if not name.endswith(".json"): name = name + ".json" return os.path.join(_samples_dir(), name)