feat(#37): 异常检测模型模板化（固定主干+配方加载，PRD 5.3 ③异常检测）
新增 core/model-framework/anomaly_detection.py：固定主干（默认 iforest 隔离森林， sklearn 可选，无依赖时退化确定性 stub）+ Recipe 声明式配方加载 + Metrics 验收口径（检出率≥95%、误报率≤5%）。配套 samples/（Ti 炉层杂质预警 + 树脂 两套样例配方）、tests/（33 用例全过）、_sanity_check.py。同框架加载两套 配方均跑通，模型代码零改动。
This commit is contained in:
@@ -0,0 +1,28 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""测试引导：把连字符目录 ``core/model-framework`` 加载为可导入包
|
||||
``model_framework``，使测试可 ``from model_framework import ...``。
|
||||
|
||||
与仓库内各 core 模块的测试引导同款模式（importlib 完整加载包，执行
|
||||
``__init__.py``，保持顶层导出可用）。
|
||||
"""
|
||||
import importlib.util
|
||||
import os
|
||||
import sys
|
||||
|
||||
PKG_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
|
||||
REPO_ROOT = os.path.dirname(os.path.dirname(
|
||||
os.path.dirname(os.path.dirname(os.path.abspath(__file__)))))
|
||||
|
||||
|
||||
def _load_package(name: str, path: str) -> None:
|
||||
if name in sys.modules:
|
||||
return
|
||||
init_py = os.path.join(path, "__init__.py")
|
||||
spec = importlib.util.spec_from_file_location(
|
||||
name, init_py, submodule_search_locations=[path])
|
||||
module = importlib.util.module_from_spec(spec)
|
||||
sys.modules[name] = module
|
||||
spec.loader.exec_module(module)
|
||||
|
||||
|
||||
_load_package("model_framework", PKG_DIR)
|
||||
@@ -0,0 +1,333 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""``anomaly_detection`` 单元测试（issue #37）。
|
||||
|
||||
覆盖：
|
||||
- 配方（Recipe）不可变性 / 序列化往返 / 非法主干、非法阈值策略与越界校验；
|
||||
- 主干工厂注册表 + 自定义主干注册（PRD 5.3「新增结构走插件注册」）；
|
||||
- stub / iforest / lof 三类主干的 fit/decision_function 契约；
|
||||
- 固定主干 + 配方加载：同框架加载 Ti / 树脂两套配方均跑通（PRD 5.3
|
||||
验收口径）；
|
||||
- Metrics 验收口径（PRD 5.3 / 里程碑：检出率 ≥ 95%、误报率 ≤ 5%）；
|
||||
- 阈值策略（contamination 高分位 / sigma Nσ 法则）；
|
||||
- 零外部强依赖：无 sklearn 时 stub 退化仍可加载与校验。
|
||||
"""
|
||||
import json
|
||||
import os
|
||||
import sys
|
||||
import unittest
|
||||
|
||||
HERE = os.path.dirname(os.path.abspath(__file__))
|
||||
sys.path.insert(0, HERE)
|
||||
import _bootstrap # noqa: E402 注册 model_framework 包
|
||||
|
||||
from model_framework import ( # noqa: E402
|
||||
AnomalyDetectionError,
|
||||
AnomalyDetectionModel,
|
||||
BACKBONES,
|
||||
Metrics,
|
||||
ModelHandle,
|
||||
Recipe,
|
||||
build_from_recipe,
|
||||
iforest_backbone,
|
||||
list_sample_recipes,
|
||||
load_recipe,
|
||||
lof_backbone,
|
||||
register_backbone,
|
||||
sample_recipe_path,
|
||||
stub_backbone,
|
||||
)
|
||||
|
||||
|
||||
def _normal_dataset(n=40, n_feat=2, seed=0):
|
||||
"""构造一组「正常」样本（围绕均值的确定性点）。"""
|
||||
X = []
|
||||
for i in range(n):
|
||||
row = []
|
||||
for j in range(n_feat):
|
||||
base = float(i % 7) + 1.0 + 0.1 * j
|
||||
row.append(base)
|
||||
X.append(row)
|
||||
return X
|
||||
|
||||
|
||||
def _labeled_dataset(n_normal=40, n_anomaly=5, n_feat=2):
|
||||
"""构造正常 + 离群点数据集，返回 (X, y_true)，1=异常。"""
|
||||
X = _normal_dataset(n_normal, n_feat)
|
||||
y = [0] * n_normal
|
||||
for k in range(n_anomaly):
|
||||
# 明显远离正常区的离群点
|
||||
X.append([100.0 + k for _ in range(n_feat)])
|
||||
y.append(1)
|
||||
return X, y
|
||||
|
||||
|
||||
class TestRecipe(unittest.TestCase):
|
||||
"""配方数据对象与校验。"""
|
||||
|
||||
def test_defaults_and_immutability(self):
|
||||
r = Recipe(name="t")
|
||||
self.assertEqual(r.backbone, "iforest")
|
||||
self.assertEqual(r.threshold_policy, "contamination")
|
||||
self.assertAlmostEqual(r.recall_floor, 0.95)
|
||||
self.assertAlmostEqual(r.false_alarm_ceil, 0.05)
|
||||
with self.assertRaises(Exception):
|
||||
r.name = "other" # frozen
|
||||
|
||||
def test_roundtrip(self):
|
||||
r = Recipe(name="t", backbone="lof",
|
||||
hyperparams={"n_neighbors": 15},
|
||||
feature_columns=("a", "b"),
|
||||
threshold_policy="sigma",
|
||||
contamination=0.1, sigma=2.5,
|
||||
recall_floor=0.9, false_alarm_ceil=0.1,
|
||||
industry="树脂", notes="n")
|
||||
d = r.to_dict()
|
||||
r2 = Recipe.from_dict(d)
|
||||
self.assertEqual(r, r2)
|
||||
# JSON 往返
|
||||
r3 = Recipe.from_dict(json.loads(json.dumps(d)))
|
||||
self.assertEqual(r, r3)
|
||||
|
||||
def test_invalid_backbone_raises(self):
|
||||
with self.assertRaises(AnomalyDetectionError):
|
||||
Recipe(name="t", backbone="svm")
|
||||
|
||||
def test_invalid_threshold_policy_raises(self):
|
||||
with self.assertRaises(AnomalyDetectionError):
|
||||
Recipe(name="t", threshold_policy="quantile")
|
||||
|
||||
def test_contamination_out_of_range(self):
|
||||
with self.assertRaises(AnomalyDetectionError):
|
||||
Recipe(name="t", contamination=0.0)
|
||||
with self.assertRaises(AnomalyDetectionError):
|
||||
Recipe(name="t", contamination=1.0)
|
||||
|
||||
def test_sigma_nonpositive_raises(self):
|
||||
with self.assertRaises(AnomalyDetectionError):
|
||||
Recipe(name="t", sigma=0)
|
||||
|
||||
def test_recall_floor_out_of_range(self):
|
||||
with self.assertRaises(AnomalyDetectionError):
|
||||
Recipe(name="t", recall_floor=1.5)
|
||||
|
||||
def test_missing_name(self):
|
||||
with self.assertRaises(AnomalyDetectionError):
|
||||
Recipe(name="")
|
||||
|
||||
def test_load_recipe_from_file(self):
|
||||
path = sample_recipe_path("recipe.ti.json")
|
||||
r = load_recipe(path)
|
||||
self.assertEqual(r.name, "ti-cl4-furnace-impurity-anomaly")
|
||||
self.assertEqual(r.backbone, "iforest")
|
||||
self.assertIn("furnace_temp", r.feature_columns)
|
||||
|
||||
|
||||
class TestBackbones(unittest.TestCase):
|
||||
"""主干工厂与注册表。"""
|
||||
|
||||
def test_builtin_backbones_registered(self):
|
||||
for name in ("iforest", "lof", "stub"):
|
||||
self.assertIn(name, BACKBONES)
|
||||
|
||||
def test_register_custom_backbone(self):
|
||||
class _Custom(ModelHandle):
|
||||
def __init__(self, p):
|
||||
super().__init__("custom", p)
|
||||
self._v = 1.0
|
||||
|
||||
def _fit_impl(self, X):
|
||||
self._v = sum(sum(r) for r in X) / (len(X) * len(X[0]))
|
||||
|
||||
def _score_one(self, row):
|
||||
# 离均值越远分数越高
|
||||
return abs(sum(float(v) for v in row) - self._v)
|
||||
|
||||
register_backbone("custom_test", lambda p: _Custom(p))
|
||||
m = AnomalyDetectionModel(backbone="custom_test")
|
||||
X = _normal_dataset()
|
||||
m.fit(X)
|
||||
self.assertEqual(len(m.predict(X)), len(X))
|
||||
# 清理避免污染其它用例
|
||||
BACKBONES.pop("custom_test", None)
|
||||
|
||||
def test_unknown_backbone_raises(self):
|
||||
with self.assertRaises(AnomalyDetectionError):
|
||||
AnomalyDetectionModel(backbone="not_a_backbone")
|
||||
|
||||
def test_stub_score_is_deterministic_and_nonneg(self):
|
||||
h = stub_backbone({})
|
||||
X = _normal_dataset()
|
||||
h.fit(X)
|
||||
s1 = h.decision_function(X)
|
||||
s2 = h.decision_function(X)
|
||||
self.assertEqual(s1, s2)
|
||||
self.assertTrue(all(isinstance(v, float) for v in s1))
|
||||
self.assertTrue(all(v >= 0 for v in s1))
|
||||
|
||||
def test_decision_before_fit_raises(self):
|
||||
h = stub_backbone({})
|
||||
with self.assertRaises(AnomalyDetectionError):
|
||||
h.decision_function([[1.0, 2.0]])
|
||||
|
||||
def test_fit_empty_raises(self):
|
||||
h = stub_backbone({})
|
||||
with self.assertRaises(AnomalyDetectionError):
|
||||
h.fit([])
|
||||
|
||||
def test_iforest_factory_runs_with_or_without_sklearn(self):
|
||||
# 无论 sklearn 是否存在都不应报错
|
||||
h = iforest_backbone({"n_estimators": 20})
|
||||
X = _normal_dataset()
|
||||
h.fit(X)
|
||||
scores = h.decision_function(X)
|
||||
self.assertEqual(len(scores), len(X))
|
||||
|
||||
|
||||
class TestModelContract(unittest.TestCase):
|
||||
"""模型 fit/decision_function/predict 契约。"""
|
||||
|
||||
def test_fit_predict_shapes(self):
|
||||
m = AnomalyDetectionModel(backbone="stub")
|
||||
X = _normal_dataset(20)
|
||||
m.fit(X)
|
||||
self.assertTrue(m.fitted)
|
||||
self.assertIsNotNone(m.threshold)
|
||||
preds = m.predict(X)
|
||||
self.assertEqual(len(preds), len(X))
|
||||
self.assertTrue(all(p in (0, 1) for p in preds))
|
||||
|
||||
def test_predict_before_fit_raises(self):
|
||||
m = AnomalyDetectionModel(backbone="stub")
|
||||
with self.assertRaises(AnomalyDetectionError):
|
||||
m.predict([[1.0, 2.0]])
|
||||
|
||||
def test_decision_before_fit_raises(self):
|
||||
m = AnomalyDetectionModel(backbone="stub")
|
||||
with self.assertRaises(AnomalyDetectionError):
|
||||
m.decision_function([[1.0, 2.0]])
|
||||
|
||||
def test_fit_empty_raises(self):
|
||||
m = AnomalyDetectionModel(backbone="stub")
|
||||
with self.assertRaises(AnomalyDetectionError):
|
||||
m.fit([])
|
||||
|
||||
def test_to_dict_roundtrip_meta(self):
|
||||
m = AnomalyDetectionModel(backbone="iforest",
|
||||
hyperparams={"n_estimators": 5},
|
||||
feature_columns=["a"],
|
||||
threshold_policy="sigma", sigma=2.0)
|
||||
d = m.to_dict()
|
||||
self.assertEqual(d["recipe_meta"]["backbone"], "iforest")
|
||||
self.assertEqual(d["recipe_meta"]["threshold_policy"], "sigma")
|
||||
self.assertIn("handle", d)
|
||||
|
||||
def test_threshold_contamination_isolate_outliers(self):
|
||||
"""contamination 阈值应把注入的离群点判为异常。"""
|
||||
m = AnomalyDetectionModel(
|
||||
backbone="stub", threshold_policy="contamination",
|
||||
contamination=0.10)
|
||||
X, y_true = _labeled_dataset(n_normal=40, n_anomaly=5)
|
||||
m.fit(X)
|
||||
preds = m.predict(X)
|
||||
# 注入的 5 个离群点应被全部判异常
|
||||
self.assertEqual(sum(preds[40:]), 5)
|
||||
|
||||
def test_threshold_sigma_isolate_outliers(self):
|
||||
"""sigma 阈值也应把注入的极端离群点判为异常。"""
|
||||
m = AnomalyDetectionModel(
|
||||
backbone="stub", threshold_policy="sigma", sigma=2.0)
|
||||
X, y_true = _labeled_dataset(n_normal=40, n_anomaly=5)
|
||||
m.fit(X)
|
||||
preds = m.predict(X)
|
||||
self.assertEqual(sum(preds[40:]), 5)
|
||||
|
||||
|
||||
class TestMetrics(unittest.TestCase):
|
||||
"""验收口径（PRD 5.3：检出率 ≥ 95%、误报率 ≤ 5%）。"""
|
||||
|
||||
def test_perfect_predictions_pass(self):
|
||||
y = [1, 1, 0, 0, 0]
|
||||
met = Metrics.compute(y, y, recall_floor=0.95, false_alarm_ceil=0.05)
|
||||
self.assertAlmostEqual(met.recall, 1.0)
|
||||
self.assertAlmostEqual(met.false_alarm_rate, 0.0)
|
||||
self.assertAlmostEqual(met.f1, 1.0)
|
||||
self.assertTrue(met.passed)
|
||||
|
||||
def test_all_miss_fails(self):
|
||||
y_true = [1, 1, 0, 0]
|
||||
y_pred = [0, 0, 0, 0] # 漏检全部异常
|
||||
met = Metrics.compute(y_true, y_pred)
|
||||
self.assertAlmostEqual(met.recall, 0.0)
|
||||
self.assertFalse(met.passed)
|
||||
|
||||
def test_high_false_alarm_fails(self):
|
||||
y_true = [1, 0, 0, 0, 0]
|
||||
y_pred = [1, 1, 1, 1, 1] # 全判异常：检出但误报爆表
|
||||
met = Metrics.compute(y_true, y_pred, false_alarm_ceil=0.05)
|
||||
self.assertAlmostEqual(met.recall, 1.0)
|
||||
self.assertGreater(met.false_alarm_rate, 0.05)
|
||||
self.assertFalse(met.passed)
|
||||
|
||||
def test_length_mismatch_raises(self):
|
||||
with self.assertRaises(AnomalyDetectionError):
|
||||
Metrics.compute([1, 0], [1])
|
||||
|
||||
def test_empty_raises(self):
|
||||
with self.assertRaises(AnomalyDetectionError):
|
||||
Metrics.compute([], [])
|
||||
|
||||
def test_no_anomaly_in_true_recall_zero_div_safe(self):
|
||||
# 无真实异常时 recall 定义为 0，不应抛 ZeroDivision
|
||||
met = Metrics.compute([0, 0, 0], [0, 0, 0])
|
||||
self.assertEqual(met.recall, 0.0)
|
||||
self.assertEqual(met.n_anomaly_true, 0)
|
||||
|
||||
def test_evaluate_end_to_end(self):
|
||||
m = AnomalyDetectionModel(backbone="stub", threshold_policy="sigma",
|
||||
sigma=2.0)
|
||||
X, y_true = _labeled_dataset(n_normal=40, n_anomaly=5)
|
||||
m.fit(X)
|
||||
met = m.evaluate(X, y_true)
|
||||
self.assertIsInstance(met, Metrics)
|
||||
# 离群点应被检出（stub 在极端离群点上召回=1）
|
||||
self.assertEqual(met.recall, 1.0)
|
||||
|
||||
|
||||
class TestSampleRecipes(unittest.TestCase):
|
||||
"""样例协议：同框架加载 Ti / 树脂两套配方均跑通（PRD 5.3 验收口径）。"""
|
||||
|
||||
def test_samples_present(self):
|
||||
names = list_sample_recipes()
|
||||
self.assertIn("recipe.ti.json", names)
|
||||
self.assertIn("recipe.resin.json", names)
|
||||
|
||||
def test_build_from_each_sample_runs(self):
|
||||
for name in ("recipe.ti.json", "recipe.resin.json"):
|
||||
m = build_from_recipe(sample_recipe_path(name))
|
||||
self.assertIn(
|
||||
m.recipe_meta["backbone"], ("iforest", "lof", "stub"))
|
||||
feat = m.recipe_meta["feature_columns"]
|
||||
n_feat = len(feat)
|
||||
self.assertGreater(n_feat, 0)
|
||||
X = [[float(i + j) for j in range(n_feat)] for i in range(30)]
|
||||
# 注入离群点
|
||||
for k in range(3):
|
||||
X.append([100.0 + k for _ in range(n_feat)])
|
||||
y_true = [0] * 30 + [1] * 3
|
||||
m.fit(X)
|
||||
preds = m.predict(X)
|
||||
self.assertEqual(len(preds), len(y_true))
|
||||
met = m.evaluate(X, y_true)
|
||||
self.assertIsInstance(met, Metrics)
|
||||
|
||||
def test_two_recipes_share_same_code(self):
|
||||
"""切换模板仅改配方，模型代码零改动（PRD 5.3）。"""
|
||||
m1 = build_from_recipe(sample_recipe_path("recipe.ti.json"))
|
||||
m2 = build_from_recipe(sample_recipe_path("recipe.resin.json"))
|
||||
self.assertEqual(type(m1), type(m2))
|
||||
self.assertNotEqual(m1.recipe_meta.get("recipe_name"),
|
||||
m2.recipe_meta.get("recipe_name"))
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
unittest.main(verbosity=2)
|
||||
Reference in New Issue
Block a user