Files
Jyotisha/scripts/research/nadi_seconds_lib.py
T
jesse-uxandJesse_Chen c40f23c20e research(rectification): nadi second-level study helpers and N0 checks (BUG-1240)
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.
2026-10-05 23:26:05 +08:00

570 lines
20 KiB
Python

#!/usr/bin/env python3
"""Offline falsification helpers for second-scale nadi / sookshma / prana claims.
Research only. This module imports production chart and Vimshottari functions
and does not change them. Fitters and rankers never read ``true_minute``.
The five-level lords are the production chain
(``dasha_analyzer.build_antardasha`` applied twice more past pratyantar).
That is the same recursion ``_active_vimshottari`` uses for the first three
levels, including its midnight date anchor and 365.25-day year.
``calculate_five_level_dasha`` stays available as a side check; it is not the
scoring chain, because its 365.25636-day year does not reconcile to zero
against the production lords.
Classical D150 order (movable direct, fixed reverse, dual from the middle)
is marked ``variant_unverified``. No Chandra Kala Nadi verse text is stored.
"""
from __future__ import annotations
import hashlib
import json
import math
from collections.abc import Mapping, Sequence
from datetime import date, datetime, timedelta
from pathlib import Path
from random import Random
from typing import Any
import dasha_analyzer
import divisional_charts_extended
import domain_calculation_service
import narayana_dasha
from scripts.active_rectification_event_engine import (
AYANAMSA,
DOMAIN_CONFIG,
NODE_MODE,
_active_vimshottari,
_event_datetime,
_house_lords,
)
ROOT = Path(__file__).resolve().parents[2]
HOLDOUT_V5 = ROOT / "references" / "real_case_calibration" / "minute_rectification_holdout_v5.json"
PREREG_PATH = ROOT / "docs" / "research" / "nadi_seconds_preregistration_2026_10_05.json"
EVENT_CUTOFF = date(2026, 9, 14)
LEVELS = ("md", "ad", "pd", "sookshma", "prana")
# Structural reading only. See classical_nadi_index.
CLASSICAL_NADI_STATUS = "variant_unverified"
KM_PER_DEGREE_LAT = 111.32
_VARGA = divisional_charts_extended.DivisionalChartsCalculator()
_SKY_CACHE: dict[tuple, dict[str, Any]] = {}
def load_cases(path: Path = HOLDOUT_V5) -> list[dict[str, Any]]:
return list(json.loads(path.read_text(encoding="utf-8"))["cases"])
def canonical_bytes(payload: Any) -> bytes:
return (
json.dumps(payload, ensure_ascii=False, sort_keys=True, separators=(",", ":"))
+ "\n"
).encode("utf-8")
def sha256_bytes(raw: bytes) -> str:
return hashlib.sha256(raw).hexdigest()
def file_sha256(path: Path) -> str:
return hashlib.sha256(path.read_bytes()).hexdigest()
def round6(value: float) -> float:
return round(float(value), 6)
def subdivide(period: Mapping[str, Any]) -> list[dict[str, Any]]:
"""One production antardasha split. Adjacent rows abut by construction."""
return list(dasha_analyzer.build_antardasha(dict(period)))
def chain_at(birth_date: str, moon_longitude: float, event_at: datetime) -> list[dict[str, Any]]:
"""Five nested production periods active at ``event_at``.
``birth_date`` is the candidate's local calendar date (YYYY-MM-DD), the
same anchor ``_active_vimshottari`` passes through.
"""
return _chain_from_timeline(_timeline(birth_date, moon_longitude), event_at)
def five_lords(birth_date: str, moon_longitude: float, event_at: datetime) -> tuple[str, str, str, str, str]:
return tuple(str(period["lord"]) for period in chain_at(birth_date, moon_longitude, event_at)) # type: ignore[return-value]
def production_three(birth_date: str, moon_longitude: float, event_at: datetime) -> tuple[str, str, str]:
return _active_vimshottari(birth_date, float(moon_longitude) % 360.0, event_at)
def d150_equal(longitude: float) -> dict[str, Any]:
"""Equal-slice D150 via the repository ``calc_custom_varga`` (N=150)."""
row = _VARGA.calc_custom_varga(float(longitude) % 360.0, 150)
sign = str(row["sign"])
return {
"part_index": int(row["part_index"]),
"sign": sign,
"sign_idx": int(row["sign_idx"]),
"sign_lord": narayana_dasha.SIGN_LORDS[sign],
"source": "calc_custom_varga",
}
def classical_nadi_index(longitude: float) -> dict[str, Any]:
"""Reorder the 150 equal slices. Not a gate input.
Movable / fixed / dual follow ``DivisionalChartsCalculator`` sign classes.
Movable: part 0..149 in longitude order. Fixed: 149 down to 0.
Dual: start at the middle index 75 and wrap. The dual start is a
structural reading of "从中段起排", not a quotation, so the status is
``variant_unverified``. Named nadi verses are not used.
"""
longitude = float(longitude) % 360.0
sign_index = int(longitude // 30.0) % 12
part = int(d150_equal(longitude)["part_index"])
if sign_index in _VARGA.MOVABLE_SIGNS:
index = part
order = "direct"
elif sign_index in _VARGA.FIXED_SIGNS:
index = 149 - part
order = "reverse"
else:
index = (part + 75) % 150
order = "from_middle"
return {
"index": index,
"equal_part": part,
"sign_index": sign_index,
"order": order,
"status": CLASSICAL_NADI_STATUS,
}
def domain_target_lords(ascendant_index: int, domain: str) -> frozenset[str]:
houses = DOMAIN_CONFIG[domain][1]
return frozenset(_house_lords(int(ascendant_index), houses))
def explained(lords: Sequence[str], targets: frozenset[str], rule: str, segment_lord: str | None = None) -> bool:
"""Binary 'this second explains this event' for one pre-registered rule.
A4 / A5: level 4 or 5 lord is a domain house lord.
B1..B5: at least that many of the five levels are domain house lords.
N5A: equal-division D150 lagna sign lord is a domain house lord.
N5B: that same sign lord appears in any of the five levels.
"""
lords = tuple(lords)
if rule == "A4":
return lords[3] in targets
if rule == "A5":
return lords[4] in targets
if rule.startswith("B") and rule[1:].isdigit():
need = int(rule[1:])
return sum(lord in targets for lord in lords) >= need
if rule == "N5A":
return segment_lord is not None and segment_lord in targets
if rule == "N5B":
return segment_lord is not None and segment_lord in set(lords)
raise ValueError(f"unknown rule {rule}")
def fit_count(
lord_rows: Sequence[Sequence[str]],
domains: Sequence[str],
ascendant_index: int,
rule: str,
segment_lord: str | None = None,
) -> int:
"""How many events ``rule`` explains. No clock and no truth label."""
total = 0
for lords, domain in zip(lord_rows, domains, strict=True):
targets = domain_target_lords(ascendant_index, domain)
if explained(lords, targets, rule, segment_lord):
total += 1
return total
def rank_offsets(counts: Mapping[int, int]) -> list[int]:
"""Offsets tied for the highest count, ascending. Ties are all kept."""
if not counts:
return []
best = max(counts.values())
return sorted(offset for offset, count in counts.items() if count == best)
def candidate_moment(birth: Mapping[str, Any], offset_seconds: int) -> datetime:
base = datetime.strptime(f"{birth['date']} {str(birth['time'])[:5]}", "%Y-%m-%d %H:%M")
return base + timedelta(seconds=int(offset_seconds))
def sky_at(
birth: Mapping[str, Any],
moment: datetime,
*,
ayanamsa: str = AYANAMSA,
node_mode: str = NODE_MODE,
longitude: float | None = None,
) -> dict[str, Any]:
"""Moon and ascendant from the production chart. Seconds are passed through."""
lat = round(float(birth["latitude"]), 6)
lon = round(float(birth["longitude"] if longitude is None else longitude), 6)
tz = round(float(birth["timezone_offset"]), 4)
key = (moment.year, moment.month, moment.day, moment.hour, moment.minute, moment.second, lat, lon, tz, str(ayanamsa), str(node_mode))
cached = _SKY_CACHE.get(key)
if cached is not None:
return cached
chart = domain_calculation_service.compute_chart({
"year": moment.year,
"month": moment.month,
"day": moment.day,
"hour": moment.hour,
"minute": moment.minute,
"second": moment.second,
"lat": lat,
"lon": lon,
"tz": tz,
"ayanamsa": ayanamsa,
"node_mode": node_mode,
})
moon = float(chart["planets"]["Moon"]["lon"]) % 360.0
asc = float(chart["ascendant"]["lon"]) % 360.0
rahu = chart.get("planets", {}).get("Rahu") or {}
rahu_lon = float(rahu["lon"]) % 360.0 if isinstance(rahu.get("lon"), (int, float)) else None
sky = {
"birth_date": moment.date().isoformat(),
"moon_lon": moon,
"asc_lon": asc,
"asc_index": int(asc // 30.0) % 12,
"rahu_lon": rahu_lon,
}
_SKY_CACHE[key] = sky
return sky
def shift_longitude_km(longitude: float, latitude: float, kilometers_east: float) -> float:
"""Move the birth longitude by an east-west ground distance.
One degree of longitude is ``111.32 * cos(latitude)`` kilometres.
Near a pole the east-west degree collapses and the shift is refused.
"""
scale = KM_PER_DEGREE_LAT * math.cos(math.radians(float(latitude)))
if abs(scale) < 1e-3:
raise ValueError("east-west shift is undefined this close to a pole")
moved = float(longitude) + float(kilometers_east) / scale
return ((moved + 180.0) % 360.0) - 180.0
def day_events(case: Mapping[str, Any]) -> list[dict[str, Any]]:
"""Day-precision events copied without truth labels or narrative text."""
rows = []
for event in case["events"]:
if event.get("precision") != "day":
continue
rows.append({
"id": str(event["id"]),
"domain": str(event["domain"]),
"date": str(event["date"])[:10],
"precision": "day",
})
return rows
def _clamp_life_date(day: date, birth: date, cutoff: date) -> date:
guard = 0
while day <= birth and guard < 160:
day = day + timedelta(days=365)
guard += 1
guard = 0
while day > cutoff and guard < 160:
day = day - timedelta(days=365)
guard += 1
if birth < day <= cutoff:
return day
fallback = birth + timedelta(days=400)
if fallback > cutoff:
fallback = cutoff
if fallback <= birth:
fallback = birth + timedelta(days=1)
return fallback
def placebo_shift_dates(
events: Sequence[Mapping[str, Any]],
birth_date: str,
*,
seed: str,
low_days: int,
high_days: int,
cutoff: date = EVENT_CUTOFF,
) -> list[dict[str, Any]]:
"""Shift each day-event date by a seeded ±[low, high] day offset.
Domain stays. A result outside (birth, cutoff] is flipped, then repaired
by 365-day steps. The seed includes the case id so case order cannot
change one case's offsets.
"""
rng = Random(seed)
birth = date.fromisoformat(birth_date)
shifted = []
for event in events:
original = date.fromisoformat(str(event["date"])[:10])
magnitude = rng.randint(int(low_days), int(high_days))
sign = rng.choice((-1, 1))
candidate = original + timedelta(days=sign * magnitude)
if not (birth < candidate <= cutoff):
candidate = original + timedelta(days=-sign * magnitude)
if not (birth < candidate <= cutoff):
candidate = _clamp_life_date(original + timedelta(days=magnitude), birth, cutoff)
shifted.append({**event, "date": candidate.isoformat()})
return shifted
def placebo_swap_dates(
cases: Sequence[Mapping[str, Any]],
*,
seed: int,
cutoff: date = EVENT_CUTOFF,
) -> dict[str, list[dict[str, Any]]]:
"""Permute day-event dates across cases. Domains stay on the recipient event."""
slots: list[tuple[str, dict[str, Any]]] = []
for case in cases:
for event in day_events(case):
slots.append((str(case["case_id"]), event))
slots.sort(key=lambda item: (item[0], item[1]["id"]))
dates = [item[1]["date"] for item in slots]
order = list(range(len(dates)))
Random(int(seed)).shuffle(order)
by_case: dict[str, list[dict[str, Any]]] = {}
births = {str(case["case_id"]): str(case["birth"]["date"]) for case in cases}
for index, (case_id, event) in enumerate(slots):
raw = date.fromisoformat(dates[order[index]])
clamped = _clamp_life_date(raw, date.fromisoformat(births[case_id]), cutoff)
by_case.setdefault(case_id, []).append({**event, "date": clamped.isoformat()})
return by_case
def event_moments(events: Sequence[Mapping[str, Any]]) -> list[datetime]:
return [_event_datetime(event) for event in events] # type: ignore[arg-type]
def _timeline(birth_date: str, moon_longitude: float) -> list[dict[str, Any]]:
nakshatra, progress, _pada = dasha_analyzer.lon_to_nakshatra(float(moon_longitude) % 360.0)
timeline, _elapsed, _remaining, _lord = dasha_analyzer.build_dasha_timeline(
birth_date, nakshatra, progress,
)
return timeline
def _chain_from_timeline(timeline: Sequence[Mapping[str, Any]], event_at: datetime) -> list[dict[str, Any]]:
_index, major = dasha_analyzer.find_current(list(timeline), event_at)
levels = [major]
for _depth in range(4):
levels.append(dasha_analyzer.find_current_sub(subdivide(levels[-1]), event_at))
return levels
def lords_for_events(
birth_date: str,
moon_longitude: float,
events: Sequence[Mapping[str, Any]],
) -> list[tuple[str, str, str, str, str]]:
"""Five lords for every event. The mahadasha timeline is built once."""
timeline = _timeline(birth_date, moon_longitude)
rows = []
for moment in event_moments(events):
rows.append(tuple(str(period["lord"]) for period in _chain_from_timeline(timeline, moment))) # type: ignore[arg-type]
return rows
def counts_by_offset(
skies: Mapping[int, Mapping[str, Any]],
events: Sequence[Mapping[str, Any]],
rule: str,
*,
use_segment_lord: bool,
) -> dict[int, int]:
domains = [str(event["domain"]) for event in events]
counts: dict[int, int] = {}
for offset, sky in skies.items():
rows = lords_for_events(str(sky["birth_date"]), float(sky["moon_lon"]), events)
segment_lord = str(d150_equal(float(sky["asc_lon"]))["sign_lord"]) if use_segment_lord else None
counts[int(offset)] = fit_count(rows, domains, int(sky["asc_index"]), rule, segment_lord)
return counts
def mean_rate(counts: Mapping[int, int], offsets: Sequence[int], event_count: int) -> float | None:
if event_count <= 0 or not offsets:
return None
total = sum(counts[offset] for offset in offsets)
return total / (len(list(offsets)) * event_count)
def fraction_explaining_all(counts: Mapping[int, int], offsets: Sequence[int], event_count: int) -> float | None:
if event_count <= 0 or not offsets:
return None
hits = sum(1 for offset in offsets if counts[offset] == event_count)
return hits / len(list(offsets))
def paired_sign_flip_p(
differences: Sequence[float],
*,
seed: int,
permutations: int,
) -> dict[str, Any]:
"""One-sided paired permutation: share of sign-flips with mean >= observed.
p = (count + 1) / (permutations + 1), counting the observed assignment.
"""
values = [float(item) for item in differences]
n = len(values)
if n == 0:
return {"n": 0, "mean": None, "p": None, "extreme": None, "permutations": permutations}
observed = sum(values) / n
rng = Random(int(seed))
extreme = 0
for _ in range(int(permutations)):
total = 0.0
for value in values:
total += value if rng.randrange(2) == 0 else -value
if total / n >= observed - 1e-15:
extreme += 1
return {
"n": n,
"mean": round6(observed),
"p": round6((extreme + 1) / (int(permutations) + 1)),
"extreme": extreme,
"permutations": int(permutations),
}
def bootstrap_mean_ci(
values: Sequence[float],
*,
seed: int,
resamples: int,
) -> dict[str, Any]:
"""Percentile 95% interval of the mean. Passes only when the lower bound is > 0."""
pool = [float(item) for item in values]
n = len(pool)
if n == 0:
return {"n": 0, "mean": None, "low": None, "high": None, "excludes_zero_positive": False}
rng = Random(int(seed))
means = []
for _ in range(int(resamples)):
draw = 0.0
for _item in range(n):
draw += pool[rng.randrange(n)]
means.append(draw / n)
means.sort()
low_index = int(0.025 * (len(means) - 1))
high_index = int(0.975 * (len(means) - 1))
low = means[low_index]
high = means[high_index]
return {
"n": n,
"mean": round6(sum(pool) / n),
"low": round6(low),
"high": round6(high),
"excludes_zero_positive": bool(low > 0.0),
}
def minute_of(clock: str) -> int:
return int(str(clock)[3:5])
def rounded_to_five_minutes(clock: str) -> bool:
return minute_of(clock) % 5 == 0
def clock_delta_minutes(stamp: str, recorded: str) -> int:
def _clock(value: str) -> int:
hours, minutes = str(value)[:5].split(":")
return int(hours) * 60 + int(minutes)
delta = (_clock(stamp) - _clock(recorded)) % 1440
return delta - 1440 if delta > 720 else delta
def leaders_within(offsets: Sequence[int], limit_seconds: int) -> bool:
"""True only when every tied leader is inside the limit. An empty tie is a miss."""
return bool(offsets) and all(abs(int(offset)) <= int(limit_seconds) for offset in offsets)
def truth_audit(cases: Sequence[Mapping[str, Any]]) -> dict[str, Any]:
"""Evaluator-side audit. This function is allowed to read the recorded clock.
It does not read ``true_minute``; the recorded minute is ``birth.time``.
"""
rounded: list[str] = []
unrounded: list[str] = []
minute_hist: dict[str, int] = {}
per_case = []
for case in cases:
clock = str(case["birth"]["time"])[:5]
minute = f"{minute_of(clock):02d}"
minute_hist[minute] = minute_hist.get(minute, 0) + 1
case_id = str(case["case_id"])
(rounded if rounded_to_five_minutes(clock) else unrounded).append(case_id)
day_count = sum(1 for event in case["events"] if event.get("precision") == "day")
per_case.append({"case_id": case_id, "minute": minute, "day_events": day_count})
day_counts = [row["day_events"] for row in per_case]
return {
"cases": len(per_case),
"rounded_5min_count": len(rounded),
"unrounded_count": len(unrounded),
"rounded_5min_case_ids": rounded,
"unrounded_case_ids": unrounded,
"minute_histogram": minute_hist,
"day_event_total": sum(day_counts),
"day_events_ge_3": sum(1 for count in day_counts if count >= 3),
"day_events_ge_4": sum(1 for count in day_counts if count >= 4),
"day_events_ge_5": sum(1 for count in day_counts if count >= 5),
"day_events_zero": sum(1 for count in day_counts if count == 0),
"per_case": per_case,
}
def reconcile_case(case: Mapping[str, Any]) -> dict[str, Any]:
"""First three lords versus ``_active_vimshottari`` for every event at the recorded second."""
moment = candidate_moment(case["birth"], 0)
sky = sky_at(case["birth"], moment)
mismatches = []
for event in case["events"]:
event_at = _event_datetime(event) # type: ignore[arg-type]
ours = five_lords(sky["birth_date"], sky["moon_lon"], event_at)
theirs = production_three(sky["birth_date"], sky["moon_lon"], event_at)
if ours[:3] != theirs:
mismatches.append(str(event["id"]))
return {
"case_id": str(case["case_id"]),
"events": len(case["events"]),
"mismatches": mismatches,
}
def public_fields(case: Mapping[str, Any]) -> dict[str, Any]:
"""Copy the birth clock, place, and events a fitter may see."""
birth = case["birth"]
return {
"case_id": str(case["case_id"]),
"birth": {
"date": str(birth["date"]),
"time": str(birth["time"])[:5],
"latitude": float(birth["latitude"]),
"longitude": float(birth["longitude"]),
"timezone_offset": float(birth["timezone_offset"]),
},
"events": [
{
"id": str(event["id"]),
"domain": str(event["domain"]),
"date": str(event["date"]),
"precision": str(event["precision"]),
}
for event in case["events"]
],
}