"""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 # few-shot threshold: best balanced accuracy on calibration 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 = [] # ---------- policy ---------- 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) # ---------- GRPO training ---------- 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) # [n, K] 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) # group-relative advantage P = self.probs(X[idx]) # [b, K] 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) # weight decay == entropy regularization for a softmax policy 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 # ---------- decision ---------- def fused_scores(self, view_scores, cats, normalizer): X = state_features(view_scores) P = self.probs(X) # [n, K] 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) # scale-invariant hedge reward self.w *= np.exp(self.eta * r) self.w /= self.w.sum() self.path.append(self.w.copy())