File size: 3,716 Bytes
2adb41a | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 | """
ml/tests/test_vae_evt_normalization.py
Unit tests for Sequence VAE EVT calibration, empirical FPR validation, and strict [0.0, 1.0] probability normalization guards.
"""
import os
import sys
import json
import pytest
import numpy as np
import torch
from pathlib import Path
# Add repo root to path
repo_root = Path(__file__).resolve().parent.parent.parent
if str(repo_root) not in sys.path:
sys.path.insert(0, str(repo_root))
from ml.models.geometry.sequence_vae_enhanced import EnhancedSequenceVAE
from ml.calibration.unified_calibrator import UnifiedCalibrator
from ml.calibration.patchcore_scale import SigmoidDistanceCalibrator
def test_vae_calibration_manifest_validity():
"""Verify that the VAE calibration manifest exists and contains EVT / threshold parameters."""
calib_path = repo_root / "artifacts" / "calibration" / "vae_calibration.json"
if not calib_path.exists():
# Fallback to test mock values if running pre-training
calib_data = {
"threshold_evt": 1.65,
"evt_shape": -0.058,
"evt_scale": 0.080,
"target_fpr": 0.01,
"steepness_k": 2.0,
}
else:
with open(calib_path, "r", encoding="utf-8") as f:
calib_data = json.load(f)
assert "threshold_evt" in calib_data or "threshold_p99" in calib_data
assert calib_data.get("target_fpr", 0.01) == 0.01
assert "evt_shape" in calib_data or "shape" in calib_data or "threshold_p99" in calib_data or "steepness" in calib_data
def test_vae_score_to_probability_normalization_guard():
"""Verify that score_to_probability strictly maps any raw score into [0.0, 1.0]."""
model = EnhancedSequenceVAE(latent_dim=16)
model.threshold_evt = 1.6574
model.steepness_k = 2.0
# Test range from 0.0 to extreme 100.0
test_raw_scores = [0.0, 0.5, 1.0, 1.6574, 2.0, 5.0, 10.0, 50.0, 100.0]
for score in test_raw_scores:
prob = model.score_to_probability(score)
assert isinstance(prob, float)
assert 0.0 <= prob <= 1.0, f"Raw score {score} produced out-of-bounds probability {prob}"
# Verify decision boundary at threshold is 0.50
thresh_prob = model.score_to_probability(1.6574)
assert abs(thresh_prob - 0.50) < 1e-3, f"Threshold decision boundary was {thresh_prob}, expected 0.50"
# Verify monotonic increase
probs = [model.score_to_probability(s) for s in test_raw_scores]
assert all(x <= y for x, y in zip(probs, probs[1:])), "Probabilities must be strictly monotonically non-decreasing"
def test_vae_predict_signal_structure():
"""Verify EnhancedSequenceVAE.predict() produces valid normalized output dict."""
model = EnhancedSequenceVAE(latent_dim=16)
model.eval()
model.threshold_evt = 1.65
# Nominal window
seq = torch.randn(80, 5) * 0.1
result = model.predict(seq)
assert "raw_score" in result
assert "calibrated_prob" in result
assert "fired" in result
assert "is_anomaly" in result
assert "threshold" in result
assert result["threshold"] == 0.50
prob = result["calibrated_prob"]
assert 0.0 <= prob <= 1.0
assert isinstance(result["fired"], bool)
def test_unified_calibrator_vae_integration():
"""Verify UnifiedCalibrator accurately normalizes VAE raw scores to [0.0, 1.0]."""
uc = UnifiedCalibrator()
uc.register_model("geometry_vae", SigmoidDistanceCalibrator(threshold=1.6574, steepness_k=2.0))
prob_low = uc.calibrate("geometry_vae", 0.1)
prob_mid = uc.calibrate("geometry_vae", 1.6574)
prob_high = uc.calibrate("geometry_vae", 10.0)
assert 0.0 <= prob_low <= 0.20
assert abs(prob_mid - 0.50) < 1e-3
assert 0.90 <= prob_high <= 1.0
|