"""生成正向、信息账、入口/余额/种子三份预期值 JSON。 用法（在仓库根目录）： python3 tests/fixtures/rules/gen_forward_info.py 只用标准库；依赖同目录的 rules_ref.py。几秒内跑完。重跑结果逐位相同。 对应 docs/方法.md： 6.1 方向、6.2 三个数→折线、6.3 开放预留与合并、6.4 入池/选批/分数尺读数、6.5 采纳线 6.9 比特/惊讶/分歧量/落点、6.10 第 4 条战绩、4.2 入口分配、第七部分余额、 以及约定表第 1–16、31–40 条。 生成的文件（都在本目录）： forward.json —— 折线、换算、合并、合并分位数、入池、V(S)、选批（批内最好期望、均匀随机）、分数尺读数、采纳线、采纳判断 info.json —— 世界比特、惊讶、分歧量、落点四段、覆盖率、战绩 misc.json —— 入口分配、余额、种子与随机抽样 JSON 格式（三份相同）： { "generator": "gen_forward_info.py", "tolerance": {"abs": 1e-12, "note": "..."}, "scenarios": [ {"name": 场景名, "section": "方法.md 小节 / 约定 N", "params": {eps, lambda, direction, ...}, "input": {...}, "expected": {...}, "note": "说明"} ] } - 分数一律是原值；"Y" 前缀的字段是改进量（越大越好）刻度。 - 三个数写成 {"q1","q5","q9"}，严格递增；点列写成 [[x, y], ...]。 - 正向候选的各人范围写成 [{"researcher", "range"}]。 - cdf = 低于 x 的把握 F(x)（折线连续，低于与不高于相同）。 - 预期数值由有理数精确算出后转成浮点；与实现比较时用绝对误差 tolerance.abs， 布尔、编号、标签、状态必须完全相同。 """ import hashlib import json import os import sys from fractions import Fraction as Fr from itertools import combinations sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) import rules_ref as R # noqa: E402 HERE = os.path.dirname(os.path.abspath(__file__)) F = float # 种子与随机（约定 39–40）：选批的均匀随机场景和 misc.json 都用。 def canonical(obj): return json.dumps(obj, sort_keys=True, separators=(",", ":"), ensure_ascii=False) def seed_of(study, rnd, ver): s = canonical({"studyId": study, "round": rnd, "version": ver}) return s, hashlib.sha256(s.encode("utf-8")).hexdigest() class Rng: def __init__(self, seed): self.seed, self.counter = seed, 0 def uniform_int(self, n): limit = 2 ** 48 - (2 ** 48 % n) while True: h = hashlib.sha256(f"{self.seed}:{self.counter}".encode("utf-8")).digest() self.counter += 1 v = int.from_bytes(h[:6], "big") if v < limit: return v % n def string_order(s): """JS 默认字符串序。只用来核对自然序的场景确实分得出两种排序。""" return s.encode("utf-16-be") def sample(seed, ids, size, key=R.id_key): """约定 40 的抽样。编号（候选、因子）一律按自然序排（c2 < c10、p2 < p10）。""" a = sorted(ids, key=key) rng = Rng(seed) n = len(a) size = min(size, n) for i in range(size): j = i + rng.uniform_int(n - i) a[i], a[j] = a[j], a[i] return sorted(a[:size], key=key) MANY_C = [f"c{i}" for i in range(1, 13)] MANY_P = [f"p{i}" for i in range(1, 13)] def pts_json(pts): return [[F(x), F(y)] for x, y in pts] def tri(t): return {"q1": t[0], "q5": t[1], "q9": t[2]} def ranges_json(ts): """正向一个候选的各人范围：[{researcher, range}]，研究者按位置编号 researcher-1、researcher-2……""" return [{"researcher": f"researcher-{i + 1}", "range": tri(t)} for i, t in enumerate(ts)] def write(name, scenarios, note, extra=None): """tolerance.abs 是默认的绝对误差；个别量另有容差时放在 extra 里，note 只解释，不另写数值。""" doc = {"generator": "gen_forward_info.py", "tolerance": {"abs": 1e-12, **(extra or {}), "note": note}, "scenarios": scenarios} with open(os.path.join(HERE, name), "w", encoding="utf-8") as f: json.dump(doc, f, ensure_ascii=False, indent=1) f.write("\n") # =========================================================== forward.json S = [] # ---- 1. 折线 CDF def cdf_case(name, t, xs, note, dens_xs=None): p = R.polyline(*t) d = R.single(p) S.append({ "name": name, "section": "6.2；约定 1–4", "params": {}, "input": {"range": tri(t), "x": xs, "densityAt": dens_xs or xs}, "expected": { "points": pts_json(p), "cdf": [F(d.left(x)) for x in xs], "density": [F(R.pts_density(p, x)) for x in (dens_xs or xs)], "mean": F(R.mean(d)), }, "note": note}) cdf_case("cdf_method_example_2.44_2.52_2.62", (2.44, 2.52, 2.62), [2.30, 2.36, 2.40, 2.44, 2.48, 2.52, 2.57, 2.62, 2.67, 2.72, 2.80], "方法 6.2 示例：低于 2.57 的把握 0.70；尾巴 2.36–2.44、2.62–2.72。") cdf_case("cdf_asymmetric_narrow", (1.2999, 1.30, 1.3005), [1.2997, 1.2998, 1.29995, 1.3, 1.3002, 1.3005, 1.30075, 1.301], "方法第五部分的窄范围写法（不对称）：低端尾巴 1.2998–1.2999，高端尾巴 1.3005–1.3010。折点上密度取右边那段。") # ---- 越小越好的换算 b_min = 0.40 t_min = (0.30, 0.35, 0.45) p_raw = R.polyline(*t_min) p_y = R.improvement_pts(p_raw, b_min, "minimize") p_y2 = R.polyline(*(R.fr(b_min) - R.fr(t_min[2]), R.fr(b_min) - R.fr(t_min[1]), R.fr(b_min) - R.fr(t_min[0]))) assert p_y == p_y2 ys = [-0.2, -0.05, 0.0, 0.05, 0.08, 0.10, 0.25] dy = R.single(p_y) S.append({ "name": "minimize_to_improvement", "section": "6.1；约定 5", "params": {"direction": "minimize"}, "input": {"b": b_min, "range": tri(t_min), "y": ys}, "expected": { "Yrange": {"q1": F(p_y[1][0]), "q5": F(p_y[2][0]), "q9": F(p_y[3][0])}, "Ypoints": pts_json(p_y), "Ycdf": [F(dy.left(y)) for y in ys], }, "note": "Y = b − 分数，一成九成对调：Y 的三个数 = (b−q9, b−q5, b−q1) = (−0.05, 0.05, 0.10)；" "低端尾巴宽 = 原 q9−q5 = 0.10，高端尾巴宽 = 原 q5−q1 = 0.05。"}) # ---- 2. 合并（正向、消融格） EPS = 0.01 LAM = 0.1 B_R1 = 0.9597642169962064 R1_FWD = { # 15-real-1 第 1 轮真实的三人预测（只当输入） "r1-v1-c1": [(2.0, 2.45, 2.6), (2.2, 2.5, 2.62), (2.2, 2.5, 2.63)], "r1-v1-c2": [(1.85, 2.35, 2.6), (1.6, 2.35, 2.6), (1.8, 2.4, 2.62)], "r1-v1-c3": [(2.3, 2.5, 2.62), (2.42, 2.52, 2.6), (2.1, 2.5, 2.6)], } R1_RESEARCHERS = ["researcher-1", "researcher-2", "researcher-3"] # 越小越好的合成正向（b = 0.40，分数越低越好），两位研究者，c3 预期落在池外 B_MIN = 0.40 MIN_FWD = { "c1": [(0.36, 0.38, 0.40), (0.35, 0.37, 0.39)], "c2": [(0.33, 0.36, 0.41), (0.37, 0.39, 0.42)], "c3": [(0.41, 0.43, 0.46), (0.40, 0.42, 0.45)], "c4": [(0.37, 0.39, 0.40), (0.38, 0.395, 0.41)], } PROBE_Y = [-1.0, -0.01, 0.0, 0.5, 1.0, 1.4, 1.5, 1.6, 1.7, 2.0] def forward_merge_case(name, cands, b, direction, probe, note): dys, draws, (ylo, yhi), (rlo, rhi) = R.forward_dists(cands, b, direction, LAM) exp = {"reserveY": [F(ylo), F(yhi)], "reserveRaw": [F(rlo), F(rhi)], "candidates": {}} for c, d in dys.items(): yq = [R.quantile(d, p) for p in (Fr(1, 10), Fr(1, 2), Fr(9, 10))] if direction == "maximize": rawq = [R.fr(b) + q for q in yq] else: rawq = [R.fr(b) - q for q in reversed(yq)] exp["candidates"][c] = { "Ycdf": [F(d.left(y)) for y in probe], "Yquantiles": {"q1": F(yq[0]), "q5": F(yq[1]), "q9": F(yq[2])}, "rawQuantiles": {"q1": F(rawq[0]), "q5": F(rawq[1]), "q9": F(rawq[2])}, "Ymean": F(R.mean(d)), } S.append({ "name": name, "section": "6.3、6.1；约定 6–8", "params": {"lambda": LAM, "direction": direction, "eps": EPS}, "input": {"b": b, "candidates": {c: ranges_json(ts) for c, ts in cands.items()}, "Yprobe": probe}, "expected": exp, "note": note}) return dys dys_r1 = forward_merge_case( "forward_merge_r1_real", R1_FWD, B_R1, "maximize", PROBE_Y, "真实第 1 轮三候选×三人（每人一份，等权）。开放预留取三候选九条折线（含尾巴）的最低首点和最高末点。" "合并分位数在改进量上用广义逆 inf{y: P(Y≤y) ≥ p} 求，再加回 b。") dys_min = forward_merge_case( "forward_merge_minimize", MIN_FWD, B_MIN, "minimize", [-0.1, -0.02, -0.01, 0.0, 0.01, 0.02, 0.05], "越小越好：逐人换成 Y = b − 分数再合并；原值分位数 = b − Y 的分位数，一成九成对调。") # 消融一格的合并：真实第 1 轮消融 p1 格 R1_ABL_AUTH = ["researcher-1", "researcher-1", "researcher-2", "researcher-2", "researcher-3", "researcher-3"] R1_ABL_IDS = ["r1-e1-1", "r1-e1-2", "r1-e2-1", "r1-e2-2", "r1-e3-1", "r1-e3-2"] R1_ABL = { "p1": [(1.2, 2, 2.44), (1.22, 1.4, 1.52), (1.05, 1.35, 1.95), (0.95, 1.3, 2.1), (1.15, 1.36, 1.55), (1.2, 1.42, 2.46)], "p2": [(1.49, 1.54, 1.58), (1.49, 1.54, 1.58), (1.44, 1.53, 1.59), (1.4, 1.52, 1.62), (1.49, 1.535, 1.58), (1.5, 1.53, 1.58)], "p3": [(2.3, 2.44, 2.45), (1.2, 1.42, 1.55), (0.85, 2.4, 2.45), (0.6, 1.45, 2.45), (1.8, 2.42, 2.58), (1.9, 2.44, 2.57)], } cd, (clo, chi) = R.cell_dist(R1_ABL["p1"], R1_ABL_AUTH, LAM) probe = [0.5, 1.0, 1.3, 1.5, 2.0, 2.5, 3.0] S.append({ "name": "ablation_cell_merge_r1_p1", "section": "6.3、6.7 成分与权重；约定 6–8、24", "params": {"lambda": LAM}, "input": {"explanations": [{"id": i, "author": a, "range": tri(t)} for i, a, t in zip(R1_ABL_IDS, R1_ABL_AUTH, R1_ABL["p1"])], "x": probe}, "expected": { "reserve": [F(clo), F(chi)], "weights": [F(w) for w in R.explanation_weights(R1_ABL_AUTH, LAM)] + [LAM], "cdf": [F(cd.left(x)) for x in probe], "quantiles": {k: F(R.quantile(cd, p)) for k, p in (("q1", Fr(1, 10)), ("q5", Fr(1, 2)), ("q9", Fr(9, 10)))}, }, "note": "真实第 1 轮消融 p1 格（关掉 p1）。三人各两条解释：每条 0.9/3/2 = 0.15，预留 0.1（权重最后一项）。" "开放预留只看这一格六条折线。分位数在原值上求。"}) # 合并分位数：平段取左端（两条不重叠的窄折线各占一半，用 λ=0 去掉预留） Q_PARTS = [(0.99, 1.0, 1.01), (1.99, 2.0, 2.01)] dq = R.Dist([(Fr(1, 2), R.polyline(*Q_PARTS[0])), (Fr(1, 2), R.polyline(*Q_PARTS[1]))]) S.append({ "name": "quantile_flat_left_end", "section": "约定 8", "params": {}, "input": {"mixture": [{"weight": 0.5, "range": tri(t)} for t in Q_PARTS], "p": [0.1, 0.5, 0.9]}, "expected": {"quantiles": [F(R.quantile(dq, p)) for p in (Fr(1, 10), Fr(1, 2), Fr(9, 10))]}, "note": "两条窄折线各占一半，中间 [1.02, 1.98] 把握走平在 0.5：一半取平段左端 1.02；一成落在第一条的" "[0.99, 1.0] 段里（第一条自己的 0.2 分位），九成落在第二条的 [2.0, 2.01] 段里。"}) # ---- 3. 入池 def pool_case(name, cands, b, direction, note): dys, _, _, _ = R.forward_dists(cands, b, direction, LAM) exp = {} for c, d in dys.items(): pr = R.pool_prob(d, EPS) exp[c] = {"probBelowMinusEps": F(pr), "inPool": pr < Fr(1, 2)} S.append({"name": name, "section": "6.4 入池；约定 9", "params": {"eps": EPS, "lambda": LAM, "direction": direction}, "input": {"b": b, "candidates": {c: ranges_json(ts) for c, ts in cands.items()}}, "expected": exp, "note": note}) return exp pool_case("pool_r1_real", R1_FWD, B_R1, "maximize", "真实第 1 轮：三个候选都远高于 b，都入池。") pool_min = pool_case("pool_minimize", MIN_FWD, B_MIN, "minimize", "越小越好：P(Y < −ε) = P(分数 > b+ε)。c3 的大部分把握在 0.41 以上，不入池。") assert not pool_min["c3"]["inPool"] # 刚好 0.5：一位研究者、一个候选，Y 的中位数正好在 −ε，折线与预留都关于 −ε 对称 EPS_H = 0.25 half = {"c1": [(0.5, 0.75, 1.0)]} # b = 1，Y 三个数 = (−0.5, −0.25, 0) dys_h, _, _, _ = R.forward_dists(half, 1.0, "maximize", LAM) ph = R.pool_prob(dys_h["c1"], EPS_H) assert ph == Fr(1, 2) float_ph = (1 - LAM) * 0.5 + LAM * 0.5 assert float_ph == 0.5 half2 = {"c1": [(0.5, 0.75 + 2 ** -10, 1.0)]} dys_h2, _, _, _ = R.forward_dists(half2, 1.0, "maximize", LAM) ph2 = R.pool_prob(dys_h2["c1"], EPS_H) S.append({"name": "pool_exactly_half", "section": "6.4 入池；约定 9", "params": {"eps": EPS_H, "lambda": LAM, "direction": "maximize"}, "input": {"b": 1.0, "cases": [{"candidates": {"c1": ranges_json(half["c1"])}}, {"candidates": {"c1": ranges_json(half2["c1"])}}]}, "expected": {"cases": [{"probBelowMinusEps": F(ph), "inPool": ph < Fr(1, 2)}, {"probBelowMinusEps": F(ph2), "inPool": ph2 < Fr(1, 2)}]}, "note": "第一例：Y 三个数 (−0.5,−0.25,0)，预留 [−1.25,0.75]，都以 −ε=−0.25 为中点，" "P(Y<−ε) = 0.9×0.5+0.1×0.5 = 0.5（浮点下也正好 0.5），不严格小于一半，不入池。" "第二例把 q5 往上挪 2⁻¹⁰，把握略小于 0.5，入池。ε 取 0.25 是为了全部是二进制精确数。"}) # ---- 4. V(S) V_IN = {"X": [(0.5, (-0.001, 0.0, 0.001)), (0.5, (0.019, 0.02, 0.021))], "Y": [(1, (0.009, 0.01, 0.011))]} VD = {k: R.Dist([(w, R.polyline(*t)) for w, t in comps]) for k, comps in V_IN.items()} S.append({"name": "V_method_example_X_Y", "section": "6.4 批内最好期望", "params": {}, "input": {"Ymixtures": {k: [{"weight": w, "range": tri(t)} for w, t in comps] for k, comps in V_IN.items()}}, "expected": {"V": {"X": F(R.V([VD["X"]])), "Y": F(R.V([VD["Y"]])), "X,Y": F(R.V([VD["X"], VD["Y"]]))}}, "note": "方法示例换成窄范围（三个数严格递增）：X 一半在 0 附近、一半在 0.02 附近；Y 在 0.01 附近。" "输入是改进量上的混合（每个成分一个三个数）。V(X) ≈ 0.010175（0 附近那一半有一点低于 0、被截成 0），" "V(Y) ≈ 0.01，V(X,Y) ≈ 0.015。"}) def batch_case(name, cands, b, direction, K, note, rand=None): dys, _, _, _ = R.forward_dists(cands, b, direction, LAM) pool = [c for c, d in dys.items() if R.pool_prob(d, EPS) < Fr(1, 2)] table, best = R.select_batch(dys, pool, K) pool_sorted = sorted(pool, key=R.id_key) full = R.V([dys[c] for c in pool_sorted]) ruler = full - best[1] if pool else Fr(0) S.append({"name": name, "section": "6.4 选批、分数尺读数；约定 10–14", "params": {"eps": EPS, "lambda": LAM, "direction": direction, "batchCap": K}, "input": {"b": b, "balance": 100, "candidates": {c: ranges_json(ts) for c, ts in cands.items()}}, "expected": { "pool": pool_sorted, "k": min(K, 100), "combos": [{"combo": c, "V": F(v)} for c, v in table], "selected": best[0], "selectedV": F(best[1]), "poolAllV": F(full), "rulerReading": F(ruler)}, "note": note}) batch_case("batch_r1_real_k2", R1_FWD, B_R1, "maximize", 2, "真实第 1 轮，每批上限 2：三选二，列出三个组合的 V。分数尺读数 = E[max(0, max_池 Y)] − V(S*)。") batch_case("batch_minimize_out_of_pool", MIN_FWD, B_MIN, "minimize", 2, "越小越好，c3 在池外：只在池内 {c1,c2,c4} 里选二；分数尺读数只按池内候选算（约定 13），池外的 c3 不进。") # V 完全相等：两个候选输入完全相同，编号按自然序，"c2" < "c10"。 # 预期值曾按 JS 字符串序选 c10；审阅指出方法说的"编号最小"是数字意义上的大小，改为自然序后选 c2，组合列表也变成先 c2 后 c10。 tie = {"c2": [(2.40, 2.50, 2.60)], "c10": [(2.40, 2.50, 2.60)]} batch_case("batch_tie_by_id", tie, 2.45, "maximize", 1, "两个候选的范围完全相同，V 完全相等；按编号的自然序，c2 排在 c10 前面，选 c2。") # 均匀随机选批（第十一部分；约定 12、13）：从全部候选（不看入池）抽 min(k, 候选数) 个；S* 是抽中的组合，V(S*) 照算， # 分数尺读数仍按候选池 V(池) − V(S*)。挑一个能抽中池外候选 c3 的种子，核对池外候选可以被抽中。 def uniform_case(name, cands, b, direction, K, must, note): dys, _, _, _ = R.forward_dists(cands, b, direction, LAM) pool = sorted([c for c, d in dys.items() if R.pool_prob(d, EPS) < Fr(1, 2)], key=R.id_key) k = min(K, 100) for rnd in range(1, 100): study = "uniform-batch" _, seed = seed_of(study, rnd, 1) chosen = sample(seed, list(cands), min(k, len(cands))) if must in chosen: break else: raise AssertionError("找不到能抽中池外候选的种子") assert must not in pool v = R.V([dys[c] for c in chosen]) full = R.V([dys[c] for c in pool]) S.append({"name": name, "section": "第十一部分 均匀随机；约定 12–14", "params": {"eps": EPS, "lambda": LAM, "direction": direction, "batchCap": K}, "input": {"b": b, "balance": 100, "seed": seed, "seedFrom": {"studyId": study, "round": rnd, "version": 1}, "candidates": {c: ranges_json(ts) for c, ts in cands.items()}}, "expected": {"pool": pool, "k": k, "selected": chosen, "selectedV": F(v), "poolAllV": F(full), "rulerReading": F(full - v)}, "note": note}) uniform_case("batch_uniform_random_minimize", MIN_FWD, B_MIN, "minimize", 2, "c3", "越小越好四候选，c3 在池外。均匀随机从全部四个候选里抽两个（编号按 JS 字符串序排好再做部分 Fisher–Yates），" "这个种子抽中了池外的 c3。selectedV = V(S*)；分数尺读数 = V(候选池 c1,c2,c4) − V(S*)。不列组合。") # ---- 5. 采纳线与采纳 adopt = [] for B, K, d, note in [ (2.60, None, "maximize", "没有已知最好：ε"), (2.60, 2.635983, "maximize", "正常：min(0.01, 0.035983/4=0.00899575)"), (2.6359, 2.635983, "maximize", "下限：(K*−B)/4 < ε/100，取 ε/100 = 0.0001"), (2.64, 2.635983, "maximize", "B 超过 K*：取 ε/100"), (2.50, 2.635983, "maximize", "差距大：取上限 ε"), (0.40, 0.38, "minimize", "越小越好：(B−K*)/4 = 0.005"), (0.37, 0.38, "minimize", "越小越好、B 已比 K* 好：ε/100"), ]: adopt.append({"B": B, "knownBest": K, "direction": d, "expected": F(R.adoption_line(EPS, B, K, d)), "note": note}) S.append({"name": "adoption_line_cases", "section": "6.5；约定 15", "params": {"eps": EPS}, "input": {"cases": [{k: v for k, v in a.items() if k not in ("expected",)} for a in adopt]}, "expected": {"lines": [a["expected"] for a in adopt]}, "note": "逐例对应 input.cases。"}) def adopt_decide(B, K, d, results): line = R.adoption_line(EPS, B, K, d) best = None for cid in sorted(results): v = results[cid] if v is None: continue if best is None or R.improve(v, results[best], d) > 0: best = cid if best is None: return {"line": F(line), "best": None, "adopt": False} gain = R.improve(results[best], B, d) return {"line": F(line), "best": best, "gain": F(gain), "adopt": gain > line} dec = [] for B, K, d, res, note in [ (2.60, 2.635983, "maximize", {"c1": 2.605, "c2": 2.6185, "c3": None}, "c2 涨 0.0185 > 0.00899575，采纳；c3 没有有效成绩"), (2.60, None, "maximize", {"c1": 2.61, "c2": 2.605}, "正好高 ε：不采纳"), (2.60, None, "maximize", {"c1": 2.6125, "c2": 2.6125}, "同分取编号小的 c1，涨 0.0125 > 0.01"), (0.40, None, "minimize", {"c1": 0.385, "c2": 0.395}, "越小越好：c1 改进 0.015 > 0.01"), ]: dec.append({"input": {"B": B, "knownBest": K, "direction": d, "results": res}, "expected": adopt_decide(B, K, d, res), "note": note}) S.append({"name": "adoption_decision_cases", "section": "6.5；约定 16", "params": {"eps": EPS}, "input": {"cases": [x["input"] for x in dec]}, "expected": {"cases": [x["expected"] for x in dec]}, "note": "改进量 > 采纳线（严格）才采纳；同分取编号小的。第二例是'正好高 ε'：浮点输入 2.61、2.60 的精确差 " "= 0.009999999999999787（浮点减法 2.61-2.6 也是这个数），不大于 0.01，不采纳。"}) write("forward.json", S, "预期值由有理数精确算出；与浮点实现比较用绝对误差 1e-12。") # =========================================================== info.json S = [] DELTA = EPS / 100 # 正向：r1 真实三候选，原值上切格（b 起切），惊讶用 c3 的真实成绩 _, draws_r1, _, _ = R.forward_dists(R1_FWD, B_R1, "maximize", LAM) fw = {} for c, d in draws_r1.items(): fw[c] = {"worldBits": R.world_bits(d, B_R1, DELTA)} v_c3 = 1.552176048093269 fw["r1-v1-c3"]["surprise"] = R.surprise(draws_r1["r1-v1-c3"], v_c3, B_R1, DELTA) far = R.surprise(draws_r1["r1-v1-c3"], 10.0, B_R1, DELTA) dis, hb, each = R.disagreement([R.single(R.polyline(*t)) for t in R1_FWD["r1-v1-c3"]], B_R1, DELTA) S.append({"name": "bits_forward_r1_real", "section": "6.9；约定 31–33", "params": {"eps": EPS, "lambda": LAM, "delta": DELTA, "direction": "maximize"}, "input": {"b": B_R1, "candidates": {c: ranges_json(ts) for c, ts in R1_FWD.items()}, "actual": {"r1-v1-c3": v_c3}, "farValue": 10.0}, "expected": {"candidates": fw, "surpriseFarValue": far, "disagreementC3": {"disagreement": dis, "avgBits": hb, "eachBits": each}}, "note": "原值上从 b 起切 Δ=ε/100 的小格，合并分布 = 三人折线 0.9 + 共同预留 0.1。" "c3 的实测 1.552176 远低于各人范围，只落在预留里。farValue=10 在预留之外，事前把握为 0，" "状态 priorZero（不是 0 或无穷大）。分歧量只用三人折线（不掺预留）。"}) # 消融格：r1 p1，b = 本次消融的基线 B_ABL = 0.9597642169962064 v_p1 = 2.250011787062861 persons = {} for a, t in zip(R1_ABL_AUTH, R1_ABL["p1"]): persons.setdefault(a, []).append(R.polyline(*t)) pd = [R.Dist([(Fr(1, len(ps)), p) for p in ps]) for ps in persons.values()] dis2, hb2, each2 = R.disagreement(pd, B_ABL, DELTA) S.append({"name": "bits_ablation_cell_r1_p1", "section": "6.9；约定 31–33", "params": {"eps": EPS, "lambda": LAM, "delta": DELTA}, "input": {"b": B_ABL, "explanations": [{"id": i, "author": a, "range": tri(t)} for i, a, t in zip(R1_ABL_IDS, R1_ABL_AUTH, R1_ABL["p1"])], "actual": v_p1}, "expected": {"worldBits": R.world_bits(cd, B_ABL, DELTA), "surprise": R.surprise(cd, v_p1, B_ABL, DELTA), "disagreement": dis2, "avgBits": hb2, "eachPersonBits": each2}, "note": "消融格的合并分布 = 6 条解释（每条 0.15）+ 这一格的预留 0.1。分歧量：每人取自己几条解释的等权平均作为" "'这个人的折线'，再比较三人平均的比特与各人比特的平均（不掺预留）。"}) # 越小越好的格子边界：原值上从 b 起切，b 以下是负的 j d_min = R.forward_dists(MIN_FWD, B_MIN, "minimize", LAM)[1]["c1"] edge_m1 = R.edge(B_MIN, DELTA, -1) cases = [] for v in [0.3999, edge_m1, 0.4, 0.38, 0.38005]: cases.append({"value": v, "result": R.surprise(d_min, v, B_MIN, DELTA)}) S.append({"name": "bits_minimize_bin_edges", "section": "6.9；约定 31–32", "params": {"eps": EPS, "lambda": LAM, "delta": DELTA, "direction": "minimize"}, "input": {"b": B_MIN, "candidates": {c: ranges_json(ts) for c, ts in MIN_FWD.items()}, "candidate": "c1", "values": [c["value"] for c in cases]}, "expected": {"worldBits": R.world_bits(d_min, B_MIN, DELTA), "surprise": [c["result"] for c in cases], "edgeMinus1": edge_m1}, "note": "越小越好也在分数原值上从 b 起切，第 j 格 [b+jΔ, b+(j+1)Δ)，j 可为负。边界一律按浮点 b + j*Δ 算：" f"b+(-1)Δ = {edge_m1!r}，所以字面 0.3999 落在 j=-2 格，而值正好等于该边界时落在 j=-1 格。" "合并分布是越小越好的那套（四候选共同预留，在原值上等价于原值折线的最低/最高点）。"}) # 落点四段、覆盖率 lc = [(2.44, 2.52, 2.62, v) for v in [2.40, 2.44, 2.50, 2.52, 2.60, 2.62, 2.65]] S.append({"name": "landing_segments_endpoints", "section": "6.9；约定 34", "params": {}, "input": {"cases": [{"range": tri(c[:3]), "value": c[3]} for c in lc]}, "expected": {"segments": [R.landing(*c) for c in lc], "coverage": sum(1 for c in lc if c[0] <= c[3] <= c[2]) / len(lc)}, "note": "vq9 above90。端点：v=q1 归一成到一半，" "v=q5 和 v=q9 归一半到九成。覆盖率 = 落在 [q1,q9]（含端点）的比例 = 5/7。"}) S.append({"name": "coverage_no_samples", "section": "6.9；约定 34", "params": {}, "input": {"cases": []}, "expected": {"coverage": {"status": "noSamples"}}, "note": "没有样本时返回明确状态，不是 0 或 NaN。"}) # 战绩 samples = [((2.44, 2.52, 2.62), 2.52), ((2.44, 2.52, 2.62), 2.62), ((2.44, 2.52, 2.62), 2.70), ((2.44, 2.52, 2.62), 2.44), ((2.44, 2.52, 2.62), 2.30), ((0.99, 1.0, 1.01), 1.0)] def track(ss): out = {"medianHigh": 0, "medianLow": 0, "pierceUpper": 0, "pierceLower": 0, "covered": 0} for (q1, q5, q9), v in ss: if q5 > v: out["medianHigh"] += 1 if q5 < v: out["medianLow"] += 1 if v > q9: out["pierceUpper"] += 1 if v < q1: out["pierceLower"] += 1 if q1 <= v <= q9: out["covered"] += 1 return out S.append({"name": "track_record_counts", "section": "6.10 第 4 条；约定 37", "params": {}, "input": {"forward": [{"range": tri(t), "value": v} for t, v in samples[:3]], "ablation": [{"range": tri(t), "value": v} for t, v in samples[3:]]}, "expected": {"forward": track(samples[:3]), "ablation": track(samples[3:])}, "note": "分数原值、不按方向。q5>v 中位数偏高；q5q9 穿上沿；v0 作者更乐观）。" "合并范围取 forward.json 里 forward_merge_r1_real 的原值分位数。作者自评三个数是示意输入。"}) write("info.json", S, "abs 用于事前把握、覆盖率、作者乐观差等普通数值。" "bits 用于世界比特和分歧量：本脚本按有理数算出每格质量后用浮点求熵（fsum），浮点实现每格质量是两个浮点 F 值之差，" "约 2 万到 5 万个小格累加后末位误差积到 1e-12 量级（第二轮审阅实测最大差 3.5e-12），留约 30 倍余量取 1e-10。" "surprise 是惊讶的容差公式：本脚本的 p 是有理数精确差，浮点实现只能算 F(右端) − F(左端)，" "两个不超过 1 的浮点数相减已有几个 2⁻⁵³ 的绝对误差，p 很小时（如 2.5e-6）换成 −log₂p 就放大成约 1e-12 量级的差。" "状态、格号、落点、计数必须完全相同。", {"bits": 1e-10, "surprise": "1e-12 + 2^-50 / (p * ln 2)"}) # =========================================================== misc.json S = [] ENT = [{"id": "e1", "author": "r1"}, {"id": "e2", "author": "r1"}, {"id": "e3", "author": "r2"}, {"id": "e4", "author": "r2"}, {"id": "e5", "author": "r3"}, {"id": "e6", "author": "r3"}] def entry_case(name, rs, ents, prefs, note): res = R.allocate_entries(rs, ents, prefs) S.append({"name": name, "section": "4.2 分配规则；约定 38", "params": {}, "input": {"researchers": rs, "entries": ents, "preferences": prefs}, "expected": {"assignments": res}, "note": note}) for name, rs, prefs, note in [ ("entries_rule1_unique", ["r1", "r2", "r3"], {"r1": ["e1", "e3", "e5"], "r2": ["e3", "e1", "e5"], "r3": ["e6", "e1", "e2"]}, "规则 1：第 1 志愿互不相同，各得第 1 志愿。"), ("entries_rule2_author_wins", ["r1", "r2", "r3"], {"r1": ["e3", "e1", "e2"], "r2": ["e3", "e4", "e1"], "r3": ["e3", "e5", "e1"]}, "规则 2：三人都选 e3，作者 r2 在其中，归 r2；r1、r3 进入第 2 轮，各得 e1、e5（第 2 志愿）。"), ("entries_rule2_smallest_index", ["r1", "r2", "r3"], {"r1": ["e5", "e2", "e3"], "r2": ["e1", "e4", "e5"], "r3": ["e1", "e4", "e2"]}, "规则 2：r2、r3 同选 e1，作者 r1 不在其中，归编号小的 r2；第 2 轮 r3 选 e4 得到。"), ("entries_identical_prefs", ["r1", "r2", "r3"], {"r1": ["e1", "e2", "e3"], "r2": ["e1", "e2", "e3"], "r3": ["e1", "e2", "e3"]}, "三人志愿完全相同：第 1 轮 e1 归作者 r1；第 2 轮 r2、r3 选 e2，作者 r1 已分到不在其中，归 r2；" "第 3 轮 r3 选 e3 得到。"), ("entries_rule3_exhausted", ["r1", "r2", "r3"], {"r1": ["e3", "e5"], "r2": ["e3", "e5"], "r3": ["e3", "e5"]}, "志愿用完：第 1 轮 e3 归作者 r2；第 2 轮 r1、r3 选 e5，作者 r3 在其中归 r3；r1 志愿用完，" "按规则 3 拿剩下编号最小的 e1，choice 为 null。"), ("entries_N1", ["r1"], {"r1": ["e2", "e1"]}, "N=1：直接给第一志愿。"), ]: ents = ENT if len(rs) > 1 else ENT[:2] entry_case(name, rs, ents, prefs, note) # "编号最小"按自然序判（researcher-2 < researcher-10、e2 < e10），与传入数组的次序无关：输入故意乱序给。 NAT_ENT = [{"id": "e10", "author": "researcher-2"}, {"id": "e1", "author": "researcher-3"}, {"id": "e2", "author": "researcher-3"}] entry_case("entries_natural_rule2", ["researcher-10", "researcher-3", "researcher-2"], NAT_ENT, {"researcher-10": ["e1"], "researcher-2": ["e1"], "researcher-3": ["e10"]}, "规则 2：researcher-10 与 researcher-2 同选 e1，作者 researcher-3 不在其中，归编号小的 researcher-2" "（按字符串序会错给 researcher-10）；researcher-3 独选 e10；researcher-10 志愿用完，按规则 3 拿剩下的 e2。") entry_case("entries_natural_rule3", ["researcher-10", "researcher-3", "researcher-2"], NAT_ENT, {"researcher-10": ["e1"], "researcher-2": ["e1"], "researcher-3": ["e1"]}, "三人同选 e1，作者 researcher-3 在其中归他；researcher-2、researcher-10 志愿用完，按编号从小到大拿剩下" "编号最小的入口：researcher-2 拿 e2，researcher-10 拿 e10（按字符串序会把 e10 排在 e2 前面）。") # 余额：第七部分示例表 rows = [("基线", ["started"]), ("第 1 轮正向", ["started"] * 2), ("第 1 轮消融", ["started"] * 2), ("第 2 轮正向", ["started"] * 2), ("第 3 轮正向", ["started"] * 2), ("第 3 轮消融", []), ("第 4 轮正向", ["started"] * 2), ("第 5 轮正向", ["started"] * 2)] runs, out = [], [] for label_, add in rows: runs = runs + add b = R.balance(13, runs) out.append({"step": label_, "added": add, **b, "forwardK": min(2, b["balance"]), "ablationMax": max(0, b["balance"] - 1)}) S.append({"name": "balance_part7_table", "section": "第七部分示例；约定 35–36", "params": {"totalRuns": 13, "batchCap": 2}, "input": {"steps": [{"step": l, "added": a} for l, a in rows]}, "expected": {"afterEachStep": out}, "note": "每一步追加的运行记录状态，累计后算余额。余额序列 12,10,8,6,4,4,2,0。" "forwardK = min(每批上限, 余额)；ablationMax = 余额−1（余额 ≤ 1 时 0）。"}) mix = ["started", "reserved", "notStarted", "started", "reserved"] b = R.balance(6, mix) S.append({"name": "balance_mixed_states", "section": "第七部分；约定 35–36", "params": {"totalRuns": 6, "batchCap": 2}, "input": {"runs": mix}, "expected": {**b, "forwardK": min(2, b["balance"]), "ablationMax": max(0, b["balance"] - 1)}, "note": "started 扣次；reserved（含状态未知）占着名额；notStarted 退还。已扣 2，占用 4，余额 2。"}) for tot, rs in [(3, ["started", "started", "started"]), (3, ["started", "started"]), (3, ["started"])]: b = R.balance(tot, rs) S.append({"name": f"balance_{b['balance']}", "section": "第七部分；约定 36", "params": {"totalRuns": tot, "batchCap": 2}, "input": {"runs": rs}, "expected": {**b, "forwardK": min(2, b["balance"]), "ablationMax": max(0, b["balance"] - 1)}, "note": "余额 0、1、2 时的正向批量与消融上限。"}) for study, rnd, ver in [("study-a", 1, 1), ("15-real-1", 3, 1)]: can, seed = seed_of(study, rnd, ver) rng = Rng(seed) draws = [rng.uniform_int(10) for _ in range(5)] # 十个以上的编号：候选、因子都按自然序排好再抽；和按字符串序排抽出的结果不同，能分出用错排序的实现。 # 预期值曾让候选按字符串序抽，改为自然序后 batchSampleMany 随之变化。 many_c = sample(seed, MANY_C, 4) many_p = sample(seed, MANY_P, 4) assert many_c != sample(seed, MANY_C, 4, string_order), "这个种子分不出候选的排序" assert many_p != sample(seed, MANY_P, 4, string_order), "这个种子分不出因子的排序" first6 = [hashlib.sha256(f"{seed}:{c}".encode()).digest()[:6].hex() for c in range(3)] S.append({"name": f"seed_{study}_r{rnd}", "section": "约定 39–40", "params": {}, "input": {"studyId": study, "round": rnd, "version": ver, "candidates": ["c1", "c2", "c3", "c4", "c5"], "batchSize": 2, "cells": ["p1", "p2", "p3", "p4"], "cellCount": 2, "candidatesMany": MANY_C, "cellsMany": MANY_P, "manyCount": 4}, "expected": {"canonical": can, "seed": seed, "first6BytesHex": first6, "uniformInt10First5": draws, "batchSample": sample(seed, ["c1", "c2", "c3", "c4", "c5"], 2), "cellSample": sample(seed, ["p1", "p2", "p3", "p4"], 2), "batchSampleMany": many_c, "cellSampleMany": many_p}, "note": "种子 = sha256(规范 JSON {round, studyId, version})。uniformInt：sha256(`${seed}:${counter}`) 前 6 字节大端，" "counter 从 0 起；uniformInt10First5 是同一个生成器连续 5 次。batchSample、cellSample 各自用一个从 counter=0 " "开始的新生成器，部分 Fisher–Yates 后取前 size 个再排序。候选、因子编号都按自然序（c2 < c10、p2 < p10）；" "*Many 两项用 12 个编号核对这一点。"}) write("misc.json", S, "全是离散结果，必须完全相同。") print("ok")