#!/usr/bin/env python3 """No.03 投票者モデル（完全グラフ）の実験器。 標準ライブラリだけを使う。各更新で異なる2人（聞き手、話し手）を 一様に選び、聞き手が話し手の意見をコピーする。結果は JSON に書く。 本番前は predictions.template.json を埋め、別名で凍結すること。 """ from __future__ import annotations import argparse import json import math import platform import random import statistics import time from datetime import datetime, timezone from pathlib import Path from typing import Any DEFAULT_SEEDS = list(range(3001, 3101)) PRODUCTION_SEEDS = list(range(3001, 4001)) def run_once(n_people: int, initial_a: int, seed: int, max_steps: int = 2_000_000) -> dict: if not (1 <= n_people and 0 <= initial_a <= n_people): raise ValueError("0 <= initial_a <= n_people が必要") rng = random.Random(seed) opinions = [1] * initial_a + [0] * (n_people - initial_a) rng.shuffle(opinions) steps = 0 a_count = initial_a while 0 < a_count < n_people and steps < max_steps: listener = rng.randrange(n_people) speaker = rng.randrange(n_people - 1) if speaker >= listener: speaker += 1 before = opinions[listener] after = opinions[speaker] opinions[listener] = after a_count += after - before steps += 1 return { "seed": seed, "initial_a": initial_a, "final_a": a_count, "winner": "A" if a_count == n_people else "B" if a_count == 0 else "timeout", "steps": steps, } def run_experiment(n_people: int, initial_values: list[int], seeds: list[int], max_steps: int = 2_000_000) -> dict: conditions = [] for initial_a in initial_values: runs = [run_once(n_people, initial_a, seed, max_steps) for seed in seeds] wins_a = sum(row["winner"] == "A" for row in runs) timeouts = sum(row["winner"] == "timeout" for row in runs) conditions.append({ "initial_a": initial_a, "n": len(runs), "wins_a": wins_a, "win_share_a": wins_a / len(runs) if runs else None, "closed_form_probability_a": initial_a / n_people, "timeouts": timeouts, "runs": runs, }) return { "model": "complete_graph_voter", "config": {"n_people": n_people, "initial_a_values": initial_values, "seeds": seeds, "max_steps": max_steps}, "conditions": conditions, } def pair_once(seed: int, n: int = 100, base_a: int = 50, max_steps: int = 2_000_000) -> dict[str, Any]: """Run the base and one-more-A worlds using one shared random stream.""" if not (1 <= n and 0 <= base_a < n): raise ValueError("0 <= base_a < n が必要") rng = random.Random(seed) idx = list(range(n)) rng.shuffle(idx) base = [int(value < base_a) for value in idx] one = [int(value < base_a + 1) for value in idx] changed = idx.index(base_a) count_base, count_one = base_a, base_a + 1 steps = steps_base = steps_one = 0 d = 1 d_max = 1 vanish = None while ((0 < count_base < n) or (0 < count_one < n)) and steps < max_steps: listener = rng.randrange(n) speaker = rng.randrange(n - 1) if speaker >= listener: speaker += 1 active_base = 0 < count_base < n active_one = 0 < count_one < n old_diff = base[listener] != one[listener] if active_base: before = base[listener] base[listener] = base[speaker] count_base += base[listener] - before steps_base += 1 if active_one: before = one[listener] one[listener] = one[speaker] count_one += one[listener] - before steps_one += 1 new_diff = base[listener] != one[listener] d += int(new_diff) - int(old_diff) steps += 1 d_max = max(d_max, d) if d == 0 and vanish is None: vanish = steps winner = lambda count: "A" if count == n else "B" if count == 0 else "timeout" return { "seed": seed, "changed": changed, "winner_base": winner(count_base), "winner_one": winner(count_one), "steps_base": steps_base, "steps_one": steps_one, "d_max": d_max, "vanish": vanish, } def _snap_step_list(end: int, early: int = 120, late: int = 120) -> list[int]: values = list(range(0, min(early, end) + 1)) if end > early: values.extend(round(early + (end - early) * k / late) for k in range(1, late + 1)) return sorted(set(values)) def snapshots(seed: int, n: int = 100, base_a: int = 50, max_steps: int = 2_000_000, snap_early: int = 120, snap_late: int = 120) -> dict[str, Any]: """Return paired state snapshots at the contractually specified steps.""" result = pair_once(seed, n, base_a, max_steps) end = max(result["steps_base"], result["steps_one"]) wanted = _snap_step_list(end, snap_early, snap_late) rng = random.Random(seed) idx = list(range(n)) rng.shuffle(idx) base = [int(value < base_a) for value in idx] one = [int(value < base_a + 1) for value in idx] cb, co = base_a, base_a + 1 rows: dict[int, tuple[str, str, int]] = {} def save(step: int) -> None: bs = "".join("A" if value else "B" for value in base) os = "".join("A" if value else "B" for value in one) rows[step] = (bs, os, sum(x != y for x, y in zip(base, one))) save(0) for step in range(1, end + 1): listener = rng.randrange(n) speaker = rng.randrange(n - 1) if speaker >= listener: speaker += 1 if 0 < cb < n: before = base[listener] base[listener] = base[speaker] cb += base[listener] - before if 0 < co < n: before = one[listener] one[listener] = one[speaker] co += one[listener] - before if step in wanted: save(step) result.update({ "snap_steps": wanted, "base": [rows[step][0] for step in wanted], "one": [rows[step][1] for step in wanted], "d": [rows[step][2] for step in wanted], }) return result def expected_steps(n: int) -> list[float]: """Solve the birth-death-chain equations with a tridiagonal sweep.""" if n < 1: raise ValueError("n >= 1 が必要") if n == 1: return [0.0, 0.0] size = n - 1 lower = [0.0] * size diag = [0.0] * size upper = [0.0] * size rhs = [1.0] * size for i, a in enumerate(range(1, n)): q = a * (n - a) / (n * (n - 1)) lower[i], diag[i], upper[i] = -q, 2 * q, -q for i in range(1, size): factor = lower[i] / diag[i - 1] diag[i] -= factor * upper[i - 1] rhs[i] -= factor * rhs[i - 1] solution = [0.0] * size solution[-1] = rhs[-1] / diag[-1] for i in range(size - 2, -1, -1): solution[i] = (rhs[i] - upper[i] * solution[i + 1]) / diag[i] return [0.0, *solution, 0.0] def metric_value(summary: dict[str, Any], metric: str) -> Any: value: Any = summary for part in metric.split("."): value = value[part] return value def prediction_hit(prediction: dict[str, Any], observed: Any) -> bool: if observed is None: return False kind = prediction["kind"] if kind == "eq": return observed == prediction["value"] if kind == "bool": return isinstance(observed, bool) and observed is prediction["value"] if kind == "range": return prediction["lo"] <= observed <= prediction["hi"] if kind == "max": return observed <= prediction["hi"] if kind == "min": return observed >= prediction["lo"] if kind == "in": return observed in prediction["values"] raise ValueError(f"unknown prediction kind: {kind}") def check_predictions(summary: dict[str, Any], predictions: dict[str, Any] | Path) -> dict[str, Any]: if isinstance(predictions, Path): predictions = json.loads(predictions.read_text(encoding="utf-8")) checks = [] for prediction in predictions["predictions"]: observed = metric_value(summary, prediction["metric"]) text = f"{observed:.10g}" if isinstance(observed, float) else str(observed) checks.append({ "id": prediction["id"], "text": prediction["text"], "metric": prediction["metric"], "observed": observed, "observed_text": text, "hit": prediction_hit(prediction, observed), }) return {"n": len(checks), "hits": sum(row["hit"] for row in checks), "checks": checks} def build_summary(seeds: list[int], n: int = 100, base_a: int = 50, max_steps: int = 2_000_000) -> dict[str, Any]: started = time.perf_counter() pairs = [pair_once(seed, n, base_a, max_steps) for seed in seeds] count = len(seeds) flips = [i for i, row in enumerate(pairs) if row["winner_base"] == "B" and row["winner_one"] == "A"] contrary = sum(row["winner_base"] == "A" and row["winner_one"] == "B" for row in pairs) vanished = sorted(row["vanish"] for row in pairs if row["vanish"] is not None) sweep_a = list(range(10, 91, 10)) exact = expected_steps(n) by_a: dict[str, Any] = {} for a in sweep_a: runs = [run_once(n, a, seed, max_steps) for seed in seeds] wins = sum(row["winner"] == "A" for row in runs) mean = statistics.fmean(row["steps"] for row in runs) by_a[str(a)] = { "a": a, "n": count, "wins_a": wins, "win_share_a": wins / count, "theory": a / n, "dev": wins / count - a / n, "steps_mean": mean, "steps_exact": exact[a], } quiet = next(i for i, row in enumerate(pairs) if row["winner_base"] == row["winner_one"] and row["d_max"] >= 3 and row["vanish"] is not None and row["vanish"] <= 120) def featured(index: int, rule: str) -> dict[str, Any]: row = snapshots(seeds[index], n, base_a, max_steps) row.update({"rule": rule, "index": index}) return row steps_base = [row["steps_base"] for row in pairs] steps_one = [row["steps_one"] for row in pairs] mean_base, mean_one = statistics.fmean(steps_base), statistics.fmean(steps_one) pair = { "n": count, "seeds_first": seeds[0], "seeds_last": seeds[-1], "winner_base": "".join(row["winner_base"][0] if row["winner_base"] != "timeout" else "T" for row in pairs), "winner_one": "".join(row["winner_one"][0] if row["winner_one"] != "timeout" else "T" for row in pairs), "steps_base": steps_base, "steps_one": steps_one, "changed": [row["changed"] for row in pairs], "d_max": [row["d_max"] for row in pairs], "vanish": [row["vanish"] for row in pairs], "wins_base": sum(row["winner_base"] == "A" for row in pairs), "wins_one": sum(row["winner_one"] == "A" for row in pairs), "flips": len(flips), "same": sum(row["winner_base"] == row["winner_one"] for row in pairs), "contrary": contrary, "timeouts": sum("timeout" in (row["winner_base"], row["winner_one"]) for row in pairs), "flip_idx": flips, "flip_seeds": [seeds[i] for i in flips], "flip_share": len(flips) / count, "vanish_median": statistics.median_low(vanished), "vanish_p90": vanished[math.ceil(0.9 * len(vanished)) - 1], "vanish_max": max(vanished), "vanish_within_100_share": sum(value <= 100 for value in vanished) / count, "d_k": 10, "d_ge10": sum(row["d_max"] >= 10 for row in pairs), "d_ge10_share": sum(row["d_max"] >= 10 for row in pairs) / count, "steps_base_mean": mean_base, "steps_base_median": statistics.median_low(steps_base), "steps_one_mean": mean_one, "steps_one_ratio": mean_one / mean_base, "flip_steps_ratio": (statistics.fmean(steps_one[i] for i in flips) / mean_one) if flips else None, } return { "model": "complete_graph_voter", "config": {"n_people": n, "base_a": base_a, "one_a": base_a + 1, "seeds": seeds, "n_per_condition": count, "sweep_a": sweep_a, "max_steps": max_steps, "d_k": 10, "snap_early": 120, "snap_late": 120, "python": platform.python_version(), "rng": "random.Random(seed)"}, "groups": {"pair": {f"a{base_a}": {"n": count}, f"a{base_a + 1}": {"n": count}}, "sweep": {f"a{a}": {"n": count} for a in sweep_a}}, "pair": pair, "featured": { "flip": featured(flips[0], "結末が変わった組のうち最小の種"), "quiet": featured(quiet, "結末が変わらず d_max ≥ 3 かつ vanish ≤ 120 の組のうち最小の種"), }, "sweep": {"a": sweep_a, "by_a": by_a, "max_abs_dev": max(abs(row["dev"]) for row in by_a.values()), "monotone": all(by_a[str(a)]["win_share_a"] < by_a[str(b)]["win_share_a"] for a, b in zip(sweep_a, sweep_a[1:])), "argmax_steps": max(sweep_a, key=lambda a: by_a[str(a)]["steps_mean"])}, "theory": {"p_base": base_a / n, "p_one": (base_a + 1) / n, "flip_share": 1 / n, "d_ge10_share": 0.1, "steps_exact": {str(a): exact[a] for a in sorted(set(sweep_a + [base_a + 1]))}, "method": "出生死滅連鎖の連立一次方程式を三重対角の掃き出しで解く"}, "generated_at": datetime.now(timezone.utc).isoformat(), "runtime_s": time.perf_counter() - started, } def dump_json(path: Path, value: dict[str, Any]) -> None: path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(value, ensure_ascii=False, sort_keys=True, indent=1) + "\n", encoding="utf-8") def main(argv: list[str] | None = None) -> int: parser = argparse.ArgumentParser() parser.add_argument("--output") parser.add_argument("--results-dir", type=Path) parser.add_argument("--n-people", type=int, default=100) parser.add_argument("--initial-a", type=int, nargs="+", default=[49, 50, 51]) parser.add_argument("--seeds", type=int, nargs="+", default=DEFAULT_SEEDS) parser.add_argument("--max-steps", type=int, default=2_000_000) args = parser.parse_args(argv) if args.results_dir: summary = build_summary(PRODUCTION_SEEDS, args.n_people, 50, args.max_steps) checks = check_predictions(summary, Path(__file__).with_name("predictions.json")) dump_json(args.results_dir / "summary.json", summary) dump_json(args.results_dir / "predictions_check.json", checks) return 0 if not args.output: parser.error("--output または --results-dir が必要") result = run_experiment(args.n_people, args.initial_a, args.seeds, args.max_steps) path = Path(args.output) path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(result, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") return 0 if __name__ == "__main__": raise SystemExit(main())