From 42d206e50301a4d426aa7cb0f44a3a7ca3543f2a Mon Sep 17 00:00:00 2001 From: bot_dev1 Date: Wed, 5 Aug 2026 01:59:42 +0800 Subject: [PATCH] =?UTF-8?q?feat(#70):=20=E7=82=89=E5=B1=82=E6=9D=82?= =?UTF-8?q?=E8=B4=A8=E9=A2=84=E8=AD=A6=E7=89=B9=E5=BE=81=E5=B7=A5=E7=A8=8B?= =?UTF-8?q?=EF=BC=88=E5=A3=B0=E6=98=8E=E5=BC=8FFeatureSpec=E5=BC=95?= =?UTF-8?q?=E6=93=8E=EF=BC=8CPRD=205.3=20=E2=91=A2=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 新增 templates/ti-cl4/impurity-forecast：声明式特征工程引擎，特征以 FeatureSpec 描述（EMA/RollingStd/RateOfChange 等 7 算子），换行业只改模板配置 features.template.yaml，引擎零改动（PRD 5.3：特征工程层跨行业差异落在 FeatureSpec，不落代码）。 - features.py：FeatureSpec 声明 + 校验 + 7 算子 + 时序对齐 + 阈值 breach + 零依赖 YAML 解析 - config/features.template.yaml：炉温/氯气/炉压/炉层 9 条特征（对齐点位字典 point_id） - tests/test_features.py：24 项单测（校验/算子/对齐/breach/配置/端到端提前量）全通过 - _sanity_check.py：冒烟脚本（配置加载 + transform + breach 可观测） 验收：一期阈值+无监督上线，提前量信号可观测（PRD 提前≥30min、误报率≤8%口径）。 --- templates/ti-cl4/impurity-forecast/README.md | 81 +++ .../ti-cl4/impurity-forecast/__init__.py | 29 + .../ti-cl4/impurity-forecast/_sanity_check.py | 67 +++ .../config/features.template.yaml | 73 +++ .../ti-cl4/impurity-forecast/features.py | 559 ++++++++++++++++++ .../impurity-forecast/tests/_bootstrap.py | 26 + .../impurity-forecast/tests/test_features.py | 313 ++++++++++ 7 files changed, 1148 insertions(+) create mode 100644 templates/ti-cl4/impurity-forecast/README.md create mode 100644 templates/ti-cl4/impurity-forecast/__init__.py create mode 100644 templates/ti-cl4/impurity-forecast/_sanity_check.py create mode 100644 templates/ti-cl4/impurity-forecast/config/features.template.yaml create mode 100644 templates/ti-cl4/impurity-forecast/features.py create mode 100644 templates/ti-cl4/impurity-forecast/tests/_bootstrap.py create mode 100644 templates/ti-cl4/impurity-forecast/tests/test_features.py diff --git a/templates/ti-cl4/impurity-forecast/README.md b/templates/ti-cl4/impurity-forecast/README.md new file mode 100644 index 0000000..0db2496 --- /dev/null +++ b/templates/ti-cl4/impurity-forecast/README.md @@ -0,0 +1,81 @@ +# iAOP-Template-Ti 一期 · 炉层杂质预警特征工程（Impurity Forecast） + +对应 PRD 5.3「③ 炉层杂质预警」与 Issue #70「[Ti-1] 炉层杂质预警特征工程」。 + +PRD 5.3 明确：**特征工程层的跨行业差异落在 FeatureSpec，不落代码**。本模块 +实现一个**声明式特征工程引擎**——特征以 `FeatureSpec` 描述（如 `EMA(炉温, 5min)`、 +`RollingStd(氯气流量, 10)`、`RateOfChange(炉压)`），由工艺模板定义、引擎解释执行； +**换行业只改模板配置（`config/features.template.yaml`），引擎零改动**。 + +## 验收口径（PRD 5.3 ③ / 验收表） + +- 炉层杂质预警提前 ≥ 30 分钟； +- 误报率 ≤ 8%； +- 一期：阈值 + 无监督上线（数据门槛低）；3 个月后转监督。 + +## 模块结构 + +``` +templates/ti-cl4/impurity-forecast/ +├── __init__.py 包入口（导出 FeatureEngine / FeatureSpec 等） +├── features.py 声明式特征工程引擎（FeatureSpec + 算子 + YAML 解析） +├── _sanity_check.py 冒烟脚本（零环境依赖，直接运行） +├── config/ +│ └── features.template.yaml 模板特征配置资产（炉温/氯气/炉压/炉层 9 条特征） +└── tests/ + ├── _bootstrap.py 测试引导（连字符目录挂载为可导入包） + └── test_features.py 24 项单测（校验/算子/对齐/breach/配置/端到端） +``` + +## 内置算子（声明式 FeatureSpec 的 `kind`） + +| kind | 含义 | 参数 | 示例 | +|------|------|------|------| +| `raw` | 原始值透传 | — | `炉温_raw` | +| `ema` | 指数滑动平均（平滑去噪） | `alpha ∈ (0,1]` | `炉温_ema5 = ema(炉温, alpha=0.2)` | +| `rolling_std` | 滚动标准差（波动度） | `window > 0` | `氯气流量_std10` | +| `rolling_mean` | 滚动均值 | `window > 0` | `炉层状态_mean10` | +| `rolling_min` / `rolling_max` | 滚动极值 | `window > 0` | 配合阈值 | +| `rate_of_change` | 变化率 `(x[t]-x[t-w])/x[t-w]` | `window > 0` | `炉压_rate10` | + +## 用法 + +```python +from impurity_forecast import FeatureEngine, load_feature_config + +# 1) 从模板资产构建引擎 +engine = FeatureEngine.from_template_config("config/features.template.yaml") + +# 2) 时序样本流 → 按时间对齐的特征向量序列 +samples = [ + {"ts": 1, "CLF-01.TEMP": 850.0, "CLF-01.CL2": 100.0, + "CLF-01.PRES": 10.0, "CLF-01.BED": 60.0}, + # ... +] +vectors = engine.transform(samples) + +# 3) 无监督阈值判定（一期上线口径） +for vec in vectors: + if vec.is_complete: + breach = engine.breach(vec) # 超阈值的 (特征名, 当前值) 列表 + if breach: + ... # 触发预警（提前量信号） +``` + +## 运行测试 / 冒烟 + +```bash +cd templates/ti-cl4/impurity-forecast +python -m unittest discover -s tests # 24 项单测 +python _sanity_check.py # 冒烟：配置加载 + transform + breach +``` + +## 设计要点 + +1. **零第三方依赖**：纯 Python 标准库；YAML 子集自解析（对齐 data-bus / rag-kb）。 +2. **模板化**：测点对齐 `templates/ti-cl4/point-dict/point_dict.default.csv` 的 + `point_id`；特征声明集中在 `config/features.template.yaml`。 +3. **缺失值治理**：缺失测点用 `NaN` 占位并记录缺失率，作为误报率治理输入。 +4. **预热语义**：滚动/变化率类算子在窗口未满时输出 `NaN`（"尚不足以计算"）， + 避免冷启动误报。 +5. **配套**：本特征矩阵供 #71「炉层杂质预警模型训练」消费（监督/无监督均可用）。 diff --git a/templates/ti-cl4/impurity-forecast/__init__.py b/templates/ti-cl4/impurity-forecast/__init__.py new file mode 100644 index 0000000..48e2b22 --- /dev/null +++ b/templates/ti-cl4/impurity-forecast/__init__.py @@ -0,0 +1,29 @@ +# -*- coding: utf-8 -*- +"""iAOP-Template-Ti 一期 · 炉层杂质预警特征工程包（Issue #70）。 + +导出声明式 FeatureSpec 特征工程引擎，供预警模型训练（#71）与驾驶舱 +告警面板复用。换行业只改模板配置，引擎零改动（PRD 5.3 特征工程层）。 +""" +from __future__ import annotations + +from .features import ( + FeatureKind, + FeatureSpec, + FeatureSpecError, + FeatureValue, + FeatureVector, + FeatureEngine, + FeatureTemplateConfig, + load_feature_config, +) + +__all__ = [ + "FeatureKind", + "FeatureSpec", + "FeatureSpecError", + "FeatureValue", + "FeatureVector", + "FeatureEngine", + "FeatureTemplateConfig", + "load_feature_config", +] diff --git a/templates/ti-cl4/impurity-forecast/_sanity_check.py b/templates/ti-cl4/impurity-forecast/_sanity_check.py new file mode 100644 index 0000000..f8c0cf0 --- /dev/null +++ b/templates/ti-cl4/impurity-forecast/_sanity_check.py @@ -0,0 +1,67 @@ +# -*- coding: utf-8 -*- +"""炉层杂质预警特征工程冒烟脚本（Issue #70）。 + +直接运行 ``python _sanity_check.py`` 验证：模板资产可加载、引擎可对 +合成时序产出完整特征向量、阈值 breach 可观测。零第三方依赖。 +""" +import importlib.util +import os +import sys + +# impurity-forecast 目录名含连字符，不能作为 Python 包名直接 import； +# 用 importlib 按文件路径加载为合法包 impurity_forecast（同 tests/_bootstrap.py）。 +_PKG_DIR = os.path.dirname(os.path.abspath(__file__)) + + +def _load_pkg(name, path): + if name in sys.modules: + return + spec = importlib.util.spec_from_file_location( + name, os.path.join(path, "__init__.py"), + submodule_search_locations=[path]) + mod = importlib.util.module_from_spec(spec) + sys.modules[name] = mod + spec.loader.exec_module(mod) + + +_load_pkg("impurity_forecast", _PKG_DIR) + +from impurity_forecast import FeatureEngine, load_feature_config # noqa: E402 + +CONFIG = os.path.join(os.path.dirname(os.path.abspath(__file__)), + "config", "features.template.yaml") + + +def main() -> int: + cfg = load_feature_config(CONFIG) + print(f"[OK] 模板特征配置加载: template={cfg.template} " + f"version={cfg.version} features={len(cfg.specs)}") + for line in FeatureEngine(cfg.specs).describe(): + print(" -", line) + + eng = FeatureEngine.from_template_config(CONFIG) + # 合成 30 步温和时序（不触发 breach），验证预热后向量完整 + samples = [ + {"ts": i, "CLF-01.TEMP": 850.0, "CLF-01.CL2": 100.0, + "CLF-01.PRES": 10.0, "CLF-01.BED": 60.0} + for i in range(30) + ] + vecs = eng.transform(samples) + last = vecs[-1] + assert last.is_complete, "预热后特征向量应完整" + assert eng.breach(last) == [], "温和时序不应触发 breach" + print(f"[OK] transform: 30 步时序 → 末向量完整，缺失率={last.missing_rate:.2%}") + + # 合成急升温序列，验证阈值 breach 可观测 + hot = [{"ts": i, "CLF-01.TEMP": 850.0 + 8.0 * i, "CLF-01.CL2": 100.0, + "CLF-01.PRES": 10.0, "CLF-01.BED": 60.0} for i in range(30)] + hot_vecs = eng.transform(hot) + breached = any(eng.breach(v) for v in hot_vecs) + assert breached, "急升温序列应触发 breach" + print("[OK] 急升温序列触发 breach（提前量信号可观测）") + print("炉层杂质预警特征工程冒烟通过 ✅") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/templates/ti-cl4/impurity-forecast/config/features.template.yaml b/templates/ti-cl4/impurity-forecast/config/features.template.yaml new file mode 100644 index 0000000..f2ebc4b --- /dev/null +++ b/templates/ti-cl4/impurity-forecast/config/features.template.yaml @@ -0,0 +1,73 @@ +# iAOP-Template-Ti 一期 · 炉层杂质预警特征工程模板资产（Issue #70 / PRD 5.3 ③） +# +# 换行业只改本配置，特征引擎（features.py）零改动（PRD 5.3：特征工程层 +# 跨行业差异落在 FeatureSpec，不落代码）。 +# +# 测点对齐 templates/ti-cl4/point-dict/point_dict.default.csv 的 point_id： +# CLF-01.TEMP 炉温 CLF-01.PRES 炉压 +# CLF-01.CL2 氯气流量 CLF-01.CO CO含量 +# CLF-01.CO2 CO₂含量 CLF-01.BED 炉层状态 +# +# 验收口径（PRD 5.3 ③ / 表验收）：炉层杂质预警提前 ≥ 30min，误报率 ≤ 8%。 +# 一期阈值 + 无监督上线；阈值参考工艺规范（沸腾氯化炉温 850±50℃，超 900℃ +# 触发降流减料），3 个月后转监督再调参。 +template: ti-cl4 +version: 1.0.0 +description: 炉层杂质预警特征工程（声明式 FeatureSpec，PRD 5.3 ③） + +specs: + # ---- 炉温：原始 + 平滑 + 变化率（趋势预警） ------------------------- + - name: 炉温_raw + kind: raw + point: CLF-01.TEMP + unit: ℃ + threshold: 900.0 # 工艺上限：超 900℃ 触发降流减料（SOP-CL-001） + - name: 炉温_ema5 + kind: ema + point: CLF-01.TEMP + unit: ℃ + params: {alpha: 0.2} + threshold: 900.0 # 平滑后更稳，避免毛刺误报 + - name: 炉温_rate10 + kind: rate_of_change + point: CLF-01.TEMP + unit: "1/min" + params: {window: 10} + threshold: 0.05 # 5%/10min 急升趋势，提前量信号 + + # ---- 氯气流量：波动度（流态化异常先兆） ------------------------------ + - name: 氯气流量_raw + kind: raw + point: CLF-01.CL2 + unit: m³/h + - name: 氯气流量_std10 + kind: rolling_std + point: CLF-01.CL2 + unit: m³/h + params: {window: 10} + threshold: 8.0 # 滚动标准差越界 → 供料不稳 + + # ---- 炉压：变化率（压力异常是炉层状态恶化强信号） -------------------- + - name: 炉压_raw + kind: raw + point: CLF-01.PRES + unit: kPa + - name: 炉压_rate10 + kind: rate_of_change + point: CLF-01.PRES + unit: "1/min" + params: {window: 10} + threshold: 0.08 # 8%/10min 压力急变 + + # ---- 炉层状态：原始值 + 滚动均值（杂质富集表征） --------------------- + - name: 炉层状态_raw + kind: raw + point: CLF-01.BED + unit: "%" + threshold: 85.0 + - name: 炉层状态_mean10 + kind: rolling_mean + point: CLF-01.BED + unit: "%" + params: {window: 10} + threshold: 85.0 diff --git a/templates/ti-cl4/impurity-forecast/features.py b/templates/ti-cl4/impurity-forecast/features.py new file mode 100644 index 0000000..a91e6c9 --- /dev/null +++ b/templates/ti-cl4/impurity-forecast/features.py @@ -0,0 +1,559 @@ +# -*- coding: utf-8 -*- +"""炉层杂质预警 · 声明式特征工程引擎（Issue #70 / PRD 5.3 异常·杂质预警）。 + +PRD 5.3 明确：**特征工程层的跨行业差异落在 FeatureSpec，不落代码**。 +即特征以声明式规格描述（如 ``EMA(炉温, 5min)``、``RollingStd(氯气流量, 10)``、 +``RateOfChange(炉压)``），由工艺模板定义、本引擎解释执行；换行业只改模板 +配置（``config/features.template.yaml``），引擎零改动。 + +一期（Template-Ti）落地「炉层杂质预警」（PRD 5.3 ③，验收：提前 ≥ 30min、 +误报率 ≤ 8%）。数据门槛低，先以**阈值 + 无监督**上线，3 个月后转监督 +（PRD 4.1）；故本期特征工程面向无监督异常评分，同时产出监督可用的特征矩阵。 + +设计要点 +-------- +1. **声明式 FeatureSpec**：每条特征声明 ``kind``（算子）+ ``point``（来源测点） + + ``params``（窗口/周期等），引擎按 kind 分派到内置算子。 +2. **内置算子**（零第三方依赖，纯标准库）： + - ``raw`` 原始值透传； + - ``ema`` 指数滑动平均（平滑、去噪）； + - ``rolling_std`` 滚动标准差（波动度）； + - ``rate_of_change`` 变化率（速率预警）； + - ``rolling_mean`` 滚动均值； + - ``rolling_min`` / ``rolling_max`` 滚动极值（配合阈值）。 +3. **时序对齐**：按时间戳对齐多测点为特征向量，缺失测点用 ``NaN`` 占位 + 并记录缺失率（误报率治理输入）。 +4. **零依赖 YAML 子集解析**（与 data-bus / rag-kb 同款），解析模板资产。 +""" +from __future__ import annotations + +import math +import os +from dataclasses import dataclass, field +from enum import Enum +from typing import Callable, Dict, List, Optional, Sequence, Tuple + +# 缺失值统一用 float('nan')，便于上层用 math.isnan 判定与屏蔽。 +NAN = float("nan") + + +class FeatureSpecError(ValueError): + """FeatureSpec 声明或执行错误（未知算子 / 缺参 / 窗口非法等）。""" + + +class FeatureKind(str, Enum): + """内置特征算子（声明式 FeatureSpec 的 ``kind`` 取值）。""" + + RAW = "raw" # 原始值透传 + EMA = "ema" # 指数滑动平均：params={"alpha": 0.2} + ROLLING_STD = "rolling_std" # 滚动标准差：params={"window": 10} + ROLLING_MEAN = "rolling_mean" # 滚动均值 + ROLLING_MIN = "rolling_min" # 滚动最小值 + ROLLING_MAX = "rolling_max" # 滚动最大值 + RATE_OF_CHANGE = "rate_of_change" # 变化率：(x[t]-x[t-w])/x[t-w] + + @property + def label(self) -> str: + return { + FeatureKind.RAW: "原始值", + FeatureKind.EMA: "指数滑动平均", + FeatureKind.ROLLING_STD: "滚动标准差", + FeatureKind.ROLLING_MEAN: "滚动均值", + FeatureKind.ROLLING_MIN: "滚动最小值", + FeatureKind.ROLLING_MAX: "滚动最大值", + FeatureKind.RATE_OF_CHANGE: "变化率", + }[self] + + +# 算子注册表：kind 名 → 算子实现。未知 kind 在注册阶段即拒绝（避免拼写漂移）。 +KIND_REGISTRY: Dict[str, FeatureKind] = {k.value: k for k in FeatureKind} + + +@dataclass +class FeatureSpec: + """单条声明式特征规格（模板配置中的一行特征声明）。 + + Attributes: + name: 特征输出名（特征向量列名，工艺可读，如 ``炉温_ema5``）。 + kind: 算子（见 :class:`FeatureKind`）。 + point: 来源测点 id（对齐点位字典 point_id，如 ``CLF-01.TEMP``）。 + params: 算子参数（如 EMA 的 alpha、rolling_* 的 window）。 + unit: 特征单位（可选，用于驾驶舱展示）。 + threshold: 预警阈值（可选，无监督阈值上线的判定边界）。 + """ + + name: str + kind: FeatureKind + point: str + params: Dict[str, float] = field(default_factory=dict) + unit: str = "" + threshold: Optional[float] = None + + def __post_init__(self) -> None: + if not self.name: + raise FeatureSpecError("FeatureSpec.name 不能为空") + if not self.point: + raise FeatureSpecError(f"特征 {self.name!r} 缺少 point（来源测点）") + # 参数合法性校验：滚动/变化率类必须有正整数 window + if self.kind in (FeatureKind.ROLLING_STD, FeatureKind.ROLLING_MEAN, + FeatureKind.ROLLING_MIN, FeatureKind.ROLLING_MAX, + FeatureKind.RATE_OF_CHANGE): + w = self.params.get("window") + if w is None: + raise FeatureSpecError( + f"特征 {self.name!r}（{self.kind.value}）缺少 window 参数") + try: + wf = float(w) + except (TypeError, ValueError) as exc: + raise FeatureSpecError( + f"特征 {self.name!r} window 必须是整数，实际 {w!r}") from exc + if wf != int(wf): + raise FeatureSpecError( + f"特征 {self.name!r} window 必须是整数，实际 {w!r}") + wi = int(wf) + if wi <= 0: + raise FeatureSpecError( + f"特征 {self.name!r} window 必须 > 0，实际 {wi}") + self.params["window"] = wi + if self.kind is FeatureKind.EMA: + alpha = self.params.get("alpha") + if alpha is None: + raise FeatureSpecError(f"特征 {self.name!r}（ema）缺少 alpha 参数") + try: + af = float(alpha) + except (TypeError, ValueError) as exc: + raise FeatureSpecError( + f"特征 {self.name!r} alpha 必须是数值，实际 {alpha!r}") from exc + if not (0.0 < af <= 1.0): + raise FeatureSpecError( + f"特征 {self.name!r} alpha 须在 (0,1]，实际 {af}") + self.params["alpha"] = af + + def describe(self) -> str: + """工艺可读描述，如 ``炉温_ema5 = ema(CLF-01.TEMP, alpha=0.2)``。""" + pa = ", ".join(f"{k}={v}" for k, v in self.params.items()) + return f"{self.name} = {self.kind.value}({self.point}{', ' + pa if pa else ''})" + + +# --------------------------------------------------------------------------- +# 算子实现：输入为按时间排序的标量序列（可能含 NAN），输出等长变换序列。 +# 滚动窗口在序列前段（样本不足 window 个）输出 NAN，表示"尚不足以计算"。 +# --------------------------------------------------------------------------- + +def _is_num(x: object) -> bool: + return isinstance(x, (int, float)) and not (isinstance(x, float) and math.isnan(x)) + + +def _rolling_window(values: Sequence[float], window: int, + reducer) -> List[float]: + """通用滚动归约：前 window-1 个位置输出 NAN。""" + out: List[float] = [] + buf: List[float] = [] + for v in values: + if _is_num(v): + buf.append(float(v)) + # 非数值视为缺失，不进缓冲区（窗口按"有效样本数"计数，更稳健） + if len(buf) >= window: + out.append(float(reducer(buf[-window:]))) + else: + out.append(NAN) + return out + + +def _op_raw(values: Sequence[float], params: Dict[str, float]) -> List[float]: + return [float(v) if _is_num(v) else NAN for v in values] + + +def _op_ema(values: Sequence[float], params: Dict[str, float]) -> List[float]: + alpha = float(params["alpha"]) + out: List[float] = [] + prev: Optional[float] = None + for v in values: + if not _is_num(v): + out.append(NAN) + continue + x = float(v) + prev = x if prev is None else (alpha * x + (1.0 - alpha) * prev) + out.append(prev) + return out + + +def _op_rate_of_change(values: Sequence[float], + params: Dict[str, float]) -> List[float]: + window = int(params["window"]) + out: List[float] = [] + num: List[float] = [] + for v in values: + if _is_num(v): + num.append(float(v)) + if len(num) >= window + 1: + base = num[-(window + 1)] + cur = num[-1] + out.append((cur - base) / base if base else NAN) + else: + out.append(NAN) + return out + + +# kind → 算子函数 注册（FeatureEngine 分派用） +OPERATORS: Dict[FeatureKind, Callable[[Sequence[float], Dict[str, float]], List[float]]] = { + FeatureKind.RAW: _op_raw, + FeatureKind.EMA: _op_ema, + FeatureKind.ROLLING_STD: lambda v, p: _rolling_window(v, int(p["window"]), + lambda w: _std(w)), + FeatureKind.ROLLING_MEAN: lambda v, p: _rolling_window(v, int(p["window"]), + lambda w: sum(w) / len(w)), + FeatureKind.ROLLING_MIN: lambda v, p: _rolling_window(v, int(p["window"]), min), + FeatureKind.ROLLING_MAX: lambda v, p: _rolling_window(v, int(p["window"]), max), + FeatureKind.RATE_OF_CHANGE: _op_rate_of_change, +} + + +def _std(samples: Sequence[float]) -> float: + """总体标准差（无监督波动度特征；零依赖实现）。""" + n = len(samples) + if n == 0: + return NAN + mean = sum(samples) / n + var = sum((x - mean) ** 2 for x in samples) / n + return math.sqrt(var) + + +# --------------------------------------------------------------------------- +# 特征值 / 特征向量 / 引擎 +# --------------------------------------------------------------------------- + +FeatureValue = float # 单个特征值（可能为 NAN） + + +@dataclass +class FeatureVector: + """某时刻对齐后的特征向量（多特征列 + 时间戳 + 缺失率）。""" + + timestamp: float + values: Dict[str, FeatureValue] # name → 特征值 + missing_rate: float = 0.0 # 本时刻缺失特征占比（误报率治理输入） + + @property + def is_complete(self) -> bool: + """所有特征均非缺失（监督训练样本需完整向量）。""" + return all(_is_num(v) for v in self.values.values()) + + def breach_features(self) -> List[str]: + """返回超阈值 breach 的特征名（无监督阈值上线判定）。""" + # 阈值判定由 FeatureEngine 注入（见 engine.breach），这里仅占位。 + return [] + + +class FeatureEngine: + """声明式特征工程引擎：解释 FeatureSpec 列表，对时序样本计算特征矩阵。 + + 换行业只改模板配置（FeatureSpec 列表），引擎零改动（PRD 5.3）。 + + 用法:: + + engine = FeatureEngine(specs) + vectors = engine.transform(samples) + for vec in vectors: + if vec.is_complete: + ... # 喂给无监督评分器或监督训练 + """ + + def __init__(self, specs: Sequence[FeatureSpec]): + if not specs: + raise FeatureSpecError("FeatureEngine 至少需要一条 FeatureSpec") + # 同名特征直接拒绝（避免特征矩阵列冲突） + seen = set() + for s in specs: + if s.name in seen: + raise FeatureSpecError(f"特征名重复：{s.name!r}") + seen.add(s.name) + self.specs: List[FeatureSpec] = list(specs) + # 按来源测点聚合，减少重复取数 + self._by_point: Dict[str, List[FeatureSpec]] = {} + for s in self.specs: + self._by_point.setdefault(s.point, []).append(s) + + # -- 配置资产 --------------------------------------------------------- + + @classmethod + def from_template_config(cls, path: str) -> "FeatureEngine": + """从模板特征配置 YAML 资产构建引擎（零第三方依赖）。""" + return cls(load_feature_config(path).specs) + + # -- 计算 ------------------------------------------------------------- + + def required_points(self) -> List[str]: + """引擎依赖的全部来源测点 id（去重保序）。""" + seen, out = set(), [] + for s in self.specs: + if s.point not in seen: + seen.add(s.point) + out.append(s.point) + return out + + def transform(self, samples: Sequence[Dict[str, object]], + ts_key: str = "ts") -> List[FeatureVector]: + """把时序样本流变换为按时间对齐的特征向量序列。 + + Args: + samples: 按 时间升序 排列的样本列表；每条样本是 ``{ts_key: epoch秒, + point_id: value, ...}`` 形态的 dict（对齐点位字典 point_id）。 + ts_key: 时间戳键名（默认 ``ts``）。 + + Returns: + 与 samples 等长的 FeatureVector 列表（按时间对齐）。 + """ + if not samples: + return [] + # 1) 按测点抽取时间序列（保持原顺序） + point_series: Dict[str, List[float]] = {p: [] for p in self._by_point} + timestamps: List[float] = [] + for sample in samples: + ts = sample.get(ts_key) + try: + timestamps.append(float(ts) if ts is not None else NAN) + except (TypeError, ValueError): + timestamps.append(NAN) + for p in point_series: + v = sample.get(p) + point_series[p].append(float(v) if _is_num(v) else NAN) + + # 2) 对每个测点的序列逐特征计算 + # feature_columns[name] = 与时间等长的特征值序列 + feature_columns: Dict[str, List[float]] = {} + for point, series in point_series.items(): + for spec in self._by_point[point]: + op = OPERATORS.get(spec.kind) + if op is None: # 理论上 __post_init__ 已拦截，防御性 + raise FeatureSpecError(f"未实现的算子 {spec.kind.value!r}") + feature_columns[spec.name] = op(series, spec.params) + + # 3) 按时间戳对齐为特征向量 + n = len(samples) + names = [s.name for s in self.specs] + vectors: List[FeatureVector] = [] + for i in range(n): + row = {name: feature_columns[name][i] for name in names} + missing = sum(1 for v in row.values() if not _is_num(v)) + vectors.append(FeatureVector( + timestamp=timestamps[i], + values=row, + missing_rate=missing / len(names) if names else 0.0, + )) + return vectors + + # -- 无监督阈值判定（一期上线口径） ----------------------------------- + + def breach(self, vector: FeatureVector) -> List[Tuple[str, float]]: + """返回超阈值的 ``(特征名, 当前值)`` 列表（无监督阈值上线判定）。 + + 一期 PRD 5.3 ③：阈值 + 无监督先上线；3 个月后转监督。本方法支持 + FeatureSpec 声明的 ``threshold``（绝对值越界即 breach）。 + """ + out: List[Tuple[str, float]] = [] + spec_by_name = {s.name: s for s in self.specs} + for name, val in vector.values.items(): + if not _is_num(val): + continue + spec = spec_by_name.get(name) + if spec is None or spec.threshold is None: + continue + if val > spec.threshold: + out.append((name, val)) + return out + + def describe(self) -> List[str]: + """返回全部特征的工艺可读描述（文档/审计用）。""" + return [s.describe() for s in self.specs] + + +# --------------------------------------------------------------------------- +# 模板配置资产（零依赖 YAML 子集解析，对齐 data-bus / rag-kb） +# --------------------------------------------------------------------------- + +@dataclass +class FeatureTemplateConfig: + """模板特征配置：模板元信息 + FeatureSpec 列表。""" + + template: str + version: str + specs: List[FeatureSpec] + description: str = "" + + +def _parse_scalar(text: str) -> str: + """去掉标量两侧引号与行内注释。""" + t = text.split(" #", 1)[0].strip() + if len(t) >= 2 and t[0] == t[-1] and t[0] in ("'", '"'): + return t[1:-1] + return t + + +def _parse_flow_value(text: str): + """解析 ``key: value`` 右侧的值，支持行内 flow map ``{k: v, k: v}``。 + + 其余（标量 / 引号串）退化为 :func:`_parse_scalar`。flow map 用于 + ``params: {alpha: 0.2, window: 10}`` 这种紧凑声明。 + """ + t = text.split(" #", 1)[0].strip() + if t.startswith("{") and t.endswith("}"): + inner = t[1:-1].strip() + out: Dict[str, object] = {} + if not inner: + return out + for part in inner.split(","): + if ":" not in part: + raise FeatureSpecError(f"flow map 项不是键值对：{part!r}") + k, _, v = part.partition(":") + out[k.strip()] = _parse_scalar(v) + return out + return _parse_scalar(text) + + +def _strip_comments(lines: List[str]) -> List[Tuple[str, int]]: + out = [] + for i, ln in enumerate(lines): + s = ln.strip() + if not s or s.startswith("#"): + continue + out.append((ln, i + 1)) + return out + + +def _parse_node(lines: List[Tuple[str, int]], i: int, indent: int): + """递归解析 YAML 节点（map / list / scalar）。返回 (value, next_i)。""" + text, _ = lines[i] + # ---- list 节点 ---- + if text.lstrip(" ").startswith("- "): + items: List[object] = [] + while i < len(lines): + t, no = lines[i] + stripped = t.lstrip(" ") + if not stripped.startswith("- "): + break + lead_j = len(t) - len(t.lstrip(" ")) + if lead_j != indent: + break + item_text = stripped[2:].strip() + if not item_text: + raise FeatureSpecError(f"features.yaml 第 {no} 行：list 项为空") + if ":" in item_text: + map_indent = len(t) - len(t.lstrip(" ")) + 2 + lines[i] = (" " * map_indent + item_text, no) + v, i = _parse_node(lines, i, map_indent) + items.append(v) + else: + items.append(_parse_flow_value(item_text)) + i += 1 + return items, i + # ---- map 节点 ---- + result: Dict[str, object] = {} + while i < len(lines): + t, no = lines[i] + lead_j = len(t) - len(t.lstrip(" ")) + if lead_j < indent or t.lstrip(" ").startswith("- "): + break + if lead_j > indent: + raise FeatureSpecError( + f"features.yaml 第 {no} 行缩进异常（期望 {indent}，实际 {lead_j}）") + if ":" not in t: + raise FeatureSpecError(f"features.yaml 第 {no} 行不是合法键值对：{t!r}") + key, _, rest = t.partition(":") + key = key.strip() + rest = rest.strip() + if rest: + result[key] = _parse_flow_value(rest) + i += 1 + continue + if i + 1 >= len(lines): + raise FeatureSpecError(f"features.yaml 第 {no} 行 {key!r} 缺少值") + sub_indent = len(lines[i + 1][0]) - len(lines[i + 1][0].lstrip(" ")) + if sub_indent <= indent: + raise FeatureSpecError(f"features.yaml 第 {no} 行 {key!r} 缺少值（无嵌套）") + v, i = _parse_node(lines, i + 1, sub_indent) + result[key] = v + return result, i + + +def _load_yaml_text(text: str) -> Dict[str, object]: + lines = _strip_comments(text.splitlines()) + if not lines: + return {} + top_indent = len(lines[0][0]) - len(lines[0][0].lstrip(" ")) + value, next_i = _parse_node(lines, 0, top_indent) + if not isinstance(value, dict): + raise FeatureSpecError("features.yaml 顶层必须是 map") + if next_i < len(lines): + raise FeatureSpecError( + f"features.yaml 第 {lines[next_i][1]} 行：顶层存在多个节点") + return value + + +def load_feature_config(path: str) -> FeatureTemplateConfig: + """从模板特征 YAML 资产加载配置。 + + 期望结构（详见 ``config/features.template.yaml``）:: + + template: ti-cl4 + version: 1.0.0 + description: 炉层杂质预警特征工程 + specs: + - name: 炉温_raw + kind: raw + point: CLF-01.TEMP + - name: 炉温_ema5 + kind: ema + point: CLF-01.TEMP + params: {alpha: 0.2} + threshold: 900.0 + """ + with open(path, "r", encoding="utf-8") as fh: + data = _load_yaml_text(fh.read()) + + template = str(data.get("template", "")).strip() + if not template: + raise FeatureSpecError("features.yaml 缺少 template 字段") + version = str(data.get("version", "1.0.0")).strip() or "1.0.0" + description = str(data.get("description", "")).strip() + + raw_specs = data.get("specs") or [] + if not isinstance(raw_specs, list): + raise FeatureSpecError("features.yaml specs 必须是 list") + specs: List[FeatureSpec] = [] + for idx, item in enumerate(raw_specs): + if not isinstance(item, dict): + raise FeatureSpecError(f"features.yaml specs[{idx}] 必须是 map") + name = str(item.get("name", "")).strip() + kind_name = str(item.get("kind", "")).strip() + if kind_name not in KIND_REGISTRY: + raise FeatureSpecError( + f"features.yaml specs[{idx}] 未知算子 {kind_name!r}" + f"（应为 {sorted(KIND_REGISTRY)}）") + point = str(item.get("point", "")).strip() + unit = str(item.get("unit", "")).strip() + raw_params = item.get("params") or {} + if not isinstance(raw_params, dict): + raise FeatureSpecError(f"features.yaml specs[{idx}] params 必须是 map") + params: Dict[str, float] = {} + for pk, pv in raw_params.items(): + try: + params[pk] = float(pv) + except (TypeError, ValueError) as exc: + raise FeatureSpecError( + f"features.yaml specs[{idx}] 参数 {pk}={pv!r} 不是数值") from exc + threshold = item.get("threshold") + if threshold not in (None, ""): + try: + threshold = float(threshold) + except (TypeError, ValueError) as exc: + raise FeatureSpecError( + f"features.yaml specs[{idx}] threshold 不是数值") from exc + else: + threshold = None + specs.append(FeatureSpec( + name=name, kind=KIND_REGISTRY[kind_name], point=point, + params=params, unit=unit, threshold=threshold, + )) + return FeatureTemplateConfig( + template=template, version=version, specs=specs, description=description) diff --git a/templates/ti-cl4/impurity-forecast/tests/_bootstrap.py b/templates/ti-cl4/impurity-forecast/tests/_bootstrap.py new file mode 100644 index 0000000..81d83bb --- /dev/null +++ b/templates/ti-cl4/impurity-forecast/tests/_bootstrap.py @@ -0,0 +1,26 @@ +# -*- coding: utf-8 -*- +"""测试引导：把连字符目录挂载为可导入包（与 core 模块同款模式）。 + +- ``templates/ti-cl4/impurity-forecast`` → 包名 ``impurity_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("impurity_forecast", PKG_DIR) diff --git a/templates/ti-cl4/impurity-forecast/tests/test_features.py b/templates/ti-cl4/impurity-forecast/tests/test_features.py new file mode 100644 index 0000000..37607c2 --- /dev/null +++ b/templates/ti-cl4/impurity-forecast/tests/test_features.py @@ -0,0 +1,313 @@ +# -*- coding: utf-8 -*- +"""炉层杂质预警特征工程引擎单元测试（Issue #70）。 + +覆盖： +- 声明式 FeatureSpec 校验（缺参 / 非法窗口 / alpha 越界 / 未知算子）； +- 各算子数学正确性（raw / ema / rolling_std / rolling_mean / rolling_min/max / + rate_of_change），含缺失值处理； +- 时序对齐与缺失率； +- 无监督阈值 breach 判定； +- 模板配置 YAML 加载（含 flow map / 错误 YAML 拒绝）； +- 端到端：模板资产加载 → 引擎 → transform → 提前量信号可观测。 +""" +import math +import os +import sys +import unittest + +sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) +import _bootstrap # noqa: F401 挂载 impurity_forecast 包 + +from impurity_forecast import ( # noqa: E402 + FeatureEngine, + FeatureKind, + FeatureSpec, + FeatureSpecError, + FeatureTemplateConfig, + load_feature_config, +) +from impurity_forecast.features import ( # noqa: E402 + NAN, + _op_ema, + _op_rate_of_change, + _op_raw, + _rolling_window, + _std, +) + +NAN = float("nan") + +CONFIG_PATH = os.path.join( + os.path.dirname(os.path.dirname(os.path.abspath(__file__))), + "config", "features.template.yaml") + + +def _approx(a: float, b: float, eps: float = 1e-9) -> bool: + if math.isnan(a) and math.isnan(b): + return True + return abs(a - b) <= eps + + +# --------------------------------------------------------------------------- +# 1. FeatureSpec 声明校验 +# --------------------------------------------------------------------------- + +class FeatureSpecValidationTest(unittest.TestCase): + + def test_minimal_raw_spec_ok(self): + s = FeatureSpec(name="t", kind=FeatureKind.RAW, point="P1") + self.assertEqual(s.describe(), "t = raw(P1)") + + def test_rolling_requires_window(self): + with self.assertRaises(FeatureSpecError): + FeatureSpec(name="x", kind=FeatureKind.ROLLING_STD, point="P1") + + def test_rolling_window_must_be_positive_int(self): + with self.assertRaises(FeatureSpecError): + FeatureSpec(name="x", kind=FeatureKind.ROLLING_MEAN, point="P1", + params={"window": 0}) + # 非整数（2.5）必须被拒绝，避免窗口语义歧义 + with self.assertRaises(FeatureSpecError): + FeatureSpec(name="x", kind=FeatureKind.ROLLING_MEAN, point="P1", + params={"window": 2.5}) + + def test_ema_alpha_range(self): + with self.assertRaises(FeatureSpecError): + FeatureSpec(name="x", kind=FeatureKind.EMA, point="P1", + params={"alpha": 0}) # 不含 0 + with self.assertRaises(FeatureSpecError): + FeatureSpec(name="x", kind=FeatureKind.EMA, point="P1", + params={"alpha": 1.5}) # 超 1 + # 合法边界 1.0 通过 + s = FeatureSpec(name="x", kind=FeatureKind.EMA, point="P1", + params={"alpha": 1.0}) + self.assertEqual(s.params["alpha"], 1.0) + + def test_empty_name_or_point_rejected(self): + with self.assertRaises(FeatureSpecError): + FeatureSpec(name="", kind=FeatureKind.RAW, point="P1") + with self.assertRaises(FeatureSpecError): + FeatureSpec(name="x", kind=FeatureKind.RAW, point="") + + +# --------------------------------------------------------------------------- +# 2. 算子数学正确性 +# --------------------------------------------------------------------------- + +class OperatorMathTest(unittest.TestCase): + + def test_raw_passes_through_with_nan(self): + out = _op_raw([1.0, NAN, 3.0], {}) + self.assertTrue(_approx(out[0], 1.0)) + self.assertTrue(math.isnan(out[1])) + self.assertTrue(_approx(out[2], 3.0)) + + def test_ema_recurrence(self): + # alpha=0.5: ema[t] = 0.5*x + 0.5*ema[t-1]，首项为 x[0] + out = _op_ema([10.0, 20.0, 30.0], {"alpha": 0.5}) + self.assertTrue(_approx(out[0], 10.0)) + self.assertTrue(_approx(out[1], 0.5 * 20 + 0.5 * 10)) # 15 + self.assertTrue(_approx(out[2], 0.5 * 30 + 0.5 * 15)) # 22.5 + + def test_ema_skips_nan_without_reset(self): + # 缺失样本不进缓冲区且不重置状态 + out = _op_ema([10.0, NAN, 20.0], {"alpha": 1.0}) + self.assertTrue(_approx(out[0], 10.0)) + self.assertTrue(math.isnan(out[1])) + self.assertTrue(_approx(out[2], 20.0)) # alpha=1 即 raw + + def test_rolling_std_window_warmup(self): + vals = [2.0, 4.0, 6.0] + out = _rolling_window(vals, 2, _std) + self.assertTrue(math.isnan(out[0])) # 不足 window + # 窗口 [2,4] 总体标准差 = sqrt(((2-3)^2+(4-3)^2)/2)=sqrt(1)=1 + self.assertTrue(_approx(out[1], 1.0)) + self.assertTrue(_approx(out[2], 1.0)) # [4,6] 同样 + + def test_rolling_mean_min_max(self): + vals = [1.0, 2.0, 3.0, 4.0] + mean = _rolling_window(vals, 2, lambda w: sum(w) / len(w)) + self.assertTrue(_approx(mean[0], NAN)) + self.assertTrue(_approx(mean[1], 1.5)) + self.assertTrue(_approx(mean[2], 2.5)) + self.assertTrue(_approx(mean[3], 3.5)) + self.assertTrue(_approx(_rolling_window(vals, 2, min)[3], 3.0)) + self.assertTrue(_approx(_rolling_window(vals, 2, max)[3], 4.0)) + + def test_rate_of_change_warmup(self): + # window=1: roc[t] = (x[t]-x[t-1])/x[t-1] + out = _op_rate_of_change([100.0, 110.0, 99.0], {"window": 1}) + self.assertTrue(math.isnan(out[0])) # 需 window+1 个样本 + self.assertTrue(_approx(out[1], 0.10)) + self.assertTrue(_approx(out[2], -0.10)) + + def test_rate_of_change_zero_base_is_nan(self): + out = _op_rate_of_change([0.0, 10.0, 20.0], {"window": 1}) + # base=0 → 除零，返回 NAN 而非崩溃 + self.assertTrue(math.isnan(out[1])) + + +# --------------------------------------------------------------------------- +# 3. 引擎：时序对齐 / 缺失率 / 重复名拒绝 +# --------------------------------------------------------------------------- + +class FeatureEngineTest(unittest.TestCase): + + def _engine(self) -> FeatureEngine: + return FeatureEngine([ + FeatureSpec(name="炉温_raw", kind=FeatureKind.RAW, point="CLF-01.TEMP"), + FeatureSpec(name="炉温_ema5", kind=FeatureKind.EMA, + point="CLF-01.TEMP", params={"alpha": 0.5}, + threshold=900.0), + FeatureSpec(name="氯气_std3", kind=FeatureKind.ROLLING_STD, + point="CLF-01.CL2", params={"window": 3}), + ]) + + def test_empty_specs_rejected(self): + with self.assertRaises(FeatureSpecError): + FeatureEngine([]) + + def test_duplicate_name_rejected(self): + with self.assertRaises(FeatureSpecError): + FeatureEngine([ + FeatureSpec(name="dup", kind=FeatureKind.RAW, point="P1"), + FeatureSpec(name="dup", kind=FeatureKind.RAW, point="P2"), + ]) + + def test_required_points_dedup(self): + eng = self._engine() + self.assertEqual(eng.required_points(), ["CLF-01.TEMP", "CLF-01.CL2"]) + + def test_transform_aligns_and_missing_rate(self): + eng = self._engine() + samples = [ + {"ts": 1, "CLF-01.TEMP": 800.0, "CLF-01.CL2": 100.0}, + {"ts": 2, "CLF-01.TEMP": 850.0, "CLF-01.CL2": 120.0}, + {"ts": 3, "CLF-01.TEMP": 910.0, "CLF-01.CL2": 90.0}, + ] + vecs = eng.transform(samples) + self.assertEqual(len(vecs), 3) + # 第一个时刻：rolling_std window=3 不足 → 该列缺失 + self.assertAlmostEqual(vecs[0].missing_rate, 1.0 / 3, places=6) + self.assertFalse(vecs[0].is_complete) # @property + # 第三个时刻所有列就绪 + self.assertTrue(vecs[2].is_complete) + self.assertAlmostEqual(vecs[2].values["炉温_raw"], 910.0) + # ema 第三个 = 0.5*910 + 0.5*(0.5*850+0.5*800) = 455+0.5*825=455+412.5 + self.assertAlmostEqual(vecs[2].values["炉温_ema5"], 867.5, places=4) + + def test_transform_missing_point_value(self): + eng = self._engine() + samples = [ + {"ts": 1, "CLF-01.TEMP": 800.0}, # CL2 缺失 + {"ts": 2, "CLF-01.TEMP": 850.0, "CLF-01.CL2": 100.0}, + {"ts": 3, "CLF-01.TEMP": 900.0, "CLF-01.CL2": 110.0}, + ] + vecs = eng.transform(samples) + self.assertTrue(math.isnan(vecs[0].values["氯气_std3"])) + + def test_breach_threshold(self): + eng = self._engine() + vec = type("V", (), {"values": { + "炉温_raw": 800.0, + "炉温_ema5": 950.0, # 超 900 + "氯气_std3": 5.0, + }})() + breach = eng.breach(vec) + names = [b[0] for b in breach] + self.assertEqual(names, ["炉温_ema5"]) + + def test_describe_lists_all_specs(self): + eng = self._engine() + self.assertEqual(len(eng.describe()), 3) + self.assertIn("alpha=0.5", eng.describe()[1]) + + +# --------------------------------------------------------------------------- +# 4. 模板配置 YAML 加载 +# --------------------------------------------------------------------------- + +class ConfigLoadTest(unittest.TestCase): + + def test_load_template_config(self): + cfg = load_feature_config(CONFIG_PATH) + self.assertIsInstance(cfg, FeatureTemplateConfig) + self.assertEqual(cfg.template, "ti-cl4") + self.assertTrue(len(cfg.specs) >= 5) + names = [s.name for s in cfg.specs] + # PRD 示例三件套均存在 + self.assertIn("炉温_ema5", names) + self.assertIn("氯气流量_std10", names) + self.assertIn("炉压_rate10", names) + + def test_flow_map_params_parsed(self): + cfg = load_feature_config(CONFIG_PATH) + ema = next(s for s in cfg.specs if s.name == "炉温_ema5") + # flow map {alpha: 0.2} 解析为数值参数 + self.assertAlmostEqual(ema.params["alpha"], 0.2) + self.assertEqual(ema.threshold, 900.0) + + def test_engine_from_template_config(self): + eng = FeatureEngine.from_template_config(CONFIG_PATH) + vecs = eng.transform([ + {"ts": i, "CLF-01.TEMP": 850.0 + i, "CLF-01.CL2": 100.0, + "CLF-01.PRES": 10.0, "CLF-01.BED": 60.0} + for i in range(20) + ]) + # 充分预热后所有特征列就绪 + self.assertTrue(vecs[-1].is_complete) + # 无 breach（值均在阈值内） + self.assertEqual(eng.breach(vecs[-1]), []) + + def test_unknown_kind_rejected(self): + import tempfile + bad = ( + "template: ti-cl4\n" + "version: 1.0.0\n" + "specs:\n" + " - name: x\n" + " kind: not_a_real_kind\n" + " point: P1\n" + ) + with tempfile.NamedTemporaryFile("w", suffix=".yaml", delete=False, + encoding="utf-8") as fh: + fh.write(bad) + path = fh.name + try: + with self.assertRaises(FeatureSpecError): + load_feature_config(path) + finally: + os.unlink(path) + + +# --------------------------------------------------------------------------- +# 5. 端到端：提前量信号可观测（PRD 验收：提前 ≥ 30min） +# --------------------------------------------------------------------------- + +class EndToEndEarlySignalTest(unittest.TestCase): + + def test_rising_temperature_triggers_breach_before_peak(self): + """模拟炉温阶跃爬升：ema 平滑值应在持续攀升阶段 breach 阈值， + 早于物理峰值时刻 —— 体现"提前量"（PRD 5.3 ③：提前 ≥ 30min）。""" + eng = FeatureEngine([ + FeatureSpec(name="炉温_ema5", kind=FeatureKind.EMA, + point="CLF-01.TEMP", params={"alpha": 0.4}, + threshold=900.0), + ]) + # 前 10 步平稳 850℃，第 10 步起每步 +8℃ 攀升，第 25 步到峰值 970℃ + temps = [850.0] * 10 + [850.0 + 8.0 * (i - 9) for i in range(10, 25)] + samples = [{"ts": i, "CLF-01.TEMP": temps[i]} for i in range(len(temps))] + vecs = eng.transform(samples) + # 第一个 breach 的时刻 + first_breach = None + for idx, v in enumerate(vecs): + if eng.breach(v): + first_breach = idx + break + self.assertIsNotNone(first_breach, "未观察到任何 breach") + # breach 应在物理峰值（最后一刻）之前出现 → 提前量可观测 + self.assertLess(first_breach, len(vecs) - 1) + + +if __name__ == "__main__": + unittest.main(verbosity=2)