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