spb/prime-mystery-engine

1#!/usr/bin/env python32# =============================================================================3# merge_c4.py — Cycle 3: merge distributed worker partials into checkpoints4# Author: Simon-Pierre Boucher — contact@spboucher.ai5# =============================================================================6# Reads all worker partial JSONs, verifies the chunks tile [2, LIMIT) exactly,7# merges cumulatively in range order, and emits checkpoint statistics at every8# requested boundary. Cross-validates overlapping checkpoints against the9# cycle-2 single-machine scan when available (5-decimal tolerance: summation10# order differs, float64 pairwise sums agree to ~1e-9).11# Usage: python3 merge_c4.py <partials_dir> <limit> [out.json]12# =============================================================================1314import json15import sys16from math import log17from pathlib import Path1819import numpy as np2021DATA = Path(__file__).resolve().parent.parent / "data"22CHECKPOINTS = [10**10, 2 * 10**10, 4 * 10**10,23               10**11, 2 * 10**11, 4 * 10**11, 10**12,24               2 * 10**12, 4 * 10**12, 10**13,25               2 * 10**13, 4 * 10**13, 10**14]262728def pearson(S):29    n = S["n"]30    cov = S["xy"] / n - (S["x"] / n) * (S["y"] / n)31    vx = S["xx"] / n - (S["x"] / n) ** 232    vy = S["yy"] / n - (S["y"] / n) ** 233    return cov / (vx * vy) ** 0.5343536def main():37    pdir = Path(sys.argv[1])38    limit = int(float(sys.argv[2]))39    out_path = Path(sys.argv[3]) if len(sys.argv) > 3 else DATA / f"cycle3_c4_{limit:.0e}.json".replace("+", "").replace("e0", "e")40    parts = [json.load(open(p)) for p in sorted(pdir.glob("part_*.json"))]41    parts.sort(key=lambda d: d["lo"])42    # tiling check43    assert parts[0]["lo"] == 2, "first chunk must start at 2"44    for a, b in zip(parts, parts[1:]):45        assert a["hi"] == b["lo"], f"gap/overlap between chunks at {a['hi']} vs {b['lo']}"46    assert parts[-1]["hi"] == limit, f"last chunk ends at {parts[-1]['hi']}, expected {limit}"47    print(f"{len(parts)} chunks tile [2, {limit:.0e}) exactly; "48          f"hosts: {sorted(set(p['host'] for p in parts))}")4950    hist = np.zeros(6000, dtype=np.int64)51    S1 = dict(n=0, x=0.0, y=0.0, xx=0.0, yy=0.0, xy=0.0)52    S2 = dict(n=0, x=0.0, y=0.0, xx=0.0, yy=0.0, xy=0.0)53    D = 054    records = []55    best = 056    cps = {}57    want = [c for c in CHECKPOINTS if c <= limit]58    for part in parts:59        for g, c in part["hist"].items():60            hist[int(g)] += c61        for k in S1:62            S1[k] += part["S1"][k]63            S2[k] += part["S2"][k]64        D += part["D"]65        for g, p in part["records"]:66            if g > best:67                best = g68                records.append((g, p))69        if want and part["hi"] == want[0]:70            c = want.pop(0)71            nz = np.nonzero(hist)[0]72            champ = int(nz[np.argmax(hist[nz])])73            g6 = 074            for g in range(6, int(nz.max()) - 2, 6):75                if hist[g] > hist[g - 2] and hist[g] > hist[g + 2]:76                    g6 = g77                else:78                    break79            r1, r2 = pearson(S1), pearson(S2)80            cps[str(c)] = {"rho1": round(r1, 7), "rho1_lnx": round(r1 * log(c), 5),81                           "rho2": round(r2, 7), "rho2_lnx": round(r2 * log(c), 5),82                           "champion": champ, "G6": g6, "D_N2_minus_N4": D}83            print(f"cp {c:.0e}: rho1*lnx={r1*log(c):+.5f} rho2*lnx={r2*log(c):+.5f} "84                  f"champ={champ} G6={g6} D={D}")85    assert not want, f"missing chunk boundaries for checkpoints {want}"8687    # cross-validation against previous cycles' scans88    xval = {}89    for ref_name in ("cycle2_c4_4e10_M3U96a.json", "cycle3_c4_1e12.json"):90        ref_file = DATA / ref_name91        if not ref_file.exists():92            continue93        ref = json.load(open(ref_file))["checkpoints"]94        for c in list(ref):95            if c not in cps:96                continue97            d = abs(cps[c]["rho1_lnx"] - ref[c]["rho1_lnx"])98            ok = (d < 2e-5 and cps[c]["D_N2_minus_N4"] == ref[c]["D_N2_minus_N4"]99                  and cps[c]["champion"] == ref[c]["champion"])100            xval[c] = {"ref": ref_name, "rho1_lnx_diff": round(d, 7), "exact_D_match": ok}101            print(f"x-val {float(c):.0e} vs {ref_name}: |drho1_lnx|={d:.2e} "102                  f"D {'==' if ok else '!='}  -> {'OK' if ok else 'FAIL'}")103104    total_s = sum(p["seconds"] for p in parts)105    out = {"limit": limit, "n_chunks": len(parts),106           "hosts": sorted(set(p["host"] for p in parts)),107           "total_core_seconds": round(total_s, 1),108           "checkpoints": cps,109           "maximal_gaps": [{"gap": g, "after": p, "csg": round(g / log(p) ** 2, 5)}110                            for g, p in records],111           "cross_validation_vs_cycle2": xval}112    with open(out_path, "w") as f:113        json.dump(out, f, indent=2)114    print("written:", out_path, f"(total {total_s:.0f} core-seconds)")115116117if __name__ == "__main__":118    main()119