File size: 6,741 Bytes
acc1556 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 | """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())
|