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.

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

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)