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",
+]

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())

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)

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 超参。"
+}

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对接后补标）。"
+}

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)

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)