Calibrated Compact-Support Anomaly Score / compact_anomaly.py

Failed on benchmark

Raw ⬇ ZIP
  1"""Calibrated compact-support anomaly score and numerical verification."""
  2import json
  3from pathlib import Path
  4import numpy as np
  5from scipy.special import gammaln
  6from scipy.integrate import quad
  7from sklearn.metrics import roc_auc_score
  8
  9
 10def parameters(d, R2):
 11    if R2 <= d + 2:
 12        raise ValueError("R2 must exceed d+2")
 13    gamma = 2.0 / (R2 - d - 2.0)
 14    a = 1.0 / gamma
 15    logC = (gammaln(d / 2 + 1 + a) - d / 2 * np.log(np.pi)
 16            - d / 2 * np.log(R2) - gammaln(1 + a))
 17    return gamma, a, logC
 18
 19
 20def compact_score(X, mu, cov, R2, barrier=False):
 21    """Negative log density; outside support is +inf (or finite training barrier)."""
 22    X = np.asarray(X)
 23    L = np.linalg.cholesky(cov)
 24    v = np.linalg.solve(L, (X - mu).T).T
 25    r2 = np.sum(v * v, axis=1)
 26    gamma, a, logC = parameters(X.shape[1], R2)
 27    u = 1.0 - r2 / R2
 28    if barrier:
 29        # Stable surrogate specified in the idea for optimization.
 30        return -logC - a * np.log(np.maximum(u, 1e-6)) + np.logaddexp(0., r2 - R2), r2
 31    score = np.full(len(X), np.inf)
 32    inside = r2 < R2
 33    score[inside] = -logC - a * np.log1p(-r2[inside] / R2)
 34    return score, r2
 35
 36
 37def gaussian_score(X, mu, cov):
 38    L = np.linalg.cholesky(cov)
 39    v = np.linalg.solve(L, (X - mu).T).T
 40    return np.sum(v * v, axis=1)
 41
 42
 43def sample_compact(rng, n, d, R2):
 44    """Exact sampler: r2/R2 ~ Beta(d/2, 1+1/gamma), direction uniform."""
 45    _, a, _ = parameters(d, R2)
 46    y = rng.beta(d / 2, a + 1, size=n)
 47    direction = rng.normal(size=(n, d))
 48    direction /= np.linalg.norm(direction, axis=1, keepdims=True)
 49    return direction * np.sqrt(R2 * y)[:, None]
 50
 51
 52def verify(d=5, seed=1013):
 53    rng = np.random.default_rng(seed)
 54    radii = [d + 2.5, d + 6, d + 20, d + 80]
 55    rows = []
 56    # Predictions: direct normalization is 1 and calibrated E[r^2] is d.
 57    for R2 in radii:
 58        gamma, a, logC = parameters(d, R2)
 59        z = sample_compact(rng, 250000, d, R2)
 60        r2 = np.sum(z*z, axis=1)
 61        surface = 2 * np.pi ** (d / 2) / np.exp(gammaln(d / 2))
 62        integral = quad(lambda t: surface * t ** (d - 1) * np.exp(logC)
 63                        * (1 - t * t / R2) ** a, 0, np.sqrt(R2),
 64                        epsabs=1e-10)[0]
 65        rows.append({"R2": R2, "gamma": gamma, "a": a,
 66                     "mean_r2": float(r2.mean()), "target_d": d,
 67                     "max_r2": float(r2.max()),
 68                     "normalization": float(integral),
 69                     "predicted_boundary_slope": -a})
 70    # Prediction: score near boundary has slope -1/gamma against log(epsilon).
 71    boundary_sweep = []
 72    eps = np.logspace(-2, -10, 9)
 73    for R2 in radii:
 74        _, a, logC = parameters(d, R2)
 75        x = np.sqrt(R2 * (1 - eps))[:, None] * np.array([[1.] + [0.] * (d - 1)])
 76        scores, _ = compact_score(x, np.zeros(d), np.eye(d), R2)
 77        observed = np.polyfit(np.log(eps), scores + logC, 1)[0]
 78        boundary_sweep.append({"R2": R2, "predicted_slope": -a,
 79                               "observed_slope": float(observed),
 80                               "abs_error": float(abs(observed + a))})
 81    R2 = d + 10
 82    outside = compact_score(np.array([[np.sqrt(R2) * 1.001] + [0.]*(d-1)]),
 83                            np.zeros(d), np.eye(d), R2)[0][0]
 84    rows_boundary = {"sweep": boundary_sweep,
 85                     "outside_is_infinite": bool(np.isinf(outside))}
 86    # Affine geometry prediction.
 87    A = np.array([[1.5, .2, 0, 0, 0], [.1, .8, .1, 0, 0],
 88                  [0, .1, 1.2, .1, 0], [0, 0, .1, .9, .2],
 89                  [0, 0, 0, .1, 1.1]])
 90    z = sample_compact(rng, 2000, d, d + 10)
 91    cov = A @ A.T
 92    _, r_orig = compact_score(z, np.zeros(d), np.eye(d), d + 10)
 93    _, r_aff = compact_score(z @ A.T, np.zeros(d), cov, d + 10)
 94    affine_err = float(np.max(np.abs(r_orig-r_aff)))
 95    # Secondary equal-feature OOD comparison.
 96    nominal = sample_compact(rng, 4000, d, d + 10)
 97    anomalies = rng.normal(size=(4000, d)) + np.array([2.2, 0, 0, 0, 0])
 98    X = np.vstack([nominal, anomalies])
 99    y = np.r_[np.zeros(len(nominal)), np.ones(len(anomalies))]
100    gs = gaussian_score(X, np.zeros(d), np.eye(d))
101    cs, _ = compact_score(X, np.zeros(d), np.eye(d), d + 10)
102    cs_rank = np.where(np.isinf(cs), np.max(np.where(np.isfinite(cs), cs, 0)) + 1e6, cs)
103    auc_g = roc_auc_score(y, gs); auc_c = roc_auc_score(y, cs_rank)
104    qg = np.quantile(gs[:len(nominal)], .95); qc = np.quantile(cs_rank[:len(nominal)], .95)
105    fpr_g = np.mean(gs[len(nominal):] <= qg); fpr_c = np.mean(cs_rank[len(nominal):] <= qc)
106    return {"calibration_rows": rows, "boundary": rows_boundary,
107            "affine_max_abs_radius_error": affine_err,
108            "ood": {"gaussian_auc": float(auc_g), "compact_auc": float(auc_c),
109                    "gaussian_anomaly_accept_rate_at_95pct_nominal": float(fpr_g),
110                    "compact_anomaly_accept_rate_at_95pct_nominal": float(fpr_c)}}
111
112if __name__ == "__main__":
113    out = verify()
114    Path("results.json").write_text(json.dumps(out, indent=2))
115    print(json.dumps(out, indent=2))