| """GRPO-style reinforcement-learning arbiter for multi-view fusion. |
| |
| The arbiter observes swarm statistics of a sample's per-view scores (state) |
| and outputs a policy over the pooling strategies (actions). Training follows |
| the group-relative idea of GRPO: for every episode (a mini-batch of samples), |
| ALL K candidate poolings are evaluated, their rewards (batch AUROC of each |
| strategy) are standardized *within the group of strategies* to form |
| advantages, and the softmax policy is updated with REINFORCE toward |
| high-advantage actions, with weight decay acting as entropy regularization. |
| |
| Final decision = policy-gated mixture of per-strategy anomaly scores, where |
| each strategy's score is category-normalized with label-free statistics |
| (median / IQR over normal calibration samples) so mixing is meaningful. |
| """ |
| from __future__ import annotations |
|
|
| import numpy as np |
|
|
| from .fusion import N_ACTIONS, STRATEGY_NAMES, STRATEGIES, fuse_all, state_features |
| from .evalmetrics import auroc |
|
|
|
|
| class CalibNormalizer: |
| """Label-free per-(category, strategy) score normalization + thresholds.""" |
|
|
| def fit(self, scores_by_cat, labels_by_cat): |
| """scores_by_cat: cat -> {strategy: [n]} (calibration split).""" |
| self.med, self.iqr, self.thr = {}, {}, {} |
| for cat, per in scores_by_cat.items(): |
| y = labels_by_cat[cat] |
| for name, s in per.items(): |
| ok = s[y == 0] |
| med = np.median(ok) |
| iqr = np.subtract(*np.percentile(ok, [75, 25])) + 1e-9 |
| self.med[(cat, name)] = med |
| self.iqr[(cat, name)] = iqr |
| |
| zn = (s - med) / iqr |
| cand = np.unique(zn) |
| if len(cand) > 200: |
| cand = np.quantile(zn, np.linspace(0, 1, 200)) |
| best_t, best_b = 0.5, -1 |
| for t in cand: |
| pred = zn > t |
| tpr = pred[y == 1].mean() if (y == 1).any() else 0 |
| tnr = 1 - pred[y == 0].mean() |
| b = 0.5 * (tpr + tnr) |
| if b > best_b: |
| best_b, best_t = b, t |
| self.thr[(cat, name)] = float(best_t) |
| return self |
|
|
| def normalize(self, cat, name, s): |
| return (np.asarray(s) - self.med[(cat, name)]) / self.iqr[(cat, name)] |
|
|
|
|
| def _normalized_matrix(view_scores, cats, normalizer): |
| """[n, K] category-normalized strategy score matrix.""" |
| n = len(view_scores) |
| S = np.zeros((n, N_ACTIONS)) |
| fused = fuse_all(view_scores) |
| for k, name in enumerate(STRATEGY_NAMES): |
| col = fused[name].copy() |
| for cat in np.unique(cats): |
| m = cats == cat |
| col[m] = normalizer.normalize(cat, name, col[m]) |
| S[:, k] = col |
| return S |
|
|
|
|
| class GRPOArbiter: |
| """Linear softmax policy over pooling strategies, trained with GRPO-style |
| group-relative advantages on batch ranking rewards.""" |
|
|
| def __init__(self, lr=1.0, weight_decay=0.005, seed=0): |
| self.lr = lr |
| self.weight_decay = weight_decay |
| self.rng = np.random.default_rng(seed) |
| self.W = np.zeros((N_ACTIONS, 8)) |
| self.b = np.zeros(N_ACTIONS) |
| self.mu = np.zeros(8) |
| self.sd = np.ones(8) |
| self.history = [] |
|
|
| |
| def _fit_state_scaler(self, X): |
| self.mu, self.sd = X.mean(0), X.std(0) + 1e-9 |
|
|
| def _logits(self, X): |
| Z = (X - self.mu) / self.sd |
| return Z @ self.W.T + self.b |
|
|
| def probs(self, X): |
| z = self._logits(X) |
| z = z - z.max(axis=1, keepdims=True) |
| p = np.exp(z) |
| return p / p.sum(axis=1, keepdims=True) |
|
|
| |
| def fit(self, view_scores, labels, cats, normalizer, epochs=30, batch=96, log=print): |
| """view_scores: [n, V]; labels: [n]; cats: [n] (calibration pool).""" |
| X = state_features(view_scores) |
| self._fit_state_scaler(X) |
| S = _normalized_matrix(view_scores, cats, normalizer) |
| y = np.asarray(labels) |
| n = len(X) |
| for ep in range(epochs): |
| order = self.rng.permutation(n) |
| ep_rewards, ep_mix = [], [] |
| for st in range(0, n, batch): |
| idx = order[st:st + batch] |
| yb = y[idx] |
| if len(np.unique(yb)) < 2: |
| continue |
| r = np.array([auroc(S[idx, k], yb) for k in range(N_ACTIONS)]) |
| adv = (r - r.mean()) / (r.std() + 1e-6) |
| P = self.probs(X[idx]) |
| grad_W = np.zeros_like(self.W) |
| grad_b = np.zeros_like(self.b) |
| for j, i in enumerate(idx): |
| Z = (X[i] - self.mu) / self.sd |
| p = P[j] |
| for k in range(N_ACTIONS): |
| g = adv[k] * p[k] * (np.eye(N_ACTIONS)[k] - p) |
| grad_W += np.outer(g, Z) |
| grad_b += g |
| self.W += self.lr * grad_W / len(idx) |
| self.b += self.lr * grad_b / len(idx) |
| |
| self.W *= (1 - self.weight_decay) |
| self.b *= (1 - self.weight_decay) |
| ep_rewards.append(r) |
| ep_mix.append(auroc((P * S[idx]).sum(1), yb)) |
| P_all = self.probs(X) |
| ent = -(P_all * np.log(P_all + 1e-12)).sum(1).mean() |
| self.history.append({ |
| "epoch": ep, |
| "policy_entropy": float(ent), |
| "mean_probs": P_all.mean(0).tolist(), |
| "mean_group_reward": float(np.mean(ep_rewards)), |
| "policy_mixture_auroc": float(np.mean(ep_mix)), |
| }) |
| return self |
|
|
| |
| def fused_scores(self, view_scores, cats, normalizer): |
| X = state_features(view_scores) |
| P = self.probs(X) |
| S = _normalized_matrix(view_scores, cats, normalizer) |
| return (P * S).sum(1), P, S |
|
|
|
|
| class OnlineHedge: |
| """Multiplicative-weights online expert aggregation (delayed labels).""" |
|
|
| def __init__(self, n_actions=N_ACTIONS, eta=4.0): |
| self.w = np.ones(n_actions) / n_actions |
| self.eta = eta |
| self.path = [self.w.copy()] |
|
|
| def update(self, rewards): |
| r = np.asarray(rewards) |
| r = (r - r.mean()) / (r.std() + 1e-6) |
| self.w *= np.exp(self.eta * r) |
| self.w /= self.w.sum() |
| self.path.append(self.w.copy()) |
|
|
