feat(#73): [Ti-1] 模型部署到内核并接入驾驶舱（预测服务+告警+可解释视图+漂移监测）

新增 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 用例）全通过；纯标准库零运行时依赖。

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

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

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)