#!/usr/bin/env python3 """Run the pre-registered nadi / sookshma / prana falsification study. Reads ``docs/research/nadi_seconds_preregistration_2026_10_05.json`` and refuses to score without it. The output JSON has no timestamps, so two runs with ``PYTHONHASHSEED=0`` are byte-identical. This script does not import ``true_minute``. """ from __future__ import annotations import argparse import json import sys import traceback from datetime import date from pathlib import Path from random import Random from typing import Any, Mapping, Sequence ROOT = Path(__file__).resolve().parents[2] if str(ROOT) not in sys.path: sys.path.insert(0, str(ROOT)) from scripts.research import nadi_seconds_lib as lib # noqa: E402 REPORT = ROOT / "docs" / "research" / "nadi_seconds_results_2026_10_05.json" def load_prereg() -> dict[str, Any]: if not lib.PREREG_PATH.is_file(): raise SystemExit(f"missing preregistration: {lib.PREREG_PATH}") return json.loads(lib.PREREG_PATH.read_text(encoding="utf-8")) def offsets_of(radius: int, step: int) -> list[int]: return list(range(-int(radius), int(radius) + 1, int(step))) def truth_minute_offsets() -> list[int]: return list(range(0, 60)) def collect_skies(birth: Mapping[str, Any], offsets: Sequence[int], **kwargs: Any) -> dict[int, dict[str, Any]]: return {int(offset): lib.sky_at(birth, lib.candidate_moment(birth, int(offset)), **kwargs) for offset in offsets} def lord_bundle(skies: Mapping[int, Mapping[str, Any]], events: Sequence[Mapping[str, Any]]) -> dict[str, Any]: lords: dict[int, list] = {} asc: dict[int, int] = {} segment: dict[int, str] = {} part: dict[int, int] = {} for offset, sky in skies.items(): lords[int(offset)] = lib.lords_for_events(str(sky["birth_date"]), float(sky["moon_lon"]), events) asc[int(offset)] = int(sky["asc_index"]) row = lib.d150_equal(float(sky["asc_lon"])) segment[int(offset)] = str(row["sign_lord"]) part[int(offset)] = int(row["part_index"]) return {"lords": lords, "asc": asc, "segment": segment, "part": part} def rule_counts(bundle: Mapping[str, Any], events: Sequence[Mapping[str, Any]], rule: str) -> dict[int, int]: domains = [str(event["domain"]) for event in events] use_segment = rule.startswith("N5") counts: dict[int, int] = {} for offset, rows in bundle["lords"].items(): targets = [lib.domain_target_lords(bundle["asc"][offset], domain) for domain in domains] segment = bundle["segment"][offset] if use_segment else None counts[offset] = sum( lib.explained(lords, target, rule, segment) for lords, target in zip(rows, targets, strict=True) ) return counts def rule_bits(bundle: Mapping[str, Any], events: Sequence[Mapping[str, Any]], rule: str) -> dict[int, list[bool]]: domains = [str(event["domain"]) for event in events] use_segment = rule.startswith("N5") bits: dict[int, list[bool]] = {} for offset, rows in bundle["lords"].items(): targets = [lib.domain_target_lords(bundle["asc"][offset], domain) for domain in domains] segment = bundle["segment"][offset] if use_segment else None bits[offset] = [ lib.explained(lords, target, rule, segment) for lords, target in zip(rows, targets, strict=True) ] return bits def mean_defined(values: Sequence[float | None]) -> float | None: nums = [float(value) for value in values if value is not None] if not nums: return None return lib.round6(sum(nums) / len(nums)) def rate(flags: Sequence[bool]) -> float | None: if not flags: return None return lib.round6(sum(bool(flag) for flag in flags) / len(flags)) def split_rates(rows: Sequence[Mapping[str, Any]], key: str, rounded: set[str]) -> dict[str, float | None]: return { "all": rate([bool(row[key]) for row in rows]), "rounded_5min": rate([bool(row[key]) for row in rows if row["case_id"] in rounded]), "unrounded": rate([bool(row[key]) for row in rows if row["case_id"] not in rounded]), "n_all": len(rows), "n_rounded_5min": sum(1 for row in rows if row["case_id"] in rounded), "n_unrounded": sum(1 for row in rows if row["case_id"] not in rounded), } def delivery_offsets(start: str | None, end: str | None, recorded: str, radius_minutes: int) -> list[int]: """One-second offsets covering each delivered clock minute, clipped to the search radius.""" if not start or not end: return [] cap = int(radius_minutes) * 60 start_min = lib.clock_delta_minutes(start, recorded) end_min = lib.clock_delta_minutes(end, recorded) def clip(lo: int, hi: int) -> list[int]: lo = max(lo, -cap) hi = min(hi, cap) if lo > hi: return [] return list(range(lo, hi + 1)) if start_min <= end_min: return clip(start_min * 60, end_min * 60 + 59) return sorted(set(clip(start_min * 60, cap) + clip(-cap, end_min * 60 + 59))) def six_probe(case: Mapping[str, Any], radius: int) -> dict[str, Any]: """futile_collect_stop_replay d1: six probes, no guided-window injection.""" from scripts.active_rectification_event_engine import AYANAMSA, NODE_MODE, compute_candidate_static_contexts from scripts.rectification.event_probes import discriminating_event_probes from scripts.rectification.refinement_packet import window_scan from scripts.rectification.scoring_service import build_event_contribution_matrix, score_from_matrix from scripts.research.fewer_probes_card_replay import _outcome from scripts.research.guided_collect_holdout_replay import _hhmm, posterior_state from scripts.research.minute_resolution_sweep import MINUTE_STEP, scoring_request_for from scripts.research.probe_supply_after_six import ASK_COUNT, TODAY recorded = str(case["birth"]["time"])[:5] request = scoring_request_for(dict(case), radius) request["ayanamsa"] = AYANAMSA request["node_mode"] = NODE_MODE request["minute_step"] = MINUTE_STEP static_contexts = compute_candidate_static_contexts(request) built = build_event_contribution_matrix(request, static_contexts=static_contexts) rows = score_from_matrix(request, built) times = [stamp for row in rows if (stamp := _hhmm(row.get("time")))] probes = discriminating_event_probes( {**request, "refresh_probes": False, "asked_probe_keys": []}, built, scan=window_scan(built), candidate_times=times, representative_time=recorded, today=TODAY, )[:ASK_COUNT] state = posterior_state(rows=rows, contexts=static_contexts, probes=probes, true_time=recorded) outcome = _outcome(state, recorded) scores = state["scores"] valid = list(state["valid"]) if not valid: leaders: list[str] = [] else: def _score(row: Mapping[str, Any]) -> float: stamp = _hhmm(row.get("time")) or "" return float(scores.get(stamp, row.get("score") or 0)) best = max(_score(row) for row in valid) leaders = sorted({ str(row.get("time"))[:5] for row in valid if abs(_score(row) - best) <= 1e-9 and str(row.get("time") or "")[:5] }) deltas = [lib.clock_delta_minutes(stamp, recorded) for stamp in leaders] return { "start": outcome["start"], "end": outcome["end"], "width": outcome["width"], "truth_in_range": bool(outcome["truth_in_range"]), "leader_times": leaders, "all_within_1min": bool(deltas) and all(abs(delta) <= 1 for delta in deltas), "any_exact": recorded in leaders, "error": None, } def gate_p(result: Mapping[str, Any]) -> bool: if result.get("mean") is None or result.get("extreme") is None: return False exact = (int(result["extreme"]) + 1) / (int(result["permutations"]) + 1) return bool(result["mean"] > 0 and exact < 0.05) def summarize_rule( rows: Sequence[Mapping[str, Any]], *, seed: int, permutations: int, ) -> dict[str, Any]: diffs = [float(row["diff"]) for row in rows] test = lib.paired_sign_flip_p(diffs, seed=seed, permutations=permutations) return { **test, "pass": gate_p(test), "truth_mean": mean_defined([row["truth"] for row in rows]), "placebo_mean": mean_defined([row["placebo"] for row in rows]), "window_mean": mean_defined([row["window"] for row in rows]), "fraction_grid_explains_all_mean": mean_defined([row["fraction_all"] for row in rows]), } def n2_case( bits: Mapping[int, Sequence[bool]], coarse: Sequence[int], events: Sequence[Mapping[str, Any]], placebo_bits: Mapping[int, Sequence[bool]], *, case_id: str, seed: str, ) -> dict[str, Any]: deltas_random = [] deltas_placebo = [] held_rates = [] for index, event in enumerate(events): counts = { offset: sum(flag for event_index, flag in enumerate(flags) if event_index != index) for offset, flags in bits.items() if offset in coarse } leaders = [offset for offset in lib.rank_offsets(counts) if offset in set(coarse)] if not leaders: continue held = sum(1 for offset in leaders if bits[offset][index]) / len(leaders) rng = Random(f"{seed}:{case_id}:{event['id']}") if len(leaders) >= len(coarse): sample = list(coarse) else: sample = rng.sample(list(coarse), len(leaders)) random_rate = sum(1 for offset in sample if bits[offset][index]) / len(sample) placebo_rate = sum(1 for offset in leaders if placebo_bits[offset][index]) / len(leaders) held_rates.append(held) deltas_random.append(held - random_rate) deltas_placebo.append(held - placebo_rate) return { "held_rate": mean_defined(held_rates), "delta_random": mean_defined(deltas_random), "delta_placebo": mean_defined(deltas_placebo), } def level_change(base: Sequence[Sequence[str]], other: Sequence[Sequence[str]], index: int) -> float | None: if not base or len(base) != len(other): return None changed = sum(left[index] != right[index] for left, right in zip(base, other, strict=True)) return lib.round6(changed / len(base)) def build(prereg: Mapping[str, Any], *, limit: int, radii: Sequence[int]) -> dict[str, Any]: cases = [lib.public_fields(case) for case in lib.load_cases()] if limit: cases = cases[:limit] grids = prereg["grids"] coarse = offsets_of(grids["coarse_radius_seconds"], grids["coarse_step_seconds"]) fine = offsets_of(grids["fine_radius_seconds"], grids["fine_step_seconds"]) truth_offsets = truth_minute_offsets() union = sorted(set(coarse) | set(fine) | set(truth_offsets)) rules = list(prereg["rules"]["computed"]) primary = list(prereg["rules"]["n1_primary"]) n1_min = int(prereg["samples"]["n1_min_day_events"]) n2_min = int(prereg["samples"]["n2_min_day_events"]) permutations = int(prereg["tests"]["permutations"]) resamples = int(prereg["tests"]["bootstrap_resamples"]) shift_low = int(prereg["placebo"]["shift_low_days"]) shift_high = int(prereg["placebo"]["shift_high_days"]) swap = lib.placebo_swap_dates(cases, seed=int(prereg["placebo"]["swap_seed"])) rounded = {case_id for case_id in lib.truth_audit(cases)["rounded_5min_case_ids"]} audit = lib.truth_audit(cases) recon = [lib.reconcile_case(case) for case in cases] n1_rows: dict[str, list[dict[str, Any]]] = {rule: [] for rule in primary} n1_placebos = ("P1", "P2") n5_rows: dict[str, list[dict[str, Any]]] = {rule: [] for rule in prereg["rules"]["n5_primary"]} n2_rows = [] n3a_rows = [] n4_rows = [] fraction_rows: dict[str, list[float]] = {rule: [] for rule in ("A4", "A5", "B1", "B2")} case_fit: list[dict[str, Any]] = [] for index, case in enumerate(cases, start=1): events = lib.day_events(case) birth = case["birth"] print(f"nadi {index}/{len(cases)} {case['case_id']} day_events={len(events)}", flush=True) skies = collect_skies(birth, union) real = lord_bundle(skies, events) if events else None shifted = lib.placebo_shift_dates( events, birth["date"], seed=prereg["placebo"]["shift_seed_template"].format(case_id=case["case_id"]), low_days=shift_low, high_days=shift_high, ) if events else [] p1 = lord_bundle(skies, shifted) if shifted else None swapped = swap.get(case["case_id"], []) p2 = lord_bundle(skies, swapped) if swapped else None fit_row: dict[str, Any] = { "case_id": case["case_id"], "day_events": len(events), "rounded_5min": case["case_id"] in rounded, } if real is not None: for rule in rules: if rule.startswith("N5") and rule not in prereg["rules"]["n5_primary"] and rule not in primary: continue counts = rule_counts(real, events, rule) truth = lib.mean_rate(counts, truth_offsets, len(events)) window = lib.mean_rate(counts, coarse, len(events)) fraction = lib.fraction_explaining_all(counts, coarse, len(events)) fit_row[rule] = { "truth": None if truth is None else lib.round6(truth), "window": None if window is None else lib.round6(window), "fraction_all": None if fraction is None else lib.round6(fraction), } if rule in fraction_rows and fraction is not None and len(events) >= n1_min: fraction_rows[rule].append(fraction) if rule in primary and len(events) >= n1_min and p1 is not None and p2 is not None: for label, bundle in (("P1", p1), ("P2", p2)): placebo_rate = lib.mean_rate(rule_counts(bundle, shifted if label == "P1" else swapped, rule), truth_offsets, len(events)) n1_rows[rule].append({ "case_id": case["case_id"], "placebo": label, "truth": lib.round6(truth or 0.0), "placebo_rate": None if placebo_rate is None else lib.round6(placebo_rate), "diff": None if placebo_rate is None else lib.round6((truth or 0.0) - placebo_rate), "window": None if window is None else lib.round6(window), "fraction_all": None if fraction is None else lib.round6(fraction), }) if rule in n5_rows and len(events) >= n1_min and p1 is not None and p2 is not None: for label, bundle in (("P1", p1), ("P2", p2)): placebo_rate = lib.mean_rate(rule_counts(bundle, shifted if label == "P1" else swapped, rule), truth_offsets, len(events)) n5_rows[rule].append({ "case_id": case["case_id"], "placebo": label, "truth": lib.round6(truth or 0.0), "placebo_rate": None if placebo_rate is None else lib.round6(placebo_rate), "diff": None if placebo_rate is None else lib.round6((truth or 0.0) - placebo_rate), "window": None if window is None else lib.round6(window), "fraction_all": None if fraction is None else lib.round6(fraction), }) if len(events) >= n2_min and p1 is not None: real_bits = rule_bits(real, events, prereg["rules"]["n2_primary"]) p1_bits = rule_bits(p1, shifted, prereg["rules"]["n2_primary"]) # P1 bits are aligned to shifted events, same index as real events. n2_rows.append({ "case_id": case["case_id"], "rounded_5min": case["case_id"] in rounded, **n2_case( real_bits, coarse, events, p1_bits, case_id=case["case_id"], seed=str(prereg["tests"]["n2_random_second_seed"]), ), }) a5_counts = rule_counts(real, events, prereg["rules"]["n3_primary"]) if events else {} leaders = lib.rank_offsets({offset: a5_counts.get(offset, 0) for offset in coarse}) if events else [] n3a_rows.append({ "case_id": case["case_id"], "rounded_5min": case["case_id"] in rounded, "day_events": len(events), "hit_60s": lib.leaders_within(leaders, 60) if events else False, "hit_120s": lib.leaders_within(leaders, 120) if events else False, "tied_leaders": len(leaders), }) else: n3a_rows.append({ "case_id": case["case_id"], "rounded_5min": case["case_id"] in rounded, "day_events": 0, "hit_60s": False, "hit_120s": False, "tied_leaders": len(coarse), }) case_fit.append(fit_row) # N4 at the recorded second only. if events: base_sky = skies[0] base_lords = lib.lords_for_events(base_sky["birth_date"], base_sky["moon_lon"], events) base_part = int(lib.d150_equal(base_sky["asc_lon"])["part_index"]) perturbations: dict[str, Any] = {} for name in prereg["n4"]["ayanamsas"]: if name == prereg["ayanamsa"]: continue sky = lib.sky_at(birth, lib.candidate_moment(birth, 0), ayanamsa=name) other = lib.lords_for_events(sky["birth_date"], sky["moon_lon"], events) perturbations[f"ayanamsa:{name}"] = { "level4": level_change(base_lords, other, 3), "level5": level_change(base_lords, other, 4), "d150_changed": int(lib.d150_equal(sky["asc_lon"])["part_index"]) != base_part, } for seconds in prereg["n4"]["time_offsets_seconds"]: sky = skies.get(int(seconds)) or lib.sky_at(birth, lib.candidate_moment(birth, int(seconds))) other = lib.lords_for_events(sky["birth_date"], sky["moon_lon"], events) perturbations[f"time:{int(seconds)}"] = { "level4": level_change(base_lords, other, 3), "level5": level_change(base_lords, other, 4), "d150_changed": int(lib.d150_equal(sky["asc_lon"])["part_index"]) != base_part, } for km in prereg["n4"]["longitude_shifts_km"]: try: moved = lib.shift_longitude_km(float(birth["longitude"]), float(birth["latitude"]), float(km)) sky = lib.sky_at(birth, lib.candidate_moment(birth, 0), longitude=moved) other = lib.lords_for_events(sky["birth_date"], sky["moon_lon"], events) perturbations[f"east_km:{km}"] = { "level4": level_change(base_lords, other, 3), "level5": level_change(base_lords, other, 4), "d150_changed": int(lib.d150_equal(sky["asc_lon"])["part_index"]) != base_part, "blocked": False, } except ValueError: perturbations[f"east_km:{km}"] = {"blocked": True} node_sky = lib.sky_at(birth, lib.candidate_moment(birth, 0), node_mode="true") perturbations["node:true"] = { "moon_changed": abs(float(node_sky["moon_lon"]) - float(base_sky["moon_lon"])) > 1e-6, "asc_changed": abs(float(node_sky["asc_lon"]) - float(base_sky["asc_lon"])) > 1e-6, "rahu_changed": node_sky["rahu_lon"] != base_sky["rahu_lon"], "level5": level_change(base_lords, lib.lords_for_events(node_sky["birth_date"], node_sky["moon_lon"], events), 4), } n4_rows.append({"case_id": case["case_id"], "day_events": len(events), "perturbations": perturbations}) def pack_comparisons(bucket: Mapping[str, list[dict[str, Any]]], seed_base: int) -> dict[str, Any]: packed = {} cursor = 0 for rule, rows in bucket.items(): packed[rule] = {} for label in n1_placebos: subset = [ { "diff": row["diff"], "truth": row["truth"], "placebo": row["placebo_rate"], "window": row["window"], "fraction_all": row["fraction_all"], } for row in rows if row["placebo"] == label and row["diff"] is not None ] packed[rule][label] = summarize_rule(subset, seed=seed_base + cursor, permutations=permutations) cursor += 1 return packed n1 = pack_comparisons(n1_rows, int(prereg["tests"]["n1_permutation_seed_base"])) n5 = pack_comparisons(n5_rows, int(prereg["tests"]["n5_permutation_seed_base"])) n1_pass = all(item["pass"] for rule in n1.values() for item in rule.values()) n5_pass = all(item["pass"] for rule in n5.values() for item in rule.values()) n2_values = [float(row["delta_random"]) for row in n2_rows if row["delta_random"] is not None] n2_placebo_values = [float(row["delta_placebo"]) for row in n2_rows if row["delta_placebo"] is not None] n2_ci = lib.bootstrap_mean_ci(n2_values, seed=int(prereg["tests"]["n2_bootstrap_seed"]), resamples=resamples) n2_placebo_ci = lib.bootstrap_mean_ci( n2_placebo_values, seed=int(prereg["tests"]["n2_bootstrap_seed"]) + 1, resamples=resamples, ) uniform_60 = lib.round6(sum(1 for offset in coarse if abs(offset) <= 60) / len(coarse)) uniform_120 = lib.round6(sum(1 for offset in coarse if abs(offset) <= 120) / len(coarse)) # N3b. Separate from the fit loop so a replay failure does not drop N1. n3b_rows = [] gate_radius = int(prereg["n3_protocol"]["radius_gate"]) for index, case in enumerate(cases, start=1): print(f"replay {index}/{len(cases)} {case['case_id']}", flush=True) by_radius = {} for radius in radii: try: snapshot = six_probe(case, radius) except Exception as exc: # noqa: BLE001 — recorded as a miss, message dropped snapshot = { "start": None, "end": None, "width": None, "truth_in_range": False, "leader_times": [], "all_within_1min": False, "any_exact": False, "error": type(exc).__name__, } traceback.print_exc(limit=2) events = lib.day_events(case) span = delivery_offsets(snapshot["start"], snapshot["end"], case["birth"]["time"], radius) if events and span: skies = collect_skies(case["birth"], span) bundle = lord_bundle(skies, events) counts = rule_counts(bundle, events, prereg["rules"]["n3_primary"]) leaders = lib.rank_offsets(counts) nadi_hit = lib.leaders_within(leaders, int(prereg["n3_protocol"]["nadi_limit_seconds"])) tied = len(leaders) else: nadi_hit = False tied = 0 by_radius[str(radius)] = { "truth_in_range": snapshot["truth_in_range"], "width": snapshot["width"], "baseline_all_within_1min": snapshot["all_within_1min"], "baseline_any_exact": snapshot["any_exact"], "nadi_all_within_60s": nadi_hit, "nadi_tied_leaders": tied, "improvement": int(nadi_hit) - int(snapshot["all_within_1min"]), "error": snapshot["error"], } n3b_rows.append({ "case_id": case["case_id"], "rounded_5min": case["case_id"] in rounded, "radii": by_radius, }) def n3b_summary(radius: int) -> dict[str, Any]: key = str(radius) improvements = [int(row["radii"][key]["improvement"]) for row in n3b_rows] inside = sum(1 for row in n3b_rows if row["radii"][key]["truth_in_range"]) ci = lib.bootstrap_mean_ci(improvements, seed=int(prereg["tests"]["n3_bootstrap_seed"]) + radius, resamples=resamples) return { "radius": radius, "truth_in_range": f"{inside}/{len(n3b_rows)}", "truth_in_range_count": inside, "truth_in_range_pass": inside >= 76 if len(n3b_rows) == 77 else None, "baseline_all_within_1min": split_rates( [{"case_id": row["case_id"], "hit": row["radii"][key]["baseline_all_within_1min"]} for row in n3b_rows], "hit", rounded, ), "baseline_any_exact": rate([row["radii"][key]["baseline_any_exact"] for row in n3b_rows]), "nadi_all_within_60s": split_rates( [{"case_id": row["case_id"], "hit": row["radii"][key]["nadi_all_within_60s"]} for row in n3b_rows], "hit", rounded, ), "improvement_ci": ci, "pass": bool(ci["excludes_zero_positive"] and len(n3b_rows) == 77 and inside >= 76), } def n4_aggregate(label: str, field: str) -> dict[str, Any]: values = [] d150_flags = [] weights = [] for row in n4_rows: item = row["perturbations"].get(label) or {} if item.get("blocked") or item.get(field) is None: continue values.append(float(item[field])) weights.append(int(row["day_events"])) if "d150_changed" in item: d150_flags.append(bool(item["d150_changed"])) event_rate = None if weights and sum(weights): event_rate = lib.round6(sum(value * weight for value, weight in zip(values, weights, strict=True)) / sum(weights)) return { "event_rate": event_rate, "case_rate": rate(d150_flags) if field == "level5" else None, "d150_case_rate": rate(d150_flags), "cases": len(values), } wording_hits = [] for name in prereg["n4"]["ayanamsas"]: if name == prereg["ayanamsa"]: continue label = f"ayanamsa:{name}" level5 = n4_aggregate(label, "level5") wording_hits.append({ "switch": f"{prereg['ayanamsa']}->{name}", "level5_event_rate": level5["event_rate"], "d150_case_rate": level5["d150_case_rate"], "over_20pct": bool( (level5["event_rate"] is not None and level5["event_rate"] > 0.20) or (level5["d150_case_rate"] is not None and level5["d150_case_rate"] > 0.20) ), }) wording_ban = any(item["over_20pct"] for item in wording_hits) n3a = { "grid": "coarse", "uniform_within_60s": uniform_60, "uniform_within_120s": uniform_120, "published_engine_prior_top1_pm10": 0.18, "hit_60s": split_rates( [{"case_id": row["case_id"], "hit": row["hit_60s"]} for row in n3a_rows], "hit", rounded, ), "hit_120s": split_rates( [{"case_id": row["case_id"], "hit": row["hit_120s"]} for row in n3a_rows], "hit", rounded, ), } n3b = {str(radius): n3b_summary(radius) for radius in radii} n3_pass = bool(n3b.get(str(gate_radius), {}).get("pass")) return { "prereg_sha256": lib.file_sha256(lib.PREREG_PATH), "library_sha256": lib.file_sha256(Path(__file__).with_name("nadi_seconds_lib.py")), "runner_sha256": lib.file_sha256(Path(__file__)), "dataset_sha256": lib.file_sha256(lib.HOLDOUT_V5), "swisseph": __import__("swisseph").version, "limit": limit, "n0": { "audit": {key: value for key, value in audit.items() if key != "per_case"}, "per_case_day_events": audit["per_case"], "reconciliation_mismatch_cases": [row["case_id"] for row in recon if row["mismatches"]], "reconciliation_events": sum(row["events"] for row in recon), }, "n1": { "pass": n1_pass and not limit, "comparisons": n1, "fraction_grid_explains_all_mean": {rule: mean_defined(values) for rule, values in fraction_rows.items()}, "cases": case_fit, }, "n2": { "pass": bool(n2_ci["excludes_zero_positive"]) and not limit, "versus_random": n2_ci, "versus_placebo_date": n2_placebo_ci, "cases": n2_rows, }, "n3": { "pass": n3_pass and not limit, "n3a": n3a, "n3b": n3b, "cases_n3a": n3a_rows, "cases_n3b": n3b_rows, }, "n4": { "wording_ban_second_scale": wording_ban, "ayanamsa_switches": wording_hits, "time": {str(seconds): n4_aggregate(f"time:{int(seconds)}", "level5") for seconds in prereg["n4"]["time_offsets_seconds"]}, "time_level4": {str(seconds): n4_aggregate(f"time:{int(seconds)}", "level4") for seconds in prereg["n4"]["time_offsets_seconds"]}, "longitude_km": {str(km): n4_aggregate(f"east_km:{km}", "level5") for km in prereg["n4"]["longitude_shifts_km"]}, "longitude_km_d150": {str(km): n4_aggregate(f"east_km:{km}", "level5")["d150_case_rate"] for km in prereg["n4"]["longitude_shifts_km"]}, "node_mean_to_true": { "moon_changed_cases": sum(1 for row in n4_rows if row["perturbations"]["node:true"]["moon_changed"]), "asc_changed_cases": sum(1 for row in n4_rows if row["perturbations"]["node:true"]["asc_changed"]), "rahu_changed_cases": sum(1 for row in n4_rows if row["perturbations"]["node:true"]["rahu_changed"]), "level5_event_rate": n4_aggregate("node:true", "level5")["event_rate"], }, "cases": n4_rows, }, "n5": { "pass": n5_pass and not limit, "named_text": "blocked", "named_text_reason": "no legal Chandra Kala Nadi source in the repository", "classical_index_status": lib.CLASSICAL_NADI_STATUS, "comparisons": n5, }, "gates": { "n1": n1_pass and not limit, "n2": bool(n2_ci["excludes_zero_positive"]) and not limit, "n3": n3_pass and not limit, "n4_blocks_second_scale_wording": wording_ban, "n5": n5_pass and not limit, }, } def main() -> int: parser = argparse.ArgumentParser() parser.add_argument("--limit", type=int, default=0) parser.add_argument("--radii", default="10,30,60") parser.add_argument("--out", default=str(REPORT)) args = parser.parse_args() prereg = load_prereg() radii = tuple(int(item) for item in str(args.radii).split(",") if item.strip()) payload = build(prereg, limit=int(args.limit), radii=radii) text = json.dumps(payload, ensure_ascii=False, sort_keys=True, indent=2) + "\n" out = Path(args.out) out.parent.mkdir(parents=True, exist_ok=True) out.write_text(text, encoding="utf-8") print(json.dumps(payload["gates"], ensure_ascii=False, sort_keys=True), flush=True) print(f"wrote {out}", flush=True) return 0 if __name__ == "__main__": raise SystemExit(main())