jesse-ux
217cd4a767
Five-level Vimshottari and equal-division D150 helpers, plus a runner that refuses to score until the preregistration file exists. First three lords match the production chain on the v5 events. No production files changed.
677 lines
31 KiB
Python
677 lines
31 KiB
Python
#!/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())
|