diff --git a/core/model-framework/__init__.py b/core/model-framework/__init__.py new file mode 100644 index 0000000..1936a3b --- /dev/null +++ b/core/model-framework/__init__.py @@ -0,0 +1,50 @@ +# -*- coding: utf-8 -*- +"""iAOP-Core · 模型框架层(AI Model Framework)。 + +对应 PRD 5.3「③ AI 模型框架」与 EPIC #5(内核平台化改造)。 + +当前已交付(自包含,不依赖未合并分支): +- ``anomaly_detection``:异常检测模型模板化(固定主干 + 配方加载), + issue #37。同一主干代码不变,切换行业/工况只改配方(声明式 JSON + 超参包)——对齐 PRD 5.3「固定主干 + 可配置超参」默认模式。 + +规划(待相关 PR 合入后无缝对接,业务侧零改动): +- ``model_recipe``:Model Recipe 插件接口(issue #34,PR #102 待审核)。 + ``anomaly_detection`` 的主干届时可注册为 ``register_backbone`` 的一个 + 具名主干,配方可映射为一条 ``ModelRecipe``。 +- ``quality_forecast``:质量预测模型模板化(issue #36,PR #103 待审核)。 + 与本模块共享「主干工厂 + Recipe + 验收口径」骨架。 +""" +from model_framework.anomaly_detection import ( + AnomalyDetectionError, + AnomalyDetectionModel, + BACKBONES, + Metrics, + ModelHandle, + Recipe, + build_from_recipe, + iforest_backbone, + list_sample_recipes, + load_recipe, + lof_backbone, + register_backbone, + sample_recipe_path, + stub_backbone, +) + +__all__ = [ + "Recipe", + "Metrics", + "AnomalyDetectionError", + "AnomalyDetectionModel", + "ModelHandle", + "BACKBONES", + "register_backbone", + "iforest_backbone", + "lof_backbone", + "stub_backbone", + "load_recipe", + "build_from_recipe", + "list_sample_recipes", + "sample_recipe_path", +] diff --git a/core/model-framework/_sanity_check.py b/core/model-framework/_sanity_check.py new file mode 100644 index 0000000..7341201 --- /dev/null +++ b/core/model-framework/_sanity_check.py @@ -0,0 +1,74 @@ +# -*- coding: utf-8 -*- +"""异常检测模型模板化 sanity 检查(无构建环境下的离线基本验证)。 + +验证 PRD 5.3 验收口径「同框架加载 Ti / 树脂两套配方均跑通」: +1. 两套样例配方均可被 ``build_from_recipe`` 加载; +2. 加载后模型可 fit / predict / evaluate 走通完整链路; +3. 切换模板仅改配方,模型代码(``type(m1) == type(m2)``)零改动; +4. 两套配方的 backbone / 特征列确实不同(确属两套模板,非同一份复制)。 + +用法:python _sanity_check.py +""" +import os +import sys + +HERE = os.path.dirname(os.path.abspath(__file__)) +if HERE not in sys.path: + sys.path.insert(0, HERE) + +from anomaly_detection import ( # noqa: E402 + build_from_recipe, + list_sample_recipes, + sample_recipe_path, +) + + +def main() -> int: + failures = [] + + names = list_sample_recipes() + required = ("recipe.ti.json", "recipe.resin.json") + for r in required: + if r not in names: + failures.append(f"缺少样例配方:{r}") + + models = {} + for r in required: + try: + m = build_from_recipe(sample_recipe_path(r)) + # 用配方声明的特征数构造演示数据(含少量离群点) + n_feat = len(m.recipe_meta["feature_columns"]) or 2 + X = [[float(i + j) for j in range(n_feat)] for i in range(30)] + # 注入 3 个明显离群点 + for k in range(3): + X.append([100.0 + k for _ in range(n_feat)]) + y_true = [0] * 30 + [1] * 3 + m.fit(X) + preds = m.predict(X) + assert len(preds) == len(y_true), "预测长度异常" + m.evaluate(X, y_true) + models[r] = m + except Exception as exc: # pragma: no cover - 诊断输出 + failures.append(f"{r} 加载/训练/评估失败:{exc!r}") + + # 切换模板仅改配方,模型代码零改动 + if len(models) == 2: + ms = list(models.values()) + if type(ms[0]) is not type(ms[1]): + failures.append("两套配方使用了不同的模型类,违反「模型代码零改动」") + if (models["recipe.ti.json"].recipe_meta["feature_columns"] + == models["recipe.resin.json"].recipe_meta["feature_columns"]): + failures.append("Ti/树脂配方特征列完全相同,疑似复制") + + if failures: + print("FAIL") + for f in failures: + print(" -", f) + return 1 + print(f"OK: {len(required)} 套配方均加载/训练/评估通过," + f"模型代码零改动(type 一致),PRD 5.3 验收口径达成") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/core/model-framework/anomaly_detection.py b/core/model-framework/anomaly_detection.py new file mode 100644 index 0000000..6058ce9 --- /dev/null +++ b/core/model-framework/anomaly_detection.py @@ -0,0 +1,673 @@ +# -*- 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) diff --git a/core/model-framework/samples/anomaly-detection/recipe.resin.json b/core/model-framework/samples/anomaly-detection/recipe.resin.json new file mode 100644 index 0000000..6b0c297 --- /dev/null +++ b/core/model-framework/samples/anomaly-detection/recipe.resin.json @@ -0,0 +1,24 @@ +{ + "name": "resin-reactor-anomaly", + "backbone": "iforest", + "industry": "吸附树脂(已终验化工新材料AI平台 baseline)", + "hyperparams": { + "n_estimators": 100, + "max_samples": "auto", + "contamination": "auto", + "random_state": 7 + }, + "feature_columns": [ + "reactor_temp", + "reactor_pressure", + "flow_rate", + "ph_value", + "conversion_rate" + ], + "threshold_policy": "contamination", + "contamination": 0.05, + "sigma": 3.0, + "recall_floor": 0.95, + "false_alarm_ceil": 0.05, + "notes": "PRD 5.3 ③ 异常检测:树脂反应釜工况/质量异常预警,复用已交付化工AI平台 baseline 超参。" +} diff --git a/core/model-framework/samples/anomaly-detection/recipe.ti.json b/core/model-framework/samples/anomaly-detection/recipe.ti.json new file mode 100644 index 0000000..0cc96ce --- /dev/null +++ b/core/model-framework/samples/anomaly-detection/recipe.ti.json @@ -0,0 +1,26 @@ +{ + "name": "ti-cl4-furnace-impurity-anomaly", + "backbone": "iforest", + "industry": "海绵钛氯化车间(Template-Ti 一期)", + "hyperparams": { + "n_estimators": 150, + "max_samples": "auto", + "contamination": "auto", + "random_state": 42 + }, + "feature_columns": [ + "furnace_temp", + "furnace_pressure", + "cl2_flow", + "ti_feed_rate", + "impurity_fe", + "impurity_v", + "impurity_si" + ], + "threshold_policy": "contamination", + "contamination": 0.05, + "sigma": 3.0, + "recall_floor": 0.95, + "false_alarm_ceil": 0.05, + "notes": "PRD 5.3 ③ 异常检测:氯化车间炉层杂质/工况异常预警(关联 EPIC #10 炉层杂质预警),验收检出率≥95%、误报率≤5%(PRD 第6章里程碑)。一期数据门槛:≥6个月标注(DCS+LIMS对接后补标)。" +} diff --git a/core/model-framework/tests/_bootstrap.py b/core/model-framework/tests/_bootstrap.py new file mode 100644 index 0000000..31a9dcf --- /dev/null +++ b/core/model-framework/tests/_bootstrap.py @@ -0,0 +1,28 @@ +# -*- coding: utf-8 -*- +"""测试引导:把连字符目录 ``core/model-framework`` 加载为可导入包 +``model_framework``,使测试可 ``from model_framework import ...``。 + +与仓库内各 core 模块的测试引导同款模式(importlib 完整加载包,执行 +``__init__.py``,保持顶层导出可用)。 +""" +import importlib.util +import os +import sys + +PKG_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +REPO_ROOT = os.path.dirname(os.path.dirname( + os.path.dirname(os.path.dirname(os.path.abspath(__file__))))) + + +def _load_package(name: str, path: str) -> None: + if name in sys.modules: + return + init_py = os.path.join(path, "__init__.py") + spec = importlib.util.spec_from_file_location( + name, init_py, submodule_search_locations=[path]) + module = importlib.util.module_from_spec(spec) + sys.modules[name] = module + spec.loader.exec_module(module) + + +_load_package("model_framework", PKG_DIR) diff --git a/core/model-framework/tests/test_anomaly_detection.py b/core/model-framework/tests/test_anomaly_detection.py new file mode 100644 index 0000000..e9ca6b8 --- /dev/null +++ b/core/model-framework/tests/test_anomaly_detection.py @@ -0,0 +1,333 @@ +# -*- coding: utf-8 -*- +"""``anomaly_detection`` 单元测试(issue #37)。 + +覆盖: +- 配方(Recipe)不可变性 / 序列化往返 / 非法主干、非法阈值策略与越界校验; +- 主干工厂注册表 + 自定义主干注册(PRD 5.3「新增结构走插件注册」); +- stub / iforest / lof 三类主干的 fit/decision_function 契约; +- 固定主干 + 配方加载:同框架加载 Ti / 树脂两套配方均跑通(PRD 5.3 + 验收口径); +- Metrics 验收口径(PRD 5.3 / 里程碑:检出率 ≥ 95%、误报率 ≤ 5%); +- 阈值策略(contamination 高分位 / sigma Nσ 法则); +- 零外部强依赖:无 sklearn 时 stub 退化仍可加载与校验。 +""" +import json +import os +import sys +import unittest + +HERE = os.path.dirname(os.path.abspath(__file__)) +sys.path.insert(0, HERE) +import _bootstrap # noqa: E402 注册 model_framework 包 + +from model_framework import ( # noqa: E402 + AnomalyDetectionError, + AnomalyDetectionModel, + BACKBONES, + Metrics, + ModelHandle, + Recipe, + build_from_recipe, + iforest_backbone, + list_sample_recipes, + load_recipe, + lof_backbone, + register_backbone, + sample_recipe_path, + stub_backbone, +) + + +def _normal_dataset(n=40, n_feat=2, seed=0): + """构造一组「正常」样本(围绕均值的确定性点)。""" + X = [] + for i in range(n): + row = [] + for j in range(n_feat): + base = float(i % 7) + 1.0 + 0.1 * j + row.append(base) + X.append(row) + return X + + +def _labeled_dataset(n_normal=40, n_anomaly=5, n_feat=2): + """构造正常 + 离群点数据集,返回 (X, y_true),1=异常。""" + X = _normal_dataset(n_normal, n_feat) + y = [0] * n_normal + for k in range(n_anomaly): + # 明显远离正常区的离群点 + X.append([100.0 + k for _ in range(n_feat)]) + y.append(1) + return X, y + + +class TestRecipe(unittest.TestCase): + """配方数据对象与校验。""" + + def test_defaults_and_immutability(self): + r = Recipe(name="t") + self.assertEqual(r.backbone, "iforest") + self.assertEqual(r.threshold_policy, "contamination") + self.assertAlmostEqual(r.recall_floor, 0.95) + self.assertAlmostEqual(r.false_alarm_ceil, 0.05) + with self.assertRaises(Exception): + r.name = "other" # frozen + + def test_roundtrip(self): + r = Recipe(name="t", backbone="lof", + hyperparams={"n_neighbors": 15}, + feature_columns=("a", "b"), + threshold_policy="sigma", + contamination=0.1, sigma=2.5, + recall_floor=0.9, false_alarm_ceil=0.1, + industry="树脂", notes="n") + d = r.to_dict() + r2 = Recipe.from_dict(d) + self.assertEqual(r, r2) + # JSON 往返 + r3 = Recipe.from_dict(json.loads(json.dumps(d))) + self.assertEqual(r, r3) + + def test_invalid_backbone_raises(self): + with self.assertRaises(AnomalyDetectionError): + Recipe(name="t", backbone="svm") + + def test_invalid_threshold_policy_raises(self): + with self.assertRaises(AnomalyDetectionError): + Recipe(name="t", threshold_policy="quantile") + + def test_contamination_out_of_range(self): + with self.assertRaises(AnomalyDetectionError): + Recipe(name="t", contamination=0.0) + with self.assertRaises(AnomalyDetectionError): + Recipe(name="t", contamination=1.0) + + def test_sigma_nonpositive_raises(self): + with self.assertRaises(AnomalyDetectionError): + Recipe(name="t", sigma=0) + + def test_recall_floor_out_of_range(self): + with self.assertRaises(AnomalyDetectionError): + Recipe(name="t", recall_floor=1.5) + + def test_missing_name(self): + with self.assertRaises(AnomalyDetectionError): + Recipe(name="") + + def test_load_recipe_from_file(self): + path = sample_recipe_path("recipe.ti.json") + r = load_recipe(path) + self.assertEqual(r.name, "ti-cl4-furnace-impurity-anomaly") + self.assertEqual(r.backbone, "iforest") + self.assertIn("furnace_temp", r.feature_columns) + + +class TestBackbones(unittest.TestCase): + """主干工厂与注册表。""" + + def test_builtin_backbones_registered(self): + for name in ("iforest", "lof", "stub"): + self.assertIn(name, BACKBONES) + + def test_register_custom_backbone(self): + class _Custom(ModelHandle): + def __init__(self, p): + super().__init__("custom", p) + self._v = 1.0 + + def _fit_impl(self, X): + self._v = sum(sum(r) for r in X) / (len(X) * len(X[0])) + + def _score_one(self, row): + # 离均值越远分数越高 + return abs(sum(float(v) for v in row) - self._v) + + register_backbone("custom_test", lambda p: _Custom(p)) + m = AnomalyDetectionModel(backbone="custom_test") + X = _normal_dataset() + m.fit(X) + self.assertEqual(len(m.predict(X)), len(X)) + # 清理避免污染其它用例 + BACKBONES.pop("custom_test", None) + + def test_unknown_backbone_raises(self): + with self.assertRaises(AnomalyDetectionError): + AnomalyDetectionModel(backbone="not_a_backbone") + + def test_stub_score_is_deterministic_and_nonneg(self): + h = stub_backbone({}) + X = _normal_dataset() + h.fit(X) + s1 = h.decision_function(X) + s2 = h.decision_function(X) + self.assertEqual(s1, s2) + self.assertTrue(all(isinstance(v, float) for v in s1)) + self.assertTrue(all(v >= 0 for v in s1)) + + def test_decision_before_fit_raises(self): + h = stub_backbone({}) + with self.assertRaises(AnomalyDetectionError): + h.decision_function([[1.0, 2.0]]) + + def test_fit_empty_raises(self): + h = stub_backbone({}) + with self.assertRaises(AnomalyDetectionError): + h.fit([]) + + def test_iforest_factory_runs_with_or_without_sklearn(self): + # 无论 sklearn 是否存在都不应报错 + h = iforest_backbone({"n_estimators": 20}) + X = _normal_dataset() + h.fit(X) + scores = h.decision_function(X) + self.assertEqual(len(scores), len(X)) + + +class TestModelContract(unittest.TestCase): + """模型 fit/decision_function/predict 契约。""" + + def test_fit_predict_shapes(self): + m = AnomalyDetectionModel(backbone="stub") + X = _normal_dataset(20) + m.fit(X) + self.assertTrue(m.fitted) + self.assertIsNotNone(m.threshold) + preds = m.predict(X) + self.assertEqual(len(preds), len(X)) + self.assertTrue(all(p in (0, 1) for p in preds)) + + def test_predict_before_fit_raises(self): + m = AnomalyDetectionModel(backbone="stub") + with self.assertRaises(AnomalyDetectionError): + m.predict([[1.0, 2.0]]) + + def test_decision_before_fit_raises(self): + m = AnomalyDetectionModel(backbone="stub") + with self.assertRaises(AnomalyDetectionError): + m.decision_function([[1.0, 2.0]]) + + def test_fit_empty_raises(self): + m = AnomalyDetectionModel(backbone="stub") + with self.assertRaises(AnomalyDetectionError): + m.fit([]) + + def test_to_dict_roundtrip_meta(self): + m = AnomalyDetectionModel(backbone="iforest", + hyperparams={"n_estimators": 5}, + feature_columns=["a"], + threshold_policy="sigma", sigma=2.0) + d = m.to_dict() + self.assertEqual(d["recipe_meta"]["backbone"], "iforest") + self.assertEqual(d["recipe_meta"]["threshold_policy"], "sigma") + self.assertIn("handle", d) + + def test_threshold_contamination_isolate_outliers(self): + """contamination 阈值应把注入的离群点判为异常。""" + m = AnomalyDetectionModel( + backbone="stub", threshold_policy="contamination", + contamination=0.10) + X, y_true = _labeled_dataset(n_normal=40, n_anomaly=5) + m.fit(X) + preds = m.predict(X) + # 注入的 5 个离群点应被全部判异常 + self.assertEqual(sum(preds[40:]), 5) + + def test_threshold_sigma_isolate_outliers(self): + """sigma 阈值也应把注入的极端离群点判为异常。""" + m = AnomalyDetectionModel( + backbone="stub", threshold_policy="sigma", sigma=2.0) + X, y_true = _labeled_dataset(n_normal=40, n_anomaly=5) + m.fit(X) + preds = m.predict(X) + self.assertEqual(sum(preds[40:]), 5) + + +class TestMetrics(unittest.TestCase): + """验收口径(PRD 5.3:检出率 ≥ 95%、误报率 ≤ 5%)。""" + + def test_perfect_predictions_pass(self): + y = [1, 1, 0, 0, 0] + met = Metrics.compute(y, y, recall_floor=0.95, false_alarm_ceil=0.05) + self.assertAlmostEqual(met.recall, 1.0) + self.assertAlmostEqual(met.false_alarm_rate, 0.0) + self.assertAlmostEqual(met.f1, 1.0) + self.assertTrue(met.passed) + + def test_all_miss_fails(self): + y_true = [1, 1, 0, 0] + y_pred = [0, 0, 0, 0] # 漏检全部异常 + met = Metrics.compute(y_true, y_pred) + self.assertAlmostEqual(met.recall, 0.0) + self.assertFalse(met.passed) + + def test_high_false_alarm_fails(self): + y_true = [1, 0, 0, 0, 0] + y_pred = [1, 1, 1, 1, 1] # 全判异常:检出但误报爆表 + met = Metrics.compute(y_true, y_pred, false_alarm_ceil=0.05) + self.assertAlmostEqual(met.recall, 1.0) + self.assertGreater(met.false_alarm_rate, 0.05) + self.assertFalse(met.passed) + + def test_length_mismatch_raises(self): + with self.assertRaises(AnomalyDetectionError): + Metrics.compute([1, 0], [1]) + + def test_empty_raises(self): + with self.assertRaises(AnomalyDetectionError): + Metrics.compute([], []) + + def test_no_anomaly_in_true_recall_zero_div_safe(self): + # 无真实异常时 recall 定义为 0,不应抛 ZeroDivision + met = Metrics.compute([0, 0, 0], [0, 0, 0]) + self.assertEqual(met.recall, 0.0) + self.assertEqual(met.n_anomaly_true, 0) + + def test_evaluate_end_to_end(self): + m = AnomalyDetectionModel(backbone="stub", threshold_policy="sigma", + sigma=2.0) + X, y_true = _labeled_dataset(n_normal=40, n_anomaly=5) + m.fit(X) + met = m.evaluate(X, y_true) + self.assertIsInstance(met, Metrics) + # 离群点应被检出(stub 在极端离群点上召回=1) + self.assertEqual(met.recall, 1.0) + + +class TestSampleRecipes(unittest.TestCase): + """样例协议:同框架加载 Ti / 树脂两套配方均跑通(PRD 5.3 验收口径)。""" + + def test_samples_present(self): + names = list_sample_recipes() + self.assertIn("recipe.ti.json", names) + self.assertIn("recipe.resin.json", names) + + def test_build_from_each_sample_runs(self): + for name in ("recipe.ti.json", "recipe.resin.json"): + m = build_from_recipe(sample_recipe_path(name)) + self.assertIn( + m.recipe_meta["backbone"], ("iforest", "lof", "stub")) + feat = m.recipe_meta["feature_columns"] + n_feat = len(feat) + self.assertGreater(n_feat, 0) + X = [[float(i + j) for j in range(n_feat)] for i in range(30)] + # 注入离群点 + for k in range(3): + X.append([100.0 + k for _ in range(n_feat)]) + y_true = [0] * 30 + [1] * 3 + m.fit(X) + preds = m.predict(X) + self.assertEqual(len(preds), len(y_true)) + met = m.evaluate(X, y_true) + self.assertIsInstance(met, Metrics) + + def test_two_recipes_share_same_code(self): + """切换模板仅改配方,模型代码零改动(PRD 5.3)。""" + m1 = build_from_recipe(sample_recipe_path("recipe.ti.json")) + m2 = build_from_recipe(sample_recipe_path("recipe.resin.json")) + self.assertEqual(type(m1), type(m2)) + self.assertNotEqual(m1.recipe_meta.get("recipe_name"), + m2.recipe_meta.get("recipe_name")) + + +if __name__ == "__main__": + unittest.main(verbosity=2)