#!/usr/bin/env python3 """ Coherence-growth replication (October 11, 2026, evening): the earlier-growth check, pre-registered and run on six new starts (13 to 18). A sandbox run kept with the project. Nothing here is served by the site. What led here (all on record and left as it was) - The coherence-growth test (coherence_growth.py, starts 7 to 12, both measures set before that run). Same time, in windows of 200 time units: +0.2763 (0.1044), no clear link, so the idea it tested (more coherence, slower growth in the same window) did not hold. Later: coherence now against the growth rate 200 to 500 time units later, -0.2061 (0.0186), below 0 in all six starts. - The earlier-growth check (earlier_growth_check.py). Exploratory: chosen after the test and run on the same saved runs. With the growth over the 500 time units before held fixed, the later link was -0.1610 (0.0279) on starts 7 to 12 and -0.2223 (0.0389) on starts 1 to 6. This run asks whether the check's result holds on starts that were never run. The test Everything is in replication_prereg.md, written before any new start was run. `freeze` records the SHA-256 of that plan, of this script, of every script the analysis uses and of the saved results and runs it compares with in replication_lock.json; `run` and `report` stop if any of them has changed. In short: - Starts 13 to 18 at N = 4,000, K = 1.90, 6,000 time units: the model, settling, gap and resets of the long run at the threshold, run by nudge_test.py's own code, imported unchanged. - Primary: start by start over t = 2,000 to 6,000 and lag by lag (L = 200 to 500 time units, steps of 5), the regression G(t + L) = a + b x C(t) + c1 x g1(t) + ... + c5 x g5(t), with g1 to g5 the growth rate in the five 100-unit windows up to t. The conditional coefficient is the partial correlation of C and the later G with the earlier growth held fixed (the same sign and test as b), averaged over the lags: one number per start. Computed by earlier_growth_check.py's own code, imported unchanged. Also reported: the same without the earlier growth (the test's later measure). - Uncertainty from the six starts only, so the overlapping intervals, windows and lags inside a start cannot shrink it. Error bar: the larger of the starts resampled 400 times and SD / sqrt(6). - PASS if the average is below 0 by more than 3 error bars; otherwise FAIL (no clear link, or reversed). Usage python3 replication.py dryrun # the whole report and figure, with the saved starts 7 to 12 standing in for the # new ones, to try the code before the lock; writes only to /tmp python3 replication.py freeze # the checks, then the lock; refused once any new data exist python3 replication.py run # again and again until it saves sim_replication4000.npz; each piece stops after # about 4 minutes, and unfinished starts carry on from replication_parts/ python3 replication.py report # replication.txt, replication.json and the figure Needs Python 3, numpy, scipy (the t-test's p only) and matplotlib, coherence_growth.py, earlier_growth_check.py and their saved results, and ../nudge-test-2026-10-10, all unchanged. The figure goes to screenshots/nudge-test-coherence-growth-replication.png. """ import hashlib import json import platform import sys import time from datetime import datetime, timezone from multiprocessing import Pool from pathlib import Path from zoneinfo import ZoneInfo import numpy as np HERE = Path(__file__).resolve().parent ROOT = HERE.parent.parent sys.dont_write_bytecode = True # leave every folder as it is sys.path.insert(0, str(HERE)) import coherence_growth as cg # noqa: E402 (the test: the model's settings, span, lags, rule and seed; unchanged) import earlier_growth_check as egc # noqa: E402 (the check: the earlier growth held fixed; unchanged) nt = cg.nt nt.LYAP_PARTS = HERE / "replication_parts" # progress of unfinished new starts stays here NEW = tuple(range(12, 18)) # starts 13 to 18: never run at this size RULE = cg.RULE # PASS if more than this many error bars below 0 QUESTION = ("Does earlier coherence retain a negative association with later perturbation growth after controlling for" " earlier growth?") HALIFAX = ZoneInfo("America/Halifax") PLAN, LOCK, RUNLOG = HERE / "replication_prereg.md", HERE / "replication_lock.json", HERE / "replication_runlog.json" SAVED = HERE / "sim_replication4000.npz" OUT_TXT, OUT_JSON = HERE / "replication.txt", HERE / "replication.json" FIGURE = ROOT / "screenshots" / "nudge-test-coherence-growth-replication.png" DRY_TXT, DRY_FIGURE = Path("/tmp/replication-dryrun.txt"), Path("/tmp/replication-dryrun.png") PAGE_SHA = "b06730e7772be1f719cadae7924cdca3c2e0ebb89efb17104969363431f782e7" # nudge_test.py as the page carries it LOCKED = dict( # every file the result depends on plan=PLAN, replication=Path(__file__).resolve(), coherence_growth=HERE / "coherence_growth.py", earlier_growth_check=HERE / "earlier_growth_check.py", nudge_test=nt.HERE / "nudge_test.py", test_results=HERE / "coherence_growth.json", check_results=HERE / "earlier_growth_check.json", test16_results=nt.HERE / "lyapunov_long.json", fresh_runs=cg.SAVED, earlier_runs=cg.EARLIER, test_output=HERE / "coherence_growth.txt", test_check=cg.CHECK_FILE, test_figure=cg.FIGURE, check_output=HERE / "earlier_growth_check.txt", check_figure=egc.FIGURE) SETTINGS = dict( n=cg.N, k=cg.K, eps=nt.EPS, delta=nt.DELTA, dt=nt.DT, settle=nt.SETTLE, run=cg.RUN, starts=[s + 1 for s in NEW], gap=nt.LYAP_GAP, put_back_every=nt.LYAP_EVERY, gap_seed=nt.LYAP_SEED, span=list(cg.SPAN), lags=list(cg.LAGS), lag_step=nt.LYAP_EVERY, earlier_growth=dict(windows=egc.MAIN[0], length=egc.MAIN[1]), rule=RULE, resample_seed=cg.SEED, error_bar="the larger of: the six starts resampled 400 times; SD (n - 1) / sqrt(6)", decision="PASS if the average conditional coefficient is below 0 by more than 3 error bars") FROZEN = dict( # the plan's numbers, which the code must match n=4000, k=1.9, eps=0.1, delta=1.0, dt=0.025, settle=300.0, run=6000.0, starts=[13, 14, 15, 16, 17, 18], gap=1e-7, put_back_every=5.0, gap_seed=20261011, span=[2000.0, 6000.0], lags=[200.0, 500.0], lag_step=5.0, earlier_growth=dict(windows=5, length=100.0), rule=3.0, resample_seed=20261015) def stamp(t=None): moment = datetime.fromtimestamp(time.time() if t is None else t, timezone.utc) return dict(utc=moment.isoformat(timespec="seconds"), halifax=moment.astimezone(HALIFAX).isoformat(timespec="seconds")) def shown(moment): return moment["halifax"][:16].replace("T", " ") + " Halifax" def sha256(path): return hashlib.sha256(Path(path).read_bytes()).hexdigest() def fingerprints(): return {name: sha256(path) for name, path in LOCKED.items()} def frozen(): """The settings in the code are the plan's.""" now = json.loads(json.dumps(SETTINGS)) wrong = [key for key, value in FROZEN.items() if now[key] != value] if wrong: sys.exit(f"The settings differ from the plan: {', '.join(wrong)}") def unchanged(): """The lock, after checking that nothing it covers has changed since.""" if not LOCK.exists(): sys.exit("No lock yet: run `python3 replication.py freeze` first, before any new start is run.") lock = json.loads(LOCK.read_text()) now = fingerprints() changed = [name for name in LOCKED if lock["sha256"].get(name) != now[name]] if lock["settings"] != json.loads(json.dumps(SETTINGS)): changed.append("settings") if changed: sys.exit(f"Changed since the lock ({shown(lock['locked'])}): {', '.join(changed)}. Stopping.") return lock def model_check(): """Start 1, run by this code for 50 time units, must repeat the long run exactly (as in coherence_growth.py).""" res = nt.lyapunov((cg.N, cg.K, (0, 1), nt.SETTLE, cg.CHECK_RUN, nt.LYAP_EVERY, nt.LYAP_GAP, nt.DT, None)) old = np.load(cg.EARLIER) samples, resets = int(cg.CHECK_RUN) + 1, int(round(cg.CHECK_RUN / nt.LYAP_EVERY)) same = dict(settling=bool(np.array_equal(res["trace"], old["trace"][0, :1])), coherence=bool(np.array_equal(res["z"], old["z"][0, :1, :, :samples])), gap=bool(np.array_equal(res["logs"], old["logs"][0, :1, :resets]))) return dict(start=1, run=cg.CHECK_RUN, **same, repeats=all(same.values()), seconds=round(res["seconds"], 1)) def analysed(z, logs): """Every frozen number for one set of starts, by the check's and the test's own code.""" checked = egc.checked(z, logs, np.random.default_rng(cg.SEED)) measured = cg.measures(z, logs, np.random.default_rng(cg.SEED)) assert abs(checked["test"]["mean"] - measured["later"]["mean"]) < 1e-12 # one measure, two codes, one number return dict(checked=checked, measured=measured) def saved_sets(): """Starts 7 to 12 and 1 to 6 analysed again: the numbers must equal the saved results, which stay as they were.""" fresh_data, old = np.load(cg.SAVED), np.load(cg.EARLIER) i = [round(float(x), 2) for x in old["ks"]].index(cg.K) sets = dict(fresh=analysed(fresh_data["z"], fresh_data["logs"]), earlier=analysed(old["z"][i], old["logs"][i])) test = json.loads(LOCKED["test_results"].read_text()) check = json.loads(LOCKED["check_results"].read_text()) for name, m in sets.items(): for key in ("test", "main", "short", "other", "own"): for field in ("mean", "se"): assert abs(m["checked"][key][field] - check[name][key][field]) < 1e-12, (name, key, field) for key in ("same", "later"): for field in ("mean", "se"): assert abs(m["measured"][key][field] - test[name][key][field]) < 1e-12, (name, key, field) assert abs(m["measured"]["long_rate"] - test[name]["long_rate"]) < 1e-12, (name, "long_rate") return sets def judged(s): """Both error bars, the larger (the one the decision uses), t with its p, and the reading by the rule.""" from scipy import stats values = np.asarray(s["starts"], float) n = len(values) textbook = float(values.std(ddof=1) / np.sqrt(n)) bar = max(float(s["se"]), textbook) t = float(values.mean() / textbook) if abs(s["mean"]) <= RULE * bar: reading = "no clear link" else: reading = "below 0" if s["mean"] < 0 else "above 0" return dict(mean=float(s["mean"]), resampled_se=float(s["se"]), textbook_se=textbook, error_bar=bar, bars=float(s["mean"] / bar), t=t, df=n - 1, p=float(2 * stats.t.sf(abs(t), n - 1)), starts=[float(v) for v in values], below=int((values < 0).sum()), above=int((values > 0).sum()), reading=reading) def decision(j): if j["reading"] == "below 0": return "PASS", ("replicated: earlier coherence keeps a negative link with later growth once the earlier growth is" " held fixed") if j["reading"] == "no clear link": return "FAIL", f"no clear link: within {RULE:g} error bars of 0 once the earlier growth is held fixed" return "FAIL", f"reversed: more than {RULE:g} error bars above 0 once the earlier growth is held fixed" def build(new, sets, labels, lock=None, log=None, dry=False): """The result and its text, for the new starts (in the dry run, saved starts standing in for them).""" primary, plain = judged(new["checked"]["main"]), judged(new["checked"]["test"]) verdict, meaning = decision(primary) plain_reading = {"below 0": "below 0: more coherence now, slower growth later", "above 0": "above 0: more coherence now, faster growth later"}.get(plain["reading"], "no clear link") checked, measured = new["checked"], new["measured"] test16 = json.loads(LOCKED["test16_results"].read_text()) diff = measured["long_rate"] - test16["rate"] combined = float(np.hypot(measured["long_rate_se"], test16["rate_se"])) agree = bool(abs(diff) <= RULE * combined) reference = {name: dict(primary=judged(s["checked"]["main"]), plain=judged(s["checked"]["test"]), same=judged(s["measured"]["same"])) for name, s in sets.items()} span = f"{labels[0]} to {labels[-1]}" result = dict( pre_registered=not dry, dry_run=dry, plan=PLAN.name, question=QUESTION, settings=SETTINGS, new_starts=labels, lock=lock, run_log=log, decision=dict(verdict=verdict, meaning=meaning, rule=SETTINGS["decision"], error_bar=SETTINGS["error_bar"]), primary=primary, plain=plain, plain_reading=plain_reading, same_reading=cg.reading(measured["same"]), long_rate=dict(rate=measured["long_rate"], se=measured["long_rate_se"], starts=measured["long_rate_starts"], test16=test16["rate"], test16_se=test16["rate_se"], difference=diff, combined_se=combined, agree=agree), new=new, reference=reference, saved_curves={name: s["checked"]["curve"] for name, s in sets.items()}) lines = [] def say(text=""): print(text) lines.append(text) def cell(s): # as in the earlier outputs: the resampled error bar return f"{s['mean']:+.4f} ({s['se']:.4f})" def values(vs): return ", ".join(f"{v:+.2f}" for v in vs) say(f"Coherence-growth replication, pre-registered: October 11, 2026, N = {cg.N:,} per group, K = {cg.K:.2f}," f" eps = {nt.EPS:g}") if dry: say("DRY RUN, to try the code before the lock: the saved starts 7 to 12 stand in for the new ones. Not a result.") if not dry: say(f"Six new starts ({span}; never run at this size), {cg.RUN:,.0f} time units each, run by nudge_test.py's own" " code, imported unchanged.") if lock: say(f"Fixed before the run in {PLAN.name} and locked {shown(lock['locked'])} ({LOCK.name}: SHA-256 of the plan," " every script used and the saved results and runs compared with). Checked for this report: unchanged.") say(f"Before the lock: start 1, run by this code for {lock['check']['run']:.0f} time units, repeated the long run" f" {'exactly' if lock['check']['repeats'] else 'NOT exactly'}, and the analysis gave the saved results of the" " test and the check exactly.") if log: pieces = log["pieces"] ended = pieces[-1].get("ended") say(f"The run: {len(pieces)} pieces, from {shown(pieces[0]['began'])} to" f" {shown(ended) if ended else 'an unfinished piece'}.") say() say(f"Primary question (fixed before the run): {QUESTION}") say(f"Regression, start by start and lag by lag over t = {cg.SPAN[0]:,.0f} to {cg.SPAN[1]:,.0f} in 5-unit intervals," " least squares:") say(f" G(t + L) = a + b x C(t) + c1 x g1(t) + ... + c5 x g5(t), L = {cg.LAGS[0]:.0f} to {cg.LAGS[1]:.0f} time units" " in steps of 5") say(" C: the coherence |z| (average of A and B). G: the growth of the log of the gap. g1 to g5: the growth rate in" " the five 100-unit windows up to t.") say(" The conditional coefficient: the partial correlation of C and the later G with g1 to g5 held fixed (the same" " sign and test as b), averaged over the 61 lags: one number per start.") say("Uncertainty: from the six starts only, so the overlapping intervals, windows and lags inside a start cannot" " shrink it. Error bar: the larger of the starts resampled 400 times and SD/sqrt(6).") say(f"Decision: PASS if the average over the six starts is below 0 by more than {RULE:g} error bars; otherwise FAIL" " (no clear link, or reversed).") say() say(f"Result, starts {span}") for name, j in (("conditional coefficient (primary)", primary), ("unadjusted (the test's later measure)", plain)): side = "below" if j["mean"] < 0 else "above" say(f" {name:<39}{j['mean']:+.4f} error bar {j['error_bar']:.4f} (resampled {j['resampled_se']:.4f}," f" SD/sqrt(6) {j['textbook_se']:.4f}): {abs(j['bars']):.1f} error bars {side} 0; t = {j['t']:+.2f}," f" {j['df']} df, p = {j['p']:.2g}") say(f" {'':<39}starts {span}: {values(j['starts'])}; below 0 in {j['below']} of {len(j['starts'])}") say(f"DECISION (fixed before the run): {verdict}: {meaning}") say(f"The unadjusted link, read by the same rule: {plain_reading}") say() say(f"Also reported, fixed before the run and not used for the decision (starts {span}; error bars resampled, as in" " the earlier outputs)") rows = (("the earlier growth held fixed: the 200 before only", checked["short"]), ("check on the check: another start's earlier growth held fixed", checked["other"]), ("the growth's own link: the 200 before against 200 to 500 later", checked["own"]), ("windows of 200, the next window: plain", checked["windows"]["1"]["plain"]), ("windows of 200, the next window: its own growth held fixed (b)", checked["windows"]["1"]["held"]), ("windows of 200, 2 windows on: plain", checked["windows"]["2"]["plain"]), ("windows of 200, 2 windows on: its own growth held fixed (b)", checked["windows"]["2"]["held"]), ("same time, windows of 200 (did not hold in the test)", measured["same"])) for name, s in rows: say(f" {name:<66}{cell(s)}, below 0 in {s['below']} of {len(s['starts'])}") weights = np.array(checked["windows"]["1"]["weights"]) say(f" scaled weights, the next window: b (coherence) {weights[:, 0].mean():+.2f}, c (growth)" f" {weights[:, 1].mean():+.2f} (averages over the starts)") say(f" same time, read by the test's rule: {result['same_reading']}") say(f" long-run rate, t = {cg.LONG_FROM:,.0f} to {cg.RUN:,.0f}: {measured['long_rate']:+.4f}" f" ({measured['long_rate_se']:.4f}); Test 16: {test16['rate']:+.4f} ({test16['rate_se']:.4f}); difference" f" {diff:+.4f} ({combined:.4f}): {'agree' if agree else 'differ'}") say(f" coherence |z| over t = {cg.SPAN[0]:,.0f} to {cg.SPAN[1]:,.0f} (average over the starts):" f" {measured['coherence']:.4f}; starts 7 to 12 {sets['fresh']['measured']['coherence']:.4f}, 1 to 6" f" {sets['earlier']['measured']['coherence']:.4f}") say() say("The main measures on every set, by the same code (error bars: the larger of the two, as for the decision)") heads = (f"new starts ({span})", "starts 7 to 12", "starts 1 to 6") say(f" {'':<34}" + "".join(f"{h:<36}" for h in heads)) mine = dict(primary=primary, plain=plain, same=judged(measured["same"])) for name, key in (("conditional (primary)", "primary"), ("unadjusted", "plain"), ("same time, windows of 200", "same")): cells = [mine[key], reference["fresh"][key], reference["earlier"][key]] say(f" {name:<34}" + "".join(f"{j['mean']:+.4f} ({j['error_bar']:.4f}), {j['below']} of 6 below 0".ljust(36) for j in cells)) say(" Starts 7 to 12: the test's two measures were fixed before that run; the conditional one was chosen after it" " (exploratory). Starts 1 to 6: where the patterns were first noticed.") say() say("The link at chosen lags L (time units), averaged over the starts: unadjusted / conditional") picked = (5, 50, 100, 200, 300, 400, 500) say(" " + "".join(f"{lag:>16d}" for lag in picked)) for name, curve in (("new", checked["curve"]), ("7 to 12", sets["fresh"]["checked"]["curve"]), ("1 to 6", sets["earlier"]["checked"]["curve"])): lags = curve["lags"] say(f" {name:<10}" + "".join(f"{curve['plain'][lags.index(lag)]:>+8.3f}/{curve['held'][lags.index(lag)]:<+7.3f}" for lag in picked)) fresh_m, fresh_c = sets["fresh"]["measured"], sets["fresh"]["checked"] say() say(f"On record and unchanged: coherence_growth.txt (the test, starts 7 to 12: same time {cell(fresh_m['same'])}, no" f" clear link; later {cell(fresh_m['later'])}) and earlier_growth_check.txt (exploratory: {cell(fresh_c['main'])}" f" on starts 7 to 12, {cell(sets['earlier']['checked']['main'])} on 1 to 6).") say("A toy model: this shows structure in it, not proof about anything beyond it. A correlation, not a cause; no" " claim of damping.") return result, lines def figure(result, path): """The link at every lag for the new starts, and the decision: single starts, average and error bar.""" import matplotlib matplotlib.use("Agg") import matplotlib.pyplot as plt blue, grey, light = "#377eb8", "0.4", "0.68" labels, verdict = result["new_starts"], result["decision"] span = f"{labels[0]} to {labels[-1]}" fig = plt.figure(figsize=(16, 8.2)) grid = fig.add_gridspec(1, 2, left=0.06, right=0.985, top=0.78, bottom=0.2, wspace=0.2, width_ratios=(1.15, 1)) fig.suptitle(f"Coherence-growth replication (pre-registered, October 11, 2026): {cg.N:,} per group, K = {cg.K:.2f}," f" new starts {span}\nDoes earlier coherence keep a negative link with later growth once the earlier" " growth is held fixed?", fontsize=14, y=0.985) color = "#1a7f37" if verdict["verdict"] == "PASS" else "#b42318" fig.text(0.5, 0.85, f"Decision (fixed before the run): {verdict['verdict']}, {verdict['meaning']}", ha="center", fontsize=12.5, color=color, weight="bold") if result["dry_run"]: fig.text(0.5, 0.5, "DRY RUN: saved starts standing in\nnot a result", ha="center", va="center", fontsize=40, color="red", alpha=0.25, rotation=15) ax = fig.add_subplot(grid[0, 0]) curve = result["new"]["checked"]["curve"] lags = np.array(curve["lags"]) ax.axvspan(*cg.LAGS, color="#fff3c4", zorder=0, label=f"L = {cg.LAGS[0]:.0f} to {cg.LAGS[1]:.0f}: fixed before the run") for name, shade, width, text in (("earlier", light, 1.2, "starts 1-6"), ("fresh", grey, 1.4, "starts 7-12")): ax.plot(lags, result["saved_curves"][name]["held"], color=shade, lw=width, label=f"{text}: earlier growth held fixed") held, se = np.array(curve["held"]), np.array(curve["held_se"]) ax.fill_between(lags, held - se, held + se, color=blue, alpha=0.2, lw=0) ax.plot(lags, curve["plain"], color=blue, lw=1.6, ls="--", label=f"new starts {span}: unadjusted") ax.plot(lags, held, color=blue, lw=2.4, label=f"new starts {span}: earlier growth (500 before) held fixed") ax.axhline(0, color="0.5", lw=0.9) ax.set(xlim=(0, lags[-1]), xlabel="L, time units", ylabel="correlation (average over the starts)", title="Coherence now against the growth rate L later\n(band: resampled error bar of the new starts, held" " fixed)") ax.legend(fontsize=8.5, loc="best") ax = fig.add_subplot(grid[0, 1]) keys = (("primary", "conditional coefficient\n(the 500 before held fixed)\nprimary"), ("plain", "unadjusted\n(the test's later measure)")) for x, (key, _) in enumerate(keys): j = result[key] ax.add_patch(plt.Rectangle((x - 0.22, -RULE * j["error_bar"]), 0.44, 2 * RULE * j["error_bar"], color="0.87", lw=0)) ax.plot(np.full(len(j["starts"]), x + 0.06), j["starts"], "o", ms=5.5, color=blue, alpha=0.6) ax.errorbar(x - 0.07, j["mean"], yerr=j["error_bar"], fmt="D", ms=9, color=blue, capsize=4, mew=2) for name, marker, shade, dx in (("fresh", "D", grey, 0.25), ("earlier", "s", light, 0.36)): r = result["reference"][name][key] ax.errorbar(x + dx, r["mean"], yerr=r["error_bar"], fmt=marker, ms=7, mfc="white", mec=shade, ecolor=shade, mew=2, capsize=3) ax.plot([], [], "D", ms=8, color=blue, label=f"new starts {span}:\naverage and error bar\n(dots: single starts)") ax.plot([], [], "D", ms=7, mfc="white", mec=grey, mew=2, label="starts 7-12\n(where the check\nwas chosen)") ax.plot([], [], "s", ms=7, mfc="white", mec=light, mew=2, label="starts 1-6") ax.add_patch(plt.Rectangle((0, 0), 0, 0, color="0.87", label=f"within {RULE:g} error bars\nof 0 (new starts)")) ax.axhline(0, color="0.5", lw=0.9) ax.set(xlim=(-0.5, 2.35), xticks=[0, 1], ylabel="correlation", title=f"Coherence now against the growth {cg.LAGS[0]:.0f} to {cg.LAGS[1]:.0f} later") ax.set_xticklabels([name for _, name in keys], fontsize=9.5) ax.legend(fontsize=8.5, loc="center right", labelspacing=1.0) who = f"Saved starts {span} standing in (dry run)" if result["dry_run"] else f"Starts {span}, never run at this size" for y, text in ((0.075, f"{who}: the same model, gap (rms {nt.LYAP_GAP:g} radians, put" f" back every {nt.LYAP_EVERY:g} time units) and length ({cg.RUN:,.0f}) as the long run at the" f" threshold. Measured start by start over t = {cg.SPAN[0]:,.0f} to {cg.SPAN[1]:,.0f}."), (0.05, "Conditional coefficient: the partial correlation of coherence now and the growth rate L later," " with the growth rate in each of the five 100-unit windows before held fixed, averaged over" f" L = {cg.LAGS[0]:.0f} to {cg.LAGS[1]:.0f}."), (0.025, "Error bars from the six starts (the larger of resampled and SD/\u221a6); grey: within" f" {RULE:g} error bars of 0. Fixed and locked before the run. A toy model: this shows" " structure, not proof beyond it.")): fig.text(0.06, y, text, fontsize=9.5, color="0.25") path.parent.mkdir(exist_ok=True) fig.savefig(path, dpi=115) def dryrun(): """The whole report and figure, with the saved starts 7 to 12 standing in for the new ones. Writes only to /tmp.""" frozen() sets = saved_sets() result, lines = build(sets["fresh"], sets, [s + 1 for s in cg.FRESH], dry=True) DRY_TXT.write_text("\n".join(lines) + "\n") figure(result, DRY_FIGURE) print(f"\nDry run: {DRY_TXT} and {DRY_FIGURE} (outside the project)") def freeze(): """The checks, then the lock. Refused once any new data exist: the plan can only be fixed before the run.""" found = [path.name for path in (SAVED, RUNLOG, nt.LYAP_PARTS) if path.exists()] if found: sys.exit(f"New data already exist ({', '.join(found)}): the plan can no longer be locked.") frozen() if sha256(LOCKED["nudge_test"]) != PAGE_SHA: sys.exit("nudge_test.py is not the copy the page carries. Stopping.") check = model_check() print(f"Check: start 1 for {check['run']:.0f} time units against the long run: settling {check['settling']}," f" coherence {check['coherence']}, gap {check['gap']} ({check['seconds']:.0f} s)") if not check["repeats"]: sys.exit("The code no longer repeats the long run. Stopping.") saved_sets() print("The analysis gives the saved results of the test and the check exactly (starts 1 to 6 and 7 to 12).") import scipy lock = dict(locked=stamp(), plan=PLAN.name, question=QUESTION, settings=SETTINGS, sha256=fingerprints(), check=check, environment=dict(python=platform.python_version(), numpy=np.__version__, scipy=scipy.__version__), reproduced="the analysis gives the saved results of the coherence-growth test and the earlier-growth" " check exactly, on starts 1 to 6 and 7 to 12", new_data="none: no saved run, no run log, no progress files") LOCK.write_text(json.dumps(lock, indent=1) + "\n") print(f"Locked {shown(lock['locked'])}: {LOCK.name}") def run(): """The six new starts, in pieces of at most nudge_test's LYAP_BUDGET seconds. Prints progress only.""" lock = unchanged() frozen() if SAVED.exists(): print(f" nothing left to run: {SAVED.name} is saved") return log = json.loads(RUNLOG.read_text()) if RUNLOG.exists() else dict(locked=lock["locked"], pieces=[]) piece = dict(began=stamp()) log["pieces"].append(piece) RUNLOG.write_text(json.dumps(log, indent=1) + "\n") clock = time.time() deadline = clock + nt.LYAP_BUDGET tasks = [(cg.N, cg.K, (s, 1), nt.SETTLE, cg.RUN, nt.LYAP_EVERY, nt.LYAP_GAP, nt.DT, deadline) for s in NEW] results = [] with Pool(len(tasks)) as pool: for res in pool.imap_unordered(nt.lyapunov, tasks): results.append(res) where = "finished" if res["finished"] else f"paused at t = {res['step'] * nt.DT:,.0f}" print(f" {where}: start {res['first'] + 1} ({res['seconds']:.0f} s so far)", flush=True) done = sum(res["finished"] for res in results) piece.update(ended=stamp(), finished=done, reached={str(res["first"] + 1): round(res["step"] * nt.DT) for res in sorted(results, key=lambda r: r["first"])}) RUNLOG.write_text(json.dumps(log, indent=1) + "\n") if done < len(results): print(f" {done} of {len(results)} starts finished; run the same command again to carry on" f" (this piece {time.time() - clock:.0f} s)") return results.sort(key=lambda res: res["first"]) np.savez_compressed(SAVED, starts=np.array(NEW), n=cg.N, k=cg.K, dt=nt.DT, settle=nt.SETTLE, run=cg.RUN, every=nt.LYAP_EVERY, gap=nt.LYAP_GAP, seed=nt.LYAP_SEED, **{key: np.concatenate([res[key] for res in results]) for key in ("z", "logs", "trace")}, seconds=np.array([res["seconds"] for res in results])) for res in results: nt.lyap_progress(cg.N, cg.K, res["first"], cg.RUN).unlink() if nt.LYAP_PARTS.exists() and not any(nt.LYAP_PARTS.iterdir()): nt.LYAP_PARTS.rmdir() print(f" saved {SAVED.name} (this piece {time.time() - clock:.0f} s)") def report(): lock = unchanged() frozen() data = np.load(SAVED) assert tuple(int(s) for s in data["starts"]) == NEW and int(data["n"]) == cg.N and float(data["k"]) == cg.K assert float(data["run"]) == cg.RUN assert all(float(data[key]) == float(value) for key, value in (("dt", nt.DT), ("settle", nt.SETTLE), ("every", nt.LYAP_EVERY), ("gap", nt.LYAP_GAP), ("seed", nt.LYAP_SEED))) sets = saved_sets() new = analysed(data["z"], data["logs"]) result, lines = build(new, sets, [s + 1 for s in NEW], lock, json.loads(RUNLOG.read_text())) OUT_JSON.write_text(json.dumps(result, indent=1)) OUT_TXT.write_text("\n".join(lines) + "\n") figure(result, FIGURE) print(f"\nWrote {OUT_TXT.name}, {OUT_JSON.name} and {FIGURE.relative_to(ROOT)}") if __name__ == "__main__": commands = dict(dryrun=dryrun, freeze=freeze, run=run, report=report) command = sys.argv[1] if len(sys.argv) > 1 else "" if command not in commands: sys.exit(__doc__) commands[command]()