File size: 5,742 Bytes
515b676 | 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 | """Core harness: RDPG generation, adjacency spectral embedding, (2,inf) error.
Everything here follows the paper "On the Effect of Misspecifying the Embedding
Dimension in Low-rank Network Models" (Taing & Levin, arXiv:2601.06014),
Definitions 1-2, Lemma 2.1, Theorems 3.1-3.2 and Section 4.
"""
import warnings
import numpy as np
def matmul_sanity():
"""numpy on Apple Accelerate emits spurious divide/overflow/invalid
RuntimeWarnings from matmul on perfectly finite inputs. Before silencing
them we check a matmul against an independent reference (einsum + lstsq)."""
rng = np.random.default_rng(0)
A = rng.standard_normal((300, 120))
B = rng.standard_normal((120, 90))
with warnings.catch_warnings():
warnings.simplefilter("ignore", RuntimeWarning)
C = A @ B
ref = np.einsum("ij,jk->ik", A, B)
err_einsum = float(np.max(np.abs(C - ref)) / np.max(np.abs(ref)))
# independent route: solve A z = C column-wise, must recover B
z, *_ = np.linalg.lstsq(A, C, rcond=None)
err_lstsq = float(np.max(np.abs(z - B)))
finite = bool(np.all(np.isfinite(C)))
return {"rel_err_vs_einsum": err_einsum, "max_err_vs_lstsq": err_lstsq,
"all_finite": finite,
"pass": bool(err_einsum < 1e-12 and err_lstsq < 1e-9 and finite)}
warnings.filterwarnings("ignore", category=RuntimeWarning, message=".*encountered in matmul.*")
# ---------------------------------------------------------------- generation
def dirichlet_latent(n, r, rng, alpha=1.0):
"""Rows of X iid Dir(alpha,...,alpha). Paper Sec. 4.1 uses alpha=(1,...,1), r=5."""
return rng.dirichlet(np.full(r, alpha), size=n)
def sym_noise(n, rng, kind="normal", scale=1.0):
"""Symmetric mean-zero noise, entries iid up to symmetry (paper Sec. 4.1)."""
if kind == "normal":
Z = rng.standard_normal((n, n)) * scale
elif kind == "laplace":
# Paper (b) writes "E_ij + 1 ~ Laplace(0,1)", but immediately after states
# that only settings (c) and (d) are re-centred to have mean zero. A
# Laplace(0,1) is already mean zero, so we read (b) as E_ij ~ Laplace(0,1).
Z = rng.laplace(0.0, 1.0, (n, n)) * scale
elif kind == "exponential": # E_ij + 1 ~ Exp(1)
Z = (rng.exponential(1.0, (n, n)) - 1.0) * scale
elif kind == "poisson": # E_ij + 1 ~ Pois(1)
Z = (rng.poisson(1.0, (n, n)) - 1.0).astype(float) * scale
elif kind == "cauchy": # violates Assumption A7 (no 2nd moment)
Z = rng.standard_cauchy((n, n)) * scale
else:
raise ValueError(kind)
E = np.triu(Z)
E = E + np.triu(Z, 1).T
return E
def weighted_rdpg(n, r, rng, rho=1.0, kind="normal", scale=1.0, X=None):
"""A = rho X X^T + E (paper Eq. 14 / 16). Returns A, rho^{1/2} X."""
if X is None:
X = dirichlet_latent(n, r, rng)
A = rho * (X @ X.T) + sym_noise(n, rng, kind, scale)
return A, np.sqrt(rho) * X
def binary_rdpg(n, r, rng, rho=1.0, X=None):
"""A_ij ~ Bern(rho x_i^T x_j), i<j, hollow (paper Eq. 18)."""
if X is None:
X = dirichlet_latent(n, r, rng)
P = rho * (X @ X.T)
np.clip(P, 0.0, 1.0, out=P)
U = rng.random((n, n))
A = (np.triu(U, 1) < np.triu(P, 1)).astype(float)
A = A + A.T
return A, np.sqrt(rho) * X
def sbm(n, r, rng, p_in=0.9, p_out=0.1, alpha=1.0):
"""Paper Sec. 4.2 SBM: pi ~ Dir(1_r), B = p_out + (p_in-p_out) I,
P = Z B Z^T, X = U_{1:r} S_{1:r}^{1/2}. Returns A, X (= rho^{1/2}X, rho=1)."""
pi = rng.dirichlet(np.full(r, alpha))
z = rng.choice(r, size=n, p=pi)
B = np.full((r, r), p_out) + (p_in - p_out) * np.eye(r)
P = B[np.ix_(z, z)]
s, U = np.linalg.eigh(P)
idx = np.argsort(s)[::-1][:r]
X = U[:, idx] * np.sqrt(np.maximum(s[idx], 0.0))
U01 = rng.random((n, n))
A = (np.triu(U01, 1) < np.triu(P, 1)).astype(float)
A = A + A.T
return A, X, z
# ------------------------------------------------------------------ spectral
def full_spectrum(A):
"""Eigenvalues sorted non-increasing (paper Sec. 1.1 convention) + eigenvectors."""
s, U = np.linalg.eigh(A)
order = np.argsort(s)[::-1]
return s[order], U[:, order]
def ase_from_spectrum(s, U, d):
"""Definition 1: Xhat_{1:d} = Uhat_{1:d} |Shat|_{1:d}^{1/2}."""
return U[:, :d] * np.sqrt(np.abs(s[:d]))
def err_2inf(Xhat, Xtrue):
"""min_W || Xhat W - Xtrue ||_{2,inf} via orthogonal Procrustes (paper Eq. 15).
Zero-pads whichever of Xhat / Xtrue has fewer columns (paper Eqs. 7-8),
so the same routine covers k<0 (pad Xhat) and k>0 (pad Xtrue)."""
n, a = Xhat.shape
b = Xtrue.shape[1]
m = max(a, b)
if a < m:
Xhat = np.hstack([Xhat, np.zeros((n, m - a))])
if b < m:
Xtrue = np.hstack([Xtrue, np.zeros((n, m - b))])
Uu, _, Vt = np.linalg.svd(Xhat.T @ Xtrue)
W = Uu @ Vt
D = Xhat @ W - Xtrue
return float(np.max(np.linalg.norm(D, axis=1))), D
def two_inf(M):
return float(np.max(np.linalg.norm(M, axis=1)))
# -------------------------------------------------------------------- fitting
def loglog_slope(x, y):
"""Least-squares slope of log y on log x, with R^2 and stderr."""
x = np.asarray(x, float)
y = np.asarray(y, float)
lx, ly = np.log(x), np.log(y)
A = np.vstack([lx, np.ones_like(lx)]).T
coef, *_ = np.linalg.lstsq(A, ly, rcond=None)
pred = A @ coef
ss_res = float(((ly - pred) ** 2).sum())
ss_tot = float(((ly - ly.mean()) ** 2).sum())
r2 = 1.0 - ss_res / ss_tot if ss_tot > 0 else float("nan")
dof = max(len(x) - 2, 1)
se = float(np.sqrt(ss_res / dof / ((lx - lx.mean()) ** 2).sum()))
return float(coef[0]), float(coef[1]), r2, se
|