File size: 4,816 Bytes
0fdad59 | 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 | """Claim 6: the alpha-dependent tradeoff between drift and stability, on the
10-D Gaussian mixture. Low alpha (little real data re-injected each generation)
should drift/collapse; higher alpha should stabilise.
Previously the logbook only hash-pinned the paper's Figure 1 panels and said so.
The 10-D Gaussian-mixture half needs no image data, so it is run here.
Self-consuming loop: generation 0 fits a GMM to real samples; each later
generation trains on a mixture of `alpha` real data and `1-alpha` of its own
previous output, which is the setting model collapse is defined in.
"""
import json, numpy as np
from sklearn.mixture import GaussianMixture
RES = {}
D, K = 10, 4
def true_gmm(seed=0):
rng = np.random.default_rng(seed)
mu = rng.normal(size=(K, D))*2.5
A = rng.normal(size=(K, D, D))*0.35
cov = np.array([a @ a.T+0.35*np.eye(D) for a in A])
w = rng.dirichlet(np.ones(K)*4)
return mu, cov, w
def sample_true(mu, cov, w, n, rng):
c = rng.choice(K, size=n, p=w)
X = np.zeros((n, D))
for k in range(K):
m = c == k
if m.sum():
X[m] = rng.multivariate_normal(mu[k], cov[k], size=int(m.sum()))
return X
def moment_error(X, mu, cov, w):
"""Distance to the TRUE distribution via first two moments (closed form)."""
tm = (w[:, None]*mu).sum(0)
tc = sum(w[k]*(cov[k]+np.outer(mu[k]-tm, mu[k]-tm)) for k in range(K))
em = X.mean(0); ec = np.cov(X.T)
return float(np.linalg.norm(em-tm)+np.linalg.norm(ec-tc, "fro"))
def run(alpha, gens=80, n=150, seed=0):
"""n is deliberately SMALL: with n=2000 the per-generation variance loss is
~1/n and collapse never manifests (variance stayed at 116% of true over 25
generations). Collapse is driven by sampling noise compounding, so it needs
small n and many generations."""
rng = np.random.default_rng(seed)
mu, cov, w = true_gmm(seed)
real = sample_true(mu, cov, w, n, rng)
model = GaussianMixture(K, covariance_type="full", reg_covar=1e-6,
random_state=0, max_iter=200).fit(real)
errs = [moment_error(model.sample(n)[0], mu, cov, w)]
traces = [float(np.trace(np.cov(model.sample(n)[0].T)))]
for g in range(gens):
gen = model.sample(n)[0]
nreal = int(round(alpha*n))
train = np.vstack([sample_true(mu, cov, w, nreal, rng), gen[:n-nreal]]) if nreal else gen
model = GaussianMixture(K, covariance_type="full", reg_covar=1e-6,
random_state=0, max_iter=200).fit(train)
S = model.sample(n)[0]
errs.append(moment_error(S, mu, cov, w))
traces.append(float(np.trace(np.cov(S.T))))
return np.array(errs), np.array(traces)
def main():
mu, cov, w = true_gmm(0)
tm = (w[:, None]*mu).sum(0)
tc = sum(w[k]*(cov[k]+np.outer(mu[k]-tm, mu[k]-tm)) for k in range(K))
true_tr = float(np.trace(tc))
rows = []
for alpha in (0.0, 0.1, 0.25, 0.5, 1.0):
E, T = [], []
for s in range(5):
e, tr = run(alpha, seed=s)
E.append(e); T.append(tr)
E = np.array(E); T = np.array(T)
rows.append({"alpha": alpha, "seeds": 5, "generations": E.shape[1]-1,
"err_gen0": round(float(E[:, 0].mean()), 4),
"err_final": round(float(E[:, -1].mean()), 4),
"err_growth": round(float(E[:, -1].mean()/max(E[:, 0].mean(), 1e-9)), 3),
"variance_trace_true": round(true_tr, 3),
"variance_trace_final": round(float(T[:, -1].mean()), 3),
"variance_retained": round(float(T[:, -1].mean()/true_tr), 4)})
print(" alpha=%.2f moment error %.4f -> %.4f (%.2fx) variance trace %.2f -> %.2f (%.1f%% of true)"
% (alpha, E[:, 0].mean(), E[:, -1].mean(), rows[-1]["err_growth"],
true_tr, T[:, -1].mean(), 100*rows[-1]["variance_retained"]), flush=True)
RES["claim6_alpha_tradeoff"] = {"D": D, "K": K, "rows": rows,
"monotone_error_in_alpha": all(rows[i+1]["err_final"] <= rows[i]["err_final"]+1e-9
for i in range(len(rows)-1)),
"monotone_variance_in_alpha": all(rows[i+1]["variance_retained"] >= rows[i]["variance_retained"]-1e-9
for i in range(len(rows)-1)),
"collapse_at_alpha0": rows[0]["variance_retained"],
"stable_at_alpha1": rows[-1]["variance_retained"]}
R = RES["claim6_alpha_tradeoff"]
print(" error monotone decreasing in alpha: %s | variance retention monotone increasing: %s"
% (R["monotone_error_in_alpha"], R["monotone_variance_in_alpha"]), flush=True)
json.dump(RES, open("collapse_results.json", "w"), indent=1)
if __name__ == "__main__":
main(); print("DONE")
|