From a10b83007e2b3e9b223c2b9d9ee7ad52ab502bac Mon Sep 17 00:00:00 2001 From: bot_dev1 Date: Wed, 5 Aug 2026 05:17:36 +0800 Subject: [PATCH] =?UTF-8?q?feat(#73):=20[Ti-1]=20=E6=A8=A1=E5=9E=8B?= =?UTF-8?q?=E9=83=A8=E7=BD=B2=E5=88=B0=E5=86=85=E6=A0=B8=E5=B9=B6=E6=8E=A5?= =?UTF-8?q?=E5=85=A5=E9=A9=BE=E9=A9=B6=E8=88=B1=EF=BC=88=E9=A2=84=E6=B5=8B?= =?UTF-8?q?=E6=9C=8D=E5=8A=A1+=E5=91=8A=E8=AD=A6+=E5=8F=AF=E8=A7=A3?= =?UTF-8?q?=E9=87=8A=E8=A7=86=E5=9B=BE+=E6=BC=82=E7=A7=BB=E7=9B=91?= =?UTF-8?q?=E6=B5=8B=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 新增 templates/ti-cl4/quality-forecast/serve.py： - PredictionService：特征抽取→模型预测→告警判定全链路，部署前 validate - Thresholds/AlarmLevel：纯度min/杂质max 双向阈值，normal/warning/critical 三级 - CockpitView：渲染 kpi_card+alarm_panel+explanation_list（对齐 cockpit-layout-v1） 含 top-3 特征贡献溯源，满足 PRD 可解释可溯源要求 - DriftMonitor：累计 R²/MAE，超阈值触发重训信号（PRD §5.3/§9） - DeploymentBundle：模型+特征清单+阈值+置信带整体序列化往返 - config/thresholds.template.yaml：TiCl₄纯度下限 99.2 阈值模板 - 14 用例（累计 53 用例）全通过；纯标准库零运行时依赖。 --- .../config/thresholds.template.yaml | 10 + templates/ti-cl4/quality-forecast/serve.py | 344 ++++++++++++++++++ .../quality-forecast/tests/test_serve.py | 180 +++++++++ 3 files changed, 534 insertions(+) create mode 100644 templates/ti-cl4/quality-forecast/config/thresholds.template.yaml create mode 100644 templates/ti-cl4/quality-forecast/serve.py create mode 100644 templates/ti-cl4/quality-forecast/tests/test_serve.py diff --git a/templates/ti-cl4/quality-forecast/config/thresholds.template.yaml b/templates/ti-cl4/quality-forecast/config/thresholds.template.yaml new file mode 100644 index 0000000..b21001d --- /dev/null +++ b/templates/ti-cl4/quality-forecast/config/thresholds.template.yaml @@ -0,0 +1,10 @@ +# -*- coding: utf-8 -*- +# Ti-1 氯化车间质量预测 · 部署阈值（Issue #73）。 +# +# target 须与 model.template.yaml 的 target 一致。 +# min/max 二选一或都填：纯度类用 min（低于下限告警），杂质类用 max。 +# warning_band：距阈值多少开始预警（如纯度下限 99.2，band 0.3 → 99.2~99.5 预警）。 + +target: RF-01.PURITY +min: 99.2 +warning_band: 0.3 diff --git a/templates/ti-cl4/quality-forecast/serve.py b/templates/ti-cl4/quality-forecast/serve.py new file mode 100644 index 0000000..c98a843 --- /dev/null +++ b/templates/ti-cl4/quality-forecast/serve.py @@ -0,0 +1,344 @@ +# -*- coding: utf-8 -*- +"""Ti-1 氯化车间质量预测 · 模型部署到内核并接入驾驶舱（Issue #73 / PRD §5.3 ① / §5.5）。 + +承接 #69（模型训练）产出的 ``QualityModel``：把"训练好的模型 → 可调用的预测服务 +→ 驾驶舱可展示的预测+可解释结果"这条链路**模板化、可测试**。 + +PRD 设计口径 +------------ +- 架构表（PRD §5.3）：``质量预测 | 预测 | 入：DCS实时数据+LIMS； + 出：质量指标预测值(纯度/杂质) | ① 质量预测 | 中``。 +- 驾驶舱接入（PRD §5.5）：预测结果以 kpi_card / alarm_panel 形式渲染， + 支持按模板渲染四状态流程视图；移动端交接班摘要引用预测结论。 +- 模型漂移（PRD §5.3 / §9 NFR）：质保期监测准确率/误报率，触发重训。 + +本模块交付 +---------- +1. **预测服务 ``PredictionService``**：加载（或内存持有）QualityModel + 特征清单， + ``predict(samples)`` 返回 ``Prediction``（预测值 + 置信区间 + 命中阈值判定）。 +2. **阈值告警 ``QualityAlarm``**：按超参包的 ``thresholds``（纯度下限/杂质上限） + 判定告警等级（normal/warning/critical），供驾驶舱 alarm_panel。 +3. **驾驶舱视图 ``CockpitView``**：把预测结果 + explain() 渲染为驾驶舱布局 + 片段（kpi_card / alarm_panel / 可解释溯源列表），对齐 iAOP-cockpit-layout-v1。 +4. **模型漂移监测 ``DriftMonitor``**：累计预测的 R²/MAE，超阈值触发重训信号。 + +设计要点 +-------- +- **零运行时依赖**（纯标准库）：与 #68/#69 一致。 +- **可解释接入驾驶舱**：预测结果附带 top-N 特征贡献（来自 #69 explain）， + 满足 PRD"要求结果可解释、可溯源，要引用依据"。 +- **可校验**：部署前 ``PredictionService.validate`` 检查模型已训练、特征清单 + 与模型特征名对齐。 +""" +from __future__ import annotations + +import json +import math +import os +from dataclasses import dataclass, field +from enum import Enum +from typing import Any, Dict, List, Optional, Sequence + +from .features import FeatureExtractor, FeatureMatrix, FeatureSpec, Sample +from .model import Evaluation, ModelRecipe, QualityModel + +NAN = float("nan") + + +class ServeError(ValueError): + """模型部署/预测服务错误。""" + + +# --------------------------------------------------------------------------- +# 告警等级 +# --------------------------------------------------------------------------- + + +class AlarmLevel(str, Enum): + NORMAL = "normal" # 正常 + WARNING = "warning" # 预警（接近阈值） + CRITICAL = "critical" # 超限（质量不达标） + + +@dataclass +class Thresholds: + """质量阈值（超参包外置）。""" + + target: str + min_value: Optional[float] = None # 纯度下限（低于则告警） + max_value: Optional[float] = None # 杂质上限（高于则告警） + warning_band: float = 0.0 # 预警带（距阈值多少开始 warning） + + @classmethod + def from_dict(cls, d: Dict[str, Any]) -> "Thresholds": + return cls( + target=str(d.get("target", "")).strip(), + min_value=(float(d["min"]) if d.get("min") is not None else None), + max_value=(float(d["max"]) if d.get("max") is not None else None), + warning_band=float(d.get("warning_band", 0.0)), + ) + + def validate(self) -> List[str]: + errs = [] + if not self.target: + errs.append("thresholds.target 不能为空") + if self.min_value is None and self.max_value is None: + errs.append("thresholds 至少需 min 或 max 之一") + return errs + + +# --------------------------------------------------------------------------- +# 预测结果 +# --------------------------------------------------------------------------- + + +@dataclass +class Prediction: + """单条预测结果。""" + + target: str + value: float + unit: str + level: str # AlarmLevel 值 + message: str + confidence_band: float # ±置信带（基于训练 RMSE） + explanation: List[Dict[str, Any]] = field(default_factory=list) + timestamp: float = 0.0 + + def to_dict(self) -> Dict[str, Any]: + return { + "target": self.target, "value": round(self.value, 4), + "unit": self.unit, "level": self.level, "message": self.message, + "confidence_band": round(self.confidence_band, 4), + "lower": (round(self.value - self.confidence_band, 4) + if not math.isnan(self.value) else None), + "upper": (round(self.value + self.confidence_band, 4) + if not math.isnan(self.value) else None), + "explanation": self.explanation, + "timestamp": self.timestamp, + } + + +# --------------------------------------------------------------------------- +# 预测服务 +# --------------------------------------------------------------------------- + + +class PredictionService: + """质量预测部署服务：特征抽取 + 模型预测 + 告警判定。""" + + def __init__(self, model: QualityModel, extractor: FeatureExtractor, + thresholds: Optional[Thresholds] = None, + *, confidence_rmse: float = 0.0): + self.model = model + self.extractor = extractor + self.thresholds = thresholds + self.confidence_rmse = confidence_rmse # 训练 RMSE，作为置信带 + self.validate() + + def validate(self) -> None: + errs: List[str] = [] + if not self.model.fitted: + errs.append("模型未训练，无法部署") + # 特征清单与模型特征名对齐 + model_names = self.model.recipe.feature_names + ext_names = self.extractor.names + if model_names and model_names != ext_names: + errs.append( + f"特征清单与模型特征名不一致: 清单={ext_names} 模型={model_names}") + if self.thresholds: + errs.extend(self.thresholds.validate()) + if (self.model.recipe.target + and self.thresholds.target + and self.thresholds.target != self.model.recipe.target): + errs.append( + f"阈值 target({self.thresholds.target}) 与模型 " + f"target({self.model.recipe.target}) 不一致") + if errs: + raise ServeError("部署校验失败:\n " + "\n ".join(errs)) + + def predict(self, samples: Sequence[Sample]) -> List[Prediction]: + """对时序样本预测（每个样本时刻一条预测）。""" + matrix: FeatureMatrix = self.extractor.extract(samples) + # 模型特征名顺序（若清单与模型一致，直接用矩阵列） + X = matrix.rows + preds_raw = self.model.predict(X) + explanation = self.model.explain() + out: List[Prediction] = [] + # 样本时刻（extract 保留输入顺序） + ts_list = [s.ts for s in sorted(samples, key=lambda s: s.ts)] + for i, raw in enumerate(preds_raw): + level, msg = self._classify(raw) + ts = ts_list[i] if i < len(ts_list) else 0.0 + out.append(Prediction( + target=self.model.recipe.target, value=raw, + unit=self.model.recipe.unit, level=level, message=msg, + confidence_band=self.confidence_rmse, + explanation=explanation, timestamp=ts)) + return out + + def _classify(self, value: float) -> tuple: + """按阈值判定告警等级。""" + if math.isnan(value) or self.thresholds is None: + return AlarmLevel.NORMAL.value, "无阈值或预测缺失，未判定" + th = self.thresholds + band = th.warning_band + if th.min_value is not None: + if value < th.min_value: + return AlarmLevel.CRITICAL.value, ( + f"{th.target}={value:.3f} 低于下限 {th.min_value}") + if value < th.min_value + band: + return AlarmLevel.WARNING.value, ( + f"{th.target}={value:.3f} 接近下限 {th.min_value}") + if th.max_value is not None: + if value > th.max_value: + return AlarmLevel.CRITICAL.value, ( + f"{th.target}={value:.3f} 超过上限 {th.max_value}") + if value > th.max_value - band: + return AlarmLevel.WARNING.value, ( + f"{th.target}={value:.3f} 接近上限 {th.max_value}") + return AlarmLevel.NORMAL.value, f"{th.target}={value:.3f} 达标" + + +# --------------------------------------------------------------------------- +# 驾驶舱视图 +# --------------------------------------------------------------------------- + + +class CockpitView: + """把预测结果渲染为驾驶舱布局片段（对齐 iAOP-cockpit-layout-v1）。""" + + WIDGET_TYPES = {"kpi_card", "alarm_panel", "explanation_list"} + + def __init__(self, title: str = "质量预测"): + self.title = title + + def render(self, prediction: Prediction, + *, position: Optional[Dict[str, int]] = None) -> Dict[str, Any]: + """渲染单个预测为 kpi_card + alarm_panel + 可解释溯源列表。""" + pos = position or {"x": 0, "y": 0, "w": 6, "h": 2} + level_color = { + AlarmLevel.NORMAL.value: "green", + AlarmLevel.WARNING.value: "yellow", + AlarmLevel.CRITICAL.value: "red", + }[prediction.level] + # top-3 特征贡献（可溯源） + top = sorted(prediction.explanation, + key=lambda e: e.get("importance", 0), reverse=True)[:3] + return { + "$schema": "iAOP-cockpit-layout-v1", + "title": self.title, + "widgets": [ + { + "type": "kpi_card", + "metric": prediction.target, + "label": self.title, + "value": round(prediction.value, 3), + "unit": prediction.unit, + "level": prediction.level, + "color": level_color, + **pos, + "description": prediction.message, + }, + { + "type": "alarm_panel", + "metric": prediction.target, + "level": prediction.level, + "message": prediction.message, + "color": level_color, + "x": pos["x"], "y": pos["y"] + pos["h"], + "w": pos["w"], "h": 1, + "description": "质量预测告警面板", + }, + { + "type": "explanation_list", + "metric": prediction.target, + "items": top, + "x": pos["x"], "y": pos["y"] + pos["h"] + 1, + "w": pos["w"], "h": 2, + "description": "预测依据（top-3 特征贡献，可溯源）", + }, + ], + } + + +# --------------------------------------------------------------------------- +# 模型漂移监测 +# --------------------------------------------------------------------------- + + +class DriftMonitor: + """累计预测的评估指标，超阈值触发重训信号（PRD §5.3 / §9）。""" + + def __init__(self, *, min_r2: float = 0.8, max_mae: float = 1.0): + self.min_r2 = min_r2 + self.max_mae = max_mae + self.history: List[Evaluation] = [] + + def record(self, evaluation: Evaluation) -> None: + self.history.append(evaluation) + + def should_retrain(self) -> tuple: + """最近一次评估是否触发重训。返回 (是否重训, 原因)。""" + if not self.history: + return False, "无评估记录" + last = self.history[-1] + if math.isnan(last.r2): + return True, f"R² 异常（NaN），建议重训" + if last.r2 < self.min_r2: + return True, f"R²={last.r2:.3f} < {self.min_r2}，模型退化" + if last.mae > self.max_mae: + return True, f"MAE={last.mae:.3f} > {self.max_mae}，误差超限" + return False, f"R²={last.r2:.3f} MAE={last.mae:.3f} 达标" + + +# --------------------------------------------------------------------------- +# 部署包加载 +# --------------------------------------------------------------------------- + + +@dataclass +class DeploymentBundle: + """模型部署包：模型 + 特征清单 + 阈值 + 置信带，可整体序列化。""" + + model: QualityModel + extractor: FeatureExtractor + thresholds: Optional[Thresholds] = None + confidence_rmse: float = 0.0 + + def to_service(self) -> PredictionService: + return PredictionService( + self.model, self.extractor, self.thresholds, + confidence_rmse=self.confidence_rmse) + + def to_dict(self) -> Dict[str, Any]: + return { + "model": self.model.to_dict(), + "feature_specs": [ + {"name": s.name, "source": s.source, "transform": s.transform, + "window": s.window, "denominator": s.denominator, + "meaning": s.meaning, "unit": s.unit} + for s in self.extractor.specs], + "thresholds": ({ + "target": self.thresholds.target, + "min": self.thresholds.min_value, + "max": self.thresholds.max_value, + "warning_band": self.thresholds.warning_band} + if self.thresholds else None), + "confidence_rmse": self.confidence_rmse, + } + + @classmethod + def from_dict(cls, d: Dict[str, Any]) -> "DeploymentBundle": + from .features import FeatureSpec as FS + model = QualityModel.from_dict(d["model"]) + specs = [FS(name=s["name"], source=s["source"], + transform=s.get("transform", "raw"), + window=float(s.get("window", 60.0)), + denominator=s.get("denominator"), + meaning=s.get("meaning", ""), unit=s.get("unit", "")) + for s in d.get("feature_specs", [])] + ext = FeatureExtractor(specs, strict=False) + th_raw = d.get("thresholds") + th = Thresholds.from_dict(th_raw) if th_raw else None + return cls(model, ext, th, float(d.get("confidence_rmse", 0.0))) diff --git a/templates/ti-cl4/quality-forecast/tests/test_serve.py b/templates/ti-cl4/quality-forecast/tests/test_serve.py new file mode 100644 index 0000000..0d557eb --- /dev/null +++ b/templates/ti-cl4/quality-forecast/tests/test_serve.py @@ -0,0 +1,180 @@ +# -*- coding: utf-8 -*- +"""Ti-1 质量预测部署接入驾驶舱测试（Issue #73）。 + +覆盖： +1. PredictionService 部署校验（模型未训练/特征名不一致/阈值target不一致）； +2. predict 全链路（特征抽取→预测→告警判定）； +3. Thresholds 分级（normal/warning/critical，min/max 双向）； +4. CockpitView 渲染 kpi_card/alarm_panel/explanation_list； +5. DriftMonitor 重训触发； +6. DeploymentBundle 序列化往返。 +""" +import math +import os +import sys +import unittest + +HERE = os.path.dirname(os.path.abspath(__file__)) +sys.path.insert(0, HERE) +import _bootstrap # noqa: F401,E402 + +from quality_forecast import features as F # noqa: E402 +from quality_forecast import model as M # noqa: E402 +from quality_forecast import serve as S # noqa: E402 + + +def _trained_service(*, thresholds=None, alpha=0.001): + """构造一个已训练的服务：y = TEMP（线性，易拟合）。""" + spec = F.FeatureSpec("t", "CLF-01.TEMP", "raw") + ext = F.FeatureExtractor([spec]) + samples = [F.Sample(ts=i * 10.0, values={"CLF-01.TEMP": 100.0 + i}) + for i in range(6)] + matrix = ext.extract(samples) + # target = TEMP 本身（完美线性，R²≈1） + recipe = M.ModelRecipe(target="RF-01.PURITY", alpha=alpha, + feature_names=["t"], unit="%") + model = M.QualityModel(recipe).fit(matrix.rows, + [s.values["CLF-01.TEMP"] for s in samples]) + ev = model.evaluate(matrix.rows, [s.values["CLF-01.TEMP"] for s in samples]) + return S.PredictionService(model, ext, thresholds, + confidence_rmse=ev.rmse), samples + + +class TestPredictionServiceValidation(unittest.TestCase): + def test_unfitted_model_raises(self): + spec = F.FeatureSpec("t", "CLF-01.TEMP", "raw") + ext = F.FeatureExtractor([spec]) + model = M.QualityModel(M.ModelRecipe(target="RF-01.PURITY", + feature_names=["t"])) + with self.assertRaises(S.ServeError): + S.PredictionService(model, ext) + + def test_feature_name_mismatch_raises(self): + # 清单 [t] 但模型声明 [other] + spec = F.FeatureSpec("t", "CLF-01.TEMP", "raw") + ext = F.FeatureExtractor([spec]) + recipe = M.ModelRecipe(target="y", alpha=0.1, feature_names=["other"]) + model = M.QualityModel(recipe).fit([[1.0], [2.0]], [1.0, 2.0]) + with self.assertRaises(S.ServeError): + S.PredictionService(model, ext) + + def test_threshold_target_mismatch_raises(self): + svc, _ = _trained_service() + th = S.Thresholds(target="WRONG.TARGET", min_value=50.0) + with self.assertRaises(S.ServeError): + S.PredictionService(svc.model, svc.extractor, th) + + +class TestPredictionAndThresholds(unittest.TestCase): + def test_predict_normal(self): + th = S.Thresholds(target="RF-01.PURITY", min_value=50.0, + warning_band=5.0) + svc, samples = _trained_service(thresholds=th) + preds = svc.predict(samples) + self.assertEqual(len(preds), len(samples)) + # TEMP 100-105 → 远高于 50 → normal + self.assertEqual(preds[-1].level, "normal") + + def test_predict_critical_low(self): + th = S.Thresholds(target="RF-01.PURITY", min_value=200.0) + svc, samples = _trained_service(thresholds=th) + preds = svc.predict(samples) + # 预测 ~100-105 < 200 → critical + self.assertEqual(preds[-1].level, "critical") + self.assertIn("低于下限", preds[-1].message) + + def test_predict_warning_band(self): + th = S.Thresholds(target="RF-01.PURITY", min_value=95.0, + warning_band=10.0) + svc, samples = _trained_service(thresholds=th) + preds = svc.predict(samples) + # 100-105 在 [95, 105] 预警带内 → warning（部分） + levels = {p.level for p in preds} + self.assertTrue(levels & {"warning", "critical"} or + "warning" in levels) + + def test_max_threshold(self): + # 杂质类：超过上限 critical + spec = F.FeatureSpec("t", "RF-01.IMP", "raw") + ext = F.FeatureExtractor([spec]) + samples = [F.Sample(ts=i, values={"RF-01.IMP": 0.1 * (i + 1)}) + for i in range(5)] + matrix = ext.extract(samples) + recipe = M.ModelRecipe(target="RF-01.IMP", alpha=0.001, + feature_names=["t"]) + model = M.QualityModel(recipe).fit(matrix.rows, + [s.values["RF-01.IMP"] for s in samples]) + th = S.Thresholds(target="RF-01.IMP", max_value=0.3, warning_band=0.1) + svc = S.PredictionService(model, ext, th) + preds = svc.predict(samples) + # 最后样本 IMP=0.5 > 0.3 → critical + self.assertEqual(preds[-1].level, "critical") + + def test_confidence_band(self): + th = S.Thresholds(target="RF-01.PURITY", min_value=50.0) + svc, samples = _trained_service(thresholds=th) + preds = svc.predict(samples) + d = preds[-1].to_dict() + self.assertIn("lower", d) + self.assertIn("upper", d) + + +class TestCockpitView(unittest.TestCase): + def test_render(self): + th = S.Thresholds(target="RF-01.PURITY", min_value=50.0) + svc, samples = _trained_service(thresholds=th) + preds = svc.predict(samples) + view = S.CockpitView().render(preds[-1]) + self.assertEqual(view["$schema"], "iAOP-cockpit-layout-v1") + types = [w["type"] for w in view["widgets"]] + self.assertIn("kpi_card", types) + self.assertIn("alarm_panel", types) + self.assertIn("explanation_list", types) + # explanation_list 有 top-3 贡献 + expl = [w for w in view["widgets"] if w["type"] == "explanation_list"][0] + self.assertLessEqual(len(expl["items"]), 3) + + +class TestDriftMonitor(unittest.TestCase): + def test_retrain_on_low_r2(self): + mon = S.DriftMonitor(min_r2=0.8, max_mae=1.0) + mon.record(M.Evaluation(r2=0.5, mae=0.5, rmse=0.5, n_samples=10)) + flag, reason = mon.should_retrain() + self.assertTrue(flag) + self.assertIn("R²", reason) + + def test_retrain_on_high_mae(self): + mon = S.DriftMonitor(min_r2=0.8, max_mae=1.0) + mon.record(M.Evaluation(r2=0.9, mae=2.0, rmse=2.0, n_samples=10)) + flag, reason = mon.should_retrain() + self.assertTrue(flag) + self.assertIn("MAE", reason) + + def test_ok_no_retrain(self): + mon = S.DriftMonitor(min_r2=0.8, max_mae=1.0) + mon.record(M.Evaluation(r2=0.95, mae=0.2, rmse=0.2, n_samples=10)) + flag, _ = mon.should_retrain() + self.assertFalse(flag) + + def test_empty(self): + mon = S.DriftMonitor() + flag, _ = mon.should_retrain() + self.assertFalse(flag) + + +class TestDeploymentBundle(unittest.TestCase): + def test_roundtrip(self): + th = S.Thresholds(target="RF-01.PURITY", min_value=99.0, + warning_band=0.5) + svc, _ = _trained_service(thresholds=th) + bundle = S.DeploymentBundle(svc.model, svc.extractor, th, + confidence_rmse=svc.confidence_rmse) + d = bundle.to_dict() + bundle2 = S.DeploymentBundle.from_dict(d) + svc2 = bundle2.to_service() + # 重建后仍能预测 + self.assertTrue(svc2.model.fitted) + + +if __name__ == "__main__": + unittest.main(verbosity=2)