def make_dataset(n_pos=80, n_neg=8000, seed=0):
rng = np.random.default_rng(seed)
X_pos = rng.normal(loc=[2.0, 2.0], scale=0.4, size=(n_pos, 2))
X_neg = rng.normal(loc=[-1.0, -1.0], scale=1.5, size=(n_neg, 2))
X = np.vstack([X_pos, X_neg]).astype(np.float32)
y = np.hstack([np.ones(n_pos), np.zeros(n_neg)]).astype(np.float32)
perm = rng.permutation(len(X))
return torch.tensor(X[perm]), torch.tensor(y[perm])
Xtr, ytr = make_dataset(seed=1)
Xva, yva = make_dataset(seed=2)
print(f"train : {len(Xtr)} samples, {int(ytr.sum())} positifs ({100*ytr.mean():.2f}%)")
print(f"val : {len(Xva)} samples, {int(yva.sum())} positifs ({100*yva.mean():.2f}%)")
class MLP(nn.Module):
def __init__(self):
super().__init__()
self.net = nn.Sequential(
nn.Linear(2, 32), nn.ReLU(),
nn.Linear(32, 1)
)
def forward(self, x):
return self.net(x).squeeze(-1)
def loss_share_per_class(model, X, y, loss_kind, gamma=2.0, alpha=0.5):
"""Part de la loss totale imputable aux positifs vs negatifs (batch).
Pour chaque classe c : mean(loss_c) * |c| = perte non-réduite cumulée
sur la classe. Total = somme sur les deux classes. Le partage par
classe est (mean(loss_c) * |c|) / total — il reflète la
**contribution absolue** de chaque classe à la loss totale,
pondérée par la cardinalité. Avec 80 positifs et 8000 négatifs,
un partage 50/50 signifierait mean(loss_pos) ≈ 100 * mean(loss_neg),
ce qui est l'information recherchée.
"""
model.eval()
X = X.to(DEVICE); y = y.to(DEVICE)
idx_pos = (y > 0.5).nonzero(as_tuple=True)[0]
idx_neg = (y < 0.5).nonzero(as_tuple=True)[0]
n_pos = len(idx_pos)
n_neg = len(idx_neg)
with torch.no_grad():
if loss_kind == "bce":
mean_pos = float(F.binary_cross_entropy_with_logits(model(X[idx_pos]), y[idx_pos]).item())
mean_neg = float(F.binary_cross_entropy_with_logits(model(X[idx_neg]), y[idx_neg]).item())
else:
mean_pos = float(focal_loss(model(X[idx_pos]), y[idx_pos], gamma=gamma, alpha=alpha).item())
mean_neg = float(focal_loss(model(X[idx_neg]), y[idx_neg], gamma=gamma, alpha=alpha).item())
sum_pos = mean_pos * n_pos
sum_neg = mean_neg * n_neg
total = sum_pos + sum_neg
return sum_pos, sum_neg, (sum_pos / total if total > 0 else 0.0)
def train_one(loss_kind, epochs=30, lr=1e-2, gamma=2.0, alpha=0.5):
torch.manual_seed(42)
model = MLP().to(DEVICE)
opt = torch.optim.Adam(model.parameters(), lr=lr)
hist_loss, hist_acc, hist_grad_norm = [], [], []
for ep in range(epochs):
model.train()
opt.zero_grad()
logits = model(Xtr.to(DEVICE))
if loss_kind == "bce":
loss = F.binary_cross_entropy_with_logits(logits, ytr.to(DEVICE))
else:
loss = focal_loss(logits, ytr.to(DEVICE), gamma=gamma, alpha=alpha)
loss.backward()
grad_sq = 0.0
for p_ in model.parameters():
if p_.grad is not None:
grad_sq += float(p_.grad.detach().pow(2).sum())
hist_grad_norm.append(grad_sq ** 0.5)
opt.step()
hist_loss.append(float(loss.detach()))
model.eval()
with torch.no_grad():
preds = (torch.sigmoid(model(Xva.to(DEVICE))) > 0.5).float()
tp = (preds * yva.to(DEVICE)).sum().item()
npos = int(yva.sum().item())
recall_pos = tp / max(npos, 1)
acc = (preds == yva.to(DEVICE)).float().mean().item()
hist_acc.append((acc, recall_pos))
return hist_loss, hist_acc, hist_grad_norm, model
t0 = time.time()
bce_loss, bce_acc, bce_grad_norm, model_bce = train_one("bce", epochs=30)
fl_loss, fl_acc, fl_grad_norm, model_fl = train_one("focal", epochs=30, gamma=2.0, alpha=0.5)
print(f"2 entraînements × 30 époques en {time.time() - t0:.1f} s")
bce_lp, bce_ln, bce_lshare = loss_share_per_class(model_bce, Xtr, ytr, "bce")
fl_lp, fl_ln, fl_lshare = loss_share_per_class(model_fl, Xtr, ytr, "focal")
print(f"\nLoss share (modèles entraînés) : BCE pos = {100*bce_lshare:.1f}% Focal pos = {100*fl_lshare:.1f}%")
# Métrique gradient par classe — version honnête (cardinalité-pondéré + cosine).
# L'ancien `g_pos² / (g_pos² + g_neg²)` n'était PAS une part additive : la
# décomposition `||w_pos g_pos + w_neg g_neg||²` contient un cross term
# `2 w_pos w_neg <g_pos, g_neg>` qui peut être ± et faire basculer la "share".
# On reporte donc les deux énergies cardinalité-pondérées + la cosine similarité.
def _flat_grad(model):
grads = [p_.grad.detach().reshape(-1) for p_ in model.parameters() if p_.grad is not None]
return torch.cat(grads) if grads else torch.zeros(1)
def grad_class_metric(model, X, y, objective):
"""||w_pos g_pos||², ||w_neg g_neg||², cosine(g_pos, g_neg).
Mesure cardinalité-pondérée : w_c = |c| / n_total. Le cross term
`2 w_pos w_neg <g_pos, g_neg>` est explicitement hors du print car
il peut basculer le signe de la "share".
"""
X = X.to(DEVICE); y = y.to(DEVICE)
idx_pos = (y > 0.5).nonzero(as_tuple=True)[0]
idx_neg = (y < 0.5).nonzero(as_tuple=True)[0]
n_pos, n_neg = len(idx_pos), len(idx_neg)
n_total = n_pos + n_neg
w_pos = n_pos / n_total
w_neg = n_neg / n_total
for p_ in model.parameters():
if p_.grad is not None:
p_.grad = None
if n_pos > 0:
objective(model(X[idx_pos]), y[idx_pos]).backward()
g_pos = _flat_grad(model)
else:
g_pos = torch.zeros(0)
for p_ in model.parameters():
if p_.grad is not None:
p_.grad = None
if n_neg > 0:
objective(model(X[idx_neg]), y[idx_neg]).backward()
g_neg = _flat_grad(model)
else:
g_neg = torch.zeros(0)
g_pos_norm = float(g_pos.norm().item()) if g_pos.numel() > 0 else 0.0
g_neg_norm = float(g_neg.norm().item()) if g_neg.numel() > 0 else 0.0
e_pos = (w_pos * g_pos_norm) ** 2
e_neg = (w_neg * g_neg_norm) ** 2
if g_pos.numel() > 0 and g_neg.numel() > 0:
cos = float(torch.nn.functional.cosine_similarity(
g_pos.unsqueeze(0), g_neg.unsqueeze(0)
).item())
else:
cos = 0.0
return e_pos, e_neg, cos
bce_e_pos, bce_e_neg, bce_cos = grad_class_metric(
model_bce, Xtr, ytr, lambda l, t: F.binary_cross_entropy_with_logits(l, t))
fl_e_pos, fl_e_neg, fl_cos = grad_class_metric(
model_fl, Xtr, ytr, lambda l, t: focal_loss(l, t, gamma=2.0, alpha=0.5))
print(f"\nGradient par classe (cardinalité-pondéré, modèles entraînés) :")
print(f" BCE : ||w_pos g_pos||² = {bce_e_pos:.6f} ||w_neg g_neg||² = {bce_e_neg:.6f} cos(g_pos, g_neg) = {bce_cos:+.3f}")
print(f" FL : ||w_pos g_pos||² = {fl_e_pos:.6f} ||w_neg g_neg||² = {fl_e_neg:.6f} cos(g_pos, g_neg) = {fl_cos:+.3f}")
print(f" Note : cross term `2 w_pos w_neg <g_pos, g_neg>` est ± — `pos share` additive n existe pas.")
print(f"Grad norm epoch 0 : BCE = {bce_grad_norm[0]:.3f} Focal = {fl_grad_norm[0]:.3f}")
print(f"Grad norm epoch 29 : BCE = {bce_grad_norm[-1]:.3f} Focal = {fl_grad_norm[-1]:.3f}")
fig, axes = plt.subplots(1, 3, figsize=(15, 3.5))
axes[0].plot(bce_loss, label="BCE", color="indianred")
axes[0].plot(fl_loss, label="Focal (γ=2, α=0.5)", color="seagreen")
axes[0].set_xlabel("époque"); axes[0].set_ylabel("loss train")
axes[0].set_title("Convergence de la loss"); axes[0].legend(); axes[0].grid(alpha=0.3)
axes[1].plot([a[1] for a in bce_acc], label="BCE — rappel pos", color="indianred")
axes[1].plot([a[1] for a in fl_acc], label="Focal — rappel pos", color="seagreen")
axes[1].set_xlabel("époque"); axes[1].set_ylabel("rappel sur les positifs (val)")
axes[1].set_title("Détection des positifs — l'écart attendu")
axes[1].legend(); axes[1].grid(alpha=0.3)
axes[2].plot(bce_grad_norm, label="BCE", color="indianred")
axes[2].plot(fl_grad_norm, label="Focal", color="seagreen")
axes[2].set_yscale("log")
axes[2].set_xlabel("époque"); axes[2].set_ylabel("||grad|| (log)")
axes[2].set_title("Norme du gradient par époque (log)")
axes[2].legend(); axes[2].grid(alpha=0.3, which="both")
plt.tight_layout(); plt.show()
print(f"\nRappel final sur les positifs : BCE = {bce_acc[-1][1]:.2f} | Focal = {fl_acc[-1][1]:.2f}")
print(f"Accuracy globale finale : BCE = {bce_acc[-1][0]:.3f} | Focal = {fl_acc[-1][0]:.3f}")
