| """Multi-view decision fusion: fixed poolers + AIP mechanisms. |
| |
| A physical sample gets one anomaly score per camera view. The Real-IAD paper |
| protocol integrates views with max pooling. We treat the 5 per-view scores as |
| reports from a swarm of agents and compare the official max against |
| Byzantine-robust poolers and two AIP ("liars are information") mechanisms: |
| |
| aip_d(alpha): dissent amplification. |
| score = max + alpha * max(0, max - median) |
| The dissenting view (potential "liar") is inverted into information: |
| the larger its gap over the swarm consensus (median), the more the |
| extreme report is trusted. |
| |
| aip_c(beta): consensus shrinkage. |
| score = max - beta * median |
| When even the most extreme report coexists with a strongly normal |
| consensus, the sample's score is shrunk; pure outlier views survive. |
| |
| aip_soft(kappa): disagreement-weighted soft pooling. |
| w_v ~ exp(kappa * z_v), z = standardized scores; score = sum w_v s_v. |
| Interpolates mean (kappa=0) and max (kappa -> inf). |
| """ |
| from __future__ import annotations |
|
|
| import numpy as np |
|
|
|
|
| def _check(s): |
| s = np.asarray(s, dtype=np.float64) |
| return s |
|
|
|
|
| def f_max(s): |
| return float(np.max(s)) |
|
|
|
|
| def f_mean(s): |
| return float(np.mean(s)) |
|
|
|
|
| def f_median(s): |
| return float(np.median(s)) |
|
|
|
|
| def f_top2(s): |
| return float(np.mean(np.sort(s)[-2:])) |
|
|
|
|
| def f_trimmed(s): |
| k = max(1, len(s) // 5) |
| return float(np.mean(np.sort(s)[k:len(s) - k])) if len(s) - 2 * k > 0 else f_median(s) |
|
|
|
|
| def aip_d(s, alpha=0.5): |
| s = _check(s) |
| return float(np.max(s) + alpha * max(0.0, np.max(s) - np.median(s))) |
|
|
|
|
| def aip_c(s, beta=0.3): |
| s = _check(s) |
| return float(np.max(s) - beta * np.median(s)) |
|
|
|
|
| def aip_soft(s, kappa=2.0): |
| s = _check(s) |
| mu, sd = np.mean(s), max(np.std(s), 1e-6) |
| z = (s - mu) / sd |
| w = np.exp(kappa * (z - z.max())) |
| w = w / w.sum() |
| return float(np.dot(w, s)) |
|
|
|
|
| def aip_mt(s, gamma=0.5): |
| """Consensus-anchored dissent blend (the 'liar informs the swarm' rule). |
| |
| score = (1 - gamma) * mean + gamma * top1 |
| The swarm consensus (mean) anchors the decision; the outlier agent's |
| report is mixed in proportionally to a fixed trust gamma. gamma=0 -> mean, |
| gamma=1 -> max. |
| """ |
| s = _check(s) |
| return float((1 - gamma) * np.mean(s) + gamma * np.max(s)) |
|
|
|
|
| def aip_wt(s, gamma=0.5): |
| """Winsorized dissent blend: outlier evidence enters via the top-2 mean, |
| which is robust to a single noisy 'liar' report.""" |
| s = _check(s) |
| return float((1 - gamma) * np.mean(s) + gamma * np.mean(np.sort(s)[-2:])) |
|
|
|
|
| |
| STRATEGIES = { |
| "max": f_max, |
| "mean": f_mean, |
| "median": f_median, |
| "top2": f_top2, |
| "trimmed": f_trimmed, |
| "aip_d05": lambda s: aip_d(s, alpha=0.5), |
| "aip_c03": lambda s: aip_c(s, beta=0.3), |
| "aip_soft2": lambda s: aip_soft(s, kappa=2.0), |
| "aip_mt01": lambda s: aip_mt(s, gamma=0.1), |
| "aip_mt025": lambda s: aip_mt(s, gamma=0.25), |
| "aip_mt05": lambda s: aip_mt(s, gamma=0.5), |
| "aip_mt075": lambda s: aip_mt(s, gamma=0.75), |
| "aip_wt05": lambda s: aip_wt(s, gamma=0.5), |
| "aip_wt075": lambda s: aip_wt(s, gamma=0.75), |
| } |
| STRATEGY_NAMES = list(STRATEGIES.keys()) |
| N_ACTIONS = len(STRATEGY_NAMES) |
|
|
|
|
| def fuse_all(sample_view_scores): |
| """sample_view_scores: [n_samples, V] -> dict name -> [n_samples] scores.""" |
| out = {name: np.empty(len(sample_view_scores)) for name in STRATEGY_NAMES} |
| for i, s in enumerate(np.asarray(sample_view_scores, dtype=np.float64)): |
| for name, fn in STRATEGIES.items(): |
| out[name][i] = fn(s) |
| return out |
|
|
|
|
| def state_features(sample_view_scores): |
| """Per-sample swarm statistics that condition the arbiter policy. |
| |
| [n, 8]: sorted view scores (s1, s5), gap s1-s2, s1-s3, mean, std, |
| max/median ratio, dispersion (std/mean). |
| """ |
| X = np.asarray(sample_view_scores, dtype=np.float64) |
| n = X.shape[0] |
| srt = np.sort(X, axis=1) |
| mu = X.mean(axis=1) |
| sd = X.std(axis=1) + 1e-9 |
| med = np.median(X, axis=1) |
| return np.stack([ |
| srt[:, -1], srt[:, 0], srt[:, -1] - srt[:, -2], srt[:, -1] - srt[:, -3], |
| mu, sd, srt[:, -1] / (np.abs(med) + 1e-9), sd / (np.abs(mu) + 1e-9), |
| ], axis=1).reshape(n, 8) |
|
|
