674 lines
26 KiB
Python
674 lines
26 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""异常检测模型模板化（固定主干 + 配方加载）。
|
||
|
||
对应 issue #37（父 EPIC #5「③ AI 模型框架 配置化重构」、PRD 5.3
|
||
「网络结构策略 / 模板化技术路径」）。
|
||
|
||
PRD 5.3 的核心诉求
|
||
------------------
|
||
|
||
异常检测属于 PRD 5.3「四类模型模板」之一（③ 异常检测），同样采用
|
||
「**固定主干 + 可配置超参**」默认模式：同一主干代码不变，切换行业 /
|
||
工况只改 *配方（recipe）* —— 一个声明式 JSON 超参包。本模块与
|
||
``quality_forecast``（issue #36）同源，共享「主干工厂 + Recipe + 验收口径」
|
||
骨架，但任务语义是无监督异常检测：
|
||
|
||
* **输入**：多维工艺特征时序点（无需标注，无监督）；
|
||
* **输出**：每个样本的异常分数（越大越异常）+ 二值异常标签（由阈值决定）；
|
||
* **验收**：检出率 / 误报率 / F1（PRD 5.3 / 第 6 章里程碑：关键异常检出率
|
||
≥ 95%、误报率 ≤ 5%）。
|
||
|
||
本模块交付什么
|
||
--------------
|
||
|
||
1. **``AnomalyDetectionModel``**：固定主干的异常检测模型。默认主干是
|
||
``iforest``（隔离森林，PRD 5.3 推荐的无监督异常检测默认结构）；当运行
|
||
环境存在 ``sklearn`` 时自动升级为真实实现，否则退化为确定性 stub，
|
||
保证边缘 / 离线 / CI 环境可加载与校验——与 issue #34 / #36 的
|
||
「numpy/sklearn 可选」策略一致。
|
||
2. **``Recipe`` 配方加载器**：声明式 JSON 超参包（``load_recipe`` /
|
||
``build_from_recipe``）。配方描述「主干类型 + 超参 + 特征列 + 阈值策略
|
||
+ 验收口径」，业务侧只 ``build_from_recipe(path)`` 一行即可拿到一个
|
||
可训练 / 可推理的异常检测模型——切换模板仅改配方，模型代码零改动。
|
||
3. **``Metrics`` 验收口径**：PRD 5.3 / 第 6 章里程碑要求「关键异常检出率
|
||
≥ 95%、误报率 ≤ 5%」。``evaluate`` 直接给出检出率 / 误报率 / 精确率 /
|
||
召回率 / F1，便于配置台与 UAT 直接读取。
|
||
4. **样例配方（``samples/`` JSON）**：Ti（海绵钛氯化车间炉层杂质预警）+
|
||
树脂两套异常检测超参包样例，验证「同框架加载两套配方均跑通」的验收
|
||
口径。
|
||
|
||
与 issue #34 ``model_recipe`` / #36 ``quality_forecast`` 的关系
|
||
--------------------------------------------------------------
|
||
|
||
接口风格对齐 #34 的 ``ModelHandle`` / ``ModelRecipe``（``fit`` /
|
||
``decision_function`` / ``to_dict``、不可变声明式数据对象），以及 #36
|
||
的「主干工厂注册表 + Recipe + 验收口径」骨架。本模块**自包含、不依赖
|
||
#34 / #36 未合并分支**，待二者合入后，异常检测主干可平滑注册为
|
||
``register_backbone("iforest", ...)`` 的一个具名主干，配方可映射为一条
|
||
``ModelRecipe``——届时本模块零业务侧改动。
|
||
|
||
零外部强依赖
|
||
------------
|
||
|
||
* 主干默认走纯 Python stub（``StubBackbone``）：无 sklearn 时也能加载、
|
||
构造、（伪）拟合与打分，保证 CI 可加载与校验；
|
||
* 存在 ``sklearn`` 时，``iforest`` 主干自动升级为真实
|
||
``IsolationForest`` 实现，其余情况退化为 stub，不影响接口契约与测试。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import math
|
||
import os
|
||
from dataclasses import dataclass, field
|
||
from typing import Any, Dict, List, Optional, Sequence, Tuple
|
||
|
||
__all__ = [
|
||
# 数据对象
|
||
"Recipe",
|
||
"Metrics",
|
||
"AnomalyDetectionError",
|
||
# 模型
|
||
"AnomalyDetectionModel",
|
||
"ModelHandle",
|
||
# 主干工厂
|
||
"BACKBONES",
|
||
"register_backbone",
|
||
"iforest_backbone",
|
||
"lof_backbone",
|
||
"stub_backbone",
|
||
# 配方 API
|
||
"load_recipe",
|
||
"build_from_recipe",
|
||
"list_sample_recipes",
|
||
"sample_recipe_path",
|
||
]
|
||
|
||
|
||
class AnomalyDetectionError(Exception):
|
||
"""异常检测模板化层的统一异常（配方非法 / 主干未注册 / 校验失败）。"""
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 配方（Recipe）：声明式超参包，不可变数据对象
|
||
# ---------------------------------------------------------------------------
|
||
|
||
#: PRD 5.3 允许的固定主干类型（默认 iforest，PRD 5.3 推荐无监督异常检测默认结构）
|
||
ALLOWED_BACKBONES = ("iforest", "lof", "stub")
|
||
|
||
#: PRD 5.3 允许的阈值策略：contamination（污染率）分数阈值；sigma（Nσ 法则）
|
||
ALLOWED_THRESHOLD_POLICIES = ("contamination", "sigma")
|
||
|
||
#: PRD 5.3 / 第 6 章里程碑：关键异常检出率（召回率）验收线 ≥ 95%
|
||
DEFAULT_RECALL_FLOOR = 0.95
|
||
|
||
#: PRD 5.3 / 第 6 章里程碑：异常误报率上限 ≤ 5%（即特异性 ≥ 0.95）
|
||
DEFAULT_FALSE_ALARM_CEIL = 0.05
|
||
|
||
#: 默认污染率（预期异常比例），对齐 sklearn IsolationForest 默认值
|
||
DEFAULT_CONTAMINATION = 0.05
|
||
|
||
#: 默认 Nσ 法则阈值（3σ 覆盖 ~99.7% 正常区）
|
||
DEFAULT_SIGMA = 3.0
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class Recipe:
|
||
"""异常检测配方（声明式超参包）。
|
||
|
||
一个 Recipe 描述「用什么固定主干 + 如何从超参构造一个可训练 / 可推理
|
||
的异常检测模型 + 用哪些特征列 + 阈值策略 + 验收口径」。它是不可变数据
|
||
对象，``to_dict`` / ``from_dict`` 可序列化往返，便于配置台展示与审计。
|
||
|
||
切换行业 / 工况只改 Recipe，模型代码（``AnomalyDetectionModel``）零改动
|
||
——对齐 PRD 5.3「固定主干 + 可配置超参」默认模式。
|
||
"""
|
||
|
||
name: str
|
||
backbone: str = "iforest"
|
||
hyperparams: Dict[str, Any] = field(default_factory=dict)
|
||
feature_columns: Tuple[str, ...] = field(default_factory=tuple)
|
||
threshold_policy: str = "contamination"
|
||
contamination: float = DEFAULT_CONTAMINATION
|
||
sigma: float = DEFAULT_SIGMA
|
||
recall_floor: float = DEFAULT_RECALL_FLOOR
|
||
false_alarm_ceil: float = DEFAULT_FALSE_ALARM_CEIL
|
||
industry: str = ""
|
||
notes: str = ""
|
||
|
||
def __post_init__(self) -> None:
|
||
if not self.name:
|
||
raise AnomalyDetectionError("Recipe 缺少 name")
|
||
if self.backbone not in ALLOWED_BACKBONES:
|
||
raise AnomalyDetectionError(
|
||
f"非法主干类型 {self.backbone!r}，允许：{ALLOWED_BACKBONES}")
|
||
if self.threshold_policy not in ALLOWED_THRESHOLD_POLICIES:
|
||
raise AnomalyDetectionError(
|
||
f"非法阈值策略 {self.threshold_policy!r}，"
|
||
f"允许：{ALLOWED_THRESHOLD_POLICIES}")
|
||
if not (0.0 < self.contamination < 1.0):
|
||
raise AnomalyDetectionError(
|
||
f"contamination 越界：{self.contamination}（应在 (0,1)）")
|
||
if self.sigma <= 0:
|
||
raise AnomalyDetectionError(
|
||
f"sigma 非法：{self.sigma}（应 > 0）")
|
||
if not (0.0 <= self.recall_floor <= 1.0):
|
||
raise AnomalyDetectionError(
|
||
f"recall_floor 越界：{self.recall_floor}（应在 [0,1]）")
|
||
if not (0.0 <= self.false_alarm_ceil <= 1.0):
|
||
raise AnomalyDetectionError(
|
||
f"false_alarm_ceil 越界：{self.false_alarm_ceil}（应在 [0,1]）")
|
||
|
||
def to_dict(self) -> Dict[str, Any]:
|
||
return {
|
||
"name": self.name,
|
||
"backbone": self.backbone,
|
||
"hyperparams": dict(self.hyperparams),
|
||
"feature_columns": list(self.feature_columns),
|
||
"threshold_policy": self.threshold_policy,
|
||
"contamination": self.contamination,
|
||
"sigma": self.sigma,
|
||
"recall_floor": self.recall_floor,
|
||
"false_alarm_ceil": self.false_alarm_ceil,
|
||
"industry": self.industry,
|
||
"notes": self.notes,
|
||
}
|
||
|
||
@classmethod
|
||
def from_dict(cls, data: Dict[str, Any]) -> "Recipe":
|
||
try:
|
||
return cls(
|
||
name=data["name"],
|
||
backbone=data.get("backbone", "iforest"),
|
||
hyperparams=dict(data.get("hyperparams", {})),
|
||
feature_columns=tuple(data.get("feature_columns", [])),
|
||
threshold_policy=data.get(
|
||
"threshold_policy", "contamination"),
|
||
contamination=float(
|
||
data.get("contamination", DEFAULT_CONTAMINATION)),
|
||
sigma=float(data.get("sigma", DEFAULT_SIGMA)),
|
||
recall_floor=float(
|
||
data.get("recall_floor", DEFAULT_RECALL_FLOOR)),
|
||
false_alarm_ceil=float(
|
||
data.get("false_alarm_ceil", DEFAULT_FALSE_ALARM_CEIL)),
|
||
industry=data.get("industry", ""),
|
||
notes=data.get("notes", ""),
|
||
)
|
||
except KeyError as exc: # pragma: no cover - 防御性
|
||
raise AnomalyDetectionError(
|
||
f"配方缺少必填字段：{exc}") from exc
|
||
|
||
|
||
def load_recipe(path: str) -> Recipe:
|
||
"""从 JSON 文件加载一个异常检测配方。
|
||
|
||
配方 JSON 结构见 ``Recipe.to_dict``；样例见 ``samples/``。
|
||
"""
|
||
with open(path, "r", encoding="utf-8") as fh:
|
||
data = json.load(fh)
|
||
if not isinstance(data, dict):
|
||
raise AnomalyDetectionError(f"配方根必须是对象：{path}")
|
||
return Recipe.from_dict(data)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 主干工厂：固定主干网络（iforest / lof / stub）
|
||
# ---------------------------------------------------------------------------
|
||
|
||
class ModelHandle:
|
||
"""统一模型句柄：fit / decision_function / to_dict，与硬件和具体库无关。
|
||
|
||
业务代码只持有 ``ModelHandle``，不感知底层是 sklearn 还是 stub。
|
||
|
||
约定 ``decision_function`` 返回**异常分数**：**越大越异常**（与
|
||
sklearn ``score_samples`` 取负号一致），便于阈值策略统一处理。
|
||
"""
|
||
|
||
def __init__(self, backbone: str, params: Dict[str, Any],
|
||
fitted: bool = False, meta: Optional[Dict[str, Any]] = None):
|
||
self.backbone = backbone
|
||
self.params = dict(params)
|
||
self._fitted = fitted
|
||
self.meta: Dict[str, Any] = dict(meta or {})
|
||
|
||
@property
|
||
def fitted(self) -> bool:
|
||
return self._fitted
|
||
|
||
def fit(self, X: Sequence[Sequence[float]]) -> "ModelHandle":
|
||
"""拟合主干（无监督，仅需 X）。"""
|
||
X = list(X)
|
||
if not X:
|
||
raise AnomalyDetectionError("训练数据为空")
|
||
self._fit_impl(X)
|
||
self._fitted = True
|
||
return self
|
||
|
||
# 子类/工厂填充
|
||
def _fit_impl(self, X: Sequence[Sequence[float]]) -> None:
|
||
raise NotImplementedError
|
||
|
||
def decision_function(
|
||
self, X: Sequence[Sequence[float]]) -> List[float]:
|
||
"""返回每个样本的异常分数（越大越异常）。"""
|
||
if not self._fitted:
|
||
raise AnomalyDetectionError("模型未拟合，无法打分")
|
||
return [self._score_one(list(row)) for row in X]
|
||
|
||
def _score_one(self, row: Sequence[float]) -> float:
|
||
raise NotImplementedError
|
||
|
||
def to_dict(self) -> Dict[str, Any]:
|
||
return {
|
||
"backbone": self.backbone,
|
||
"params": dict(self.params),
|
||
"fitted": self._fitted,
|
||
"meta": dict(self.meta),
|
||
}
|
||
|
||
|
||
class _StubBackbone(ModelHandle):
|
||
"""确定性 stub 主干：无 sklearn 时的保底实现。
|
||
|
||
拟合阶段记录每维特征的均值与标准差；打分取各维偏离均值的标准差倍数
|
||
之和（马氏距离的简化版），保证可复现、可校验、可对比，便于 CI 与
|
||
配置台预览。
|
||
"""
|
||
|
||
def __init__(self, params: Dict[str, Any]):
|
||
super().__init__(backbone="stub", params=params)
|
||
self._means: List[float] = []
|
||
self._stds: List[float] = []
|
||
|
||
def _fit_impl(self, X) -> None:
|
||
n_feat = len(X[0])
|
||
self._means = [0.0] * n_feat
|
||
self._stds = [1.0] * n_feat
|
||
for j in range(n_feat):
|
||
col = [float(row[j]) for row in X]
|
||
mean = sum(col) / len(col)
|
||
var = sum((v - mean) ** 2 for v in col) / len(col)
|
||
self._means[j] = mean
|
||
self._stds[j] = math.sqrt(var) or 1.0
|
||
self.meta.update({"n_features": n_feat})
|
||
|
||
def _score_one(self, row) -> float:
|
||
# 各维偏离均值的标准差倍数之和（≥0，越大越异常）
|
||
total = 0.0
|
||
for j, v in enumerate(row):
|
||
total += abs(float(v) - self._means[j]) / (self._stds[j] or 1.0)
|
||
return total
|
||
|
||
|
||
class _SklearnIForestBackbone(ModelHandle):
|
||
"""真实隔离森林主干（sklearn IsolationForest）。
|
||
|
||
仅当运行环境存在 sklearn 时启用；与 stub 接口完全一致。
|
||
``decision_function`` 对 sklearn ``score_samples`` 取负号，
|
||
统一为「越大越异常」。
|
||
"""
|
||
|
||
def __init__(self, params: Dict[str, Any]):
|
||
super().__init__(backbone="iforest", params=params)
|
||
from sklearn.ensemble import IsolationForest # type: ignore
|
||
self._Clz = IsolationForest
|
||
self._model: Any = None
|
||
|
||
def _fit_impl(self, X) -> None:
|
||
kw = {
|
||
"n_estimators": int(self.params.get("n_estimators", 100)),
|
||
"max_samples": self.params.get("max_samples", "auto"),
|
||
"contamination": float(
|
||
self.params.get("contamination", "auto")),
|
||
"random_state": int(self.params.get("random_state", 42)),
|
||
}
|
||
self._model = self._Clz(**kw)
|
||
self._model.fit(list(X))
|
||
# 记录实际生效的关键超参（max_samples 可能是 'auto'）
|
||
self.meta.update({"n_estimators": kw["n_estimators"],
|
||
"random_state": kw["random_state"]})
|
||
|
||
def _score_one(self, row) -> float:
|
||
# score_samples 越大越正常，取负号统一为「越大越异常」
|
||
return float(-self._model.score_samples([list(row)])[0])
|
||
|
||
|
||
class _SklearnLOFBackbone(ModelHandle):
|
||
"""真实局部离群因子主干（sklearn LocalOutlierFactor）。
|
||
|
||
PRD 5.3 备选结构；仅当运行环境存在 sklearn 时启用。 novelty=True 以
|
||
支持 predict / score_samples 对新样本打分。
|
||
"""
|
||
|
||
def __init__(self, params: Dict[str, Any]):
|
||
super().__init__(backbone="lof", params=params)
|
||
from sklearn.neighbors import LocalOutlierFactor # type: ignore
|
||
self._Clz = LocalOutlierFactor
|
||
self._model: Any = None
|
||
|
||
def _fit_impl(self, X) -> None:
|
||
kw = {
|
||
"n_neighbors": int(self.params.get("n_neighbors", 20)),
|
||
"contamination": float(
|
||
self.params.get("contamination", "auto")),
|
||
"novelty": True,
|
||
}
|
||
self._model = self._Clz(**kw)
|
||
self._model.fit(list(X))
|
||
self.meta.update({"n_neighbors": kw["n_neighbors"]})
|
||
|
||
def _score_one(self, row) -> float:
|
||
return float(-self._model.score_samples([list(row)])[0])
|
||
|
||
|
||
def _has_sklearn() -> bool:
|
||
try:
|
||
import sklearn # noqa: F401
|
||
return True
|
||
except Exception:
|
||
return False
|
||
|
||
|
||
def stub_backbone(hyperparams: Dict[str, Any]) -> ModelHandle:
|
||
"""stub 主干工厂（恒可用）。"""
|
||
return _StubBackbone(hyperparams)
|
||
|
||
|
||
def iforest_backbone(hyperparams: Dict[str, Any]) -> ModelHandle:
|
||
"""iforest 主干工厂：有 sklearn 用真实隔离森林，否则退化为 stub。
|
||
|
||
PRD 5.3 推荐的无监督异常检测默认结构（隔离森林）。
|
||
"""
|
||
if _has_sklearn():
|
||
return _SklearnIForestBackbone(hyperparams)
|
||
# 无 sklearn：退化 stub 但保留声明主干名，便于审计
|
||
h = _StubBackbone(hyperparams)
|
||
h.meta["degraded_from"] = "iforest"
|
||
return h
|
||
|
||
|
||
def lof_backbone(hyperparams: Dict[str, Any]) -> ModelHandle:
|
||
"""lof 主干工厂：有 sklearn 用真实 LOF，否则退化为 stub。"""
|
||
if _has_sklearn():
|
||
return _SklearnLOFBackbone(hyperparams)
|
||
h = _StubBackbone(hyperparams)
|
||
h.meta["degraded_from"] = "lof"
|
||
return h
|
||
|
||
|
||
#: 主干注册表：新增结构走 ``register_backbone`` 注册，不动内核
|
||
#: （对齐 PRD 5.3「新增结构走插件注册」理念，风格对齐 #34 / #36）。
|
||
BACKBONES: Dict[str, Any] = {
|
||
"iforest": iforest_backbone,
|
||
"lof": lof_backbone,
|
||
"stub": stub_backbone,
|
||
}
|
||
|
||
|
||
def register_backbone(name: str, factory: Any) -> None:
|
||
"""注册一个新主干工厂 ``factory(hyperparams) -> ModelHandle``。
|
||
|
||
允许高级行业模板声明非默认主干（如自研流式异常检测），不动内核——对齐
|
||
PRD「新增结构走插件注册而非改内核」。
|
||
"""
|
||
if not callable(factory):
|
||
raise AnomalyDetectionError("主干工厂必须是可调用对象")
|
||
BACKBONES[name] = factory
|
||
|
||
|
||
def _build_backbone(backbone: str,
|
||
hyperparams: Dict[str, Any]) -> ModelHandle:
|
||
factory = BACKBONES.get(backbone)
|
||
if factory is None:
|
||
raise AnomalyDetectionError(
|
||
f"未注册的主干类型：{backbone!r}，已注册：{list(BACKBONES)}")
|
||
return factory(hyperparams)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 异常检测模型：固定主干 + 配方加载
|
||
# ---------------------------------------------------------------------------
|
||
|
||
class AnomalyDetectionModel:
|
||
"""异常检测模型（固定主干 + 配方加载）。
|
||
|
||
业务侧两种等价入口：
|
||
|
||
1. 直接构造（显式主干）::
|
||
|
||
m = AnomalyDetectionModel(backbone="iforest", hyperparams={...})
|
||
|
||
2. 配方加载（推荐，切换模板仅改配方）::
|
||
|
||
m = build_from_recipe(
|
||
"templates/.../anomaly-detection/recipe.ti.json")
|
||
"""
|
||
|
||
def __init__(self,
|
||
backbone: str = "iforest",
|
||
hyperparams: Optional[Dict[str, Any]] = None,
|
||
feature_columns: Optional[Sequence[str]] = None,
|
||
threshold_policy: str = "contamination",
|
||
contamination: float = DEFAULT_CONTAMINATION,
|
||
sigma: float = DEFAULT_SIGMA,
|
||
recall_floor: float = DEFAULT_RECALL_FLOOR,
|
||
false_alarm_ceil: float = DEFAULT_FALSE_ALARM_CEIL):
|
||
self.threshold_policy = threshold_policy
|
||
self.contamination = contamination
|
||
self.sigma = sigma
|
||
self.recall_floor = recall_floor
|
||
self.false_alarm_ceil = false_alarm_ceil
|
||
self.recipe_meta: Dict[str, Any] = {
|
||
"backbone": backbone,
|
||
"hyperparams": dict(hyperparams or {}),
|
||
"feature_columns": list(feature_columns or []),
|
||
"threshold_policy": threshold_policy,
|
||
"contamination": contamination,
|
||
"sigma": sigma,
|
||
"recall_floor": recall_floor,
|
||
"false_alarm_ceil": false_alarm_ceil,
|
||
}
|
||
self._handle: ModelHandle = _build_backbone(
|
||
backbone, hyperparams or {})
|
||
self._threshold: Optional[float] = None
|
||
|
||
@classmethod
|
||
def from_recipe(cls, recipe: Recipe) -> "AnomalyDetectionModel":
|
||
"""从一个 ``Recipe`` 构造模型（推荐入口）。"""
|
||
m = cls(
|
||
backbone=recipe.backbone,
|
||
hyperparams=recipe.hyperparams,
|
||
feature_columns=recipe.feature_columns,
|
||
threshold_policy=recipe.threshold_policy,
|
||
contamination=recipe.contamination,
|
||
sigma=recipe.sigma,
|
||
recall_floor=recipe.recall_floor,
|
||
false_alarm_ceil=recipe.false_alarm_ceil,
|
||
)
|
||
m.recipe_meta["recipe_name"] = recipe.name
|
||
m.recipe_meta["industry"] = recipe.industry
|
||
return m
|
||
|
||
# ---- 训练 / 推理 ----
|
||
|
||
def fit(self, X: Sequence[Sequence[float]]) -> "AnomalyDetectionModel":
|
||
"""拟合主干（无监督）。同时在训练集上确定异常分数阈值。"""
|
||
X = list(X)
|
||
self._handle.fit(X)
|
||
# 用训练分布确定阈值：contamination 取高分位数；sigma 取均值+Nσ
|
||
scores = self._handle.decision_function(X)
|
||
self._threshold = self._derive_threshold(scores)
|
||
return self
|
||
|
||
def _derive_threshold(self, scores: Sequence[float]) -> float:
|
||
"""根据阈值策略从训练分数分布确定异常分数阈值。
|
||
|
||
- ``contamination``：取高分位数（1 - contamination），高于即判异常；
|
||
- ``sigma``：取均值 + Nσ（N=3 默认覆盖 ~99.7% 正常区）。
|
||
"""
|
||
scores = sorted(float(s) for s in scores)
|
||
if not scores:
|
||
raise AnomalyDetectionError("训练分数为空，无法确定阈值")
|
||
if self.threshold_policy == "sigma":
|
||
mean = sum(scores) / len(scores)
|
||
var = sum((s - mean) ** 2 for s in scores) / len(scores)
|
||
std = math.sqrt(var) or 1.0
|
||
return mean + self.sigma * std
|
||
# contamination：高分位数（线性插值）
|
||
k = (1.0 - self.contamination) * (len(scores) - 1)
|
||
lo = int(math.floor(k))
|
||
hi = int(math.ceil(k))
|
||
if lo == hi:
|
||
return scores[lo]
|
||
frac = k - lo
|
||
return scores[lo] + (scores[hi] - scores[lo]) * frac
|
||
|
||
def decision_function(
|
||
self, X: Sequence[Sequence[float]]) -> List[float]:
|
||
"""返回每个样本的异常分数（越大越异常）。"""
|
||
return self._handle.decision_function(X)
|
||
|
||
def predict(self, X: Sequence[Sequence[float]]) -> List[int]:
|
||
"""返回每个样本的二值异常标签：1=异常，0=正常。
|
||
|
||
依据 ``fit`` 时确定的阈值（未拟合或阈值未定则报错）。
|
||
"""
|
||
if self._threshold is None:
|
||
raise AnomalyDetectionError(
|
||
"阈值未确定：请先 fit，或阈值策略未被应用")
|
||
scores = self.decision_function(X)
|
||
return [1 if s > self._threshold else 0 for s in scores]
|
||
|
||
@property
|
||
def fitted(self) -> bool:
|
||
return self._handle.fitted
|
||
|
||
@property
|
||
def threshold(self) -> Optional[float]:
|
||
return self._threshold
|
||
|
||
# ---- 验收口径 ----
|
||
|
||
def evaluate(self, X: Sequence[Sequence[float]],
|
||
y_true: Sequence[int]) -> "Metrics":
|
||
"""评估并返回检出率 / 误报率 / 精确率 / 召回率 / F1 与是否达标。
|
||
|
||
``y_true`` 中 1=异常、0=正常。检出率即召回率（PRD 5.3 / 里程碑：
|
||
≥ 95%）；误报率即假阳性率（1 - 特异性，里程碑：≤ 5%）。
|
||
``recall >= recall_floor`` 且 ``false_alarm <= false_alarm_ceil``
|
||
即视为达标。
|
||
"""
|
||
y_pred = self.predict(X)
|
||
return Metrics.compute(
|
||
y_true=list(y_true), y_pred=y_pred,
|
||
recall_floor=self.recall_floor,
|
||
false_alarm_ceil=self.false_alarm_ceil)
|
||
|
||
def to_dict(self) -> Dict[str, Any]:
|
||
return {
|
||
"recipe_meta": dict(self.recipe_meta),
|
||
"handle": self._handle.to_dict(),
|
||
"threshold": self._threshold,
|
||
}
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 验收：Metrics
|
||
# ---------------------------------------------------------------------------
|
||
|
||
@dataclass(frozen=True)
|
||
class Metrics:
|
||
"""异常检测验收结果（PRD 5.3 检出率 / 误报率口径）。"""
|
||
|
||
recall: float # 检出率（TP/TP+FN），里程碑 ≥ 95%
|
||
precision: float # 精确率（TP/TP+FP）
|
||
f1: float # F1
|
||
false_alarm_rate: float # 误报率（FP/FP+TN），里程碑 ≤ 5%
|
||
n_anomaly_true: int
|
||
n_normal_true: int
|
||
recall_floor: float
|
||
false_alarm_ceil: float
|
||
passed: bool
|
||
|
||
def to_dict(self) -> Dict[str, Any]:
|
||
return {
|
||
"recall": self.recall,
|
||
"precision": self.precision,
|
||
"f1": self.f1,
|
||
"false_alarm_rate": self.false_alarm_rate,
|
||
"n_anomaly_true": self.n_anomaly_true,
|
||
"n_normal_true": self.n_normal_true,
|
||
"recall_floor": self.recall_floor,
|
||
"false_alarm_ceil": self.false_alarm_ceil,
|
||
"passed": self.passed,
|
||
}
|
||
|
||
@classmethod
|
||
def compute(cls, y_true: Sequence[int], y_pred: Sequence[int],
|
||
recall_floor: float = DEFAULT_RECALL_FLOOR,
|
||
false_alarm_ceil: float = DEFAULT_FALSE_ALARM_CEIL) -> "Metrics":
|
||
if len(y_true) != len(y_pred):
|
||
raise AnomalyDetectionError(
|
||
f"y_true/y_pred 长度不一致：{len(y_true)} != {len(y_pred)}")
|
||
if not y_true:
|
||
raise AnomalyDetectionError("评估数据为空")
|
||
# 统计混淆矩阵四元
|
||
tp = fp = fn = tn = 0
|
||
for yt, yp in zip(y_true, y_pred):
|
||
if yt == 1 and yp == 1:
|
||
tp += 1
|
||
elif yt == 0 and yp == 1:
|
||
fp += 1
|
||
elif yt == 1 and yp == 0:
|
||
fn += 1
|
||
else:
|
||
tn += 1
|
||
n_anomaly = tp + fn
|
||
n_normal = fp + tn
|
||
recall = tp / n_anomaly if n_anomaly > 0 else 0.0
|
||
precision = tp / (tp + fp) if (tp + fp) > 0 else 0.0
|
||
f1 = (2 * precision * recall / (precision + recall)
|
||
if (precision + recall) > 0 else 0.0)
|
||
far = fp / n_normal if n_normal > 0 else 0.0
|
||
passed = recall >= recall_floor and far <= false_alarm_ceil
|
||
return cls(
|
||
recall=recall, precision=precision, f1=f1,
|
||
false_alarm_rate=far,
|
||
n_anomaly_true=n_anomaly, n_normal_true=n_normal,
|
||
recall_floor=recall_floor, false_alarm_ceil=false_alarm_ceil,
|
||
passed=passed,
|
||
)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 配方构建入口 + 样例协议
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def build_from_recipe(path: str) -> AnomalyDetectionModel:
|
||
"""从 JSON 配方文件加载并构造一个异常检测模型（推荐入口）。
|
||
|
||
切换模板仅改配方文件，业务代码零改动——对齐 PRD 5.3 验收口径。
|
||
"""
|
||
return AnomalyDetectionModel.from_recipe(load_recipe(path))
|
||
|
||
|
||
def _samples_dir() -> str:
|
||
return os.path.join(os.path.dirname(os.path.abspath(__file__)),
|
||
"samples", "anomaly-detection")
|
||
|
||
|
||
def list_sample_recipes() -> List[str]:
|
||
"""列出内置样例配方（树脂 + Ti 两套，验证同框架加载多套配方）。"""
|
||
d = _samples_dir()
|
||
if not os.path.isdir(d):
|
||
return []
|
||
return sorted(f for f in os.listdir(d) if f.endswith(".json"))
|
||
|
||
|
||
def sample_recipe_path(name: str) -> str:
|
||
"""返回样例配方的完整路径。"""
|
||
if not name.endswith(".json"):
|
||
name = name + ".json"
|
||
return os.path.join(_samples_dir(), name)
|