feat(#70): 炉层杂质预警特征工程(声明式FeatureSpec引擎,PRD 5.3 ③)

新增 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%口径)。
This commit is contained in:
2026-08-05 01:59:42 +08:00
parent f5d2294ee4
commit 42d206e503
7 changed files with 1148 additions and 0 deletions
@@ -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())