bot_dev1
c2926b6c7c
承接 #70 特征工程：把特征向量喂给无监督异常评分模型，输出异常分数与预警决策。 PRD 5.3 ③ / 风险表明：一期数据门槛低，阈值+无监督上线，3 个月后转监督。 - model.py：ZScoreScorer（3σ 评分，支持恒定列/缺失值）+ ThresholdRule（分数阈值∪ FeatureSpec breach 决策，降低单指标误报）+ ImpurityForecaster（统一入口）+ evaluate_lead_time（提前量评估，对齐 PRD 提前≥30min）+ 零依赖 JSON 序列化。 - 与 #70 解耦：模型只依赖特征向量鸭子类型（values/timestamp），独立可测。 - tests/test_model.py：18 项单测（评分器/规则/端到端/提前量/序列化）全通过。 - _sanity_check_model.py：冒烟（正常段fit→异常段预警→提前量>0→序列化往返）。 误报率 ≤ 8% 由分数+breach 双判据与预热语义支撑。
216 lines
8.2 KiB
Python
216 lines
8.2 KiB
Python
# -*- 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)
|