#!/usr/bin/env python3 """ Cause test: the decision rule, written once (October 12, 2026). design_numbers.py uses it to work out the chance of each decision before the run; cause_test.py is to use it, unchanged, to make the decision after the run. It is arithmetic on numbers already measured and runs nothing by itself. A sandbox script kept with the project. Nothing here is served by the site. The rule, as cause_test_prereg.md states it in words and equations For start s, method k (dense, sparse) and moment m, G is the growth rate of the gap over (T + 200, T + 500] in each branch (up, down, sham). Each arm is compared with its own method's sham, at the same moment and from the same saved state: d_up = G_up - G_sham, d_down = G_down - G_sham Within a start the moments carried out (normally all ten) are averaged with equal weights, so each start gives one D_up and one D_down per method; S = (D_down - D_up) / 2. Pooled: the two methods averaged with equal weights, start by start. A moment is not carried out (all six of its branches left out, NaN here) if a target was infeasible, the solver failed or a change fell outside its band; nothing is clipped or put in its place. Across the 42 starts: the mean of the per-start values. Its error bar is the larger of two: the starts resampled 400 times with replacement (seed 20261015, the same draws for every number; the SD of the 400 resampled means), and SD (n - 1) / sqrt(n). Nothing inside a start is treated as independent. A reading, in this order: M1 (the change as planned, the branches matched) fails INCONCLUSIVE (M1 includes the count: at most 1 moment in 20 not carried out, at least 5 of every start's 10 kept) S above 0 by more than 3 error bars, D_up below 0 and D_down above 0, each by more than 2 of its own error bars, and (pooled reading only) each method's S above 0 by more than 2 of its own: PASS if S >= delta, else SMALL otherwise PARTIAL S below 0 by more than 3 error bars REVERSED S + 2 error bars below delta, and the change lasted (M2) FAIL anything else INCONCLUSIVE The overall decision is the pooled reading, with M1 counted as failed if it fails for either method. Each method's own reading, by the same rule without the condition on the methods, is reported beside it. """ import numpy as np RULE, SIDE = 3.0, 2.0 # error bars: S must clear 3; each side, 2 RESAMPLES, RESAMPLE_SEED = 400, 20261015 # as in every earlier part ARMS = ("up", "down", "sham") METHODS = ("dense", "sparse") READINGS = ("PASS", "SMALL", "PARTIAL", "REVERSED", "FAIL", "INCONCLUSIVE") POSITIVE = ("PASS", "SMALL", "PARTIAL") # the readings that report an effect in the predicted direction MOST_MISSING, FEWEST_KEPT = 20, 5 # M1's count: at most 1 moment in 20 not carried out; 5 per start def draws(starts): """The 400 resampling draws of the starts, with replacement; the same draws for every number.""" return np.random.default_rng(RESAMPLE_SEED).integers(0, starts, (RESAMPLES, starts)) def error_bar(values, picks): """values: per-start numbers, (..., start). The larger of the resampled error bar (the SD of the 400 resampled means) and SD (n - 1) / sqrt(n).""" values = np.asarray(values, float) textbook = values.std(-1, ddof=1) / np.sqrt(values.shape[-1]) resampled = values[..., picks].mean(-1).std(-1) return np.maximum(resampled, textbook) def moment_average(x): """x: (..., start, moment, method), NaN for a moment not carried out. Each start's average over the moments carried out, with equal weights: (..., start, method); NaN for a start with none. With every moment carried out, it is the plain average over the ten, to the last digit.""" x = np.asarray(x, float) kept = ~np.isnan(x) with np.errstate(invalid="ignore", divide="ignore"): return np.where(kept, x, 0.0).sum(-2) / kept.sum(-2) def per_start(g): """g: the later growth of every branch, (..., start, moment, method, arm), arms in the order of ARMS; NaN for every branch of a moment not carried out. Each start's averages over the moments carried out, each arm minus its own method's sham: D_up and D_down, (..., start, method).""" g = np.asarray(g, float) d_up = moment_average(g[..., 0] - g[..., 2]) d_down = moment_average(g[..., 1] - g[..., 2]) return d_up, d_down def enough_moments(carried): """carried: (start, moment), True where all six branches of the moment were carried out. M1's count: at most 1 moment in 20 not carried out (21 of 420), and at least 5 of every start's 10 carried out.""" carried = np.asarray(carried, bool) missing = int((~carried).sum()) return bool(MOST_MISSING * missing <= carried.size and int(carried.sum(-1).min()) >= FEWEST_KEPT) def reading(d_up, d_down, delta, picks, m1=True, m2=True, methods=()): """One reading by the rule. d_up, d_down: per-start values, (..., start); delta: the threshold; m1, m2: whether the manipulation checks hold; methods: for the pooled reading, each method's per-start (D_up, D_down), whose S must each be above 0 by more than 2 of its own error bars for PASS or SMALL. Returns the reading and every number behind it.""" d_up, d_down = np.asarray(d_up, float), np.asarray(d_down, float) s = (d_down - d_up) / 2 mean_s, mean_up, mean_down = s.mean(-1), d_up.mean(-1), d_down.mean(-1) bar_s, bar_up, bar_down = error_bar(s, picks), error_bar(d_up, picks), error_bar(d_down, picks) clear = mean_s > RULE * bar_s sides = (mean_up < -SIDE * bar_up) & (mean_down > SIDE * bar_down) agree = np.ones(np.shape(mean_s), bool) for up_k, down_k in methods: s_k = (np.asarray(down_k, float) - np.asarray(up_k, float)) / 2 agree = agree & (s_k.mean(-1) > SIDE * error_bar(s_k, picks)) out = np.where(np.asarray(m2, bool) & (mean_s + SIDE * bar_s < delta), "FAIL", "INCONCLUSIVE") out = np.where(mean_s < -RULE * bar_s, "REVERSED", out) out = np.where(clear, np.where(sides & agree, np.where(mean_s >= delta, "PASS", "SMALL"), "PARTIAL"), out) out = np.where(np.asarray(m1, bool), out, "INCONCLUSIVE") return out, dict(s=mean_s, s_bar=bar_s, up=mean_up, up_bar=bar_up, down=mean_down, down_bar=bar_down, clear=clear, sides=sides, methods_agree=agree, delta=delta) def decide(d_up, d_down, delta, picks, m1, m2): """The overall decision and each method's own reading. d_up, d_down: (..., start, method), methods in the order of METHODS; delta, m1, m2: dicts with the keys "pooled", "dense" and "sparse" (m1["pooled"] is M1 for both methods).""" d_up, d_down = np.asarray(d_up, float), np.asarray(d_down, float) pooled = reading(d_up.mean(-1), d_down.mean(-1), delta["pooled"], picks, m1["pooled"], m2["pooled"], methods=[(d_up[..., k], d_down[..., k]) for k in range(len(METHODS))]) each = {name: reading(d_up[..., k], d_down[..., k], delta[name], picks, m1[name], m2[name]) for k, name in enumerate(METHODS)} return pooled, each