Files
iomgaa 60c21ae346 feat: 研究规则 rules 模块(任务书 02)
rules 是纯计算,src/rules 共 1761 行:方法.md 第六部分的全部公式、第七部分的余额与扣次、4.2 的入口分配,以及种子与规则登记表。

预期值有三个来源:方法.md 的手算例;tests/fixtures/rules/ 下不看实现写的 Python 脚本;参考项目 calc15、calc14b 里与现行方法一致的部分。审阅做了两轮,每轮都换一个全新上下文的子 Agent,提出的问题都已修复或写明不修的理由。

用户确认的约定(原话"1–8 都同意"):入池、采纳、消融中发现都用严格不等号;成本用 max(0, ĝ₀);没测的格附"含它的最好组合"的净值;战绩在分数原值上判。三个数改为严格递增后,删掉了尖峰的全部专门处理(bfb96ba)。vocab 里 Range 的说明同步改为严格递增。

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-02 01:50:21 -04:00

654 lines
36 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""生成正向、信息账、入口/余额/种子三份预期值 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": "v<q1 below10;q1≤v<q5 from10to50;q5≤v≤q9 from50to90;v>q9 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 中位数偏高;q5<v 偏低;v>q9 穿上沿;v<q1 穿下沿;q1≤v≤q9 包住。"
"v=q5 既不偏高也不偏低;v=q9 算包住、不算穿上沿;窄范围 (0.99,1,1.01) 实测 1 只算包住。正向与消融分开数。"})
# 作者自评 vs 合并
mq = {}
for c, d in dys_r1.items():
yq = [R.quantile(d, p) for p in (Fr(1, 10), Fr(1, 2), Fr(9, 10))]
mq[c] = tuple(F(R.fr(B_R1) + q) for q in yq)
av = [
{"candidate": "r1-v1-c3", "author": (2.1, 2.5, 2.6), "merged": mq["r1-v1-c3"], "value": v_c3, "ran": True},
{"candidate": "r1-v1-c1", "author": (2.2, 2.5, 2.63), "merged": mq["r1-v1-c1"], "value": None, "ran": False},
]
rows = []
for a in av:
if not a["ran"] or a["value"] is None:
continue
rows.append({"candidate": a["candidate"],
"q5Diff": F(R.fr(a["author"][1]) - R.fr(a["merged"][1])),
"authorSegment": R.landing(*a["author"], a["value"]),
"mergedSegment": R.landing(*a["merged"], a["value"])})
S.append({"name": "author_vs_merged", "section": "6.10 第 4 条;约定 37",
"params": {"direction": "maximize"},
"input": {"items": [{"candidate": a["candidate"], "author": tri(a["author"]), "merged": tri(a["merged"]),
"value": a["value"], "ran": a["ran"]} for a in av]},
"expected": {"rows": rows},
"note": "只收跑过且有有效成绩的候选(c1 没跑,不进)。q5Diff = 按方向的 作者q5 − 合并q5(>0 作者更乐观)。"
"合并范围取 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")