"""ML anomaly detection engine: multiple detector families + score ensemble. Families (spec §13): - density: KNN distance, Local Outlier Factor - isolation: Isolation Forest, One-Class SVM, DBSCAN noise/centroid distance - deep: Autoencoder reconstruction error (lightweight numpy MLP) - generative: VAE (numpy, PCA-latent Gaussian likelihood), GAN (numpy Mahalanobis "discriminator" surrogate) - metric: few-shot prototype embeddings (support/query distance) The deep/generative detectors are intentionally lightweight numpy implementations (no torch in this environment): clearly marked, deterministic, and still produce meaningful scores on the synthetic features. All detectors implement Detector.fit(X) / .score(X) where HIGHER = more anomalous. Ensemble normalizes scores before combining and records detector_scores, normalized_scores, ensemble_strategy, ensemble_score and the decision_threshold on every decision (spec §14). """ from __future__ import annotations from typing import Dict, List, Optional, Tuple import numpy as np from sklearn.cluster import DBSCAN from sklearn.ensemble import IsolationForest from sklearn.linear_model import LogisticRegression from sklearn.neighbors import LocalOutlierFactor, NearestNeighbors from sklearn.preprocessing import StandardScaler from sklearn.svm import OneClassSVM class Detector: name = "detector" family = "generic" def fit(self, X: np.ndarray) -> "Detector": raise NotImplementedError def score(self, X: np.ndarray) -> np.ndarray: """Higher = more anomalous.""" raise NotImplementedError class KNNDetector(Detector): name = "knn" family = "density" def __init__(self, k: int = 10): self.k = k self.nn: Optional[NearestNeighbors] = None def fit(self, X: np.ndarray) -> "KNNDetector": self.nn = NearestNeighbors(n_neighbors=min(self.k + 1, len(X))) self.nn.fit(X) return self def score(self, X: np.ndarray) -> np.ndarray: dists, _ = self.nn.kneighbors(X) return dists[:, 1:].mean(axis=1) # skip self-distance class LOFDetector(Detector): name = "lof" family = "density" def __init__(self, n_neighbors: int = 20): self.n_neighbors = n_neighbors self._scores: Optional[np.ndarray] = None self._X: Optional[np.ndarray] = None def fit(self, X: np.ndarray) -> "LOFDetector": lof = LocalOutlierFactor(n_neighbors=min(self.n_neighbors, max(2, len(X) - 1))) lof.fit(X) self._scores = -lof.negative_outlier_factor_ self._X = X return self def score(self, X: np.ndarray) -> np.ndarray: if self._X is not None and X.shape == self._X.shape and np.allclose(X, self._X): return self._scores # Novel points (e.g. a re-scored corrected label): approximate LOF by # the ratio of the query's kNN distance to the training population's. nn = NearestNeighbors(n_neighbors=min(self.n_neighbors, len(self._X))) nn.fit(self._X) d_query, _ = nn.kneighbors(X) d_train, _ = nn.kneighbors(self._X) ref = max(float(d_train[:, 1:].mean()), 1e-9) return d_query.mean(axis=1) / ref class IsolationForestDetector(Detector): name = "isolation_forest" family = "isolation" def __init__(self, n_estimators: int = 100, seed: int = 7): self.model = IsolationForest(n_estimators=n_estimators, random_state=seed) def fit(self, X: np.ndarray) -> "IsolationForestDetector": self.model.fit(X) return self def score(self, X: np.ndarray) -> np.ndarray: return -self.model.score_samples(X) class OCSVMDetector(Detector): name = "ocsvm" family = "isolation" def __init__(self, nu: float = 0.05): self.model = OneClassSVM(nu=nu, kernel="rbf", gamma="scale") def fit(self, X: np.ndarray) -> "OCSVMDetector": self.model.fit(X) return self def score(self, X: np.ndarray) -> np.ndarray: return -self.model.decision_function(X) class DBSCANDetector(Detector): name = "dbscan" family = "isolation" def __init__(self, eps: float = 1.2, min_samples: int = 5): self.eps = eps self.min_samples = min_samples self.centroids: Optional[np.ndarray] = None def fit(self, X: np.ndarray) -> "DBSCANDetector": labels = DBSCAN(eps=self.eps, min_samples=self.min_samples).fit_predict(X) cents = [X[labels == c].mean(axis=0) for c in sorted(set(labels)) if c != -1] self.centroids = np.array(cents) if cents else X.mean(axis=0, keepdims=True) return self def score(self, X: np.ndarray) -> np.ndarray: d = np.linalg.norm(X[:, None, :] - self.centroids[None, :, :], axis=2) return d.min(axis=1) class AutoencoderDetector(Detector): """Lightweight numpy MLP autoencoder (in-repo substitute for a torch AE). Single hidden bottleneck, tanh activation, plain SGD; deterministic seed. Score = per-sample reconstruction MSE. """ name = "autoencoder" family = "deep" def __init__(self, latent_dim: int = 4, epochs: int = 60, lr: float = 0.05, seed: int = 7): self.latent_dim = latent_dim self.epochs = epochs self.lr = lr self.seed = seed self.W1 = self.b1 = self.W2 = self.b2 = None def fit(self, X: np.ndarray) -> "AutoencoderDetector": rng = np.random.default_rng(self.seed) n, d = X.shape k = min(self.latent_dim, d) self.W1 = rng.normal(0, 0.1, (d, k)) self.b1 = np.zeros(k) self.W2 = rng.normal(0, 0.1, (k, d)) self.b2 = np.zeros(d) for _ in range(self.epochs): idx = rng.permutation(n)[: min(n, 512)] xb = X[idx] z = np.tanh(xb @ self.W1 + self.b1) xh = z @ self.W2 + self.b2 err = xh - xb # (B, d) gW2 = z.T @ err / len(xb) gb2 = err.mean(axis=0) dz = (err @ self.W2.T) * (1 - z ** 2) gW1 = xb.T @ dz / len(xb) gb1 = dz.mean(axis=0) self.W2 -= self.lr * gW2 self.b2 -= self.lr * gb2 self.W1 -= self.lr * gW1 self.b1 -= self.lr * gb1 return self def score(self, X: np.ndarray) -> np.ndarray: z = np.tanh(X @ self.W1 + self.b1) xh = z @ self.W2 + self.b2 return ((xh - X) ** 2).mean(axis=1) class VAEDetector(Detector): """Lightweight generative surrogate: Gaussian likelihood in a PCA latent space (stands in for a torch VAE ELBO). Score = negative log-likelihood. """ name = "vae" family = "generative" def __init__(self, latent_dim: int = 5): self.latent_dim = latent_dim self.mean = self.components = self.latent_var = self.noise_var = None def fit(self, X: np.ndarray) -> "VAEDetector": self.mean = X.mean(axis=0) Xc = X - self.mean _, s, vt = np.linalg.svd(Xc, full_matrices=False) k = min(self.latent_dim, vt.shape[0]) self.components = vt[:k] z = Xc @ self.components.T self.latent_var = z.var(axis=0) + 1e-6 recon = z @ self.components self.noise_var = max(float(((Xc - recon) ** 2).mean()), 1e-6) return self def score(self, X: np.ndarray) -> np.ndarray: Xc = X - self.mean z = Xc @ self.components.T recon = z @ self.components nll_latent = 0.5 * ((z ** 2) / self.latent_var).sum(axis=1) nll_recon = 0.5 * ((Xc - recon) ** 2).sum(axis=1) / self.noise_var return nll_latent + nll_recon class GANDetector(Detector): """Lightweight generative-adversarial surrogate: discriminator score approximated by Mahalanobis distance under a shrunk covariance of the "real" data distribution (stands in for a torch GAN discriminator). """ name = "gan" family = "generative" def __init__(self, shrinkage: float = 0.1): self.shrinkage = shrinkage self.mean = self.prec = None def fit(self, X: np.ndarray) -> "GANDetector": self.mean = X.mean(axis=0) cov = np.cov(X, rowvar=False) cov = (1 - self.shrinkage) * cov + self.shrinkage * np.eye(X.shape[1]) * np.trace(cov) / X.shape[1] self.prec = np.linalg.pinv(cov) return self def score(self, X: np.ndarray) -> np.ndarray: d = X - self.mean return np.sqrt(np.maximum((d @ self.prec * d).sum(axis=1), 0)) class FewShotDetector(Detector): """Metric-learning detector: class prototypes from a small support set of presumed-normal samples; query score = distance to nearest prototype. """ name = "few_shot" family = "metric" def __init__(self, support_per_class: int = 20, n_prototypes: int = 8, seed: int = 7): self.support_per_class = support_per_class self.n_prototypes = n_prototypes self.seed = seed self.prototypes: Optional[np.ndarray] = None def fit(self, X: np.ndarray) -> "FewShotDetector": rng = np.random.default_rng(self.seed) # Support set: samples nearest to the data median (most "typical"). med = np.median(X, axis=0) d = np.linalg.norm(X - med, axis=1) order = np.argsort(d) support = X[order[: max(self.n_prototypes * self.support_per_class, 10)]] # Simple k-means style prototypes over the support set. k = min(self.n_prototypes, len(support)) protos = support[rng.choice(len(support), k, replace=False)] for _ in range(10): assign = np.argmin(np.linalg.norm(support[:, None] - protos[None], axis=2), axis=1) for j in range(k): pts = support[assign == j] if len(pts): protos[j] = pts.mean(axis=0) self.prototypes = protos return self def score(self, X: np.ndarray) -> np.ndarray: d = np.linalg.norm(X[:, None, :] - self.prototypes[None, :, :], axis=2) return d.min(axis=1) # ------------------------------------------------------------------ ensemble def normalize_scores(scores: np.ndarray) -> np.ndarray: """Rank-based normalization to [0, 1]; robust to scale differences.""" order = scores.argsort().argsort().astype(np.float64) denom = max(len(scores) - 1, 1) return order / denom DEFAULT_WEIGHTS = { "knn": 1.0, "lof": 1.0, "isolation_forest": 1.2, "ocsvm": 0.8, "dbscan": 0.8, "autoencoder": 1.0, "vae": 0.9, "gan": 0.7, "few_shot": 0.9, } class AnomalyEnsemble: """Runs enabled detectors on standardized features and combines them.""" def __init__(self, config: Optional[Dict] = None, seed: int = 7): self.config = config or {} self.seed = seed self.detectors = self._build() self.scaler = StandardScaler() # detector name -> error string for detectors that failed in the last # run() and were excluded from the ensemble (never silently zeroed). self.last_failures: Dict[str, str] = {} def _cfg(self, *path, default=None): node = self.config for p in path: if not isinstance(node, dict) or p not in node: return default node = node[p] return node def _build(self) -> List[Detector]: dets: List[Detector] = [] if self._cfg("detectors", "knn", "enabled", default=True): dets.append(KNNDetector(k=int(self._cfg("detectors", "knn", "k", default=10)))) if self._cfg("detectors", "lof", "enabled", default=True): dets.append(LOFDetector(n_neighbors=int(self._cfg("detectors", "lof", "n_neighbors", default=20)))) if self._cfg("detectors", "isolation_forest", "enabled", default=True): dets.append(IsolationForestDetector( n_estimators=int(self._cfg("detectors", "isolation_forest", "n_estimators", default=100)), seed=self.seed)) if self._cfg("detectors", "ocsvm", "enabled", default=True): dets.append(OCSVMDetector(nu=float(self._cfg("detectors", "ocsvm", "nu", default=0.05)))) if self._cfg("detectors", "dbscan", "enabled", default=True): dets.append(DBSCANDetector( eps=float(self._cfg("detectors", "dbscan", "eps", default=1.2)), min_samples=int(self._cfg("detectors", "dbscan", "min_samples", default=5)))) if self._cfg("deep", "autoencoder", "enabled", default=True): dets.append(AutoencoderDetector( latent_dim=int(self._cfg("deep", "autoencoder", "latent_dim", default=4)), epochs=int(self._cfg("deep", "autoencoder", "epochs", default=60)), seed=self.seed)) if self._cfg("deep", "vae", "enabled", default=True): dets.append(VAEDetector(latent_dim=int(self._cfg("deep", "vae", "latent_dim", default=5)))) if self._cfg("deep", "gan", "enabled", default=False): dets.append(GANDetector()) if self._cfg("advanced", "few_shot", "enabled", default=True): dets.append(FewShotDetector( support_per_class=int(self._cfg("advanced", "few_shot", "support_per_class", default=20)), seed=self.seed)) return dets @property def strategy(self) -> str: return str(self._cfg("advanced", "ensemble_strategy", default="weighted_average")) @property def threshold(self) -> float: return float(self._cfg("advanced", "decision_threshold", default=0.9)) def run( self, X: np.ndarray, supervision: Optional[np.ndarray] = None, ) -> Tuple[np.ndarray, Dict[str, np.ndarray], Dict[str, np.ndarray]]: """Returns (ensemble_scores, raw_scores_by_detector, normalized_by_detector). supervision (optional 0/1 labels) is used ONLY by the meta_classifier strategy, trained on a held-out half to avoid trivially memorizing. """ Xs = self.scaler.fit_transform(X) raw: Dict[str, np.ndarray] = {} norm: Dict[str, np.ndarray] = {} self.last_failures: Dict[str, str] = {} for det in self.detectors: try: det.fit(Xs) s = det.score(Xs) except Exception as exc: # A failed detector is recorded and EXCLUDED from the ensemble. # Substituting zeros (the previous behavior) silently diluted # every other detector's signal and masked anomalies. self.last_failures[det.name] = f"{type(exc).__name__}: {exc}" continue raw[det.name] = s norm[det.name] = normalize_scores(s) if not norm: return np.zeros(len(X)), raw, norm M = np.stack([norm[k] for k in norm], axis=1) strategy = self.strategy if strategy == "majority_vote": votes = (M >= self.threshold).astype(float) ensemble = votes.mean(axis=1) elif strategy == "meta_classifier" and supervision is not None and supervision.sum() >= 4: rng = np.random.default_rng(self.seed) idx = rng.permutation(len(M)) half = len(M) // 2 tr, te = idx[:half], idx[half:] clf = LogisticRegression(max_iter=500) ensemble = np.zeros(len(M)) try: clf.fit(M[tr], supervision[tr]) ensemble[te] = clf.predict_proba(M[te])[:, 1] clf2 = LogisticRegression(max_iter=500).fit(M[te], supervision[te]) ensemble[tr] = clf2.predict_proba(M[tr])[:, 1] except Exception: ensemble = M.mean(axis=1) else: weights = np.array([DEFAULT_WEIGHTS.get(k, 1.0) for k in norm]) ensemble = (M * weights).sum(axis=1) / weights.sum() return ensemble, raw, norm