From a0b21866fe58c0e540bdffd2c18c7f909538ba27 Mon Sep 17 00:00:00 2001 From: bot_dev2 Date: Wed, 5 Aug 2026 08:21:41 +0800 Subject: [PATCH] =?UTF-8?q?Merge=20PR=20#112/#113=20(feat=20#71=20?= =?UTF-8?q?=E6=97=A0=E7=9B=91=E7=9D=A3=E6=A8=A1=E5=9E=8B=20+=20#72=20?= =?UTF-8?q?=E5=91=8A=E8=AD=A6=E8=A7=84=E5=88=99=EF=BC=8C=E4=B8=8E=20#70=20?= =?UTF-8?q?=E7=89=B9=E5=BE=81=E5=B7=A5=E7=A8=8B=E8=81=94=E5=90=88)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../ti-cl4/impurity-forecast/__init__.py | 47 ++- .../impurity-forecast/_sanity_check_model.py | 74 ++++ .../impurity-forecast/_sanity_check_rules.py | 59 +++ .../ti-cl4/impurity-forecast/alert_rules.py | 356 ++++++++++++++++++ .../config/alert_rules.template.yaml | 64 ++++ templates/ti-cl4/impurity-forecast/model.py | 252 +++++++++++++ .../tests/test_alert_rules.py | 197 ++++++++++ .../impurity-forecast/tests/test_model.py | 215 +++++++++++ 8 files changed, 1261 insertions(+), 3 deletions(-) create mode 100644 templates/ti-cl4/impurity-forecast/_sanity_check_model.py create mode 100644 templates/ti-cl4/impurity-forecast/_sanity_check_rules.py create mode 100644 templates/ti-cl4/impurity-forecast/alert_rules.py create mode 100644 templates/ti-cl4/impurity-forecast/config/alert_rules.template.yaml create mode 100644 templates/ti-cl4/impurity-forecast/model.py create mode 100644 templates/ti-cl4/impurity-forecast/tests/test_alert_rules.py create mode 100644 templates/ti-cl4/impurity-forecast/tests/test_model.py diff --git a/templates/ti-cl4/impurity-forecast/__init__.py b/templates/ti-cl4/impurity-forecast/__init__.py index 48e2b22..056c025 100644 --- a/templates/ti-cl4/impurity-forecast/__init__.py +++ b/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", ] diff --git a/templates/ti-cl4/impurity-forecast/_sanity_check_model.py b/templates/ti-cl4/impurity-forecast/_sanity_check_model.py new file mode 100644 index 0000000..c1512bc --- /dev/null +++ b/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()) diff --git a/templates/ti-cl4/impurity-forecast/_sanity_check_rules.py b/templates/ti-cl4/impurity-forecast/_sanity_check_rules.py new file mode 100644 index 0000000..8bde4b0 --- /dev/null +++ b/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()) diff --git a/templates/ti-cl4/impurity-forecast/alert_rules.py b/templates/ti-cl4/impurity-forecast/alert_rules.py new file mode 100644 index 0000000..0e6650c --- /dev/null +++ b/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) diff --git a/templates/ti-cl4/impurity-forecast/config/alert_rules.template.yaml b/templates/ti-cl4/impurity-forecast/config/alert_rules.template.yaml new file mode 100644 index 0000000..8616cba --- /dev/null +++ b/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} diff --git a/templates/ti-cl4/impurity-forecast/model.py b/templates/ti-cl4/impurity-forecast/model.py new file mode 100644 index 0000000..c6a8268 --- /dev/null +++ b/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) diff --git a/templates/ti-cl4/impurity-forecast/tests/test_alert_rules.py b/templates/ti-cl4/impurity-forecast/tests/test_alert_rules.py new file mode 100644 index 0000000..46e7ebd --- /dev/null +++ b/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) diff --git a/templates/ti-cl4/impurity-forecast/tests/test_model.py b/templates/ti-cl4/impurity-forecast/tests/test_model.py new file mode 100644 index 0000000..e120f00 --- /dev/null +++ b/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)