From c6dc7d2344772b48cbd95d4ab9b1fbadcc6a6fa6 Mon Sep 17 00:00:00 2001 From: bot_dev1 Date: Wed, 5 Aug 2026 05:14:39 +0800 Subject: [PATCH] =?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)