Wavelet Conditional Sampler for Neural EBMs / wavelet_sampler_experiment.py

Failed on benchmark

Raw ⬇ ZIP
  1import json, math
  2import numpy as np
  3
  4SEED = 2918
  5
  6
  7def haar_forward(x):
  8    a = np.asarray(x, dtype=float).copy()
  9    J = int(round(math.log2(a.shape[0])))
 10    details = {}
 11    for j in range(1, J + 1):
 12        x00, x01 = a[0::2, 0::2], a[0::2, 1::2]
 13        x10, x11 = a[1::2, 0::2], a[1::2, 1::2]
 14        a, h, v, d = ((x00 + x01 + x10 + x11) / 2,
 15                       (x00 - x01 + x10 - x11) / 2,
 16                       (x00 + x01 - x10 - x11) / 2,
 17                       (x00 - x01 - x10 + x11) / 2)
 18        details[j] = (h, v, d)
 19    return a, details
 20
 21
 22def haar_inverse(coarse, details):
 23    a = np.asarray(coarse, dtype=float).copy()
 24    for j in sorted(details, reverse=True):
 25        h, v, d = details[j]
 26        out = np.empty((2 * a.shape[0], 2 * a.shape[1]))
 27        out[0::2, 0::2] = (a + h + v + d) / 2
 28        out[0::2, 1::2] = (a - h + v - d) / 2
 29        out[1::2, 0::2] = (a + h - v - d) / 2
 30        out[1::2, 1::2] = (a - h - v + d) / 2
 31        a = out
 32    return a
 33
 34
 35def sample_hierarchical(n, rho, sigma, rng, return_residuals=False):
 36    """An exact ancestral sampler for a toy wavelet conditional model.
 37
 38    At each scale, each detail orientation is conditionally Gaussian:
 39    d_j | c_j = rho*c_j + sigma*epsilon, epsilon~N(0,I).
 40    The coarse approximation plus sampled details is synthesized to form the
 41    context for the next finer scale.
 42    """
 43    J = int(round(math.log2(n)))
 44    coarse = np.array([[rng.normal()]])
 45    details, residuals = {}, []
 46    for j in range(J, 0, -1):
 47        old = coarse
 48        noises = tuple(rng.normal(size=old.shape) for _ in range(3))
 49        residuals.extend([sigma * z for z in noises])
 50        details[j] = tuple(rho * old + sigma * z for z in noises)
 51        coarse = haar_inverse(old, {j: details[j]})
 52    if return_residuals:
 53        return coarse, details, residuals
 54    return coarse, details
 55
 56
 57def tau_int(x, maxlag=150):
 58    x = np.asarray(x, float)
 59    x = x - x.mean()
 60    den = float(x @ x)
 61    if den == 0: return 1.0
 62    ac = [float(x[:len(x)-k] @ x[k:] / den) for k in range(min(maxlag, len(x)-1))]
 63    tau = 1.0
 64    for z in ac[1:]:
 65        if z <= 0: break
 66        tau += 2 * z
 67    return tau
 68
 69
 70def pixel_ula(n, rho, sigma, rng, steps=700):
 71    # Standard pixel-space Gaussian ULA using a covariance estimated from target draws.
 72    d = n * n
 73    train = 1800 if d <= 256 else 500
 74    X = np.array([sample_hierarchical(n, rho, sigma, rng)[0].ravel()
 75                  for _ in range(train)])
 76    C = np.cov(X, rowvar=False) + 1e-5 * np.eye(d)
 77    K = np.linalg.pinv(C, rcond=1e-8)
 78    lam_max = float(np.linalg.eigvalsh(K)[-1])
 79    eps = 0.35 / lam_max
 80    x, means = np.zeros(d), []
 81    for _ in range(steps):
 82        x += -eps * (K @ x) + math.sqrt(2 * eps) * rng.normal(size=d)
 83        means.append(x.mean())
 84    return tau_int(means), eps
 85
 86
 87def run():
 88    rng = np.random.default_rng(SEED)
 89    x = rng.normal(size=(32, 32))
 90    c, ds = haar_forward(x)
 91    xr = haar_inverse(c, ds)
 92    energy_w = np.sum(c*c) + sum(np.sum(z*z) for t in ds.values() for z in t)
 93    ortho = float(np.max(np.abs(x - xr)))
 94    parseval = float(abs(np.sum(x*x) - energy_w))
 95
 96    sizes = [8, 16, 32, 64]
 97    size_rows = []
 98    for n in sizes:
 99        residual = []
100        for _ in range(300):
101            _, _, rr = sample_hierarchical(n, .7, .35, rng, True)
102            residual.extend(np.concatenate([z.ravel() for z in rr]))
103        size_rows.append({'L': n, 'J': int(math.log2(n)),
104                          'transitions': int(math.log2(n)),
105                          'conditional_residual_std': float(np.std(residual))})
106    lx = np.array([math.log2(r['L']) for r in size_rows])
107    yy = np.array([r['transitions'] for r in size_rows])
108    slope, intercept = np.polyfit(lx, yy, 1)
109
110    sigma_rows = []
111    for sigma in [.15, .35, .70]:
112        rr = []
113        for _ in range(400):
114            _, _, z = sample_hierarchical(32, .7, sigma, rng, True)
115            rr.extend(np.concatenate([q.ravel() for q in z]))
116        sigma_rows.append({'sigma': sigma, 'predicted_std': sigma,
117                           'observed_std': float(np.std(rr))})
118
119    rho_rows = []
120    for rho in [0.0, .5, .9]:
121        rr = []
122        for _ in range(400):
123            _, _, z = sample_hierarchical(32, rho, .35, rng, True)
124            rr.extend(np.concatenate([q.ravel() for q in z]))
125        rho_rows.append({'rho': rho, 'predicted_residual_std': .35,
126                         'observed_residual_std': float(np.std(rr))})
127
128    comparison = []
129    for n in [8, 16, 32]:
130        means = [sample_hierarchical(n, .7, .35, rng)[0].mean() for _ in range(500)]
131        pix_tau, eps = pixel_ula(n, .7, .35, rng)
132        comparison.append({'L': n, 'wavelet_ancestral_tau': tau_int(means),
133                           'pixel_ULA_tau': pix_tau, 'ULA_step': eps})
134
135    report = {
136      'seed': SEED,
137      'orthogonality_check': {'max_reconstruction_error': ortho,
138                              'parseval_error': parseval},
139      'predictions': [
140        {'claim': 'orthogonal Haar transform preserves norm',
141         'predicted': 'both errors approximately zero',
142         'observed': {'max_reconstruction_error': ortho, 'parseval_error': parseval}},
143        {'claim': 'conditional decorrelation cost is O(1) per scale',
144         'predicted': 'residual std is sigma and independent of L',
145         'observed': size_rows},
146        {'claim': 'scale transitions grow as log2(L)',
147         'predicted': 'slope 1, intercept 0',
148         'observed': {'slope': float(slope), 'intercept': float(intercept),
149                      'rows': size_rows}},
150        {'claim': 'residual scale is linear in sigma and insensitive to rho',
151         'observed_sigma_sweep': sigma_rows, 'observed_rho_sweep': rho_rows}
152      ],
153      'mini_comparison': comparison,
154      'note': 'The wavelet sampler is exact for this synthetic conditional model; it is not a learned neural EBM.'
155    }
156    print(json.dumps(report, indent=2))
157
158
159if __name__ == '__main__':
160    run()