From 0a9cf12212937c1a4f3ba4c5db42bac538d00e30 Mon Sep 17 00:00:00 2001 From: CarterPerez-dev Date: Sat, 28 Feb 2026 05:59:36 -0500 Subject: [PATCH] feat(ml): add ensemble validation with PR-AUC and F1 gates Loads ONNX models, normalizes per-model scores, fuses them, and computes classification metrics with configurable quality gates. --- .../backend/ml/validation.py | 154 ++++++++++++++ .../backend/tests/test_validation.py | 189 ++++++++++++++++++ 2 files changed, 343 insertions(+) create mode 100644 PROJECTS/advanced/ai-threat-detection/backend/ml/validation.py create mode 100644 PROJECTS/advanced/ai-threat-detection/backend/tests/test_validation.py diff --git a/PROJECTS/advanced/ai-threat-detection/backend/ml/validation.py b/PROJECTS/advanced/ai-threat-detection/backend/ml/validation.py new file mode 100644 index 00000000..d1755005 --- /dev/null +++ b/PROJECTS/advanced/ai-threat-detection/backend/ml/validation.py @@ -0,0 +1,154 @@ +""" +©AngelaMos | 2026 +validation.py +""" + +import logging +from dataclasses import dataclass, field +from pathlib import Path + +import numpy as np +from sklearn.metrics import ( + average_precision_score, + confusion_matrix, + f1_score, + precision_score, + recall_score, + roc_auc_score, +) + +from app.core.detection.ensemble import ( + fuse_scores, + normalize_ae_score, + normalize_if_score, +) +from app.core.detection.inference import InferenceEngine + +logger = logging.getLogger(__name__) + +DEFAULT_ENSEMBLE_WEIGHTS: dict[str, float] = { + "ae": 0.4, + "rf": 0.4, + "if": 0.2, +} + +BINARY_THRESHOLD = 0.5 + + +@dataclass +class ValidationResult: + """ + Ensemble validation metrics and gate results + """ + + precision: float + recall: float + f1: float + pr_auc: float + roc_auc: float + confusion_matrix: list[list[int]] + passed_gates: bool + gate_details: dict[str, bool] = field(default_factory=dict) + + +def validate_ensemble( + model_dir: Path, + X_test: np.ndarray, + y_test: np.ndarray, + ensemble_weights: dict[str, float] | None = None, + pr_auc_gate: float = 0.85, + f1_gate: float = 0.80, +) -> ValidationResult: + """ + Run all 3 models on test data and compute classification metrics + """ + weights = ensemble_weights or DEFAULT_ENSEMBLE_WEIGHTS + + engine = InferenceEngine(model_dir=str(model_dir)) + if not engine.is_loaded: + raise RuntimeError( + f"Failed to load models from {model_dir}" + ) + + raw_scores = engine.predict( + X_test.astype(np.float32), + ) + if raw_scores is None: + raise RuntimeError("Inference returned None") + + fused = _compute_fused_scores( + raw_scores, engine.threshold, weights + ) + + y_pred = (fused >= BINARY_THRESHOLD).astype(np.int32) + + prec = float( + precision_score(y_test, y_pred, zero_division=0) + ) + rec = float( + recall_score(y_test, y_pred, zero_division=0) + ) + f1_val = float( + f1_score(y_test, y_pred, zero_division=0) + ) + pr_auc_val = float( + average_precision_score(y_test, fused) + ) + roc_auc_val = float(roc_auc_score(y_test, fused)) + + cm = confusion_matrix(y_test, y_pred).tolist() + + pr_auc_passed = pr_auc_val >= pr_auc_gate + f1_passed = f1_val >= f1_gate + gate_details = { + "pr_auc": pr_auc_passed, + "f1": f1_passed, + } + + logger.info( + "Validation: precision=%.3f recall=%.3f " + "f1=%.3f pr_auc=%.3f roc_auc=%.3f", + prec, + rec, + f1_val, + pr_auc_val, + roc_auc_val, + ) + + return ValidationResult( + precision=prec, + recall=rec, + f1=f1_val, + pr_auc=pr_auc_val, + roc_auc=roc_auc_val, + confusion_matrix=cm, + passed_gates=pr_auc_passed and f1_passed, + gate_details=gate_details, + ) + + +def _compute_fused_scores( + raw_scores: dict[str, list[float]], + threshold: float, + weights: dict[str, float], +) -> np.ndarray: + """ + Normalize and fuse per-model raw scores into a single score array + """ + n_samples = len(raw_scores["ae"]) + fused = np.zeros(n_samples, dtype=np.float64) + + for i in range(n_samples): + per_model: dict[str, float] = {} + + per_model["ae"] = normalize_ae_score( + raw_scores["ae"][i], threshold + ) + per_model["rf"] = raw_scores["rf"][i] + per_model["if"] = normalize_if_score( + raw_scores["if"][i] + ) + + fused[i] = fuse_scores(per_model, weights) + + return fused diff --git a/PROJECTS/advanced/ai-threat-detection/backend/tests/test_validation.py b/PROJECTS/advanced/ai-threat-detection/backend/tests/test_validation.py new file mode 100644 index 00000000..14926e43 --- /dev/null +++ b/PROJECTS/advanced/ai-threat-detection/backend/tests/test_validation.py @@ -0,0 +1,189 @@ +""" +©AngelaMos | 2026 +test_validation.py +""" + +import json +from pathlib import Path + +import numpy as np +import pytest +from sklearn.ensemble import IsolationForest, RandomForestClassifier + +from ml.autoencoder import ThreatAutoencoder +from ml.export_onnx import ( + export_autoencoder, + export_isolation_forest, + export_random_forest, +) +from ml.scaler import FeatureScaler +from ml.validation import ValidationResult, validate_ensemble + + +@pytest.fixture +def trained_model_dir(tmp_path: Path) -> Path: + """ + Create a temp directory with trained ONNX models for validation testing + """ + rng = np.random.default_rng(42) + X_normal = rng.standard_normal((200, 35)).astype(np.float32) + X_attack = rng.standard_normal((80, 35)).astype(np.float32) + 3.0 + X = np.vstack([X_normal, X_attack]) + y = np.array([0] * 200 + [1] * 80, dtype=np.int32) + + ae = ThreatAutoencoder(input_dim=35) + export_autoencoder(ae, tmp_path / "ae.onnx") + + rf = RandomForestClassifier(n_estimators=10, random_state=42) + rf.fit(X, y) + export_random_forest(rf, 35, tmp_path / "rf.onnx") + + iso = IsolationForest(n_estimators=10, random_state=42) + iso.fit(X_normal) + export_isolation_forest(iso, 35, tmp_path / "if.onnx") + + scaler = FeatureScaler() + scaler.fit(X_normal) + scaler.save_json(tmp_path / "scaler.json") + + (tmp_path / "threshold.json").write_text( + json.dumps({"threshold": 0.05}) + ) + + return tmp_path + + +@pytest.fixture +def separable_test_data() -> tuple[np.ndarray, np.ndarray]: + """ + Test data where normal and attack clusters are well-separated + """ + rng = np.random.default_rng(99) + X_normal = rng.standard_normal((50, 35)).astype(np.float32) + X_attack = ( + rng.standard_normal((30, 35)).astype(np.float32) + 3.0 + ) + X = np.vstack([X_normal, X_attack]) + y = np.array([0] * 50 + [1] * 30, dtype=np.int32) + return X, y + + +class TestValidateEnsemble: + """ + Test ensemble validation with metric gates + """ + + def test_returns_validation_result( + self, + trained_model_dir: Path, + separable_test_data: tuple[np.ndarray, np.ndarray], + ) -> None: + """ + validate_ensemble returns a ValidationResult instance + """ + X_test, y_test = separable_test_data + result = validate_ensemble(trained_model_dir, X_test, y_test) + + assert isinstance(result, ValidationResult) + + def test_result_has_all_metrics( + self, + trained_model_dir: Path, + separable_test_data: tuple[np.ndarray, np.ndarray], + ) -> None: + """ + ValidationResult contains precision, recall, f1, pr_auc, roc_auc + """ + X_test, y_test = separable_test_data + result = validate_ensemble(trained_model_dir, X_test, y_test) + + assert 0.0 <= result.precision <= 1.0 + assert 0.0 <= result.recall <= 1.0 + assert 0.0 <= result.f1 <= 1.0 + assert 0.0 <= result.pr_auc <= 1.0 + assert 0.0 <= result.roc_auc <= 1.0 + + def test_confusion_matrix_shape( + self, + trained_model_dir: Path, + separable_test_data: tuple[np.ndarray, np.ndarray], + ) -> None: + """ + Confusion matrix is 2x2 + """ + X_test, y_test = separable_test_data + result = validate_ensemble(trained_model_dir, X_test, y_test) + + assert len(result.confusion_matrix) == 2 + assert len(result.confusion_matrix[0]) == 2 + assert len(result.confusion_matrix[1]) == 2 + + def test_gate_details_contains_both_gates( + self, + trained_model_dir: Path, + separable_test_data: tuple[np.ndarray, np.ndarray], + ) -> None: + """ + gate_details has pr_auc and f1 entries + """ + X_test, y_test = separable_test_data + result = validate_ensemble(trained_model_dir, X_test, y_test) + + assert "pr_auc" in result.gate_details + assert "f1" in result.gate_details + + def test_gates_pass_with_low_thresholds( + self, + trained_model_dir: Path, + separable_test_data: tuple[np.ndarray, np.ndarray], + ) -> None: + """ + passed_gates is True when gates are set very low + """ + X_test, y_test = separable_test_data + result = validate_ensemble( + trained_model_dir, + X_test, + y_test, + pr_auc_gate=0.01, + f1_gate=0.01, + ) + + assert result.passed_gates is True + + def test_gates_fail_with_high_thresholds( + self, + trained_model_dir: Path, + separable_test_data: tuple[np.ndarray, np.ndarray], + ) -> None: + """ + passed_gates is False when gates are impossibly high + """ + X_test, y_test = separable_test_data + result = validate_ensemble( + trained_model_dir, + X_test, + y_test, + pr_auc_gate=1.0, + f1_gate=1.0, + ) + + assert result.passed_gates is False + + def test_custom_ensemble_weights( + self, + trained_model_dir: Path, + separable_test_data: tuple[np.ndarray, np.ndarray], + ) -> None: + """ + Custom ensemble weights are accepted without error + """ + X_test, y_test = separable_test_data + result = validate_ensemble( + trained_model_dir, + X_test, + y_test, + ensemble_weights={"ae": 0.5, "rf": 0.3, "if": 0.2}, + ) + + assert isinstance(result, ValidationResult)