From c6dc7d2344772b48cbd95d4ab9b1fbadcc6a6fa6 Mon Sep 17 00:00:00 2001 From: bot_dev1 Date: Wed, 5 Aug 2026 05:14:39 +0800 Subject: [PATCH 1/3] =?UTF-8?q?feat(#68):=20[Ti-1]=20=E6=B0=AF=E5=8C=96?= =?UTF-8?q?=E8=BD=A6=E9=97=B4=E8=B4=A8=E9=87=8F=E9=A2=84=E6=B5=8B=E7=89=B9?= =?UTF-8?q?=E5=BE=81=E5=B7=A5=E7=A8=8B=EF=BC=88=E5=9F=BA=E4=BA=8E=E7=82=B9?= =?UTF-8?q?=E4=BD=8D=E5=AD=97=E5=85=B8=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 新增 templates/ti-cl4/quality-forecast/features.py： - PointDict：解析 point_dict CSV（9 列，对齐 core/edge-gateway），检索/存在性校验 - FeatureSpec：声明式特征（source/transform/window/meaning），换行业只改清单 - FeatureExtractor：按清单从时序样本抽取特征矩阵，缺失值 NaN 占位 - 9 算子：raw/mean/std/min/max/range/diff/slope/ratio - 零依赖 YAML 子集加载（与 recipe-optim/data-bus 同款） - config/features.template.yaml：7 个默认特征（炉温/配比/CO波动/炉层/TiCl₄纯度） - 22 用例全通过；纯标准库零运行时依赖。 --- templates/ti-cl4/quality-forecast/README.md | 37 ++ templates/ti-cl4/quality-forecast/__init__.py | 14 + .../ti-cl4/quality-forecast/_sanity_check.py | 57 ++ .../config/features.template.yaml | 63 ++ templates/ti-cl4/quality-forecast/features.py | 629 ++++++++++++++++++ .../quality-forecast/tests/_bootstrap.py | 26 + .../quality-forecast/tests/test_features.py | 228 +++++++ 7 files changed, 1054 insertions(+) create mode 100644 templates/ti-cl4/quality-forecast/README.md create mode 100644 templates/ti-cl4/quality-forecast/__init__.py create mode 100644 templates/ti-cl4/quality-forecast/_sanity_check.py create mode 100644 templates/ti-cl4/quality-forecast/config/features.template.yaml create mode 100644 templates/ti-cl4/quality-forecast/features.py create mode 100644 templates/ti-cl4/quality-forecast/tests/_bootstrap.py create mode 100644 templates/ti-cl4/quality-forecast/tests/test_features.py diff --git a/templates/ti-cl4/quality-forecast/README.md b/templates/ti-cl4/quality-forecast/README.md new file mode 100644 index 0000000..abf5fee --- /dev/null +++ b/templates/ti-cl4/quality-forecast/README.md @@ -0,0 +1,37 @@ +# Ti-1 氯化车间质量预测（quality-forecast） + +> 父 Issue：#10「[Template-Ti 一期] ① 质量预测 + ③ 炉层杂质预警」 +> 子任务：#68 特征工程 / #69 模型训练与评估 / #73 模型部署到内核并接入驾驶舱 + +基于 `templates/ti-cl4/point-dict/point_dict.default.csv` 默认点位集（CLF-01 沸腾氯化炉 / +RF-01 精制还原 / E-01 能源 / ST-01 蒸汽 / CW-01 循环水），交付「DCS 点表 → 可训练特征 +矩阵 → 质量预测 → 部署接入」全链路，对齐 PRD §5.3 ①「质量预测」。 + +## 设计要点 + +- **模板化**：特征清单（`config/features.template.yaml`）外置，换行业/换模板只改特征 + 清单 + 点位字典，特征工程代码零改动（PRD §5.3「换行业只改 Recipe」）。 +- **纯标准库零运行时依赖**：CSV/YAML 子集/统计全部自实现（不依赖 numpy/pandas/pyyaml）， + 便于离线/隔离网部署，与 `recipe-optim` / `impurity-forecast` 内核模块一致。 +- **可解释前置**：每个 `FeatureSpec` 带 `meaning`（工艺含义），供 #69 模型可解释性引用。 + +## 模块 + +| 文件 | 说明 | Issue | +| --- | --- | --- | +| `features.py` | 点位字典 `PointDict` + 特征规格 `FeatureSpec` + 抽取器 `FeatureExtractor` | #68 | +| `model.py` | 质量预测模型训练与评估（岭回归 + R²） | #69 | +| `serve.py` | 模型部署到内核并接入驾驶舱（预测服务） | #73 | +| `config/features.template.yaml` | 默认特征清单（7 个特征，对齐默认点位集） | #68 | + +## 算子 + +`raw / mean / std / min / max / range / diff / slope / ratio`（`ratio` 需 `denominator`）。 +缺失点位统一用 `NaN` 占位，便于上层判空屏蔽（`FeatureMatrix.drop_nan_rows()`）。 + +## 运行 + +```bash +# 单测（嵌入式 Python 3.12 即可，无第三方依赖） +python -m unittest discover -s templates/ti-cl4/quality-forecast/tests -p "test_*.py" -v +``` diff --git a/templates/ti-cl4/quality-forecast/__init__.py b/templates/ti-cl4/quality-forecast/__init__.py new file mode 100644 index 0000000..3325b47 --- /dev/null +++ b/templates/ti-cl4/quality-forecast/__init__.py @@ -0,0 +1,14 @@ +# -*- coding: utf-8 -*- +"""Ti-1 氯化车间质量预测模板包（Issue #68/#69/#73）。 + +子模块： + +- ``features``：基于点位字典的特征工程（#68）； +- ``model``：质量预测模型训练与评估（#69）； +- ``serve``：模型部署到内核并接入驾驶舱（#73）。 + +设计口径：纯标准库、零运行时依赖，便于离线/隔离网部署（与既有 +``recipe-optim`` / ``impurity-forecast`` 内核模块一致）。 +""" + +__all__ = ["features", "model", "serve"] diff --git a/templates/ti-cl4/quality-forecast/_sanity_check.py b/templates/ti-cl4/quality-forecast/_sanity_check.py new file mode 100644 index 0000000..9ffd009 --- /dev/null +++ b/templates/ti-cl4/quality-forecast/_sanity_check.py @@ -0,0 +1,57 @@ +# -*- coding: utf-8 -*- +"""Ti-1 质量预测 sanity 检查（无构建环境下的离线基本验证）。 + +检查项： +1. 默认特征清单 features.template.yaml 可加载且通过校验； +2. 特征 source/denominator 的点位都在默认点位字典内； +3. 全部测试用例通过。 + +用法：python _sanity_check.py +退出码：0 全通过，非 0 有失败。 +""" +import os +import sys +import unittest + +HERE = os.path.dirname(os.path.abspath(__file__)) +TESTS = os.path.join(HERE, "tests") + + +def main() -> int: + failures = [] + + # 1) 加载特征清单并校验点位 + try: + sys.path.insert(0, TESTS) + import _bootstrap # noqa: F401 挂载 quality_forecast + from quality_forecast import features as F + ti_cl4 = os.path.dirname(HERE) + pd = F.PointDict.from_csv( + os.path.join(ti_cl4, "point-dict", "point_dict.default.csv")) + with open(os.path.join(HERE, "config", "features.template.yaml"), + "r", encoding="utf-8") as fh: + ext = F.load_feature_specs(fh.read(), pd) + if not ext.names: + failures.append("特征清单为空") + except Exception as exc: # noqa: BLE001 + failures.append(f"特征清单加载失败: {exc}") + + # 2) 跑测试 + loader = unittest.TestLoader() + suite = loader.discover(TESTS, pattern="test_*.py") + runner = unittest.TextTestRunner(verbosity=1) + result = runner.run(suite) + if not result.wasSuccessful(): + failures.append(f"{len(result.failures)} 失败, {len(result.errors)} 错误") + + if failures: + print("\n[sanity] 失败:") + for f in failures: + print(" -", f) + return 1 + print("\n[sanity] 全部通过") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/templates/ti-cl4/quality-forecast/config/features.template.yaml b/templates/ti-cl4/quality-forecast/config/features.template.yaml new file mode 100644 index 0000000..387de57 --- /dev/null +++ b/templates/ti-cl4/quality-forecast/config/features.template.yaml @@ -0,0 +1,63 @@ +# -*- coding: utf-8 -*- +# Ti-1 氯化车间质量预测 · 特征清单模板（Issue #68 / PRD §5.3 ①）。 +# +# 模板化技术路径（PRD §5.3）：特征清单外置，换行业/换模板只改本文件 + 点位字典， +# 特征工程代码零改动。本清单基于 templates/ti-cl4/point-dict/point_dict.default.csv +# 的默认点位集（CLF-01 沸腾氯化炉 / RF-01 精制还原）。 +# +# 算子说明： +# raw 取窗口内最新值 +# mean 窗口均值 +# std 窗口标准差（样本无偏） +# min/max 窗口极值 +# range 窗口极差 +# diff 最近两点差分 +# slope 最小二乘斜率（值/秒） +# ratio source/denominator 比值（需 denominator） +# +# 缺省窗口 60s；meaning 字段供 #69 模型可解释性引用。 + +features: + - name: clf_temp_mean + source: CLF-01.TEMP + transform: mean + window: 60 + unit: "℃" + meaning: 沸腾氯化炉炉温窗口均值（反应强度主控变量） + - name: clf_temp_slope + source: CLF-01.TEMP + transform: slope + window: 120 + unit: "℃/s" + meaning: 炉温变化趋势（升温过快影响 TiCl₄纯度） + - name: clf_cl2_feed_ratio + source: CLF-01.CL2 + transform: ratio + denominator: CLF-01.FEED + window: 60 + unit: "m³/t" + meaning: 氯气/进料配比（配比偏离是杂质主因） + - name: clf_co_std + source: CLF-01.CO + transform: std + window: 120 + unit: "%" + meaning: CO 含量波动（燃烧不稳信号） + - name: clf_bed_range + source: CLF-01.BED + transform: range + window: 300 + unit: "%" + meaning: 炉层状态波动范围（偏钛酸铁预警信号） + - name: rf_purity_last + source: RF-01.PURITY + transform: raw + window: 60 + unit: "%" + meaning: TiCl₄ 纯度最新读数（LIMS 低频，质量标签候选） + - name: rf_imp_mean + source: RF-01.IMP + transform: mean + window: 60 + unit: "%" + meaning: 杂质含量窗口均值（质量标签候选） diff --git a/templates/ti-cl4/quality-forecast/features.py b/templates/ti-cl4/quality-forecast/features.py new file mode 100644 index 0000000..430fcd5 --- /dev/null +++ b/templates/ti-cl4/quality-forecast/features.py @@ -0,0 +1,629 @@ +# -*- coding: utf-8 -*- +"""Ti-1 氯化车间质量预测 · 特征工程（基于点位字典）（Issue #68 / PRD §5.3 ①）。 + +承接 PRD §5.3 ①「① 质量预测」与父 Issue #10「[Template-Ti 一期] ① 质量预测 + +③ 炉层杂质预警」：把"DCS 点表 → 可训练的特征矩阵"这条链路**模板化、可配置、 +可测试**，且与 #69 模型训练、#73 模型部署解耦。 + +PRD 设计口径 +------------ +- 架构表（PRD §5.3）：``质量预测 | 预测 | 入：DCS实时数据+LIMS； + 出：质量指标预测值(纯度/杂质) | ① 质量预测 | 中``。 +- 模板化技术路径：特征清单（``FeatureSpec``）外置为 YAML/JSON 超参包， + 切换模板/行业只改特征清单，特征工程代码零改动（PRD §5.3「换行业只改 Recipe」）。 +- 数据门槛：一期客户 DCS 点表未到位时启用默认通用点位集完成框架验证 + （PRD §13 缺省策略），故本模块**不依赖真实历史数据**——用合成/默认点位即可 + 完整跑通特征抽取，单测零外部数据依赖。 + +本模块交付 +---------- +1. **点位字典加载 ``PointDict``**：解析 ``point_dict.default.csv`` + （device_id/point_id/name/unit/...，与 ``core/edge-gateway`` 同款 9 列）， + 提供 ``by_point_id`` / ``by_device`` 检索与点位存在性校验。 +2. **特征规格 ``FeatureSpec``**：声明式特征——``name``、``source``（点位 point_id + 或常量）、``transform``（聚合算子 raw/mean/std/min/max/diff/ratio/…）、 + ``window``（时间窗，秒）、``meaning``（工艺含义，供 #69/#73 可解释引用）。 +3. **特征抽取器 ``FeatureExtractor``**：按特征清单从时序样本（``Sample`` 列表） + 抽取特征向量；缺失值用 ``NaN`` 占位（与 impurity-forecast / recipe-optim 一致， + 便于上层判空屏蔽）；输出有序 ``FeatureMatrix``（行=样本时刻，列=特征）。 +4. **声明式加载**：从 YAML/JSON 特征清单加载（零第三方依赖 YAML 子集解析， + 与 recipe-optim / data-bus / rag-kb 同款）。 + +设计要点 +-------- +- **零运行时依赖**（纯标准库）：CSV 用 ``csv``、YAML 子集自实现、统计用 ``math`` + 与手写聚合（不依赖 numpy/pandas），便于隔离网部署。 +- **可解释前置**：特征 ``meaning`` 字段，为 #69 模型可解释性预留引用依据。 +- **可校验**：``FeatureSpec.validate`` 聚合列出全部错误（未知点位/非法算子/负窗 + 口），便于配置台一次性反馈。 +""" +from __future__ import annotations + +import csv +import math +import os +from dataclasses import dataclass, field +from enum import Enum +from typing import Any, Dict, Iterable, List, Optional, Sequence, Tuple + +# 缺失值统一用 float('nan')，与 impurity-forecast / recipe-optim 一致。 +NAN = float("nan") + +# 点位字典 CSV 表头（与 core/edge-gateway/config/point_dict.example.csv 对齐，9 列） +POINT_COLUMNS = [ + "device_id", "point_id", "name", "unit", "dataType", + "sampleRate", "qualityCode", "opcNode", "protocol", +] + +# 允许的特征变换算子（与 impurity-forecast 特征口径对齐） +ALLOWED_TRANSFORMS = { + "raw", "mean", "std", "min", "max", "range", "diff", "ratio", "slope", +} + + +class FeatureError(ValueError): + """特征工程错误（未知点位 / 非法算子 / 窗口非法 / 重复特征名等）。""" + + +# --------------------------------------------------------------------------- +# 点位字典 +# --------------------------------------------------------------------------- + + +@dataclass +class Point: + """点位字典一行。""" + + device_id: str + point_id: str + name: str + unit: str + data_type: str = "float" + sample_rate: int = 1000 + quality_code: str = "true" + opc_node: str = "" + protocol: str = "" + + @classmethod + def from_row(cls, row: Dict[str, str]) -> "Point": + return cls( + device_id=(row.get("device_id") or "").strip(), + point_id=(row.get("point_id") or "").strip(), + name=(row.get("name") or "").strip(), + unit=(row.get("unit") or "").strip(), + data_type=(row.get("dataType") or "float").strip(), + sample_rate=int(float(row.get("sampleRate") or 1000)), + quality_code=(row.get("qualityCode") or "true").strip(), + opc_node=(row.get("opcNode") or "").strip(), + protocol=(row.get("protocol") or "").strip(), + ) + + +class PointDict: + """点位字典：解析 CSV，提供检索与存在性校验。""" + + def __init__(self, points: Sequence[Point]): + self._by_id: Dict[str, Point] = {p.point_id: p for p in points} + self._by_device: Dict[str, List[Point]] = {} + for p in points: + self._by_device.setdefault(p.device_id, []).append(p) + self.points: Tuple[Point, ...] = tuple(points) + + @classmethod + def from_csv(cls, path: str) -> "PointDict": + with open(path, "r", encoding="utf-8") as fh: + rows = list(csv.DictReader(fh)) + if not rows: + raise FeatureError(f"点位字典为空: {path}") + header = list(rows[0].keys()) + missing = [c for c in POINT_COLUMNS if c not in header] + if missing: + raise FeatureError(f"点位字典缺列: {missing}") + return cls([Point.from_row(r) for r in rows]) + + def has(self, point_id: str) -> bool: + return point_id in self._by_id + + def by_point_id(self, point_id: str) -> Point: + if point_id not in self._by_id: + raise FeatureError(f"未知点位: {point_id}") + return self._by_id[point_id] + + def by_device(self, device_id: str) -> List[Point]: + return list(self._by_device.get(device_id, [])) + + def point_ids(self) -> List[str]: + return list(self._by_id.keys()) + + +# --------------------------------------------------------------------------- +# 特征规格 +# --------------------------------------------------------------------------- + + +class Transform(str, Enum): + RAW = "raw" + MEAN = "mean" + STD = "std" + MIN = "min" + MAX = "max" + RANGE = "range" + DIFF = "diff" + RATIO = "ratio" + SLOPE = "slope" + + +@dataclass +class FeatureSpec: + """声明式特征规格。 + + - ``source`` 形如 ``CLF-01.TEMP``（点位 point_id）或常量数值； + - ``transform`` 聚合算子（raw/mean/std/min/max/range/diff/ratio/slope）； + - ``window`` 时间窗（秒，仅滚动窗算子有意义；raw/diff 用最近两点）； + - ``denominator`` 仅 ratio 算子使用（另一个 point_id 或常量）； + - ``meaning`` 工艺含义（#69/#73 可解释性引用）。 + """ + + name: str + source: str + transform: str = "raw" + window: float = 60.0 + denominator: Optional[str] = None + meaning: str = "" + unit: str = "" + + def validate(self, point_dict: Optional[PointDict] = None) -> List[str]: + errors: List[str] = [] + if not self.name: + errors.append("特征 name 不能为空") + if self.transform not in ALLOWED_TRANSFORMS: + errors.append(f"特征 {self.name}: 非法 transform={self.transform}") + if self.window < 0: + errors.append(f"特征 {self.name}: window 不能为负 (={self.window})") + if self.transform == "ratio" and not self.denominator: + errors.append(f"特征 {self.name}: ratio 算子需指定 denominator") + # 点位存在性（source/denominator 形如 point_id 时校验） + if point_dict is not None: + for label, val in (("source", self.source), + ("denominator", self.denominator)): + if val and not _is_constant(val) and not point_dict.has(val): + errors.append(f"特征 {self.name}: {label}={val} 不在点位字典") + return errors + + @classmethod + def from_dict(cls, d: Dict[str, Any]) -> "FeatureSpec": + return cls( + name=str(d.get("name", "")).strip(), + source=str(d.get("source", "")).strip(), + transform=str(d.get("transform", "raw")).strip(), + window=float(d.get("window", 60.0)), + denominator=(str(d.get("denominator")).strip() + if d.get("denominator") else None), + meaning=str(d.get("meaning", "")).strip(), + unit=str(d.get("unit", "")).strip(), + ) + + +def _is_constant(val: str) -> bool: + """source/denominator 是否为常量数值（而非 point_id）。""" + try: + float(val) + return True + except (TypeError, ValueError): + return False + + +# --------------------------------------------------------------------------- +# 时序样本 +# --------------------------------------------------------------------------- + + +@dataclass +class Sample: + """一个采样时刻的多点位读数。 + + - ``ts`` 时间戳（秒，单调不减）； + - ``values`` point_id → 数值；缺失点位视为无读数。 + """ + + ts: float + values: Dict[str, float] = field(default_factory=dict) + + +# --------------------------------------------------------------------------- +# 特征抽取 +# --------------------------------------------------------------------------- + + +class FeatureExtractor: + """按特征清单从时序样本抽取特征向量。 + + 用法:: + + ext = FeatureExtractor(specs, point_dict) + matrix = ext.extract(samples) + # matrix.rows[i] 是一个有序特征向量；matrix.names 是列名 + """ + + def __init__(self, specs: Sequence[FeatureSpec], + point_dict: Optional[PointDict] = None, + *, strict: bool = True): + self.specs: Tuple[FeatureSpec, ...] = tuple(specs) + self.point_dict = point_dict + if strict: + errors = self.validate() + if errors: + raise FeatureError("特征清单校验失败:\n " + "\n ".join(errors)) + # 重复特征名检查 + names = [s.name for s in self.specs] + dup = {n for n in names if names.count(n) > 1} + if dup and strict: + raise FeatureError(f"重复特征名: {sorted(dup)}") + + def validate(self) -> List[str]: + errors: List[str] = [] + for s in self.specs: + errors.extend(s.validate(self.point_dict)) + return errors + + @property + def names(self) -> List[str]: + return [s.name for s in self.specs] + + def extract(self, samples: Sequence[Sample]) -> "FeatureMatrix": + rows: List[List[float]] = [] + # 滚动窗：按 window 秒选取 <= ts 的历史样本 + win = [s.window for s in self.specs] + max_window = max(win) if win else 0.0 + ordered = sorted(samples, key=lambda s: s.ts) + for cur in ordered: + window_samples = [ + s for s in ordered + if cur.ts - max_window <= s.ts <= cur.ts + ] + row = [self._compute(spec, cur, window_samples) + for spec in self.specs] + rows.append(row) + return FeatureMatrix(names=self.names, rows=rows) + + # 单特征计算 ------------------------------------------------------------ + def _compute(self, spec: FeatureSpec, cur: Sample, + window_samples: Sequence[Sample]) -> float: + series = _series(spec.source, window_samples, self.point_dict) + denom_series = ( + _series(spec.denominator, window_samples, self.point_dict) + if spec.denominator else [] + ) + tf = spec.transform + if tf == "raw": + return _last_or_nan(series) + if tf == "mean": + return _mean(series) + if tf == "std": + return _std(series) + if tf == "min": + return _min(series) + if tf == "max": + return _max(series) + if tf == "range": + return _range(series) + if tf == "diff": + return _diff(series) + if tf == "slope": + return _slope(series, spec.window) + if tf == "ratio": + return _ratio(_last_or_nan(series), _last_or_nan(denom_series)) + # 不应到达（已 validate） + return NAN + + +@dataclass +class FeatureMatrix: + """特征抽取结果：有序特征名 + 行向量集合。""" + + names: List[str] + rows: List[List[float]] + + def column(self, name: str) -> List[float]: + idx = self.names.index(name) + return [r[idx] for r in self.rows] + + def to_records(self) -> List[Dict[str, float]]: + return [dict(zip(self.names, row)) for row in self.rows] + + def drop_nan_rows(self) -> "FeatureMatrix": + """丢弃任一特征为 NaN 的行（数据门槛不足时常用）。""" + clean = [r for r in self.rows if not any(math.isnan(v) for v in r)] + return FeatureMatrix(names=list(self.names), rows=clean) + + +# --------------------------------------------------------------------------- +# 聚合算子（纯标准库） +# --------------------------------------------------------------------------- + + +def _series(source: str, samples: Sequence[Sample], + point_dict: Optional[PointDict]) -> List[Tuple[float, float]]: + """取一个 source 的 (ts, value) 序列。常量源展开为各样本时刻。""" + if _is_constant(source): + const = float(source) + return [(s.ts, const) for s in samples] + return [(s.ts, s.values[source]) for s in samples + if source in s.values and not math.isnan(s.values[source])] + + +def _last_or_nan(series: Sequence[Tuple[float, float]]) -> float: + return series[-1][1] if series else NAN + + +def _values(series: Sequence[Tuple[float, float]]) -> List[float]: + return [v for _, v in series] + + +def _mean(series: Sequence[Tuple[float, float]]) -> float: + vs = _values(series) + return sum(vs) / len(vs) if vs else NAN + + +def _std(series: Sequence[Tuple[float, float]]) -> float: + vs = _values(series) + n = len(vs) + if n < 2: + return NAN if n == 0 else 0.0 + mu = sum(vs) / n + var = sum((v - mu) ** 2 for v in vs) / (n - 1) + return math.sqrt(var) + + +def _min(series: Sequence[Tuple[float, float]]) -> float: + vs = _values(series) + return min(vs) if vs else NAN + + +def _max(series: Sequence[Tuple[float, float]]) -> float: + vs = _values(series) + return max(vs) if vs else NAN + + +def _range(series: Sequence[Tuple[float, float]]) -> float: + vs = _values(series) + return (max(vs) - min(vs)) if vs else NAN + + +def _diff(series: Sequence[Tuple[float, float]]) -> float: + if len(series) < 2: + return NAN + return series[-1][1] - series[-2][1] + + +def _slope(series: Sequence[Tuple[float, float]], window: float) -> float: + """最小二乘斜率（值/秒）；样本不足返回 NaN。""" + if len(series) < 2: + return NAN + xs = [t for t, _ in series] + # 时间窗外的样本不参与（已由 caller 截窗，这里再以 window 收敛） + if window and window > 0: + tmax = max(xs) + kept = [(t, v) for t, v in series if t >= tmax - window] + if len(kept) < 2: + return NAN + xs = [t for t, _ in kept] + ys = [v for _, v in kept] + else: + ys = [v for _, v in series] + n = len(xs) + xbar = sum(xs) / n + ybar = sum(ys) / n + num = sum((xs[i] - xbar) * (ys[i] - ybar) for i in range(n)) + den = sum((xs[i] - xbar) ** 2 for i in range(n)) + return num / den if den else NAN + + +def _ratio(a: float, b: float) -> float: + if math.isnan(a) or math.isnan(b) or b == 0: + return NAN + return a / b + + +# --------------------------------------------------------------------------- +# 声明式加载（零第三方依赖 YAML 子集解析，与 recipe-optim 同款） +# --------------------------------------------------------------------------- + + +def load_feature_specs(text: str, + point_dict: Optional[PointDict] = None, + *, strict: bool = True) -> FeatureExtractor: + """从 YAML/JSON 文本加载特征清单并构造 FeatureExtractor。 + + 支持的 YAML 子集：``features:`` 顶层键，下为 ``- name/source/transform/...`` + 列表项。也兼容 JSON（``{"features": [...]}``）。 + """ + text = text.strip() + data: Any + if text.startswith("{") or text.startswith("["): + import json + data = json.loads(text) + else: + data = _parse_yaml_subset(text) + if not isinstance(data, dict): + raise FeatureError("特征清单顶层应为映射（含 features 键）") + raw_features = data.get("features") + if not isinstance(raw_features, list): + raise FeatureError("特征清单缺少 features 列表") + specs = [FeatureSpec.from_dict(f) for f in raw_features if isinstance(f, dict)] + if not specs: + raise FeatureError("特征清单 features 为空") + return FeatureExtractor(specs, point_dict, strict=strict) + + +def _parse_yaml_subset(text: str) -> Any: + """极简 YAML 子集解析器（仅供模板资产，非通用 YAML）。 + + 支持：注释（# ...）、映射（key: value）、列表（- item）、嵌套缩进、 + 基本标量（int/float/str/bool/null）。与 recipe-optim / data-bus 同款。 + """ + lines: List[str] = [] + for raw in text.splitlines(): + stripped = raw.rstrip() + if not stripped.strip(): + continue + if stripped.lstrip().startswith("#"): + continue + hi = _find_inline_comment(stripped) + if hi is not None: + stripped = stripped[:hi].rstrip() + if stripped: + lines.append(stripped) + parser = _YamlParser(lines) + return parser.parse_block(0)[0] if lines else {} + + +def _find_inline_comment(line: str) -> Optional[int]: + depth = 0 + in_str = False + for i, ch in enumerate(line): + if ch == '"': + in_str = not in_str + elif not in_str: + if ch in "[{": + depth += 1 + elif ch in "]}": + depth = max(0, depth - 1) + elif ch == "#" and depth == 0: + if i == 0 or line[i - 1] in (" ", "\t"): + return i + return None + + +def _parse_scalar(raw: str) -> Any: + raw = raw.strip() + if not raw: + return None + if raw.startswith('"') and raw.endswith('"'): + return raw[1:-1] + if raw.startswith("[") or raw.startswith("{"): + import json + try: + return json.loads(raw) + except Exception: + return raw + low = raw.lower() + if low == "true": + return True + if low == "false": + return False + if low in ("null", "~", "none"): + return None + try: + return int(raw) + except ValueError: + pass + try: + return float(raw) + except ValueError: + pass + return raw + + +class _YamlParser: + """递归下降的 YAML 子集解析器（按缩进分层）。""" + + def __init__(self, lines: List[str]) -> None: + self.lines = lines + self.i = 0 + + def _indent(self, line: str) -> int: + return len(line) - len(line.lstrip(" ")) + + def parse_block(self, indent: int) -> Tuple[Any, bool]: + if self.i >= len(self.lines): + return {}, False + line = self.lines[self.i] + cur = self._indent(line) + if cur < indent: + return {}, False + stripped = line.strip() + if stripped.startswith("- ") or stripped == "-": + return self._parse_list(cur), True + return self._parse_mapping(cur), False + + def _parse_mapping(self, indent: int) -> Dict[str, Any]: + result: Dict[str, Any] = {} + effective = indent + if self.i < len(self.lines): + first = self._indent(self.lines[self.i]) + if first > indent: + effective = first + while self.i < len(self.lines): + line = self.lines[self.i] + cur = self._indent(line) + if cur < effective: + break + if cur > effective: + self.i += 1 + continue + stripped = line.strip() + if stripped.startswith("- "): + break + key, sep, rest = stripped.partition(":") + if not sep: + self.i += 1 + continue + key = key.strip() + rest = rest.strip() + self.i += 1 + if rest: + result[key] = _parse_scalar(rest) + else: + # 子块 + if self.i < len(self.lines): + nxt = self._indent(self.lines[self.i]) + if nxt > effective: + val, _ = self.parse_block(nxt) + result[key] = val + return result + + def _parse_list(self, indent: int) -> List[Any]: + items: List[Any] = [] + while self.i < len(self.lines): + line = self.lines[self.i] + cur = self._indent(line) + if cur < indent: + break + if cur > indent: + self.i += 1 + continue + stripped = line.strip() + if not stripped.startswith("-"): + break + item_text = stripped[1:].strip() + if not item_text: + # 子块（嵌套映射/列表） + if self.i + 1 < len(self.lines): + nxt = self._indent(self.lines[self.i + 1]) + if nxt > cur: + self.i += 1 + val, _ = self.parse_block(nxt) + items.append(val) + continue + self.i += 1 + items.append(None) + continue + # "- key: value" 形式 → 该 item 是映射 + if ":" in item_text and not item_text.startswith('"'): + k, sep, v = item_text.partition(":") + if sep: + item: Dict[str, Any] = {k.strip(): _parse_scalar(v.strip())} + self.i += 1 + # 后续同缩进的 key 归入同一 item + if self.i < len(self.lines): + child_indent = self._indent(self.lines[self.i]) + if child_indent > cur: + sub, _ = self.parse_block(child_indent) + if isinstance(sub, dict): + item.update(sub) + items.append(item) + continue + items.append(_parse_scalar(item_text)) + self.i += 1 + return items diff --git a/templates/ti-cl4/quality-forecast/tests/_bootstrap.py b/templates/ti-cl4/quality-forecast/tests/_bootstrap.py new file mode 100644 index 0000000..ff00902 --- /dev/null +++ b/templates/ti-cl4/quality-forecast/tests/_bootstrap.py @@ -0,0 +1,26 @@ +# -*- coding: utf-8 -*- +"""测试引导：把连字符目录挂载为可导入包（与 core 模块同款模式）。 + +- ``templates/ti-cl4/quality-forecast`` → 包名 ``quality_forecast``。 +本模块零内核依赖（纯标准库），仅挂载自身包即可。 +""" +import importlib.util +import os +import sys + +PKG_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) + + +def _load_package(name: str, path: str) -> None: + """按文件路径完整加载一个包（执行其 __init__.py）。""" + 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("quality_forecast", PKG_DIR) diff --git a/templates/ti-cl4/quality-forecast/tests/test_features.py b/templates/ti-cl4/quality-forecast/tests/test_features.py new file mode 100644 index 0000000..28fc0a7 --- /dev/null +++ b/templates/ti-cl4/quality-forecast/tests/test_features.py @@ -0,0 +1,228 @@ +# -*- coding: utf-8 -*- +"""Ti-1 质量预测特征工程测试（Issue #68）。 + +覆盖： +1. 点位字典加载（CSV 解析、检索、存在性校验、缺列报错）； +2. FeatureSpec 校验（非法算子/未知点位/负窗口/ratio 缺 denominator）； +3. 各 transform 算子（raw/mean/std/min/max/range/diff/slope/ratio）数值正确； +4. 缺失点位 → NaN 占位； +5. 滚动窗：window 外的样本不参与； +6. 声明式加载（YAML 子集 + JSON）； +7. 重复特征名报错； +8. 模板资产 features.template.yaml 可加载并通过校验（对齐默认点位集）。 +""" +import math +import os +import sys +import unittest + +HERE = os.path.dirname(os.path.abspath(__file__)) # .../quality-forecast/tests +PKG_DIR = os.path.dirname(HERE) # .../quality-forecast +TI_CL4_DIR = os.path.dirname(PKG_DIR) # .../ti-cl4 +sys.path.insert(0, HERE) +import _bootstrap # noqa: F401,E402 挂载 quality_forecast 包 + +from quality_forecast import features as F # noqa: E402 + +PDICT_DEFAULT = os.path.join( + TI_CL4_DIR, "point-dict", "point_dict.default.csv") +FEATURES_TPL = os.path.join( + PKG_DIR, "config", "features.template.yaml") + + +def _pdict(): + return F.PointDict.from_csv(PDICT_DEFAULT) + + +def _samples(values, ts0=0.0, step=10.0): + """构造样本序列：values 是 [{point_id: v}, ...]。""" + out = [] + for i, vmap in enumerate(values): + out.append(F.Sample(ts=ts0 + i * step, values=dict(vmap))) + return out + + +class TestPointDict(unittest.TestCase): + def test_load_default_csv(self): + pd = _pdict() + self.assertTrue(pd.has("CLF-01.TEMP")) + self.assertIn("CLF-01", {p.device_id for p in pd.points}) + + def test_by_point_id_unknown_raises(self): + pd = _pdict() + with self.assertRaises(F.FeatureError): + pd.by_point_id("NOPE") + + def test_by_device(self): + pd = _pdict() + clf = pd.by_device("CLF-01") + self.assertTrue(all(p.device_id == "CLF-01" for p in clf)) + self.assertGreater(len(clf), 0) + + +class TestFeatureSpecValidate(unittest.TestCase): + def test_bad_transform(self): + s = F.FeatureSpec(name="x", source="CLF-01.TEMP", transform="bogus") + self.assertIn("非法 transform", "\n".join(s.validate())) + + def test_negative_window(self): + s = F.FeatureSpec(name="x", source="CLF-01.TEMP", window=-1) + self.assertIn("window 不能为负", "\n".join(s.validate())) + + def test_ratio_needs_denominator(self): + s = F.FeatureSpec(name="x", source="CLF-01.CL2", transform="ratio") + self.assertIn("denominator", "\n".join(s.validate())) + + def test_unknown_point_with_dict(self): + pd = _pdict() + s = F.FeatureSpec(name="x", source="UNKNOWN.PT") + errs = s.validate(pd) + self.assertTrue(any("不在点位字典" in e for e in errs)) + + def test_constant_source_ok(self): + pd = _pdict() + s = F.FeatureSpec(name="x", source="1.5") + self.assertEqual(s.validate(pd), []) + + +class TestTransforms(unittest.TestCase): + def setUp(self): + self.pd = _pdict() + # 4 个样本，TEMP 单调上升 + self.samples = _samples([ + {"CLF-01.TEMP": 100.0}, + {"CLF-01.TEMP": 110.0}, + {"CLF-01.TEMP": 120.0}, + {"CLF-01.TEMP": 130.0}, + ], step=10.0) + + def test_raw(self): + ext = F.FeatureExtractor( + [F.FeatureSpec("t", "CLF-01.TEMP", "raw")], self.pd) + m = ext.extract(self.samples) + self.assertEqual(m.column("t")[-1], 130.0) + + def test_mean(self): + ext = F.FeatureExtractor( + [F.FeatureSpec("t", "CLF-01.TEMP", "mean", window=1000)], + self.pd) + m = ext.extract(self.samples) + self.assertAlmostEqual(m.column("t")[-1], 115.0) + + def test_min_max_range(self): + ext = F.FeatureExtractor([ + F.FeatureSpec("mn", "CLF-01.TEMP", "min", window=1000), + F.FeatureSpec("mx", "CLF-01.TEMP", "max", window=1000), + F.FeatureSpec("rg", "CLF-01.TEMP", "range", window=1000), + ], self.pd) + m = ext.extract(self.samples) + last = m.rows[-1] + self.assertEqual(last[0], 100.0) # min + self.assertEqual(last[1], 130.0) # max + self.assertEqual(last[2], 30.0) # range + + def test_std(self): + ext = F.FeatureExtractor( + [F.FeatureSpec("s", "CLF-01.TEMP", "std", window=1000)], + self.pd) + m = ext.extract(self.samples) + # 无偏样本标准差：100,110,120,130 → 12.9099... + self.assertAlmostEqual(m.column("s")[-1], + math.sqrt(500.0 / 3), places=4) + + def test_diff(self): + ext = F.FeatureExtractor( + [F.FeatureSpec("d", "CLF-01.TEMP", "diff")], self.pd) + m = ext.extract(self.samples) + self.assertEqual(m.column("d")[-1], 10.0) + + def test_slope(self): + ext = F.FeatureExtractor( + [F.FeatureSpec("sl", "CLF-01.TEMP", "slope", window=1000)], + self.pd) + m = ext.extract(self.samples) + # 每 10s +10 → 斜率 1.0 + self.assertAlmostEqual(m.column("sl")[-1], 1.0, places=6) + + def test_ratio(self): + ext = F.FeatureExtractor([ + F.FeatureSpec("r", "CLF-01.CL2", "ratio", + denominator="CLF-01.FEED"), + ], self.pd) + samples = _samples([ + {"CLF-01.CL2": 30.0, "CLF-01.FEED": 10.0}, + {"CLF-01.CL2": 60.0, "CLF-01.FEED": 20.0}, + ], step=10.0) + m = ext.extract(samples) + self.assertAlmostEqual(m.column("r")[-1], 3.0) + + +class TestMissingAndWindow(unittest.TestCase): + def test_missing_point_is_nan(self): + ext = F.FeatureExtractor( + [F.FeatureSpec("t", "CLF-01.TEMP", "raw")], _pdict()) + # 样本里没有 TEMP → NaN + samples = _samples([{"CLF-01.PRES": 1.0}]) + m = ext.extract(samples) + self.assertTrue(math.isnan(m.column("t")[0])) + + def test_window_excludes_old(self): + ext = F.FeatureExtractor( + [F.FeatureSpec("t", "CLF-01.TEMP", "mean", window=15)], + _pdict()) + # window=15s 只含最近 ≤2 个样本（step=10） + samples = _samples([{"CLF-01.TEMP": 0.0}, + {"CLF-01.TEMP": 100.0}, + {"CLF-01.TEMP": 200.0}], step=10.0) + m = ext.extract(samples) + # 最后时刻 window=15 → 含 ts=20(100) 与 ts=30(200) → 均值 150 + self.assertAlmostEqual(m.column("t")[-1], 150.0) + + def test_drop_nan_rows(self): + ext = F.FeatureExtractor( + [F.FeatureSpec("t", "CLF-01.TEMP", "raw")], _pdict()) + samples = _samples([ + {"CLF-01.PRES": 1.0}, # TEMP 缺失 → NaN + {"CLF-01.TEMP": 50.0}, + ]) + m = ext.extract(samples).drop_nan_rows() + self.assertEqual(len(m.rows), 1) + + +class TestLoading(unittest.TestCase): + def test_load_template_yaml(self): + pd = _pdict() + with open(FEATURES_TPL, "r", encoding="utf-8") as fh: + text = fh.read() + ext = F.load_feature_specs(text, pd) + self.assertGreater(len(ext.names), 0) + # 抽取一次能跑通（合成样本） + samples = _samples([{"CLF-01.TEMP": 850.0, "CLF-01.CL2": 120.0, + "CLF-01.FEED": 4.0, "CLF-01.CO": 2.0, + "CLF-01.BED": 60.0, "RF-01.PURITY": 99.0, + "RF-01.IMP": 0.3}]) + m = ext.extract(samples) + self.assertEqual(len(m.names), len(ext.names)) + self.assertEqual(len(m.rows), 1) + + def test_load_json(self): + import json + text = json.dumps({"features": [ + {"name": "t", "source": "CLF-01.TEMP", "transform": "raw"}]}) + ext = F.load_feature_specs(text, _pdict()) + self.assertEqual(ext.names, ["t"]) + + def test_duplicate_names_raise(self): + with self.assertRaises(F.FeatureError): + F.FeatureExtractor([ + F.FeatureSpec("dup", "CLF-01.TEMP"), + F.FeatureSpec("dup", "CLF-01.PRES"), + ], _pdict()) + + def test_empty_features_raise(self): + with self.assertRaises(F.FeatureError): + F.load_feature_specs("features: []", _pdict()) + + +if __name__ == "__main__": + unittest.main(verbosity=2) From aeee8ef46871ced25ce8636eb99158520582defb Mon Sep 17 00:00:00 2001 From: bot_dev1 Date: Wed, 5 Aug 2026 05:16:26 +0800 Subject: [PATCH 2/3] =?UTF-8?q?feat(#69):=20[Ti-1]=20=E8=B4=A8=E9=87=8F?= =?UTF-8?q?=E9=A2=84=E6=B5=8B=E6=A8=A1=E5=9E=8B=E8=AE=AD=E7=BB=83=E4=B8=8E?= =?UTF-8?q?=E8=AF=84=E4=BC=B0=EF=BC=88=E5=B2=AD=E5=9B=9E=E5=BD=92+R=C2=B2/?= =?UTF-8?q?MAE/RMSE+=E5=8F=AF=E8=A7=A3=E9=87=8A=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 新增 templates/ti-cl4/quality-forecast/model.py： - RidgeRegression：纯标准库最小二乘+L2正则闭式解（高斯消元，不依赖numpy） - QualityModel：聚合特征+目标+岭回归，fit/predict/evaluate(R²/MAE/RMSE)/explain - explain()：尺度归一化重要性（|权重|×std），跨特征可比较，供#73可解释 - TrainingSet.from_records：自动跳过含 NaN 的行 - ModelRecipe 超参包：target/alpha/feature_names 外置 YAML/JSON - 序列化 to_dict/from_dict 往返一致 - 17 用例（累计 39 用例）全通过；纯标准库零运行时依赖。 --- .../config/model.template.yaml | 18 + templates/ti-cl4/quality-forecast/model.py | 531 ++++++++++++++++++ .../quality-forecast/tests/test_model.py | 182 ++++++ 3 files changed, 731 insertions(+) create mode 100644 templates/ti-cl4/quality-forecast/config/model.template.yaml create mode 100644 templates/ti-cl4/quality-forecast/model.py create mode 100644 templates/ti-cl4/quality-forecast/tests/test_model.py diff --git a/templates/ti-cl4/quality-forecast/config/model.template.yaml b/templates/ti-cl4/quality-forecast/config/model.template.yaml new file mode 100644 index 0000000..5d2f5f1 --- /dev/null +++ b/templates/ti-cl4/quality-forecast/config/model.template.yaml @@ -0,0 +1,18 @@ +# -*- coding: utf-8 -*- +# Ti-1 氯化车间质量预测 · 模型超参包（Issue #69 / PRD §5.3 ①）。 +# +# 模板化技术路径（PRD §5.3）：超参包外置，换行业只改本文件 + 特征清单。 +# target：质量标签列名（默认取 RF-01.PURITY = TiCl₄纯度）。 +# alpha：岭回归 L2 正则强度（防共线性/过拟合）。 +# feature_names：训练用特征名（须与 features.template.yaml 的 name 对齐）。 + +target: RF-01.PURITY +alpha: 1.0 +unit: "%" +feature_names: + - clf_temp_mean + - clf_temp_slope + - clf_cl2_feed_ratio + - clf_co_std + - clf_bed_range + - rf_imp_mean diff --git a/templates/ti-cl4/quality-forecast/model.py b/templates/ti-cl4/quality-forecast/model.py new file mode 100644 index 0000000..d48419e --- /dev/null +++ b/templates/ti-cl4/quality-forecast/model.py @@ -0,0 +1,531 @@ +# -*- coding: utf-8 -*- +"""Ti-1 氯化车间质量预测 · 模型训练与评估（Issue #69 / PRD §5.3 ①）。 + +承接 #68（特征工程）产出的 ``FeatureMatrix``：把"特征矩阵 + 质量标签 → 可评估、 +可解释的预测模型"这条链路**模板化、可测试**，且与 #73 模型部署解耦。 + +PRD 设计口径 +------------ +- 架构表（PRD §5.3）：``质量预测 | 预测 | 入：特征矩阵； + 出：质量指标预测值(纯度/杂质) | ① 质量预测 | 中``。 +- 模板化技术路径：模型超参（``alpha`` 正则强度、``target`` 标签列）外置为 + JSON/YAML 超参包，切换模板只改超参包（PRD §5.3「换行业只改 Recipe」）。 +- 验收（PRD §10 DoD）：质量预测在客户数据上达约定 R² / MAE 指标。 + +本模块交付 +---------- +1. **岭回归 ``RidgeRegression``**：纯标准库最小二乘 + L2 正则（闭式解）， + ``fit(X, y)`` / ``predict(X)`` / ``coef_`` / ``intercept_``。不依赖 numpy。 +2. **质量预测模型 ``QualityModel``**：聚合特征名 + 目标列 + 岭回归，提供 + ``fit(matrix, target)`` / ``predict(matrix)`` / ``evaluate(matrix, target)`` + （R² / MAE / RMSE）/ ``explain()``（权重 → 特征贡献，可溯源）。 +3. **超参包 ``ModelRecipe``**：``target``/``alpha``/``feature_names`` 外置 + JSON/YAML 加载（零依赖 YAML 子集解析）。 +4. **评估指标**：R²、MAE、RMSE 纯标准库实现。 + +设计要点 +-------- +- **零运行时依赖**（纯标准库）：矩阵运算手写（不依赖 numpy/sklearn）。 +- **可解释**：``explain()`` 输出每个特征的权重 × 方差贡献度，供 #73 接入驾驶舱 + 展示"为何预测这个纯度"（对齐 PRD"要求结果可解释、可溯源"）。 +- **稳健**：L2 正则避免共线性/过拟合；数据不足时返回明确错误而非崩溃。 +""" +from __future__ import annotations + +import json +import math +import os +from dataclasses import dataclass, field +from typing import Any, Dict, List, Optional, Sequence, Tuple + +NAN = float("nan") + + +class ModelError(ValueError): + """质量预测模型错误（数据不足 / 维度不匹配 / 奇异 等）。""" + + +# --------------------------------------------------------------------------- +# 纯标准库线性代数（最小二乘岭回归闭式解） +# --------------------------------------------------------------------------- + + +def _matmul_at_a(a: Sequence[Sequence[float]]) -> List[List[float]]: + """计算 AᵀA（n×n）。""" + n = len(a[0]) if a else 0 + res = [[0.0] * n for _ in range(n)] + for row in a: + for i in range(n): + ri = row[i] + if ri == 0.0: + continue + for j in range(n): + res[i][j] += ri * row[j] + return res + + +def _matvec_at_b(a: Sequence[Sequence[float]], b: Sequence[float]) -> List[float]: + """计算 Aᵀb（n）。""" + n = len(a[0]) if a else 0 + res = [0.0] * n + for row, y in zip(a, b): + for i in range(n): + res[i] += row[i] * y + return res + + +def _solve(A: List[List[float]], b: List[float]) -> List[float]: + """高斯消元解 Ax=b（带部分主元）。A 会被修改。""" + n = len(A) + # 增广 + M = [list(A[i]) + [b[i]] for i in range(n)] + for col in range(n): + # 主元 + pivot = max(range(col, n), key=lambda r: abs(M[r][col])) + if abs(M[pivot][col]) < 1e-12: + raise ModelError("矩阵奇异（特征共线或数据不足），无法求解") + M[col], M[pivot] = M[pivot], M[col] + piv = M[col][col] + for j in range(col, n + 1): + M[col][j] /= piv + for r in range(n): + if r == col: + continue + factor = M[r][col] + if factor == 0.0: + continue + for j in range(col, n + 1): + M[r][j] -= factor * M[col][j] + return [M[i][n] for i in range(n)] + + +# --------------------------------------------------------------------------- +# 岭回归 +# --------------------------------------------------------------------------- + + +class RidgeRegression: + """岭回归（L2 正则最小二乘，闭式解）。纯标准库。 + + 解：``w = (XᵀX + αI)⁻¹ Xᵀy``。 + """ + + def __init__(self, alpha: float = 1.0): + if alpha < 0: + raise ModelError(f"alpha 不能为负: {alpha}") + self.alpha = alpha + self.coef_: List[float] = [] + self.intercept_: float = 0.0 + self._n_features: int = 0 + + def fit(self, X: Sequence[Sequence[float]], y: Sequence[float]) -> "RidgeRegression": + m = len(X) + if m == 0: + raise ModelError("训练集为空") + n = len(X[0]) + if n == 0: + raise ModelError("特征数为 0") + if len(y) != m: + raise ModelError(f"X/y 行数不匹配: {m} != {len(y)}") + self._n_features = n + # 中心化（数值稳定 + 让 intercept 可独立） + x_mean = [sum(X[i][j] for i in range(m)) / m for j in range(n)] + y_mean = sum(y) / m + Xc = [[X[i][j] - x_mean[j] for j in range(n)] for i in range(m)] + yc = [y[i] - y_mean for i in range(m)] + # 正规方程 (XᵀX + αI) w = Xᵀy + A = _matmul_at_a(Xc) + for i in range(n): + A[i][i] += self.alpha + b = _matvec_at_b(Xc, yc) + self.coef_ = _solve(A, b) + self.intercept_ = y_mean - sum(self.coef_[j] * x_mean[j] for j in range(n)) + return self + + def predict(self, X: Sequence[Sequence[float]]) -> List[float]: + if not self.coef_: + raise ModelError("模型未训练") + return [self.intercept_ + sum(self.coef_[j] * row[j] + for j in range(self._n_features)) + for row in X] + + def to_dict(self) -> Dict[str, Any]: + return {"alpha": self.alpha, "coef": list(self.coef_), + "intercept": self.intercept_, + "n_features": self._n_features} + + @classmethod + def from_dict(cls, d: Dict[str, Any]) -> "RidgeRegression": + m = cls(alpha=float(d.get("alpha", 1.0))) + m.coef_ = [float(c) for c in d.get("coef", [])] + m.intercept_ = float(d.get("intercept", 0.0)) + m._n_features = int(d.get("n_features", len(m.coef_))) + return m + + +# --------------------------------------------------------------------------- +# 评估指标 +# --------------------------------------------------------------------------- + + +def r2_score(y_true: Sequence[float], y_pred: Sequence[float]) -> float: + """决定系数 R²。""" + if len(y_true) != len(y_pred): + raise ModelError("y_true/y_pred 长度不匹配") + n = len(y_true) + if n == 0: + return NAN + mean = sum(y_true) / n + ss_res = sum((y_true[i] - y_pred[i]) ** 2 for i in range(n)) + ss_tot = sum((y_true[i] - mean) ** 2 for i in range(n)) + if ss_tot == 0: + return 1.0 if ss_res == 0 else 0.0 + return 1.0 - ss_res / ss_tot + + +def mae_score(y_true: Sequence[float], y_pred: Sequence[float]) -> float: + if len(y_true) != len(y_pred): + raise ModelError("y_true/y_pred 长度不匹配") + n = len(y_true) + return sum(abs(y_true[i] - y_pred[i]) for i in range(n)) / n if n else NAN + + +def rmse_score(y_true: Sequence[float], y_pred: Sequence[float]) -> float: + if len(y_true) != len(y_pred): + raise ModelError("y_true/y_pred 长度不匹配") + n = len(y_true) + if n == 0: + return NAN + return math.sqrt(sum((y_true[i] - y_pred[i]) ** 2 for i in range(n)) / n) + + +# --------------------------------------------------------------------------- +# 质量预测模型（聚合特征 + 目标 + 岭回归） +# --------------------------------------------------------------------------- + + +@dataclass +class ModelRecipe: + """模型超参包（外置 JSON/YAML）。""" + + target: str # 质量标签列名（如 TiCl₄纯度） + alpha: float = 1.0 # L2 正则强度 + feature_names: List[str] = field(default_factory=list) + unit: str = "" + + @classmethod + def from_dict(cls, d: Dict[str, Any]) -> "ModelRecipe": + return cls( + target=str(d.get("target", "")).strip(), + alpha=float(d.get("alpha", 1.0)), + feature_names=[str(x) for x in d.get("feature_names", [])], + unit=str(d.get("unit", "")).strip(), + ) + + def validate(self) -> List[str]: + errs = [] + if not self.target: + errs.append("target 不能为空") + if self.alpha < 0: + errs.append(f"alpha 不能为负: {self.alpha}") + return errs + + +@dataclass +class Evaluation: + """评估结果。""" + + r2: float + mae: float + rmse: float + n_samples: int + + def to_dict(self) -> Dict[str, Any]: + return {"r2": round(self.r2, 6), "mae": round(self.mae, 6), + "rmse": round(self.rmse, 6), "n_samples": self.n_samples} + + def passes(self, *, min_r2: float = 0.0, max_mae: float = math.inf) -> bool: + return (self.r2 >= min_r2 and self.mae <= max_mae + and not math.isnan(self.r2)) + + +class QualityModel: + """质量预测模型：特征名 + 目标列 + 岭回归。""" + + def __init__(self, recipe: ModelRecipe): + errs = recipe.validate() + if errs: + raise ModelError("超参包校验失败: " + "; ".join(errs)) + self.recipe = recipe + self.regression: Optional[RidgeRegression] = None + self._feature_std: List[float] = [] + + @property + def fitted(self) -> bool: + return self.regression is not None + + def fit(self, X: Sequence[Sequence[float]], y: Sequence[float]) -> "QualityModel": + """X 行=样本，列=特征（顺序与 recipe.feature_names 对齐）。""" + self.regression = RidgeRegression(alpha=self.recipe.alpha).fit(X, y) + # 记录训练集每列标准差，供 explain() 计算尺度归一化重要性 + m = len(X) + n = len(X[0]) if X else 0 + if m > 1 and n: + means = [sum(X[i][j] for i in range(m)) / m for j in range(n)] + self._feature_std = [ + math.sqrt(sum((X[i][j] - means[j]) ** 2 for i in range(m)) / (m - 1)) + for j in range(n)] + else: + self._feature_std = [1.0] * n + return self + + def predict(self, X: Sequence[Sequence[float]]) -> List[float]: + if self.regression is None: + raise ModelError("模型未训练，先 fit()") + return self.regression.predict(X) + + def evaluate(self, X: Sequence[Sequence[float]], + y: Sequence[float]) -> Evaluation: + pred = self.predict(X) + return Evaluation( + r2=r2_score(y, pred), mae=mae_score(y, pred), + rmse=rmse_score(y, pred), n_samples=len(y)) + + def explain(self) -> List[Dict[str, Any]]: + """特征贡献度（尺度归一化：|权重| × 特征标准差），供可解释性。 + + 归一化动机：原始权重受特征量纲影响（温度 850℃ vs 配比 30）， + ``|权重|×std`` 才反映特征对预测的实际扰动幅度（与 sklearn + permutation importance / 标准化系数同思路），可跨特征横向比较。 + """ + if self.regression is None: + raise ModelError("模型未训练") + names = self.recipe.feature_names or [ + f"x{i}" for i in range(len(self.regression.coef_))] + weights = list(self.regression.coef_) + stds = getattr(self, "_feature_std", None) or [1.0] * len(weights) + contribs = [abs(weights[i]) * stds[i] for i in range(len(weights))] + total = sum(contribs) or 1.0 + return [{"feature": names[i], "weight": round(weights[i], 6), + "importance": round(contribs[i] / total, 4)} + for i in range(len(weights))] + + def to_dict(self) -> Dict[str, Any]: + return { + "recipe": {"target": self.recipe.target, "alpha": self.recipe.alpha, + "feature_names": list(self.recipe.feature_names), + "unit": self.recipe.unit}, + "regression": (self.regression.to_dict() + if self.regression else None), + "feature_std": list(self._feature_std), + } + + @classmethod + def from_dict(cls, d: Dict[str, Any]) -> "QualityModel": + model = cls(ModelRecipe.from_dict(d.get("recipe", {}))) + reg = d.get("regression") + if reg: + model.regression = RidgeRegression.from_dict(reg) + model._feature_std = [float(x) for x in d.get("feature_std", [])] + return model + + +# --------------------------------------------------------------------------- +# 训练集数据集（特征矩阵 + 标签） +# --------------------------------------------------------------------------- + + +@dataclass +class TrainingSet: + """训练集：特征名 + X + y。""" + + feature_names: List[str] + X: List[List[float]] + y: List[float] + + @classmethod + def from_records(cls, records: Sequence[Dict[str, float]], + feature_names: Sequence[str], + target: str) -> "TrainingSet": + X, y = [], [] + for r in records: + if target not in r or math.isnan(r[target]): + continue + row = [r.get(fn, NAN) for fn in feature_names] + if any(math.isnan(v) for v in row): + continue + X.append(row) + y.append(r[target]) + return cls(feature_names=list(feature_names), X=X, y=y) + + def __len__(self) -> int: + return len(self.X) + + +# --------------------------------------------------------------------------- +# 超参包加载（零依赖 YAML 子集 / JSON） +# --------------------------------------------------------------------------- + + +def load_recipe(text: str) -> ModelRecipe: + text = text.strip() + if text.startswith("{"): + data = json.loads(text) + else: + data = _parse_yaml_subset(text) + if not isinstance(data, dict): + raise ModelError("超参包顶层应为映射") + return ModelRecipe.from_dict(data) + + +def _parse_yaml_subset(text: str) -> Any: + """极简 YAML 子集解析（与 features.py 同款实现，避免跨模块依赖）。""" + lines: List[str] = [] + for raw in text.splitlines(): + stripped = raw.rstrip() + if not stripped.strip() or stripped.lstrip().startswith("#"): + continue + hi = _find_inline_comment(stripped) + if hi is not None: + stripped = stripped[:hi].rstrip() + if stripped: + lines.append(stripped) + parser = _YamlParser(lines) + return parser.parse_block(0)[0] if lines else {} + + +def _find_inline_comment(line: str) -> Optional[int]: + depth = 0 + in_str = False + for i, ch in enumerate(line): + if ch == '"': + in_str = not in_str + elif not in_str: + if ch in "[{": + depth += 1 + elif ch in "]}": + depth = max(0, depth - 1) + elif ch == "#" and depth == 0: + if i == 0 or line[i - 1] in (" ", "\t"): + return i + return None + + +def _parse_scalar(raw: str) -> Any: + raw = raw.strip() + if not raw: + return None + if raw.startswith('"') and raw.endswith('"'): + return raw[1:-1] + if raw.startswith("[") or raw.startswith("{"): + try: + return json.loads(raw) + except Exception: + return raw + low = raw.lower() + if low == "true": + return True + if low == "false": + return False + if low in ("null", "~", "none"): + return None + try: + return int(raw) + except ValueError: + pass + try: + return float(raw) + except ValueError: + pass + return raw + + +class _YamlParser: + def __init__(self, lines: List[str]) -> None: + self.lines = lines + self.i = 0 + + def _indent(self, line: str) -> int: + return len(line) - len(line.lstrip(" ")) + + def parse_block(self, indent: int) -> Tuple[Any, bool]: + if self.i >= len(self.lines): + return {}, False + line = self.lines[self.i] + cur = self._indent(line) + if cur < indent: + return {}, False + stripped = line.strip() + if stripped.startswith("- ") or stripped == "-": + return self._parse_list(cur), True + return self._parse_mapping(cur), False + + def _parse_mapping(self, indent: int) -> Dict[str, Any]: + result: Dict[str, Any] = {} + effective = indent + if self.i < len(self.lines): + first = self._indent(self.lines[self.i]) + if first > indent: + effective = first + while self.i < len(self.lines): + line = self.lines[self.i] + cur = self._indent(line) + if cur < effective: + break + if cur > effective: + self.i += 1 + continue + stripped = line.strip() + if stripped.startswith("- "): + break + key, sep, rest = stripped.partition(":") + if not sep: + self.i += 1 + continue + key = key.strip() + rest = rest.strip() + self.i += 1 + if rest: + result[key] = _parse_scalar(rest) + else: + if self.i < len(self.lines): + nxt = self._indent(self.lines[self.i]) + if nxt > effective: + val, _ = self.parse_block(nxt) + result[key] = val + return result + + def _parse_list(self, indent: int) -> List[Any]: + items: List[Any] = [] + while self.i < len(self.lines): + line = self.lines[self.i] + cur = self._indent(line) + if cur < indent: + break + if cur > indent: + self.i += 1 + continue + stripped = line.strip() + if not stripped.startswith("-"): + break + item_text = stripped[1:].strip() + if not item_text: + self.i += 1 + items.append(None) + continue + if ":" in item_text and not item_text.startswith('"'): + k, sep, v = item_text.partition(":") + if sep: + item: Dict[str, Any] = {k.strip(): _parse_scalar(v.strip())} + self.i += 1 + if self.i < len(self.lines): + child_indent = self._indent(self.lines[self.i]) + if child_indent > cur: + sub, _ = self.parse_block(child_indent) + if isinstance(sub, dict): + item.update(sub) + items.append(item) + continue + items.append(_parse_scalar(item_text)) + self.i += 1 + return items diff --git a/templates/ti-cl4/quality-forecast/tests/test_model.py b/templates/ti-cl4/quality-forecast/tests/test_model.py new file mode 100644 index 0000000..b865f1b --- /dev/null +++ b/templates/ti-cl4/quality-forecast/tests/test_model.py @@ -0,0 +1,182 @@ +# -*- coding: utf-8 -*- +"""Ti-1 质量预测模型训练与评估测试（Issue #69）。 + +覆盖： +1. RidgeRegression 拟合线性关系（权重/截距/预测正确）； +2. L2 正则缓解共线性（alpha>0 可解奇异阵）； +3. 评估指标 R²/MAE/RMSE 数值正确； +4. QualityModel fit/predict/evaluate/explain 全链路； +5. TrainingSet.from_records 自动跳过含 NaN 的行； +6. 超参包 YAML/JSON 加载 + 校验； +7. 序列化 to_dict/from_dict 往返一致； +8. 数据不足报错。 +""" +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 model as M # noqa: E402 +from quality_forecast import features as F # noqa: E402 + + +class TestRidgeRegression(unittest.TestCase): + def test_fit_linear(self): + # y = 2*x0 + 1，纯线性 + X = [[0.0], [1.0], [2.0], [3.0]] + y = [1.0, 3.0, 5.0, 7.0] + r = M.RidgeRegression(alpha=0.0).fit(X, y) + self.assertAlmostEqual(r.coef_[0], 2.0, places=4) + self.assertAlmostEqual(r.intercept_, 1.0, places=4) + self.assertAlmostEqual(r.predict([[5.0]])[0], 11.0, places=4) + + def test_fit_multivariate(self): + # y = 1*x0 + 2*x1 + X = [[0.0, 0.0], [1.0, 0.0], [0.0, 1.0], [1.0, 1.0], [2.0, 3.0]] + y = [0.0, 1.0, 2.0, 3.0, 8.0] + r = M.RidgeRegression(alpha=0.0).fit(X, y) + self.assertAlmostEqual(r.coef_[0], 1.0, places=3) + self.assertAlmostEqual(r.coef_[1], 2.0, places=3) + + def test_regularization_handles_collinearity(self): + # 共线性特征：x0 == x1，alpha=0 会奇异，alpha>0 可解 + X = [[1.0, 1.0], [2.0, 2.0], [3.0, 3.0], [4.0, 4.0]] + y = [2.0, 4.0, 6.0, 8.0] + with self.assertRaises(M.ModelError): + M.RidgeRegression(alpha=0.0).fit(X, y) + r = M.RidgeRegression(alpha=1.0).fit(X, y) + pred = r.predict([[5.0, 5.0]]) + # 预测应接近 10（y=2*x） + self.assertAlmostEqual(pred[0], 10.0, places=0) + + def test_empty_raises(self): + with self.assertRaises(M.ModelError): + M.RidgeRegression().fit([], []) + + def test_predict_before_fit_raises(self): + with self.assertRaises(M.ModelError): + M.RidgeRegression().predict([[1.0]]) + + +class TestMetrics(unittest.TestCase): + def test_perfect_prediction(self): + y = [1.0, 2.0, 3.0] + self.assertEqual(M.r2_score(y, y), 1.0) + self.assertEqual(M.mae_score(y, y), 0.0) + self.assertEqual(M.rmse_score(y, y), 0.0) + + def test_r2_mean_predictor(self): + # 预测恒为均值 → R²=0 + y = [1.0, 2.0, 3.0] + mean = 2.0 + self.assertAlmostEqual(M.r2_score(y, [mean, mean, mean]), 0.0, places=6) + + def test_mae_rmse(self): + y = [1.0, 3.0] + pred = [2.0, 2.0] + self.assertEqual(M.mae_score(y, pred), 1.0) + self.assertAlmostEqual(M.rmse_score(y, pred), 1.0, places=6) + + +class TestQualityModel(unittest.TestCase): + def _make_model_and_data(self): + recipe = M.ModelRecipe( + target="y", alpha=0.01, feature_names=["a", "b"]) + # y = 3*a + 0*b + 2；b 为与 a 不相关的纯噪声维度（权重应趋近 0） + X = [[0.0, 7.0], [1.0, 3.0], [2.0, 9.0], + [3.0, 1.0], [4.0, 5.0]] + y = [2.0, 5.0, 8.0, 11.0, 14.0] + return recipe, X, y + + def test_fit_predict_evaluate(self): + recipe, X, y = self._make_model_and_data() + model = M.QualityModel(recipe).fit(X, y) + ev = model.evaluate(X, y) + self.assertGreaterEqual(ev.r2, 0.99) + self.assertLess(ev.mae, 0.1) + self.assertTrue(ev.passes(min_r2=0.9)) + + def test_explain(self): + recipe, X, y = self._make_model_and_data() + model = M.QualityModel(recipe).fit(X, y) + exp = model.explain() + self.assertEqual(len(exp), 2) + # a 的贡献应远大于 b + imp = {e["feature"]: e["importance"] for e in exp} + self.assertGreater(imp["a"], imp["b"]) + + def test_predict_before_fit(self): + model = M.QualityModel(M.ModelRecipe(target="y")) + with self.assertRaises(M.ModelError): + model.predict([[1.0]]) + + def test_serialization_roundtrip(self): + recipe, X, y = self._make_model_and_data() + model = M.QualityModel(recipe).fit(X, y) + d = model.to_dict() + model2 = M.QualityModel.from_dict(d) + self.assertEqual(model.predict(X), model2.predict(X)) + + +class TestTrainingSet(unittest.TestCase): + def test_skips_nan_rows(self): + records = [ + {"a": 1.0, "b": 2.0, "y": 5.0}, + {"a": float("nan"), "b": 2.0, "y": 5.0}, # a NaN → 跳过 + {"a": 1.0, "b": float("nan"), "y": 5.0}, # b NaN → 跳过 + {"a": 3.0, "b": 4.0, "y": 7.0}, + {"a": 1.0, "b": 2.0}, # 无 y → 跳过 + ] + ts = M.TrainingSet.from_records(records, ["a", "b"], "y") + self.assertEqual(len(ts), 2) + + +class TestRecipeLoading(unittest.TestCase): + def test_load_yaml(self): + text = "target: TiCl4_purity\nalpha: 0.5\nunit: '%'\n" + r = M.load_recipe(text) + self.assertEqual(r.target, "TiCl4_purity") + self.assertEqual(r.alpha, 0.5) + + def test_load_json(self): + r = M.load_recipe('{"target": "y", "alpha": 2.0}') + self.assertEqual(r.target, "y") + self.assertEqual(r.alpha, 2.0) + + def test_bad_recipe_raises(self): + with self.assertRaises(M.ModelError): + M.QualityModel(M.ModelRecipe(target="", alpha=-1)) + + +class TestEndToEndWithFeatures(unittest.TestCase): + """端到端：特征抽取 → 训练 → 评估。""" + + def test_pipeline(self): + # 用 features 模块抽取，再训练一个能拟合的模型 + ext = F.FeatureExtractor([ + F.FeatureSpec("t", "CLF-01.TEMP", "raw"), + ]) + samples = [ + F.Sample(ts=i * 10.0, values={"CLF-01.TEMP": 100.0 + 10 * i}) + for i in range(8) + ] + matrix = ext.extract(samples) + # 目标：纯度 = 0.1*TEMP - 5（线性可分） + records = [] + for row, s in zip(matrix.rows, samples): + temp = s.values["CLF-01.TEMP"] + records.append({"t": row[0], "RF-01.PURITY": 0.1 * temp - 5.0}) + ts = M.TrainingSet.from_records(records, ["t"], "RF-01.PURITY") + model = M.QualityModel( + M.ModelRecipe(target="RF-01.PURITY", alpha=0.001, + feature_names=["t"])).fit(ts.X, ts.y) + ev = model.evaluate(ts.X, ts.y) + self.assertGreaterEqual(ev.r2, 0.99) + + +if __name__ == "__main__": + unittest.main(verbosity=2) From a10b83007e2b3e9b223c2b9d9ee7ad52ab502bac Mon Sep 17 00:00:00 2001 From: bot_dev1 Date: Wed, 5 Aug 2026 05:17:36 +0800 Subject: [PATCH 3/3] =?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)