1"""Resolution-gated dual masking MVP with numerical check and synthetic sweep."""
  2from __future__ import annotations
  3import json
  4from pathlib import Path
  5import numpy as np
  6
  7
  8def _codes(z):
  9    z = np.asarray(z, dtype=np.int64)
 10    if z.ndim == 1:
 11        return z
 12    # Binary candidate variables: encode a mask row as a mixed-radix integer.
 13    return z.dot(1 << np.arange(z.shape[1], dtype=np.int64))
 14
 15
 16def conditional_entropy(y, z=None):
 17    y = np.asarray(y, dtype=np.int64)
 18    if z is None:
 19        z = np.zeros(len(y), dtype=np.int64)
 20    keys = _codes(z)
 21    # H(Y|Z) from one joint bincount; this is substantially faster than loops.
 22    _, zi = np.unique(keys, return_inverse=True)
 23    ny = int(y.max()) + 1
 24    joint = np.bincount(zi * ny + y, minlength=int(zi.max() + 1) * ny).reshape(-1, ny)
 25    totals = joint.sum(axis=1)
 26    nz = joint > 0
 27    probs = np.divide(joint, totals[:, None], out=np.zeros_like(joint, dtype=float), where=totals[:, None] > 0)
 28    ent = -np.sum(np.where(nz, probs * np.log(np.maximum(probs, 1e-300)), 0.0), axis=1)
 29    return float(np.sum(totals / len(y) * ent))
 30
 31
 32def entropy_scores(y, X, mask=None):
 33    y, X = np.asarray(y), np.asarray(X, dtype=np.int64)
 34    mask = [] if mask is None else list(mask)
 35    h0 = conditional_entropy(y, np.zeros(len(y), dtype=np.int64) if not mask else X[:, mask])
 36    return np.array([h0 - conditional_entropy(y, X[:, mask + [j]])
 37                     for j in range(X.shape[1]) if j not in mask])
 38
 39
 40def conditional_mean_predict(train_x, train_y, test_x):
 41    train_x = np.asarray(train_x, dtype=np.int64); test_x = np.asarray(test_x, dtype=np.int64)
 42    if train_x.ndim == 1: train_x = train_x[:, None]
 43    if test_x.ndim == 1: test_x = test_x[:, None]
 44    # Mixed-radix encoding is safe here because all features are binary.
 45    kt = _codes(train_x); kv = _codes(test_x)
 46    sums = np.bincount(kt, weights=train_y)
 47    counts = np.bincount(kt)
 48    glob = float(np.mean(train_y))
 49    return np.divide(sums[kv], counts[kv], out=np.full(len(kv), glob), where=counts[kv] > 0)
 50
 51
 52def risk_scores(ytr, Xtr, yva, Xva, mask=None):
 53    mask = [] if mask is None else list(mask)
 54    bxtr = np.zeros((len(ytr), 1), dtype=np.int64) if not mask else Xtr[:, mask]
 55    bxva = np.zeros((len(yva), 1), dtype=np.int64) if not mask else Xva[:, mask]
 56    base = np.mean((yva - conditional_mean_predict(bxtr, ytr, bxva)) ** 2)
 57    out = []
 58    for j in range(Xtr.shape[1]):
 59        if j not in mask:
 60            pred = conditional_mean_predict(Xtr[:, mask + [j]], ytr, Xva[:, mask + [j]])
 61            out.append(base - np.mean((yva - pred) ** 2))
 62    return np.array(out), base
 63
 64
 65def bootstrap_gamma(y, X, B=30, seed=0):
 66    rng = np.random.default_rng(seed)
 67    vals = np.array([entropy_scores(y[ix], X[ix])
 68                     for ix in (rng.integers(0, len(y), len(y)) for _ in range(B))])
 69    observed = entropy_scores(y, X)
 70    j = int(np.argmax(observed))
 71    return float(observed[j] / (np.std(vals[:, j], ddof=1) + 1e-12)), observed
 72
 73
 74def gated_select(ytr, Xtr, yva, Xva, kappa_c=1.0, gamma_c=2.0, B=30, seed=0):
 75    kappa = conditional_entropy(ytr) / np.log(2.0)
 76    gamma, sh = bootstrap_gamma(ytr, Xtr, B=B, seed=seed)
 77    sr, _ = risk_scores(ytr, Xtr, yva, Xva)
 78    mh, mr = int(np.argmax(sh)), int(np.argmax(sr))
 79    chosen = mh if (kappa < kappa_c and gamma > gamma_c) else mr
 80    return dict(kappa=kappa, gamma=gamma, sh=sh, sr=sr, mh=mh, mr=mr, chosen=chosen)
 81
 82
 83def generate(n, p, d, seed):
 84    rng = np.random.default_rng(seed)
 85    lag = rng.integers(0, 2, n)
 86    X = np.column_stack([lag, rng.integers(0, 2, (n, d - 1))])
 87    y = lag ^ (rng.random(n) < p).astype(int)
 88    return X, y
 89
 90
 91def run():
 92    checks = []
 93    for p in (0.0, .1, .25, .4, .5):
 94        X, y = generate(100000, p, 3, 91)
 95        empirical = conditional_entropy(y) - conditional_entropy(y, X[:, 0])
 96        q = np.array([p, 1-p]); hb = -np.sum(q[q > 0] * np.log(q[q > 0]))
 97        checks.append((p, empirical, np.log(2)-hb))
 98    rows = []
 99    for n in (256, 1024, 4096):
100      for p in (0.0, .1, .25, .4, .49):
101       for d in (3, 9, 17):
102        vals = {k: [] for k in ("entropy_recovery","risk_recovery","gate_recovery","risk_h","risk_r","risk_gate","kappa","gamma")}
103        for rep in range(6):
104            X,y = generate(n + n//2, p, d, 10000 + n + 100*d + rep)
105            Xtr, Xva, ytr, yva = X[:n], X[n:], y[:n], y[n:]
106            g = gated_select(ytr, Xtr, yva, Xva, B=30, seed=rep)
107            pred = []
108            for col in (g['mh'], g['mr'], g['chosen']):
109                pred.append(conditional_mean_predict(Xtr[:, [col]], ytr, Xva[:, [col]]))
110            vals["entropy_recovery"].append(g['mh'] == 0); vals["risk_recovery"].append(g['mr'] == 0); vals["gate_recovery"].append(g['chosen'] == 0)
111            vals["risk_h"].append(np.mean((yva-pred[0])**2)); vals["risk_r"].append(np.mean((yva-pred[1])**2)); vals["risk_gate"].append(np.mean((yva-pred[2])**2)); vals["kappa"].append(g['kappa']); vals["gamma"].append(g['gamma'])
112        rows.append(dict(n=n,p=p,d=d,**{k:float(np.mean(v)) for k,v in vals.items()}))
113    out = {"math_checks":checks, "results":rows}
114    Path("results.json").write_text(json.dumps(out, indent=2))
115    print(json.dumps({"math_checks":checks, "selected_summary": rows[:3]}, indent=2))
116
117if __name__ == "__main__": run()