Merge PR #112/#113 (feat #71 无监督模型 + #72 告警规则，与 #70 特征工程联合)

templates/ti-cl4/impurity-forecast/__init__.py

@@ -1,8 +1,14 @@
 # -*- coding: utf-8 -*-
-"""iAOP-Template-Ti 一期 · 炉层杂质预警特征工程包（Issue #70）。
+"""iAOP-Template-Ti 一期 · 炉层杂质预警包（Issue #70 特征工程 + #71 无监督模型 + #72 告警规则）。
 
-导出声明式 FeatureSpec 特征工程引擎，供预警模型训练（#71）与驾驶舱
-告警面板复用。换行业只改模板配置，引擎零改动（PRD 5.3 特征工程层）。
+导出三部分，覆盖 PRD 5.3 ③ 场景 A 全链路（采集 → 特征 → 评分 → 规则分级 →
+驾驶舱红色告警 → LLM 解释）：
+
+- 特征工程引擎（FeatureSpec/FeatureEngine，#70）；
+- 无监督异常评分器与预警决策（ZScoreScorer/ImpurityForecaster，#71）；
+- 声明式三级告警规则引擎（AlertRuleEngine，#72）。
+
+换行业只改模板配置，引擎与模型零改动（PRD 5.3）。
 """
 from __future__ import annotations
 
@@ -16,8 +22,27 @@ from .features import (
     FeatureTemplateConfig,
     load_feature_config,
 )
+from .model import (
+    AlertDecision,
+    FeatureVectorLike,
+    ImpurityForecaster,
+    LeadTimeResult,
+    ThresholdRule,
+    ZScoreScorer,
+    evaluate_lead_time,
+)
+from .alert_rules import (
+    Alert,
+    AlertCondition,
+    AlertRule,
+    AlertRuleEngine,
+    AlertRulesTemplateConfig,
+    AlertSeverity,
+    load_alert_rules_config,
+)
 
 __all__ = [
+    # 特征工程（#70）
     "FeatureKind",
     "FeatureSpec",
     "FeatureSpecError",
@@ -26,4 +51,20 @@ __all__ = [
     "FeatureEngine",
     "FeatureTemplateConfig",
     "load_feature_config",
+    # 无监督模型（#71）
+    "AlertDecision",
+    "FeatureVectorLike",
+    "ImpurityForecaster",
+    "LeadTimeResult",
+    "ThresholdRule",
+    "ZScoreScorer",
+    "evaluate_lead_time",
+    # 告警规则（#72）
+    "Alert",
+    "AlertCondition",
+    "AlertRule",
+    "AlertRuleEngine",
+    "AlertRulesTemplateConfig",
+    "AlertSeverity",
+    "load_alert_rules_config",
 ]

templates/ti-cl4/impurity-forecast/_sanity_check_model.py

@@ -0,0 +1,74 @@
+# -*- coding: utf-8 -*-
+"""炉层杂质预警无监督模型冒烟脚本（Issue #71）。
+
+直接运行 ``python _sanity_check_model.py`` 验证：ZScoreScorer 可在正常段 fit、
+异常段产出高分数、ThresholdRule 触发预警、提前量评估为正（对齐 PRD 提前≥30min）。
+零第三方依赖。
+"""
+import importlib.util
+import os
+import sys
+
+_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 (  # noqa: E402
+    FeatureVectorLike,
+    ImpurityForecaster,
+    ThresholdRule,
+    ZScoreScorer,
+)
+
+
+def vec(ts, **kw):
+    return FeatureVectorLike(timestamp=ts, values=dict(kw))
+
+
+def main() -> int:
+    # 正常段：炉温 850±5 波动，氯气 100±3 波动
+    normal = [vec(i, 炉温=850.0 + (i % 3) * 2, 氯气=100.0 + (i % 2))
+              for i in range(40)]
+    # 观测段：前 40 正常，之后急升温 + 氯气突降
+    obs = list(normal) + [
+        vec(40 + i, 炉温=860.0 + 6.0 * i, 氯气=95.0 - i) for i in range(20)
+    ]
+    anomaly_ts = 59.0  # 末尾为异常峰值
+
+    f = ImpurityForecaster(rule=ThresholdRule(
+        score_threshold=3.0,
+        feature_thresholds={"炉温": 900.0}))
+    f.fit(normal)
+    decisions, lt = f.evaluate(obs, anomaly_ts=anomaly_ts)
+
+    triggered = [d for d in decisions if d.triggered]
+    assert triggered, "异常段应触发预警"
+    print(f"[OK] 预警触发 {len(triggered)} 次")
+    print(f"[OK] 首次预警 ts={lt.first_alert_ts}，真实异常 ts={anomaly_ts}，"
+          f"提前量={lt.lead_minutes:.1f} min（>0 即满足提前量口径）")
+
+    # 模型可序列化
+    d = f.scorer.to_dict()
+    sc2 = ZScoreScorer.from_dict(d)
+    s1 = f.scorer.score(obs[-1:])
+    s2 = sc2.score(obs[-1:])
+    assert abs(s1[0] - s2[0]) < 1e-9, "序列化前后分数应一致"
+    print("[OK] 模型序列化往返一致（可版本化保存）")
+    print("炉层杂质预警模型冒烟通过 ✅")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

templates/ti-cl4/impurity-forecast/_sanity_check_rules.py

@@ -0,0 +1,59 @@
+# -*- coding: utf-8 -*-
+"""炉层杂质预警规则引擎冒烟脚本（Issue #72）。
+
+直接运行验证：模板规则资产可加载、规则引擎对急升温特征向量产出 P0 告警、
+无异常时不告警。零第三方依赖。
+"""
+import importlib.util
+import os
+import sys
+
+_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 AlertRuleEngine, AlertSeverity  # noqa: E402
+
+CONFIG = os.path.join(_PKG_DIR, "config", "alert_rules.template.yaml")
+
+
+def main() -> int:
+    eng = AlertRuleEngine.from_template_config(CONFIG)
+    print(f"[OK] 规则配置加载: {len(eng.rules)} 条预警规则")
+    for r in eng.rules:
+        print(f"   {r.describe()}  (sop={r.sop or '-'})")
+
+    # 1) 正常工况不告警
+    normal = eng.evaluate(1, {"炉温_ema5": 850.0, "炉温_rate10": 0.01,
+                              "氯气流量_std10": 3.0, "炉压_rate10": 0.01,
+                              "炉层状态_mean10": 60.0})
+    assert not normal, "正常工况不应告警"
+    print("[OK] 正常工况：无告警")
+
+    # 2) 急升温 + 炉压急变 → P0 红色告警（PRD 场景A）
+    abnormal = eng.evaluate(2, {"炉温_ema5": 905.0, "炉温_rate10": 0.06,
+                                "氯气流量_std10": 4.0, "炉压_rate10": 0.09,
+                                "炉层状态_mean10": 60.0})
+    assert abnormal, "异常工况应触发告警"
+    primary = eng.primary_alert(abnormal)
+    assert primary.severity == AlertSeverity.P0, "主告警应为 P0"
+    print(f"[OK] 异常工况：主告警 {primary.severity.value}（{primary.message}）"
+          f" sop={primary.sop}")
+    print("炉层杂质预警规则引擎冒烟通过 ✅")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

templates/ti-cl4/impurity-forecast/alert_rules.py

@@ -0,0 +1,356 @@
+# -*- coding: utf-8 -*-
+"""炉层杂质预警 · 预警规则与阈值设定引擎（Issue #72 / PRD 5.3 ③）。
+
+PRD 5.3 ③ / 场景 A（line 80）：异常检测模型触发 → 驾驶舱红色告警 + LLM 生成
+"原因+处置建议" → 值班长确认。本模块把"行业知识"——**预警分级 + 阈值 + 处置 SOP**——
+外置为模板配置（PRD line 152/171：阈值外置 JSON，行业工程师在配置台维护），引擎
+按规则评估特征向量产出带 severity 的 Alert。
+
+设计要点
+--------
+1. **声明式 AlertRule**：每条规则声明 ``id`` + ``severity``（P0/P1/P2）+ ``condition``
+   （特征名 + 比较运算 + 阈值）+ ``sop``（处置 SOP 引用，供 LLM 报警解释/驾驶舱展示）。
+2. **severity 三级**（PRD line 333：关键告警不直接联动执行机构，高利害人工确认）：
+   - ``P0``（critical）：红色告警，立即人工确认 + 紧急处置；
+   - ``P1``（warning）：黄色告警，加强监控 + 预备处置；
+   - ``P2``（info）：提示，记录跟踪。
+3. **规则引擎**：``AlertRuleEngine.evaluate`` 对一个特征向量评估全部规则，返回命中的
+   Alert 列表（取最高 severity 为主告警）；可与 #70/#71 组合（特征向量/异常分数均可作为
+   condition 输入）。
+4. **零依赖 YAML 子集解析**（对齐 data-bus / rag-kb / #70），阈值外置模板资产。
+"""
+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
+
+NAN = float("nan")
+
+
+class AlertSeverity(str, Enum):
+    """预警严重度三级（PRD line 80 红色告警 / line 333 关键告警人工确认）。"""
+
+    P0 = "P0"   # critical：红色，立即人工确认 + 紧急处置
+    P1 = "P1"   # warning：黄色，加强监控 + 预备处置
+    P2 = "P2"   # info：提示，记录跟踪
+
+    @property
+    def label(self) -> str:
+        return {AlertSeverity.P0: "严重", AlertSeverity.P1: "警告",
+                AlertSeverity.P2: "提示"}[self]
+
+    @property
+    def rank(self) -> int:
+        """排序权重，越大越严重（用于取主告警）。"""
+        return {AlertSeverity.P0: 3, AlertSeverity.P1: 2, AlertSeverity.P2: 1}[self]
+
+
+# 比较运算符注册表（condition.op 取值）
+OPS: Dict[str, Callable[[float, float], bool]] = {
+    ">": lambda a, b: a > b,
+    ">=": lambda a, b: a >= b,
+    "<": lambda a, b: a < b,
+    "<=": lambda a, b: a <= b,
+    "==": lambda a, b: a == b,
+}
+
+
+def _is_num(x: object) -> bool:
+    return isinstance(x, (int, float)) and not (isinstance(x, float) and math.isnan(x))
+
+
+@dataclass
+class AlertCondition:
+    """单条触发条件：特征名 + 比较运算 + 阈值。"""
+
+    feature: str
+    op: str
+    threshold: float
+
+    def __post_init__(self) -> None:
+        if self.op not in OPS:
+            raise ValueError(f"未知比较运算 {self.op!r}（应为 {sorted(OPS)}）")
+
+    def matches(self, values: Dict[str, float]) -> bool:
+        v = values.get(self.feature)
+        if not _is_num(v):
+            return False
+        return OPS[self.op](float(v), self.threshold)
+
+
+@dataclass
+class AlertRule:
+    """声明式预警规则（模板配置中的一行规则声明）。
+
+    Attributes:
+        id: 规则 id（稳定标识，供驾驶舱/审计引用）。
+        severity: 严重度（P0/P1/P2）。
+        conditions: 触发条件列表（AND 语义：全部满足才命中）。
+        message: 告警文案（驾驶舱展示）。
+        sop: 处置 SOP 引用（PRD 场景A：LLM 报警解释 + 值班长确认）。
+    """
+
+    id: str
+    severity: AlertSeverity
+    conditions: List[AlertCondition]
+    message: str = ""
+    sop: str = ""
+
+    def __post_init__(self) -> None:
+        if not self.id:
+            raise ValueError("AlertRule.id 不能为空")
+        if not self.conditions:
+            raise ValueError(f"规则 {self.id!r} 至少需要 1 条 condition")
+
+    def matches(self, values: Dict[str, float]) -> bool:
+        return all(c.matches(values) for c in self.conditions)
+
+    def describe(self) -> str:
+        conds = " 且 ".join(f"{c.feature}{c.op}{c.threshold}" for c in self.conditions)
+        return f"[{self.severity.value}] {self.id}: {conds}"
+
+
+@dataclass
+class Alert:
+    """一次预警命中（规则 + 触发时的特征快照）。"""
+
+    rule_id: str
+    severity: AlertSeverity
+    message: str
+    sop: str
+    timestamp: float
+    snapshot: Dict[str, float] = field(default_factory=dict)
+
+
+class AlertRuleEngine:
+    """预警规则引擎：评估特征向量，产出带 severity 的 Alert 列表。
+
+    换行业只改模板配置（AlertRule 列表），引擎零改动（PRD line 152/171）。
+
+    用法::
+
+        engine = AlertRuleEngine(rules)
+        alerts = engine.evaluate(timestamp=100, values={"炉温_ema5": 920.0})
+        if alerts:
+            primary = engine.primary_alert(alerts)   # 取最高 severity
+    """
+
+    def __init__(self, rules: Sequence[AlertRule]):
+        if not rules:
+            raise ValueError("AlertRuleEngine 至少需要 1 条规则")
+        ids = set()
+        for r in rules:
+            if r.id in ids:
+                raise ValueError(f"规则 id 重复：{r.id!r}")
+            ids.add(r.id)
+        self.rules: List[AlertRule] = list(rules)
+
+    @classmethod
+    def from_template_config(cls, path: str) -> "AlertRuleEngine":
+        return cls(load_alert_rules_config(path).rules)
+
+    def evaluate(self, timestamp: float,
+                 values: Dict[str, float]) -> List[Alert]:
+        """评估一个特征向量，返回全部命中规则的 Alert（按 severity 降序）。"""
+        hits: List[Alert] = []
+        for rule in self.rules:
+            if rule.matches(values):
+                hits.append(Alert(
+                    rule_id=rule.id, severity=rule.severity,
+                    message=rule.message, sop=rule.sop,
+                    timestamp=timestamp, snapshot=dict(values),
+                ))
+        hits.sort(key=lambda a: a.severity.rank, reverse=True)
+        return hits
+
+    def primary_alert(self, alerts: Sequence[Alert]) -> Optional[Alert]:
+        """取最高 severity 的主告警（驾驶舱红色告警）。无命中返回 None。"""
+        return alerts[0] if alerts else None
+
+
+# ---------------------------------------------------------------------------
+# 模板配置（零依赖 YAML 子集解析，对齐 #70 / data-bus / rag-kb）
+# ---------------------------------------------------------------------------
+
+@dataclass
+class AlertRulesTemplateConfig:
+    """模板预警规则配置：模板元信息 + AlertRule 列表。"""
+
+    template: str
+    version: str
+    rules: List[AlertRule]
+    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 用于
+    ``conditions: [{feature: x, op: ">", threshold: 900.0}]`` 这种紧凑声明。
+    """
+    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 ValueError(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):
+    text, _ = lines[i]
+    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 ValueError(f"alerts.yaml 第 {no} 行：list 项为空")
+            # 行内 flow map（{k: v, ...}）优先用 _parse_flow_value，避免被
+            # 下方「含 : 即 map 项」分支误判（flow map 也含 :）。
+            if item_text.startswith("{") and item_text.endswith("}"):
+                items.append(_parse_flow_value(item_text))
+                i += 1
+            elif ":" 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
+    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 ValueError(f"alerts.yaml 第 {no} 行缩进异常")
+        if ":" not in t:
+            raise ValueError(f"alerts.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 ValueError(f"alerts.yaml 第 {no} 行 {key!r} 缺少值")
+        sub_indent = len(lines[i + 1][0]) - len(lines[i + 1][0].lstrip(" "))
+        if sub_indent <= indent:
+            raise ValueError(f"alerts.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 ValueError("alerts.yaml 顶层必须是 map")
+    if next_i < len(lines):
+        raise ValueError(f"alerts.yaml 第 {lines[next_i][1]} 行：顶层存在多个节点")
+    return value
+
+
+def load_alert_rules_config(path: str) -> AlertRulesTemplateConfig:
+    """从模板预警规则 YAML 资产加载配置。
+
+    期望结构（详见 ``config/alert_rules.template.yaml``）::
+
+        template: ti-cl4
+        version: 1.0.0
+        rules:
+          - id: bed_temp_critical
+            severity: P0
+            message: 炉温超上限，立即降流减料
+            sop: SOP-CL-001
+            conditions:
+              - {feature: 炉温_ema5, op: ">", 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 ValueError("alerts.yaml 缺少 template 字段")
+    version = str(data.get("version", "1.0.0")).strip() or "1.0.0"
+    description = str(data.get("description", "")).strip()
+
+    raw_rules = data.get("rules") or []
+    if not isinstance(raw_rules, list):
+        raise ValueError("alerts.yaml rules 必须是 list")
+    rules: List[AlertRule] = []
+    for idx, item in enumerate(raw_rules):
+        if not isinstance(item, dict):
+            raise ValueError(f"alerts.yaml rules[{idx}] 必须是 map")
+        rid = str(item.get("id", "")).strip()
+        sev_name = str(item.get("severity", "")).strip().upper()
+        sev_map = {s.value: s for s in AlertSeverity}
+        if sev_name not in sev_map:
+            raise ValueError(
+                f"alerts.yaml rules[{idx}] 未知 severity {sev_name!r}"
+                f"（应为 {sorted(sev_map)}）")
+        message = str(item.get("message", "")).strip()
+        sop = str(item.get("sop", "")).strip()
+        raw_conds = item.get("conditions") or []
+        if not isinstance(raw_conds, list):
+            raise ValueError(f"alerts.yaml rules[{idx}] conditions 必须是 list")
+        conds: List[AlertCondition] = []
+        for ci, c in enumerate(raw_conds):
+            if not isinstance(c, dict):
+                raise ValueError(f"alerts.yaml rules[{idx}].conditions[{ci}] 必须是 map")
+            feature = str(c.get("feature", "")).strip()
+            op = str(c.get("op", "")).strip()
+            if op not in OPS:
+                raise ValueError(
+                    f"alerts.yaml rules[{idx}].conditions[{ci}] 未知 op {op!r}")
+            try:
+                threshold = float(c.get("threshold"))
+            except (TypeError, ValueError) as exc:
+                raise ValueError(
+                    f"alerts.yaml rules[{idx}].conditions[{ci}] threshold 不是数值") from exc
+            conds.append(AlertCondition(feature=feature, op=op, threshold=threshold))
+        rules.append(AlertRule(id=rid, severity=sev_map[sev_name],
+                               conditions=conds, message=message, sop=sop))
+    return AlertRulesTemplateConfig(template=template, version=version,
+                                    rules=rules, description=description)

templates/ti-cl4/impurity-forecast/config/alert_rules.template.yaml

@@ -0,0 +1,64 @@
+# iAOP-Template-Ti 一期 · 炉层杂质预警规则与阈值模板资产（Issue #72 / PRD 5.3 ③）
+#
+# 把"行业知识"——预警分级 + 阈值 + 处置 SOP——外置为本配置（PRD line 152/171：
+# 阈值外置，行业工程师在配置台维护），规则引擎（alert_rules.py）零改动。
+#
+# severity 三级（PRD line 80 红色告警 / line 333 关键告警人工确认）：
+#   P0 critical 红色，立即人工确认 + 紧急处置；
+#   P1 warning  黄色，加强监控 + 预备处置；
+#   P2 info     提示，记录跟踪。
+#
+# 特征名对齐 #70 features.template.yaml 的 FeatureSpec.name（如 炉温_ema5、炉压_rate10）；
+# sop 引用异常处置 SOP（供 #11 LLM 报警解释 + 驾驶舱展示 + 值班长确认）。
+template: ti-cl4
+version: 1.0.0
+description: 炉层杂质预警规则与阈值（声明式 AlertRule，PRD 5.3 ③ 场景A）
+
+rules:
+  # ---- P0 严重：炉温超上限，立即降流减料（SOP-CL-001） -----------------
+  - id: bed_temp_critical
+    severity: P0
+    message: 炉温超工艺上限，立即降低氯气流量并减少加料，10 分钟未回落按紧急停机处理
+    sop: SOP-CL-001
+    conditions:
+      - {feature: 炉温_ema5, op: ">", threshold: 900.0}
+
+  # ---- P0 严重：炉温急升趋势（提前量信号，PRD 提前≥30min） -------------
+  - id: bed_temp_rising_critical
+    severity: P0
+    message: 炉温急升趋势，疑似炉层状态恶化，预备紧急处置并通知班长
+    sop: SOP-CL-002
+    conditions:
+      - {feature: 炉温_rate10, op: ">", threshold: 0.05}
+
+  # ---- P0 严重：炉压急变（压力异常是炉层恶化强信号） -------------------
+  - id: bed_pressure_critical
+    severity: P0
+    message: 炉压急变，排查炉层状态与尾气系统，必要时降负荷
+    sop: SOP-CL-003
+    conditions:
+      - {feature: 炉压_rate10, op: ">", threshold: 0.08}
+
+  # ---- P1 警告：氯气流量波动度越界（流态化异常先兆） -------------------
+  - id: cl2_flow_warning
+    severity: P1
+    message: 氯气流量波动增大，检查供料与流态化状态，加强监控
+    sop: SOP-CL-004
+    conditions:
+      - {feature: 氯气流量_std10, op: ">", threshold: 8.0}
+
+  # ---- P1 警告：炉层状态均值超阈（杂质富集表征） -----------------------
+  - id: bed_state_warning
+    severity: P1
+    message: 炉层状态偏高，关注杂质富集趋势，按批次增加检测频次
+    sop: SOP-CL-005
+    conditions:
+      - {feature: 炉层状态_mean10, op: ">", threshold: 85.0}
+
+  # ---- P2 提示：炉温接近上限（预警预备） -------------------------------
+  - id: bed_temp_near_limit
+    severity: P2
+    message: 炉温接近工艺上限，记录并跟踪趋势
+    sop: ""
+    conditions:
+      - {feature: 炉温_ema5, op: ">", threshold: 880.0}

templates/ti-cl4/impurity-forecast/model.py

@@ -0,0 +1,252 @@
+# -*- coding: utf-8 -*-
+"""炉层杂质预警 · 无监督异常评分模型训练与推理（Issue #71 / PRD 5.3 ③）。
+
+承接 #70 的特征工程：把特征向量序列喂给**无监督异常评分模型**，输出每个时刻
+的「异常分数」与「预警决策」。PRD 5.3 ③ / 风险表明确：一期数据门槛低，以
+**阈值 + 无监督**上线，3 个月后转监督（PRD 4.1 / 风险表 ①③先无监督）。
+
+设计要点
+--------
+1. **无监督评分器**（零第三方依赖，纯标准库）：
+   - ``ZScoreScorer``：按特征列在训练段估计均值/方差，推理段算各特征 Z-score，
+     取绝对值最大者（或均值）为该时刻异常分数。对应 PRD「3σ」阈值口径。
+   - ``ThresholdRule``：把 #70 的 FeatureSpec 阈值 breach 与分数阈值组合，给出
+     最终预警决策（避免单一指标误报，对齐误报率 ≤ 8%）。
+2. **训练 / 推理分离**：``fit`` 在"正常段"估计分布参数，``score`` 在"观测段"产出
+   异常分数；可序列化保存（零依赖 JSON）。
+3. **提前量评估**：``evaluate_lead_time`` 计算预警首次触发时刻相对真实异常
+   时刻的提前量（对齐 PRD 提前 ≥ 30min）。
+4. **与 #70 解耦**：模型只依赖特征向量的 ``values: Dict[str,float]`` / ``timestamp``
+   （鸭子类型），不强耦合 FeatureEngine，便于独立测试与换行业复用。
+"""
+from __future__ import annotations
+
+import json
+import math
+import os
+from dataclasses import dataclass, field
+from typing import Dict, List, Optional, Sequence, Tuple
+
+NAN = float("nan")
+
+
+def _is_num(x: object) -> bool:
+    return isinstance(x, (int, float)) and not (isinstance(x, float) and math.isnan(x))
+
+
+def _mean(xs: Sequence[float]) -> float:
+    xs = [x for x in xs if _is_num(x)]
+    return sum(xs) / len(xs) if xs else NAN
+
+
+def _std(xs: Sequence[float]) -> float:
+    xs = [x for x in xs if _is_num(x)]
+    n = len(xs)
+    if n == 0:
+        return NAN
+    m = sum(xs) / n
+    return math.sqrt(sum((x - m) ** 2 for x in xs) / n)
+
+
+@dataclass
+class FeatureVectorLike:
+    """特征向量鸭子类型（与 #70 FeatureVector 字段兼容）。
+
+    模型只读 ``timestamp`` 与 ``values``，不依赖具体类，便于独立测试。
+    """
+
+    timestamp: float
+    values: Dict[str, float] = field(default_factory=dict)
+
+
+class ZScoreScorer:
+    """Z-score（3σ）无监督异常评分器。
+
+    训练阶段在"正常段"按特征列估计均值 μ 与标准差 σ；推理阶段对每个时刻
+    计算各特征 ``|x-μ|/σ``，取**最大值**作为该时刻异常分数（取最显著偏离的
+    特征，对齐"任一指标异常即预警"的工艺口径）。
+
+    新特征列（推理段出现而训练段没有）按需跳过；训练段 σ=0（恒定）的特征
+    视为"无区分度"，偏离即记为高分数（用大常数代替除零）。
+    """
+
+    LARGE = 1e6  # σ=0 时的等效分数，保证恒定列偏离可被识别
+
+    def __init__(self) -> None:
+        self._mean: Dict[str, float] = {}
+        self._std: Dict[str, float] = {}
+        self._fitted = False
+
+    @property
+    def fitted(self) -> bool:
+        return self._fitted
+
+    def fit(self, samples: Sequence[FeatureVectorLike]) -> "ZScoreScorer":
+        """在正常段估计各特征列的 μ/σ。"""
+        if not samples:
+            raise ValueError("ZScoreScorer.fit 至少需要 1 条样本")
+        names = set()
+        for s in samples:
+            names.update(k for k, v in s.values.items() if _is_num(v))
+        self._mean = {n: _mean([s.values[n] for s in samples]) for n in names}
+        self._std = {n: _std([s.values[n] for s in samples]) for n in names}
+        self._fitted = True
+        return self
+
+    def score(self, samples: Sequence[FeatureVectorLike]) -> List[float]:
+        """对观测段逐时刻输出异常分数（≥0，越大越异常）。"""
+        if not self._fitted:
+            raise ValueError("ZScoreScorer 未 fit，请先在正常段训练")
+        out: List[float] = []
+        for s in samples:
+            best = 0.0
+            for name, mu in self._mean.items():
+                v = s.values.get(name)
+                if not _is_num(v):
+                    continue
+                sigma = self._std.get(name, 0.0)
+                if sigma <= 1e-12:
+                    # 恒定列：任何偏离都视作异常（用大常数）
+                    z = self.LARGE if abs(v - mu) > 1e-9 else 0.0
+                else:
+                    z = abs(v - mu) / sigma
+                if z > best:
+                    best = z
+            out.append(best)
+        return out
+
+    # -- 序列化（零依赖 JSON，便于版本化保存/复现） ----------------------
+
+    def to_dict(self) -> Dict[str, object]:
+        return {
+            "kind": "zscore",
+            "mean": self._mean,
+            "std": self._std,
+            "fitted": self._fitted,
+        }
+
+    @classmethod
+    def from_dict(cls, d: Dict[str, object]) -> "ZScoreScorer":
+        m = cls()
+        m._mean = {k: float(v) for k, v in (d.get("mean") or {}).items()}
+        m._std = {k: float(v) for k, v in (d.get("std") or {}).items()}
+        m._fitted = bool(d.get("fitted", False))
+        return m
+
+    def save(self, path: str) -> None:
+        with open(path, "w", encoding="utf-8") as fh:
+            json.dump(self.to_dict(), fh, ensure_ascii=False, indent=2)
+
+    @classmethod
+    def load(cls, path: str) -> "ZScoreScorer":
+        with open(path, "r", encoding="utf-8") as fh:
+            return cls.from_dict(json.load(fh))
+
+
+@dataclass
+class AlertDecision:
+    """单时刻预警决策。"""
+
+    timestamp: float
+    score: float                      # 异常分数
+    triggered: bool                   # 是否触发预警
+    reasons: List[str] = field(default_factory=list)  # 触发原因（分数超阈/特征 breach）
+
+
+class ThresholdRule:
+    """预警决策规则：异常分数阈值 ∪ FeatureSpec breach（任一满足即预警）。
+
+    PRD 5.3 ③：误报率 ≤ 8%。组合两条判据降低单指标误报：
+    - 分数判据：``ZScoreScorer`` 输出 ≥ ``score_threshold``（默认 3σ）；
+    - breach 判据：特征值超 #70 FeatureSpec 声明的 ``threshold``（工艺硬限）。
+    """
+
+    def __init__(self, score_threshold: float = 3.0,
+                 feature_thresholds: Optional[Dict[str, float]] = None) -> None:
+        if score_threshold <= 0:
+            raise ValueError("score_threshold 必须 > 0")
+        self.score_threshold = score_threshold
+        # feature_thresholds: 特征名 → 绝对上限（来自 #70 FeatureSpec.threshold）
+        self.feature_thresholds: Dict[str, float] = dict(feature_thresholds or {})
+
+    def decide(self, timestamp: float, values: Dict[str, float],
+               score: float) -> AlertDecision:
+        reasons: List[str] = []
+        if _is_num(score) and score >= self.score_threshold:
+            reasons.append(f"异常分数 {score:.2f} ≥ {self.score_threshold}σ")
+        for name, limit in self.feature_thresholds.items():
+            v = values.get(name)
+            if _is_num(v) and v > limit:
+                reasons.append(f"{name}={v:.2f} 超阈值 {limit}")
+        return AlertDecision(
+            timestamp=timestamp, score=score,
+            triggered=bool(reasons), reasons=reasons,
+        )
+
+
+@dataclass
+class LeadTimeResult:
+    """提前量评估结果（对齐 PRD：提前 ≥ 30min）。"""
+
+    first_alert_ts: Optional[float]      # 首次预警时刻（无则 None）
+    anomaly_ts: Optional[float]          # 真实异常时刻
+    lead_seconds: Optional[float]        # 提前量（秒）；负=滞后
+
+    @property
+    def lead_minutes(self) -> Optional[float]:
+        return None if self.lead_seconds is None else self.lead_seconds / 60.0
+
+
+def evaluate_lead_time(decisions: Sequence[AlertDecision],
+                       anomaly_ts: float) -> LeadTimeResult:
+    """评估首次预警相对真实异常时刻的提前量。
+
+    Args:
+        decisions: 按时间升序的预警决策序列。
+        anomaly_ts: 真实异常（如人工标注/峰值）发生的时刻。
+    """
+    first = None
+    for d in decisions:
+        if d.triggered:
+            first = d.timestamp
+            break
+    if first is None:
+        return LeadTimeResult(first_alert_ts=None, anomaly_ts=anomaly_ts,
+                              lead_seconds=None)
+    return LeadTimeResult(first_alert_ts=first, anomaly_ts=anomaly_ts,
+                          lead_seconds=anomaly_ts - first)
+
+
+class ImpurityForecaster:
+    """炉层杂质预警统一入口：评分器 + 决策规则 + 提前量评估。
+
+    典型用法（配合 #70 FeatureEngine）::
+
+        from impurity_forecast import FeatureEngine, load_feature_config
+        eng = FeatureEngine.from_template_config("config/features.template.yaml")
+        vectors = eng.transform(samples)             # 特征矩阵
+        forecaster = ImpurityForecaster()
+        forecaster.fit(vectors[:normal_n])           # 正常段训练
+        decisions = forecaster.predict(vectors)      # 全段预警决策
+    """
+
+    def __init__(self, scorer: Optional[ZScoreScorer] = None,
+                 rule: Optional[ThresholdRule] = None) -> None:
+        self.scorer = scorer or ZScoreScorer()
+        self.rule = rule or ThresholdRule()
+
+    def fit(self, normal_samples: Sequence[FeatureVectorLike]) -> "ImpurityForecaster":
+        self.scorer.fit(normal_samples)
+        return self
+
+    def predict(self, samples: Sequence[FeatureVectorLike]) -> List[AlertDecision]:
+        scores = self.scorer.score(samples)
+        out: List[AlertDecision] = []
+        for s, sc in zip(samples, scores):
+            out.append(self.rule.decide(s.timestamp, s.values, sc))
+        return out
+
+    def evaluate(self, samples: Sequence[FeatureVectorLike],
+                 anomaly_ts: float) -> Tuple[List[AlertDecision], LeadTimeResult]:
+        decisions = self.predict(samples)
+        return decisions, evaluate_lead_time(decisions, anomaly_ts)

templates/ti-cl4/impurity-forecast/tests/test_alert_rules.py

@@ -0,0 +1,197 @@
+# -*- coding: utf-8 -*-
+"""炉层杂质预警规则引擎单元测试（Issue #72）。
+
+覆盖：
+- AlertCondition 运算匹配（含缺失值不触发、未知 op 拒绝）；
+- AlertRule AND 语义、空条件拒绝、id 重复拒绝；
+- AlertSeverity 排序（primary_alert 取最高）；
+- AlertRuleEngine.evaluate 命中（多规则按 severity 降序）；
+- 模板配置 YAML 加载（含 flow map condition、错误 severity/op 拒绝）；
+- 端到端：模板资产加载 → 急升温特征向量 → P0 命中（场景A 红色告警）。
+"""
+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
+    Alert,
+    AlertCondition,
+    AlertRule,
+    AlertRuleEngine,
+    AlertSeverity,
+    load_alert_rules_config,
+)
+
+NAN = float("nan")
+CONFIG_PATH = os.path.join(
+    os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
+    "config", "alert_rules.template.yaml")
+
+
+# ---------------------------------------------------------------------------
+# 1. AlertCondition
+# ---------------------------------------------------------------------------
+
+class AlertConditionTest(unittest.TestCase):
+
+    def test_ops(self):
+        self.assertTrue(AlertCondition("x", ">", 10).matches({"x": 11}))
+        self.assertTrue(AlertCondition("x", ">=", 10).matches({"x": 10}))
+        self.assertTrue(AlertCondition("x", "<", 10).matches({"x": 9}))
+        self.assertTrue(AlertCondition("x", "<=", 10).matches({"x": 10}))
+        self.assertTrue(AlertCondition("x", "==", 10).matches({"x": 10}))
+        self.assertFalse(AlertCondition("x", ">", 10).matches({"x": 10}))
+
+    def test_missing_value_not_match(self):
+        self.assertFalse(AlertCondition("x", ">", 10).matches({"x": NAN}))
+        self.assertFalse(AlertCondition("x", ">", 10).matches({"y": 100}))
+
+    def test_unknown_op_rejected(self):
+        with self.assertRaises(ValueError):
+            AlertCondition("x", "!=", 10)
+
+
+# ---------------------------------------------------------------------------
+# 2. AlertRule
+# ---------------------------------------------------------------------------
+
+class AlertRuleTest(unittest.TestCase):
+
+    def test_and_semantics(self):
+        rule = AlertRule(id="r1", severity=AlertSeverity.P0, conditions=[
+            AlertCondition("a", ">", 10),
+            AlertCondition("b", "<", 5),
+        ])
+        self.assertTrue(rule.matches({"a": 11, "b": 4}))
+        self.assertFalse(rule.matches({"a": 11, "b": 6}))   # b 不满足
+        self.assertFalse(rule.matches({"a": 9, "b": 4}))    # a 不满足
+
+    def test_empty_conditions_rejected(self):
+        with self.assertRaises(ValueError):
+            AlertRule(id="r", severity=AlertSeverity.P0, conditions=[])
+
+    def test_empty_id_rejected(self):
+        with self.assertRaises(ValueError):
+            AlertRule(id="", severity=AlertSeverity.P0,
+                      conditions=[AlertCondition("a", ">", 1)])
+
+    def test_describe(self):
+        rule = AlertRule(id="r1", severity=AlertSeverity.P0, conditions=[
+            AlertCondition("炉温_ema5", ">", 900.0),
+        ])
+        self.assertIn("P0", rule.describe())
+        self.assertIn("炉温_ema5>900.0", rule.describe())
+
+
+# ---------------------------------------------------------------------------
+# 3. AlertRuleEngine
+# ---------------------------------------------------------------------------
+
+class AlertRuleEngineTest(unittest.TestCase):
+
+    def _engine(self) -> AlertRuleEngine:
+        return AlertRuleEngine([
+            AlertRule(id="p0_rule", severity=AlertSeverity.P0, conditions=[
+                AlertCondition("炉温", ">", 900.0)]),
+            AlertRule(id="p1_rule", severity=AlertSeverity.P1, conditions=[
+                AlertCondition("氯气", ">", 8.0)]),
+            AlertRule(id="p2_rule", severity=AlertSeverity.P2, conditions=[
+                AlertCondition("炉温", ">", 880.0)]),
+        ])
+
+    def test_empty_rules_rejected(self):
+        with self.assertRaises(ValueError):
+            AlertRuleEngine([])
+
+    def test_duplicate_id_rejected(self):
+        with self.assertRaises(ValueError):
+            AlertRuleEngine([
+                AlertRule(id="dup", severity=AlertSeverity.P0,
+                          conditions=[AlertCondition("a", ">", 1)]),
+                AlertRule(id="dup", severity=AlertSeverity.P1,
+                          conditions=[AlertCondition("b", ">", 1)]),
+            ])
+
+    def test_evaluate_returns_sorted_by_severity(self):
+        eng = self._engine()
+        # 炉温=890 同时命中 p2（>880）；氯气=10 命中 p1
+        alerts = eng.evaluate(100, {"炉温": 890.0, "氯气": 10.0})
+        ids = [a.rule_id for a in alerts]
+        self.assertEqual(ids, ["p1_rule", "p2_rule"])   # P1 > P2
+        self.assertEqual([a.severity for a in alerts],
+                         [AlertSeverity.P1, AlertSeverity.P2])
+
+    def test_primary_alert_picks_highest(self):
+        eng = self._engine()
+        # 炉温=920 命中 p0 + p2 → 主告警 P0
+        alerts = eng.evaluate(100, {"炉温": 920.0})
+        primary = eng.primary_alert(alerts)
+        self.assertIsNotNone(primary)
+        self.assertEqual(primary.severity, AlertSeverity.P0)
+
+    def test_no_hit_returns_empty(self):
+        eng = self._engine()
+        self.assertEqual(eng.evaluate(100, {"炉温": 850.0, "氯气": 5.0}), [])
+        self.assertIsNone(eng.primary_alert([]))
+
+
+# ---------------------------------------------------------------------------
+# 4. 模板配置 YAML 加载
+# ---------------------------------------------------------------------------
+
+class ConfigLoadTest(unittest.TestCase):
+
+    def test_load_template_config(self):
+        cfg = load_alert_rules_config(CONFIG_PATH)
+        self.assertEqual(cfg.template, "ti-cl4")
+        self.assertGreaterEqual(len(cfg.rules), 5)
+        ids = [r.id for r in cfg.rules]
+        self.assertIn("bed_temp_critical", ids)
+        self.assertIn("cl2_flow_warning", ids)
+
+    def test_flow_map_condition_parsed(self):
+        cfg = load_alert_rules_config(CONFIG_PATH)
+        r = next(x for x in cfg.rules if x.id == "bed_temp_critical")
+        self.assertEqual(r.severity, AlertSeverity.P0)
+        self.assertEqual(r.conditions[0].feature, "炉温_ema5")
+        self.assertEqual(r.conditions[0].op, ">")
+        self.assertEqual(r.conditions[0].threshold, 900.0)
+        self.assertEqual(r.sop, "SOP-CL-001")
+
+    def test_engine_from_template_config(self):
+        eng = AlertRuleEngine.from_template_config(CONFIG_PATH)
+        # 炉温_ema5=920 命中 bed_temp_critical (P0) + bed_temp_near_limit (P2)
+        alerts = eng.evaluate(1, {"炉温_ema5": 920.0})
+        ids = [a.rule_id for a in alerts]
+        self.assertIn("bed_temp_critical", ids)
+        self.assertEqual(eng.primary_alert(alerts).severity, AlertSeverity.P0)
+
+
+# ---------------------------------------------------------------------------
+# 5. 端到端：急升温场景命中 P0（PRD 场景A 红色告警）
+# ---------------------------------------------------------------------------
+
+class EndToEndScenarioATest(unittest.TestCase):
+
+    def test_rising_temp_triggers_p0(self):
+        """模拟炉层杂质富集的急升温：规则引擎应在 ema 平滑值越界时产出 P0 告警。"""
+        eng = AlertRuleEngine.from_template_config(CONFIG_PATH)
+        # 特征向量：炉温_ema5 越过 900 上限
+        values = {"炉温_ema5": 905.0, "炉温_rate10": 0.06,
+                  "氯气流量_std10": 5.0, "炉压_rate10": 0.02,
+                  "炉层状态_mean10": 60.0}
+        alerts = eng.evaluate(timestamp=100, values=values)
+        primary = eng.primary_alert(alerts)
+        self.assertIsNotNone(primary, "急升温应触发预警")
+        self.assertEqual(primary.severity, AlertSeverity.P0,
+                         "主告警应为 P0 红色告警")
+        # 命中的 P0 规则应有处置 SOP（供 LLM 报警解释 + 值班长确认）
+        self.assertTrue(any(a.sop for a in alerts if a.severity == AlertSeverity.P0))
+
+
+if __name__ == "__main__":
+    unittest.main(verbosity=2)

templates/ti-cl4/impurity-forecast/tests/test_model.py

@@ -0,0 +1,215 @@
+# -*- coding: utf-8 -*-
+"""炉层杂质预警无监督模型单元测试（Issue #71）。
+
+覆盖：
+- ZScoreScorer：fit 估计 μ/σ、score 异常分数（含 σ=0 恒定列、缺失值、未 fit 拒绝）；
+- ThresholdRule：分数阈值 ∪ 特征 breach 决策；
+- ImpurityForecaster：fit/predict 端到端；
+- evaluate_lead_time：提前量评估（对齐 PRD 提前 ≥ 30min）；
+- 序列化：to_dict/from_dict/save/load 可复现。
+"""
+import math
+import os
+import sys
+import tempfile
+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
+    AlertDecision,
+    FeatureVectorLike,
+    ImpurityForecaster,
+    ThresholdRule,
+    ZScoreScorer,
+    evaluate_lead_time,
+)
+
+NAN = float("nan")
+
+
+def _approx(a: float, b: float, eps: float = 1e-6) -> bool:
+    if math.isnan(a) and math.isnan(b):
+        return True
+    return abs(a - b) <= eps
+
+
+def vec(ts: float, **kw) -> FeatureVectorLike:
+    return FeatureVectorLike(timestamp=ts, values=dict(kw))
+
+
+# ---------------------------------------------------------------------------
+# 1. ZScoreScorer
+# ---------------------------------------------------------------------------
+
+class ZScoreScorerTest(unittest.TestCase):
+
+    def test_fit_estimates_mean_std(self):
+        sc = ZScoreScorer().fit([
+            vec(1, x=10.0), vec(2, x=12.0), vec(3, x=14.0), vec(4, x=12.0),
+        ])
+        self.assertTrue(sc.fitted)
+        # mean=12, std=sqrt(((10-12)^2+(12-12)^2+(14-12)^2+(12-12)^2)/4)=sqrt(2)=1.414
+        self.assertTrue(_approx(sc._mean["x"], 12.0))
+        self.assertTrue(_approx(sc._std["x"], math.sqrt(2.0)))
+
+    def test_score_normal_is_low(self):
+        sc = ZScoreScorer().fit([vec(i, x=100.0) for i in range(20)])
+        # 正常段（等于均值）分数应为 0
+        scores = sc.score([vec(100, x=100.0)])
+        self.assertTrue(_approx(scores[0], 0.0))
+
+    def test_score_anomaly_is_high(self):
+        # 正常段均值 100、std≈1.414；异常值 110 → |110-100|/1.414≈7.07
+        sc = ZScoreScorer().fit([vec(i, x=100.0 + (i % 3)) for i in range(20)])
+        scores = sc.score([vec(99, x=110.0)])
+        self.assertGreater(scores[0], 5.0)
+
+    def test_score_takes_max_across_features(self):
+        sc = ZScoreScorer().fit([
+            vec(1, a=0.0, b=0.0), vec(2, a=2.0, b=2.0), vec(3, a=1.0, b=1.0),
+        ])
+        # a/b 均值=1，std≈0.816；输入 a=1（近均值）、b=10（远）→ 取 b 的偏离
+        scores = sc.score([vec(4, a=1.0, b=10.0)])
+        # b 的 z = |10-1|/0.816 ≈ 11.02，应远大于 a 的 z≈0
+        self.assertGreater(scores[0], 10.0)
+
+    def test_constant_column_deviation_flagged(self):
+        # 训练段恒定（std=0），推理段偏离 → 用大常数识别为异常
+        sc = ZScoreScorer().fit([vec(i, c=5.0) for i in range(10)])
+        scores = sc.score([vec(11, c=5.0), vec(12, c=6.0)])
+        self.assertTrue(_approx(scores[0], 0.0))           # 不偏离
+        self.assertGreater(scores[1], 1e5)                 # 偏离 → 大常数
+
+    def test_missing_value_skipped(self):
+        sc = ZScoreScorer().fit([vec(1, x=10.0), vec(2, x=12.0)])
+        # x 缺失（NaN）不应崩溃，分数按可用特征计算（这里全缺失 → 0）
+        scores = sc.score([vec(3, x=NAN)])
+        self.assertTrue(_approx(scores[0], 0.0))
+
+    def test_not_fitted_raises(self):
+        with self.assertRaises(ValueError):
+            ZScoreScorer().score([vec(1, x=1.0)])
+
+    def test_fit_empty_raises(self):
+        with self.assertRaises(ValueError):
+            ZScoreScorer().fit([])
+
+
+# ---------------------------------------------------------------------------
+# 2. ThresholdRule
+# ---------------------------------------------------------------------------
+
+class ThresholdRuleTest(unittest.TestCase):
+
+    def test_score_below_threshold_no_alert(self):
+        rule = ThresholdRule(score_threshold=3.0)
+        d = rule.decide(1.0, {"x": 1.0}, score=2.0)
+        self.assertFalse(d.triggered)
+
+    def test_score_above_threshold_alerts(self):
+        rule = ThresholdRule(score_threshold=3.0)
+        d = rule.decide(1.0, {"x": 1.0}, score=4.5)
+        self.assertTrue(d.triggered)
+        self.assertTrue(any("异常分数" in r for r in d.reasons))
+
+    def test_feature_breach_alerts(self):
+        rule = ThresholdRule(score_threshold=3.0,
+                             feature_thresholds={"炉温_ema5": 900.0})
+        # 分数低，但特征超阈值 → 仍预警
+        d = rule.decide(1.0, {"炉温_ema5": 950.0}, score=1.0)
+        self.assertTrue(d.triggered)
+        self.assertTrue(any("炉温_ema5" in r for r in d.reasons))
+
+    def test_score_threshold_must_be_positive(self):
+        with self.assertRaises(ValueError):
+            ThresholdRule(score_threshold=0)
+        with self.assertRaises(ValueError):
+            ThresholdRule(score_threshold=-1)
+
+
+# ---------------------------------------------------------------------------
+# 3. ImpurityForecaster 端到端
+# ---------------------------------------------------------------------------
+
+class ForecasterTest(unittest.TestCase):
+
+    def test_fit_then_predict(self):
+        f = ImpurityForecaster(rule=ThresholdRule(score_threshold=3.0))
+        normal = [vec(i, x=100.0 + (i % 3)) for i in range(20)]
+        f.fit(normal)
+        decisions = f.predict(normal + [vec(99, x=200.0)])
+        # 正常段无预警；最后一条异常值预警
+        self.assertFalse(any(d.triggered for d in decisions[:-1]))
+        self.assertTrue(decisions[-1].triggered)
+
+    def test_evaluate_returns_leadtime(self):
+        f = ImpurityForecaster(rule=ThresholdRule(score_threshold=3.0))
+        f.fit([vec(i, x=100.0) for i in range(10)])
+        # 构造：ts 0..9 正常，ts 10 起开始异常（递增）
+        samples = [vec(i, x=100.0) for i in range(10)] + \
+                  [vec(i, x=100.0 + 5.0 * (i - 9)) for i in range(10, 20)]
+        decisions, lt = f.evaluate(samples, anomaly_ts=19.0)
+        # 应在 ts=19（峰值）前触发 → 提前量为正
+        self.assertIsNotNone(lt.first_alert_ts)
+        self.assertGreater(lt.lead_seconds, 0)
+        self.assertGreater(lt.lead_minutes, 0)
+
+
+# ---------------------------------------------------------------------------
+# 4. evaluate_lead_time
+# ---------------------------------------------------------------------------
+
+class LeadTimeTest(unittest.TestCase):
+
+    def test_no_alert_returns_none(self):
+        decisions = [AlertDecision(timestamp=t, score=1.0, triggered=False)
+                     for t in [1, 2, 3]]
+        lt = evaluate_lead_time(decisions, anomaly_ts=3.0)
+        self.assertIsNone(lt.first_alert_ts)
+        self.assertIsNone(lt.lead_seconds)
+
+    def test_alert_before_anomaly_positive_lead(self):
+        decisions = [
+            AlertDecision(timestamp=1, score=1.0, triggered=False),
+            AlertDecision(timestamp=5, score=4.0, triggered=True),
+            AlertDecision(timestamp=10, score=5.0, triggered=True),
+        ]
+        lt = evaluate_lead_time(decisions, anomaly_ts=10.0)
+        self.assertEqual(lt.first_alert_ts, 5)
+        # 提前量 = 10 - 5 = 5s
+        self.assertTrue(_approx(lt.lead_seconds, 5.0))
+        self.assertTrue(_approx(lt.lead_minutes, 5.0 / 60))
+
+
+# ---------------------------------------------------------------------------
+# 5. 序列化
+# ---------------------------------------------------------------------------
+
+class SerializationTest(unittest.TestCase):
+
+    def test_roundtrip_dict(self):
+        sc = ZScoreScorer().fit([vec(1, x=10.0), vec(2, x=20.0)])
+        d = sc.to_dict()
+        sc2 = ZScoreScorer.from_dict(d)
+        self.assertTrue(sc2.fitted)
+        # 复现：同一输入分数一致
+        s1 = sc.score([vec(3, x=15.0)])
+        s2 = sc2.score([vec(3, x=15.0)])
+        self.assertTrue(_approx(s1[0], s2[0]))
+
+    def test_save_load_file(self):
+        sc = ZScoreScorer().fit([vec(1, x=10.0), vec(2, x=20.0)])
+        with tempfile.NamedTemporaryFile("w", suffix=".json", delete=False) as fh:
+            path = fh.name
+        try:
+            sc.save(path)
+            sc2 = ZScoreScorer.load(path)
+            self.assertTrue(sc2.fitted)
+        finally:
+            os.unlink(path)
+
+
+if __name__ == "__main__":
+    unittest.main(verbosity=2)