diff --git a/core/model-framework/__init__.py b/core/model-framework/__init__.py new file mode 100644 index 0000000..d70d6b2 --- /dev/null +++ b/core/model-framework/__init__.py @@ -0,0 +1,48 @@ +# -*- coding: utf-8 -*- +"""iAOP-Core · 模型框架层(AI Model Framework)。 + +对应 PRD 5.3「③ AI 模型框架」与 EPIC #5(内核平台化改造)。 + +当前已交付(自包含,不依赖未合并分支): +- ``quality_forecast``:质量预测模型模板化(固定主干 + 配方加载), + issue #36。同一主干代码不变,切换行业/工况只改配方(声明式 JSON + 超参包)——对齐 PRD 5.3「固定主干 + 可配置超参」默认模式。 + +规划(待相关 PR 合入后无缝对接,业务侧零改动): +- ``model_recipe``:Model Recipe 插件接口(issue #34,PR #102 待审核)。 + ``quality_forecast`` 的主干届时可注册为 ``register_backbone`` 的一个 + 具名主干,配方可映射为一条 ``ModelRecipe``。 +""" +from model_framework.quality_forecast import ( + Accuracy, + BACKBONES, + ModelHandle, + QualityForecastError, + QualityForecastModel, + Recipe, + build_from_recipe, + dnn_backbone, + gbdt_backbone, + list_sample_recipes, + load_recipe, + register_backbone, + sample_recipe_path, + stub_backbone, +) + +__all__ = [ + "Recipe", + "Accuracy", + "QualityForecastError", + "QualityForecastModel", + "ModelHandle", + "BACKBONES", + "register_backbone", + "gbdt_backbone", + "dnn_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..d8483c0 --- /dev/null +++ b/core/model-framework/_sanity_check.py @@ -0,0 +1,71 @@ +# -*- 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 quality_forecast 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(10)] + y = [float(i % 4) + 1.0 for i in range(10)] + m.fit(X, y) + preds = m.predict(X) + assert len(preds) == len(y), "预测长度异常" + m.evaluate(X, y) + 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/quality_forecast.py b/core/model-framework/quality_forecast.py new file mode 100644 index 0000000..325df28 --- /dev/null +++ b/core/model-framework/quality_forecast.py @@ -0,0 +1,534 @@ +# -*- coding: utf-8 -*- +"""质量预测模型模板化(固定主干 + 配方加载)。 + +对应 issue #36(父 EPIC #5「③ AI 模型框架 配置化重构」、PRD 5.3 +「网络结构策略 / 模板化技术路径」)。 + +PRD 5.3 的核心诉求 +------------------ + +质量预测属于 PRD 5.3「四类模型模板」之一(① 质量预测),采用 +「**固定主干网络 + 可配置超参**」为默认模式:同一主干代码不变, +切换行业/工况只改 *配方(recipe)* —— 一个声明式 JSON 超参包。 + +本模块交付什么 +-------------- + +1. **``QualityForecastModel``**:固定主干的质量预测模型。默认主干是 + ``gbdt``(梯度提升回归,PRD 5.3 推荐的监督回归默认结构);当运行 + 环境存在 ``sklearn`` 时自动升级为真实实现,否则退化为确定性 stub, + 保证边缘 / 离线 / CI 环境可加载与校验——与 issue #34 / #35 的 + 「numpy/sklearn 可选」策略一致。 +2. **``Recipe`` 配方加载器**:声明式 JSON 超参包(``load_recipe`` / + ``build_from_recipe``)。配方描述「主干类型 + 超参 + 特征列 + 目标列 + + 验收口径」,业务侧只 ``build_from_recipe(path)`` 一行即可拿到一个 + 可训练/可推理的模型——切换模板仅改配方,模型代码零改动。 +3. **``Accuracy`` 验收口径**:PRD 5.3 / 第 6 章里程碑要求「关键质量指标 + 预测准确率 ≥ 90%」。``evaluate`` 直接给出准确率 / MAE / RMSE,便于 + 配置台与 UAT 直接读取。 +4. **样例配方(``samples/`` JSON)**:Ti(海绵钛氯化车间)+ 树脂两套 + 质量预测超参包样例,验证「同框架加载两套配方均跑通」的验收口径。 + +与 issue #34 ``model_recipe`` 的关系 +------------------------------------ + +接口风格对齐 #34 的 ``ModelHandle`` / ``ModelRecipe``(``fit`` / ``predict`` +/ ``to_dict``、不可变声明式数据对象)。本模块**自包含、不依赖 #34 未合并 +的 ``model_recipe``**,待 #34(PR #102)合入后,质量预测主干可平滑注册为 +``register_backbone("gbdt", ...)`` 的一个具名主干,配方可映射为一条 +``ModelRecipe``——届时本模块零业务侧改动。 + +零外部强依赖 +------------ + +* 主干默认走纯 Python stub(``StubBackbone``):无 sklearn 时也能加载、 + 构造、(伪)拟合与预测,保证 CI 可加载与校验; +* 存在 ``sklearn`` 时,``gbdt`` 主干自动升级为真实 + ``GradientBoostingRegressor`` 实现,其余情况退化为 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", + "Accuracy", + "QualityForecastError", + # 模型 + "QualityForecastModel", + "ModelHandle", + # 主干工厂 + "BACKBONES", + "register_backbone", + "gbdt_backbone", + "dnn_backbone", + "stub_backbone", + # 配方 API + "load_recipe", + "build_from_recipe", + "list_sample_recipes", + "sample_recipe_path", +] + + +class QualityForecastError(Exception): + """质量预测模板化层的统一异常(配方非法 / 主干未注册 / 校验失败)。""" + + +# --------------------------------------------------------------------------- +# 配方(Recipe):声明式超参包,不可变数据对象 +# --------------------------------------------------------------------------- + +#: PRD 5.3 允许的固定主干类型(默认 gbdt,PRD 5.3 推荐监督回归默认结构) +ALLOWED_BACKBONES = ("gbdt", "dnn", "stub") + +#: PRD 5.3 / 第 6 章里程碑:质量预测准确率验收线 ≥ 90% +DEFAULT_ACCURACY_FLOOR = 0.90 + + +@dataclass(frozen=True) +class Recipe: + """质量预测配方(声明式超参包)。 + + 一个 Recipe 描述「用什么固定主干 + 如何从超参构造一个可训练/可推理的 + 质量预测模型 + 用哪些特征/目标列 + 验收口径」。它是不可变数据对象, + ``to_dict`` / ``from_dict`` 可序列化往返,便于配置台展示与审计。 + + 切换行业/工况只改 Recipe,模型代码(``QualityForecastModel``)零改动 + ——对齐 PRD 5.3「固定主干 + 可配置超参」默认模式。 + """ + + name: str + backbone: str = "gbdt" + hyperparams: Dict[str, Any] = field(default_factory=dict) + feature_columns: Tuple[str, ...] = field(default_factory=tuple) + target_column: str = "quality_index" + accuracy_floor: float = DEFAULT_ACCURACY_FLOOR + industry: str = "" + notes: str = "" + + def __post_init__(self) -> None: + if not self.name: + raise QualityForecastError("Recipe 缺少 name") + if self.backbone not in ALLOWED_BACKBONES: + raise QualityForecastError( + f"非法主干类型 {self.backbone!r},允许:{ALLOWED_BACKBONES}") + if self.accuracy_floor < 0 or self.accuracy_floor > 1: + raise QualityForecastError( + f"accuracy_floor 越界:{self.accuracy_floor}(应在 [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), + "target_column": self.target_column, + "accuracy_floor": self.accuracy_floor, + "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", "gbdt"), + hyperparams=dict(data.get("hyperparams", {})), + feature_columns=tuple(data.get("feature_columns", [])), + target_column=data.get("target_column", "quality_index"), + accuracy_floor=float(data.get( + "accuracy_floor", DEFAULT_ACCURACY_FLOOR)), + industry=data.get("industry", ""), + notes=data.get("notes", ""), + ) + except KeyError as exc: # pragma: no cover - 防御性 + raise QualityForecastError(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 QualityForecastError(f"配方根必须是对象:{path}") + return Recipe.from_dict(data) + + +# --------------------------------------------------------------------------- +# 主干工厂:固定主干网络(gbdt / dnn / stub) +# --------------------------------------------------------------------------- + +class ModelHandle: + """统一模型句柄:fit / predict / to_dict,与硬件和具体库无关。 + + 业务代码只持有 ``ModelHandle``,不感知底层是 sklearn 还是 stub。 + """ + + 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]], y: Sequence[float]) -> "ModelHandle": + """拟合主干。stub 主干记录均值/极差用于确定性预测。""" + X = list(X) + y = list(y) + if not X or not y: + raise QualityForecastError("训练数据为空") + if len(X) != len(y): + raise QualityForecastError( + f"X/y 样本数不一致:{len(X)} != {len(y)}") + self._fit_impl(X, y) + self._fitted = True + return self + + # 子类/工厂填充 + def _fit_impl(self, X: Sequence[Sequence[float]], y: Sequence[float]) -> None: + raise NotImplementedError + + def predict(self, X: Sequence[Sequence[float]]) -> List[float]: + if not self._fitted: + raise QualityForecastError("模型未拟合,无法预测") + return [self._predict_one(list(row)) for row in X] + + def _predict_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._y_mean: float = 0.0 + self._y_amp: float = 1.0 + + def _fit_impl(self, X, y) -> None: + self._y_mean = sum(y) / len(y) + self._y_amp = (max(y) - min(y)) or 1.0 + self.meta.update({"y_mean": self._y_mean, "y_amp": self._y_amp}) + + def _predict_one(self, row) -> float: + # 确定性:输入和的 tanh 压缩到 [y_mean-amp/2, y_mean+amp/2] + s = sum(float(v) for v in row) if row else 0.0 + # 归一化到 [-1,1] 附近,再映射回目标域 + norm = math.tanh(s / (self._y_amp or 1.0)) + return self._y_mean + 0.5 * self._y_amp * norm + + +class _SklearnGbdtBackbone(ModelHandle): + """真实 GBDT 主干(sklearn GradientBoostingRegressor)。 + + 仅当运行环境存在 sklearn 时启用;与 stub 接口完全一致。 + """ + + def __init__(self, params: Dict[str, Any]): + super().__init__(backbone="gbdt", params=params) + # 延迟 import,避免无 sklearn 环境加载失败 + from sklearn.ensemble import GradientBoostingRegressor # type: ignore + self._Clz = GradientBoostingRegressor + self._model: Any = None + + def _fit_impl(self, X, y) -> None: + kw = { + "n_estimators": int(self.params.get("n_estimators", 100)), + "max_depth": int(self.params.get("max_depth", 3)), + "learning_rate": float(self.params.get("learning_rate", 0.1)), + "random_state": int(self.params.get("random_state", 42)), + } + self._model = self._Clz(**kw) + self._model.fit(list(X), list(y)) + self.meta.update(kw) + + def _predict_one(self, row) -> float: + return float(self._model.predict([list(row)])[0]) + + +class _SklearnDnnBackbone(ModelHandle): + """真实轻量 DNN 主干(sklearn MLPRegressor)。 + + PRD 5.3 备选结构;仅当运行环境存在 sklearn 时启用。 + """ + + def __init__(self, params: Dict[str, Any]): + super().__init__(backbone="dnn", params=params) + from sklearn.neural_network import MLPRegressor # type: ignore + self._Clz = MLPRegressor + self._model: Any = None + + def _fit_impl(self, X, y) -> None: + kw = { + "hidden_layer_sizes": tuple( + self.params.get("hidden_layer_sizes", (32, 16))), + "max_iter": int(self.params.get("max_iter", 500)), + "random_state": int(self.params.get("random_state", 42)), + } + self._model = self._Clz(**kw) + self._model.fit(list(X), list(y)) + self.meta.update({"hidden_layer_sizes": list(kw["hidden_layer_sizes"]), + "max_iter": kw["max_iter"]}) + + def _predict_one(self, row) -> float: + return float(self._model.predict([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 gbdt_backbone(hyperparams: Dict[str, Any]) -> ModelHandle: + """gbdt 主干工厂:有 sklearn 用真实 GBDT,否则退化为 stub。 + + PRD 5.3 推荐的监督回归默认结构(梯度提升回归)。 + """ + if _has_sklearn(): + return _SklearnGbdtBackbone(hyperparams) + # 无 sklearn:退化 stub 但保留声明主干名,便于审计 + h = _StubBackbone(hyperparams) + h.meta["degraded_from"] = "gbdt" + return h + + +def dnn_backbone(hyperparams: Dict[str, Any]) -> ModelHandle: + """dnn 主干工厂:有 sklearn 用真实 MLP,否则退化为 stub。""" + if _has_sklearn(): + return _SklearnDnnBackbone(hyperparams) + h = _StubBackbone(hyperparams) + h.meta["degraded_from"] = "dnn" + return h + + +#: 主干注册表:新增结构走 ``register_backbone`` 注册,不动内核 +#: (对齐 PRD 5.3「新增结构走插件注册」理念,风格对齐 #34)。 +BACKBONES: Dict[str, Any] = { + "gbdt": gbdt_backbone, + "dnn": dnn_backbone, + "stub": stub_backbone, +} + + +def register_backbone(name: str, factory: Any) -> None: + """注册一个新主干工厂 ``factory(hyperparams) -> ModelHandle``。 + + 允许高级行业模板声明非默认主干(如自研网络),不动内核——对齐 PRD + 「新增结构走插件注册而非改内核」。 + """ + if not callable(factory): + raise QualityForecastError("主干工厂必须是可调用对象") + BACKBONES[name] = factory + + +def _build_backbone(backbone: str, hyperparams: Dict[str, Any]) -> ModelHandle: + factory = BACKBONES.get(backbone) + if factory is None: + raise QualityForecastError( + f"未注册的主干类型:{backbone!r},已注册:{list(BACKBONES)}") + return factory(hyperparams) + + +# --------------------------------------------------------------------------- +# 质量预测模型:固定主干 + 配方加载 +# --------------------------------------------------------------------------- + +class QualityForecastModel: + """质量预测模型(固定主干 + 配方加载)。 + + 业务侧两种等价入口: + + 1. 直接构造(显式主干):: + + m = QualityForecastModel(backbone="gbdt", hyperparams={...}) + + 2. 配方加载(推荐,切换模板仅改配方):: + + m = build_from_recipe("templates/.../quality-forecast/recipe.ti.json") + """ + + def __init__(self, backbone: str = "gbdt", + hyperparams: Optional[Dict[str, Any]] = None, + feature_columns: Optional[Sequence[str]] = None, + target_column: str = "quality_index", + accuracy_floor: float = DEFAULT_ACCURACY_FLOOR): + self.recipe_meta: Dict[str, Any] = { + "backbone": backbone, + "hyperparams": dict(hyperparams or {}), + "feature_columns": list(feature_columns or []), + "target_column": target_column, + "accuracy_floor": accuracy_floor, + } + self._handle: ModelHandle = _build_backbone(backbone, hyperparams or {}) + + @classmethod + def from_recipe(cls, recipe: Recipe) -> "QualityForecastModel": + """从一个 ``Recipe`` 构造模型(推荐入口)。""" + m = cls( + backbone=recipe.backbone, + hyperparams=recipe.hyperparams, + feature_columns=recipe.feature_columns, + target_column=recipe.target_column, + accuracy_floor=recipe.accuracy_floor, + ) + m.recipe_meta["recipe_name"] = recipe.name + m.recipe_meta["industry"] = recipe.industry + return m + + # ---- 训练 / 推理 ---- + + def fit(self, X: Sequence[Sequence[float]], y: Sequence[float]) -> "QualityForecastModel": + self._handle.fit(X, y) + return self + + def predict(self, X: Sequence[Sequence[float]]) -> List[float]: + return self._handle.predict(X) + + @property + def fitted(self) -> bool: + return self._handle.fitted + + # ---- 验收口径 ---- + + def evaluate(self, X: Sequence[Sequence[float]], + y: Sequence[float]) -> "Accuracy": + """评估并返回准确率/MAE/RMSE 与是否达标。 + + 准确率口径(PRD 5.3 / 里程碑):相对误差在容忍带 + ``tolerance``(默认 10%)内计为命中。``accuracy >= accuracy_floor`` + 即视为达标(默认 90%)。 + """ + preds = self.predict(X) + return Accuracy.compute( + y_true=list(y), y_pred=preds, + accuracy_floor=self.recipe_meta["accuracy_floor"]) + + def to_dict(self) -> Dict[str, Any]: + return { + "recipe_meta": dict(self.recipe_meta), + "handle": self._handle.to_dict(), + } + + +# --------------------------------------------------------------------------- +# 验收:Accuracy +# --------------------------------------------------------------------------- + +@dataclass(frozen=True) +class Accuracy: + """质量预测验收结果(PRD 5.3 准确率口径)。""" + + accuracy: float + mae: float + rmse: float + tolerance: float + accuracy_floor: float + passed: bool + + def to_dict(self) -> Dict[str, Any]: + return { + "accuracy": self.accuracy, + "mae": self.mae, + "rmse": self.rmse, + "tolerance": self.tolerance, + "accuracy_floor": self.accuracy_floor, + "passed": self.passed, + } + + @classmethod + def compute(cls, y_true: Sequence[float], y_pred: Sequence[float], + tolerance: float = 0.10, + accuracy_floor: float = DEFAULT_ACCURACY_FLOOR) -> "Accuracy": + if len(y_true) != len(y_pred): + raise QualityForecastError( + f"y_true/y_pred 长度不一致:{len(y_true)} != {len(y_pred)}") + if not y_true: + raise QualityForecastError("评估数据为空") + n = len(y_true) + hits = 0 + abs_err_sum = 0.0 + sq_err_sum = 0.0 + for yt, yp in zip(y_true, y_pred): + denom = abs(yt) if abs(yt) > 1e-9 else 1.0 + rel = abs(yp - yt) / denom + if rel <= tolerance: + hits += 1 + abs_err_sum += abs(yp - yt) + sq_err_sum += (yp - yt) ** 2 + accuracy = hits / n + mae = abs_err_sum / n + rmse = math.sqrt(sq_err_sum / n) + return cls( + accuracy=accuracy, mae=mae, rmse=rmse, + tolerance=tolerance, accuracy_floor=accuracy_floor, + passed=accuracy >= accuracy_floor, + ) + + +# --------------------------------------------------------------------------- +# 配方构建入口 + 样例协议 +# --------------------------------------------------------------------------- + +def build_from_recipe(path: str) -> QualityForecastModel: + """从 JSON 配方文件加载并构造一个质量预测模型(推荐入口)。 + + 切换模板仅改配方文件,业务代码零改动——对齐 PRD 5.3 验收口径。 + """ + return QualityForecastModel.from_recipe(load_recipe(path)) + + +def _samples_dir() -> str: + return os.path.join(os.path.dirname(os.path.abspath(__file__)), + "samples", "quality-forecast") + + +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/quality-forecast/recipe.resin.json b/core/model-framework/samples/quality-forecast/recipe.resin.json new file mode 100644 index 0000000..2cc3661 --- /dev/null +++ b/core/model-framework/samples/quality-forecast/recipe.resin.json @@ -0,0 +1,21 @@ +{ + "name": "resin-quality", + "backbone": "gbdt", + "industry": "吸附树脂(已终验化工新材料AI平台 baseline)", + "hyperparams": { + "n_estimators": 100, + "max_depth": 3, + "learning_rate": 0.1, + "random_state": 7 + }, + "feature_columns": [ + "reactor_temp", + "reactor_pressure", + "flow_rate", + "ph_value", + "conversion_rate" + ], + "target_column": "resin_purity_index", + "accuracy_floor": 0.90, + "notes": "PRD 5.3 ① 质量预测:树脂纯度/合格率预测,复用已交付化工AI平台 baseline 超参。" +} diff --git a/core/model-framework/samples/quality-forecast/recipe.ti.json b/core/model-framework/samples/quality-forecast/recipe.ti.json new file mode 100644 index 0000000..c733821 --- /dev/null +++ b/core/model-framework/samples/quality-forecast/recipe.ti.json @@ -0,0 +1,22 @@ +{ + "name": "ti-cl4-quality", + "backbone": "gbdt", + "industry": "海绵钛氯化车间(Template-Ti 一期)", + "hyperparams": { + "n_estimators": 120, + "max_depth": 4, + "learning_rate": 0.08, + "random_state": 42 + }, + "feature_columns": [ + "furnace_temp", + "furnace_pressure", + "cl2_flow", + "ti_feed_rate", + "impurity_fe", + "impurity_v" + ], + "target_column": "ti_product_grade_index", + "accuracy_floor": 0.90, + "notes": "PRD 5.3 ① 质量预测:氯化车间一次合格率预测,验收准确率≥90%(PRD 第6章里程碑)。一期数据门槛:≥6个月标注(LIMS对接后补标)。" +} diff --git a/core/model-framework/tests/__init__.py b/core/model-framework/tests/__init__.py new file mode 100644 index 0000000..40a96af --- /dev/null +++ b/core/model-framework/tests/__init__.py @@ -0,0 +1 @@ +# -*- coding: utf-8 -*- 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_quality_forecast.py b/core/model-framework/tests/test_quality_forecast.py new file mode 100644 index 0000000..395e0fb --- /dev/null +++ b/core/model-framework/tests/test_quality_forecast.py @@ -0,0 +1,255 @@ +# -*- coding: utf-8 -*- +"""``quality_forecast`` 单元测试(issue #36)。 + +覆盖: +- 配方(Recipe)不可变性 / 序列化往返 / 非法主干与越界校验; +- 主干工厂注册表 + 自定义主干注册(PRD 5.3「新增结构走插件注册」); +- stub / gbdt / dnn 三类主干的 fit/predict/evaluate 契约; +- 固定主干 + 配方加载:同框架加载 Ti / 树脂两套配方均跑通(PRD 5.3 + 验收口径); +- Accuracy 验收口径(PRD 5.3 / 里程碑:准确率 ≥ 90%); +- 零外部强依赖:无 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 + Accuracy, + BACKBONES, + ModelHandle, + QualityForecastError, + QualityForecastModel, + Recipe, + build_from_recipe, + dnn_backbone, + gbdt_backbone, + list_sample_recipes, + load_recipe, + register_backbone, + sample_recipe_path, + stub_backbone, +) + + +def _linear_dataset(n=40, noise=0.0): + """构造一个 y ≈ 2*x0 + x1 的可学习数据集(带可选噪声)。""" + X, y = [], [] + for i in range(n): + x0 = float(i % 7) + 1.0 + x1 = float(i % 5) * 0.5 + 0.5 + yv = 2.0 * x0 + x1 + noise * (i % 3 - 1) + X.append([x0, x1]) + y.append(yv) + return X, y + + +class TestRecipe(unittest.TestCase): + """配方数据对象与校验。""" + + def test_defaults_and_immutability(self): + r = Recipe(name="t") + self.assertEqual(r.backbone, "gbdt") + self.assertEqual(r.target_column, "quality_index") + self.assertAlmostEqual(r.accuracy_floor, 0.90) + with self.assertRaises(Exception): + r.name = "other" # frozen + + def test_roundtrip(self): + r = Recipe(name="t", backbone="dnn", + hyperparams={"max_iter": 50}, + feature_columns=("a", "b"), + target_column="y", + accuracy_floor=0.8, 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(QualityForecastError): + Recipe(name="t", backbone="svm") + + def test_accuracy_floor_out_of_range(self): + with self.assertRaises(QualityForecastError): + Recipe(name="t", accuracy_floor=1.5) + with self.assertRaises(QualityForecastError): + Recipe(name="t", accuracy_floor=-0.1) + + def test_missing_name(self): + with self.assertRaises(QualityForecastError): + 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-quality") + self.assertEqual(r.backbone, "gbdt") + self.assertIn("furnace_temp", r.feature_columns) + + +class TestBackbones(unittest.TestCase): + """主干工厂与注册表。""" + + def test_builtin_backbones_registered(self): + for name in ("gbdt", "dnn", "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, y): + self._v = sum(y) / len(y) + + def _predict_one(self, row): + return self._v + + register_backbone("custom_test", lambda p: _Custom(p)) + m = QualityForecastModel(backbone="custom_test") + X, y = _linear_dataset() + m.fit(X, y) + self.assertEqual(len(m.predict(X)), len(X)) + # 清理避免污染其它用例 + BACKBONES.pop("custom_test", None) + + def test_unknown_backbone_raises(self): + with self.assertRaises(QualityForecastError): + QualityForecastModel(backbone="not_a_backbone") + + def test_stub_predict_is_deterministic(self): + h = stub_backbone({}) + X, y = _linear_dataset() + h.fit(X, y) + p1 = h.predict(X) + p2 = h.predict(X) + self.assertEqual(p1, p2) + self.assertTrue(all(isinstance(v, float) for v in p1)) + + def test_gbdt_factory_runs_with_or_without_sklearn(self): + # 无论 sklearn 是否存在都不应报错 + h = gbdt_backbone({"n_estimators": 20, "max_depth": 2}) + X, y = _linear_dataset() + h.fit(X, y) + preds = h.predict(X) + self.assertEqual(len(preds), len(y)) + + +class TestModelContract(unittest.TestCase): + """模型 fit/predict/evaluate 契约。""" + + def test_fit_predict_shapes(self): + m = QualityForecastModel(backbone="stub") + X, y = _linear_dataset(20) + m.fit(X, y) + self.assertTrue(m.fitted) + preds = m.predict(X) + self.assertEqual(len(preds), len(y)) + + def test_predict_before_fit_raises(self): + m = QualityForecastModel(backbone="stub") + with self.assertRaises(QualityForecastError): + m.predict([[1.0, 2.0]]) + + def test_fit_mismatched_lengths_raises(self): + m = QualityForecastModel(backbone="stub") + with self.assertRaises(QualityForecastError): + m.fit([[1.0], [2.0]], [1.0]) + + def test_fit_empty_raises(self): + m = QualityForecastModel(backbone="stub") + with self.assertRaises(QualityForecastError): + m.fit([], []) + + def test_to_dict_roundtrip_meta(self): + m = QualityForecastModel(backbone="gbdt", + hyperparams={"n_estimators": 5}, + feature_columns=["a"], + target_column="y") + d = m.to_dict() + self.assertEqual(d["recipe_meta"]["backbone"], "gbdt") + self.assertIn("handle", d) + + +class TestAccuracy(unittest.TestCase): + """验收口径(PRD 5.3:准确率 ≥ 90%)。""" + + def test_perfect_predictions_pass(self): + y = [10.0, 20.0, 30.0, 40.0] + acc = Accuracy.compute(y, y, accuracy_floor=0.9) + self.assertAlmostEqual(acc.accuracy, 1.0) + self.assertAlmostEqual(acc.mae, 0.0) + self.assertAlmostEqual(acc.rmse, 0.0) + self.assertTrue(acc.passed) + + def test_bad_predictions_fail(self): + y_true = [10.0, 20.0, 30.0, 40.0] + y_pred = [11.0, 50.0, 5.0, 80.0] # 大偏差 + acc = Accuracy.compute(y_true, y_pred, accuracy_floor=0.9) + self.assertLess(acc.accuracy, 0.9) + self.assertFalse(acc.passed) + self.assertGreater(acc.mae, 0.0) + self.assertGreater(acc.rmse, 0.0) + + def test_length_mismatch_raises(self): + with self.assertRaises(QualityForecastError): + Accuracy.compute([1.0, 2.0], [1.0]) + + def test_empty_raises(self): + with self.assertRaises(QualityForecastError): + Accuracy.compute([], []) + + def test_evaluate_end_to_end(self): + # stub 主干在确定性、低噪声线性数据上应能给出确定性的验收结果 + m = QualityForecastModel(backbone="stub", accuracy_floor=0.0) + X, y = _linear_dataset(30) + m.fit(X, y) + acc = m.evaluate(X, y) + self.assertIsInstance(acc, Accuracy) + self.assertEqual(acc.to_dict()["accuracy_floor"], 0.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"], ("gbdt", "dnn", "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(12)] + y = [float(i % 4) + 1.0 for i in range(12)] + m.fit(X, y) + preds = m.predict(X) + self.assertEqual(len(preds), len(y)) + acc = m.evaluate(X, y) + self.assertIsInstance(acc, Accuracy) + + 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)