root/Jyotisha
1
0
Fork 0
Code Issues 1 Pull Requests Actions Packages Projects Releases Wiki Activity

feat: import oracle collection and calibration parity

Browse Source
This commit is contained in:
732642856
2026-07-17 22:35:19 +08:00
parent ba9d43e128
commit 8c9b426344
18 changed files with 5344 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/collect_vedastro_official_parity_raw.py
+206
View File
@@ -0,0 +1,206 @@
#!/usr/bin/env python3
"""Collect bounded, secret-free VedAstro HTTP raw for divisional parity."""
from __future__ import annotations
import argparse
import hashlib
import json
import os
from pathlib import Path
from typing import Any
try:
from scripts import vedastro_service_adapter as adapter
except ModuleNotFoundError: # pragma: no cover - direct script execution
import vedastro_service_adapter as adapter
ROOT = Path(__file__).resolve().parents[1]
DEFAULT_PLANETS = ["Sun", "Moon", "Mars", "Mercury", "Jupiter", "Venus", "Saturn"]
VARGA_FIELDS = {
"D2": "PlanetHoraD2Signs",
"D4": "PlanetChaturthamshaD4Sign",
"D9": "PlanetNavamshaD9Sign",
"D10": "PlanetDashamamshaD10Sign",
}
SHADBALA_COMPONENT_FIELDS = [
"PlanetSthanaBala",
"PlanetDigBala",
"PlanetKalaBala",
"PlanetChestaBala",
"PlanetNaisargikaBala",
"PlanetDrikBala",
]
def _hash(payload: Any) -> str:
raw = json.dumps(payload, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode("utf-8")
return hashlib.sha256(raw).hexdigest()
def collect_chart_core(
case: dict[str, Any],
*,
case_id: str,
planets: list[str],
output_path: Path,
) -> dict[str, Any]:
endpoint = os.environ.get("VEDASTRO_API_ENDPOINT", "").strip()
if not endpoint:
raise RuntimeError("VEDASTRO_API_ENDPOINT is not configured")
manifest = adapter._official_full_snapshot_manifest(case, case_id)
chart_request = next(item for item in manifest["requests"] if item["section"] == "chart_core")
raw_responses: dict[str, Any] = {}
statuses: dict[str, str] = {}
scalar_responses: dict[str, Any] = {}
scalar_statuses: dict[str, str] = {}
component_responses: dict[str, Any] = {}
component_statuses: dict[str, str] = {}
retry_error_codes: list[int] = []
attempt_count = 0
for planet in planets:
request_item = {**chart_request, "fanout_value": planet}
try:
payload, attempts, retries = adapter._post_official_snapshot_section(endpoint, request_item)
raw_responses[planet] = payload
statuses[planet] = adapter._payload_status(payload)
attempt_count += attempts
retry_error_codes.extend(retries)
except Exception as exc: # Preserve partial raw instead of discarding the batch.
statuses[planet] = f"{type(exc).__name__}:{str(exc)[:200]}"
common_body = adapter._official_common_body(case)
scalar_requests = {
"shadbala": {
"section": "shadbala",
"endpoint_path": "/Calculate/AllPlanetStrength",
"method": "POST",
"headers": {"Content-Type": "application/json"},
"body": common_body,
"calculator_name": "AllPlanetStrength",
},
"ashtakavarga_sav": {
"section": "ashtakavarga_sav",
"endpoint_path": "/Calculate/AshtakvargaLifeMap",
"method": "POST",
"headers": {"Content-Type": "application/json"},
"body": {"birthTime": common_body["time"], "Ayanamsa": common_body["Ayanamsa"]},
"calculator_name": "AshtakvargaLifeMap",
},
"ashtakavarga_bav": {
"section": "ashtakavarga_bav",
"endpoint_path": "/Calculate/BhinnashtakavargaChart",
"method": "POST",
"headers": {"Content-Type": "application/json"},
"body": {"birthTime": common_body["time"], "Ayanamsa": common_body["Ayanamsa"]},
"calculator_name": "BhinnashtakavargaChart",
},
"ashtakavarga_sav_chart": {
"section": "ashtakavarga_sav_chart",
"endpoint_path": "/Calculate/SarvashtakavargaChart",
"method": "POST",
"headers": {"Content-Type": "application/json"},
"body": {"birthTime": common_body["time"], "Ayanamsa": common_body["Ayanamsa"]},
"calculator_name": "SarvashtakavargaChart",
},
}
for section, request_item in scalar_requests.items():
try:
payload, attempts, retries = adapter._post_official_snapshot_section(endpoint, request_item)
scalar_responses[section] = payload
scalar_statuses[section] = adapter._payload_status(payload)
attempt_count += attempts
retry_error_codes.extend(retries)
except Exception as exc:
scalar_statuses[section] = f"{type(exc).__name__}:{str(exc)[:200]}"
for planet in planets:
key = f"{planet}.chesta"
request_item = {
"section": "shadbala_chesta",
"endpoint_path": "/Calculate/PlanetChestaBala",
"method": "POST",
"headers": {"Content-Type": "application/json"},
"body": {
"planetName": {"Name": planet},
"time": common_body["time"],
"useSpecialSunMoon": False,
"Ayanamsa": common_body["Ayanamsa"],
},
"calculator_name": "PlanetChestaBala",
}
try:
payload, attempts, retries = adapter._post_official_snapshot_section(endpoint, request_item)
component_responses[key] = payload
component_statuses[key] = adapter._payload_status(payload)
attempt_count += attempts
retry_error_codes.extend(retries)
except Exception as exc:
component_statuses[key] = f"{type(exc).__name__}:{str(exc)[:200]}"
settings = {
"case_id": case_id,
"birth": {key: case.get(key) for key in ("year", "month", "day", "hour", "minute", "second", "lat", "lon", "tz")},
"ayanamsa_policy": case.get("ayanamsa_policy") or case.get("ayanamsa") or "lahiri",
"node_policy": case.get("node_policy") or case.get("node_mode") or "mean",
"endpoint_host": adapter._endpoint_host(endpoint),
"calculator": "AllPlanetData",
"planets": planets,
}
packet = {
"scope": "vedastro_official_http_divisional_parity_raw",
"status": "ok"
if statuses and scalar_statuses and component_statuses and all(value == "ok" for value in [*statuses.values(), *scalar_statuses.values(), *component_statuses.values()])
else "partial",
"source": "vedastro_official_http",
"settings": settings,
"coverage": {
"sections": [
*VARGA_FIELDS,
"shadbala_total",
"shadbala_components",
"ashtakavarga_bav",
"ashtakavarga_sav",
],
"field_map": VARGA_FIELDS,
"shadbala_component_fields": SHADBALA_COMPONENT_FIELDS,
"blocked_sections": [],
},
"fanout_statuses": dict(sorted(statuses.items())),
"scalar_statuses": dict(sorted(scalar_statuses.items())),
"component_statuses": dict(sorted(component_statuses.items())),
"attempt_count": attempt_count,
"retry_error_codes": retry_error_codes,
"raw_responses": raw_responses,
"scalar_responses": scalar_responses,
"component_responses": component_responses,
}
packet["response_hash"] = _hash({"chart_core": raw_responses, "scalar": scalar_responses, "components": component_responses})
output_path.parent.mkdir(parents=True, exist_ok=True)
output_path.write_text(json.dumps(packet, ensure_ascii=False, indent=2, sort_keys=True), encoding="utf-8")
return packet
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--case", choices=sorted(adapter.PARITY_CASES), default="beijing_first_use_demo")
parser.add_argument("--planets", default=",".join(DEFAULT_PLANETS))
parser.add_argument("--output", type=Path)
args = parser.parse_args()
planets = [item.strip() for item in args.planets.split(",") if item.strip()]
output = args.output or ROOT / "references" / "oracle" / "artifacts" / f"vedastro_{args.case}_divisional_raw.json"
report = collect_chart_core(
dict(adapter.PARITY_CASES[args.case]),
case_id=args.case,
planets=planets,
output_path=output,
)
print(json.dumps({"status": report["status"], "output": str(output), "response_hash": report["response_hash"], "fanout_statuses": report["fanout_statuses"]}, ensure_ascii=False, indent=2, sort_keys=True))
return 0 if report["status"] == "ok" else 1
if __name__ == "__main__":
raise SystemExit(main())
scripts/diagnose_external_engine_adapters.py
+52 -5
View File
@@ -25,25 +25,72 @@ REQUIRED_PARITY_OUTPUTS = ["D1", "D9", "D10", "D2", "D4", "Vimshottari", "Shadba
def _public_pyjhora_parity_manifest() -> dict:
path = Path("references/oracle/pyjhora_same_chart_parity_public_smoke_manifest.json")
try:
return json.loads(path.read_text(encoding="utf-8"))
baseline = json.loads(path.read_text(encoding="utf-8"))
except (OSError, json.JSONDecodeError):
return {"status": "not_available", "tested": False}
supplemental_path = Path("references/oracle/pyjhora_extended_parity_public_smoke_manifest.json")
try:
supplemental = json.loads(supplemental_path.read_text(encoding="utf-8"))
except (OSError, json.JSONDecodeError):
supplemental = None
result = dict(baseline)
supplements = [supplemental] if supplemental and supplemental.get("tested") else []
covered = set(baseline.get("covered_outputs", []))
sample_counts = {name: baseline.get("sample_count", 0) for name in covered}
partial_sample_counts = {}
source_reports = [baseline["source_report"]] if baseline.get("source_report") else []
for item in supplements:
item_count = item.get("sample_count", 0)
for name in item.get("covered_outputs", []):
covered.add(name)
sample_counts[name] = max(sample_counts.get(name, 0), item_count)
for name in item.get("partial_outputs", []):
partial_sample_counts[name] = max(partial_sample_counts.get(name, 0), item_count)
if item.get("source_report"):
source_reports.append(item["source_report"])
result["covered_outputs"] = [name for name in REQUIRED_PARITY_OUTPUTS if name in covered]
result["missing_required_outputs"] = [name for name in REQUIRED_PARITY_OUTPUTS if name not in covered]
result["output_sample_counts"] = sample_counts
result["partial_output_sample_counts"] = partial_sample_counts
result["source_reports"] = source_reports
result["supplemental_verifications"] = supplements
result["boundary"] = (
"Partial PyJHora verification with per-output sample counts. D2/D4/Ashtakavarga use a one-chart "
"supplemental replay; Shadbala absolute values remain mismatched. This is not JHora desktop, "
"VedAstro, jyotishganit, or predictive validation."
)
return result
def _same_chart_parity_contract(engines: dict) -> dict:
pyjhora_public = _public_pyjhora_parity_manifest()
replay_manifest = validate_parity_replay_manifest("references/oracle/three_engine_parity_replay_manifest.json")
replay_covers_all_engines = bool(
replay_manifest.get("tested") and not replay_manifest.get("missing_engines")
)
engine_states = {}
for name, engine in engines.items():
available = engine["status"] == "available"
engine_states[name] = {
"available": available,
"tested": bool(name == "PyJHora/JHora" and pyjhora_public.get("tested")),
"tested": bool(
replay_covers_all_engines
or (name == "PyJHora/JHora" and pyjhora_public.get("tested"))
),
"blocked": not available,
"blocking_reason": "" if available else engine["status"],
}
replay_manifest = validate_parity_replay_manifest("references/oracle/three_engine_parity_replay_manifest.json")
if not all(state["available"] for state in engine_states.values()):
contract_status = "blocked"
elif not all(state["tested"] for state in engine_states.values()):
contract_status = "partial"
else:
contract_status = replay_manifest.get("status", "partial")
return {
"status": "ready" if all(state["available"] for state in engine_states.values()) else "blocked",
"status": contract_status,
"required_outputs": REQUIRED_PARITY_OUTPUTS,
"expected_oracle_fields": {
"VedAstro": [
@@ -71,7 +118,7 @@ def _same_chart_parity_contract(engines: dict) -> dict:
"engine_states": engine_states,
"replay_manifest": replay_manifest,
"partial_verifications": {"PyJHora/JHora": pyjhora_public},
"boundary": "This is a parity contract. Public PyJHora partial verification does not close missing outputs or other engine raw-oracle requirements.",
"boundary": "Availability, executed raw coverage, and numerical parity are separate states; mismatch is not ready.",
}
scripts/gulika.py
+28 -3
View File
@@ -25,6 +25,18 @@ GHATIKA_END = {
6: {"day": 26, "night": 10},
}
# Saturn's zero-based share in eight equal day/night parts. Python weekday:
# Monday=0. This is the variant used by PyJHora's public black-box oracle.
SATURN_PART_START = {
0: {"day": 5, "night": 1},
1: {"day": 4, "night": 0},
2: {"day": 3, "night": 6},
3: {"day": 2, "night": 5},
4: {"day": 1, "night": 4},
5: {"day": 0, "night": 3},
6: {"day": 6, "night": 2},
}
def _sidereal_ascendant(jd_ut: float, lat: float, lon: float) -> float:
swe.set_sid_mode(swe.SIDM_LAHIRI)
@@ -38,17 +50,27 @@ def calculate_gulika(
lat: float,
lon: float,
tz: float,
method: str = "saturn_part_start",
) -> dict[str, Any]:
"""Return Gulika from local moment/location using Swiss sunrise and sunset."""
daynight = determine_daytime(moment, lat=lat, lon=lon, tz=tz)
is_day = bool(daynight["is_daytime"])
period = "day" if is_day else "night"
ghatika_end = GHATIKA_END[moment.weekday()][period]
start_jd = daynight["sunrise_jd_ut"] if is_day else daynight["sunset_jd_ut"]
end_jd = daynight["sunset_jd_ut"] if is_day else daynight["sunrise_jd_ut"] + 1.0
if end_jd <= start_jd:
end_jd += 1.0
segment_jd = start_jd + (end_jd - start_jd) * (ghatika_end / 30.0)
if method == "saturn_part_start":
part_index = SATURN_PART_START[moment.weekday()][period]
segment_fraction = part_index / 8.0
ghatika_end = None
elif method == "legacy_ghatika_end":
part_index = None
ghatika_end = GHATIKA_END[moment.weekday()][period]
segment_fraction = ghatika_end / 30.0
else:
raise ValueError("method must be saturn_part_start or legacy_ghatika_end")
segment_jd = start_jd + (end_jd - start_jd) * segment_fraction
longitude = _sidereal_ascendant(segment_jd, float(lat), float(lon))
return {
"scope": "gulika_prasna_marga",
@@ -58,10 +80,13 @@ def calculate_gulika(
"degree_in_sign": round(longitude % 30, 6),
"period": period,
"weekday": moment.weekday(),
"method": method,
"part_index": part_index,
"segment_fraction": segment_fraction,
"ghatika_end": ghatika_end,
"segment_jd_ut": segment_jd,
"daynight_evidence": daynight,
"ayanamsa": "lahiri",
"rule_source": "references/prashna-complete-guide.md#3.5",
"boundary": "Formula is implemented from the local classical guide; external JHora/PyJHora numeric parity remains required before enabling Sphuta or verdict layers.",
"boundary": "PyJHora-aligned Saturn-part-start variant. Numeric parity is evidence only and does not enable Prashna verdict layers.",
}
scripts/three_engine_parity_replay_validator.py
+22
View File
@@ -12,6 +12,17 @@ from typing import Any
REQUIRED_ENGINES = {"VedAstro", "PyJHora_JHora", "jyotishganit"}
REQUIRED_ROW_FIELDS = {"section", "field", "local_value", "oracle_values", "status"}
REQUIRED_HIGH_RIGOR_SECTIONS = {
"D1",
"D2",
"D4",
"D9",
"D10",
"ashtakavarga_bav",
"ashtakavarga_sav",
"shadbala_total",
"shadbala_components",
}
VALID_ROW_STATUSES = {"match", "mismatch", "blocked", "not_comparable"}
RAW_VERIFIED_STATUSES = {"verified", "official_verified", "imported"}
@@ -79,6 +90,12 @@ def validate_manifest(path: str | Path) -> dict[str, Any]:
for row in rows:
if isinstance(row, dict) and row.get("status") in counts:
counts[row["status"]] += 1
covered_sections = {
str(row.get("section"))
for row in rows
if isinstance(row, dict) and row.get("status") in {"match", "mismatch"}
}
missing_high_rigor_sections = sorted(REQUIRED_HIGH_RIGOR_SECTIONS - covered_sections)
if errors:
status = "invalid"
@@ -92,6 +109,9 @@ def validate_manifest(path: str | Path) -> dict[str, Any]:
elif counts["blocked"]:
status = "partial"
blocked_reason = "some_comparison_rows_blocked"
elif missing_high_rigor_sections:
status = "partial"
blocked_reason = "missing_high_rigor_sections"
else:
status = "pass"
blocked_reason = None
@@ -110,6 +130,8 @@ def validate_manifest(path: str | Path) -> dict[str, Any]:
"mismatch_count": counts["mismatch"],
"blocked_row_count": counts["blocked"],
"not_comparable_count": counts["not_comparable"],
"covered_sections": sorted(covered_sections),
"missing_high_rigor_sections": missing_high_rigor_sections,
"blocked_reason": blocked_reason,
"errors": errors,
"runtime_boundary": manifest.get("runtime_boundary", ""),
scripts/western_real_case_calibration.py
+73
View File
@@ -0,0 +1,73 @@
#!/usr/bin/env python3
"""Replay Western timing geometry for an existing public real-case manifest."""
from __future__ import annotations
import argparse
import json
from pathlib import Path
try:
from western_timing_engine import calculate_parans_status, calculate_secondary_progressions
except ImportError:
from scripts.western_timing_engine import calculate_parans_status, calculate_secondary_progressions
DEFAULT_MANIFEST = Path("references/real_case_calibration/replay_manifest.json")
def build_case(case_id: str, manifest_path: Path = DEFAULT_MANIFEST) -> dict:
manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
case = next((row for row in manifest["cases"] if row["case_id"] == case_id), None)
if case is None:
return {"status": "blocked", "reason": "case_id_not_found", "case_id": case_id}
event = case["event_outcomes"][0]
subject = case["subject"]
birth = {
"year": subject["year"], "month": subject["month"], "day": subject["day"],
"hour": subject["hour"], "minute": subject["minute"], "second": 0,
"latitude": subject["lat"], "longitude": subject["lon"], "timezone": subject["tz"],
}
progressions = calculate_secondary_progressions(target_date=event["event_date"], **birth)
parans = calculate_parans_status(target_date=event["event_date"], **birth)
return {
"scope": "western_real_case_calibration",
"status": "calculated_not_predictive_validation",
"case_id": case_id,
"subject": subject["name"],
"event": {
"date": event["event_date"],
"type": event["event_type"],
"source": event["source"],
},
"birth_source": subject["birth_source"],
"layers": {
"secondary_progressions": progressions,
"parans": parans,
},
"summary": {
"progressed_aspect_count": len(progressions["aspects"]),
"paran_event_count": parans["event_count"],
"paran_pair_count": len(parans["paran_pairs_within_4_minutes"]),
},
"boundary": "Known-event geometry replay only. One positive case cannot establish specificity, false-positive rate, or predictive accuracy.",
}
def main() -> int:
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument("--case-id", default="jobs_iphone_2007")
parser.add_argument("--manifest", type=Path, default=DEFAULT_MANIFEST)
parser.add_argument("--output", type=Path)
args = parser.parse_args()
report = build_case(args.case_id, args.manifest)
rendered = json.dumps(report, ensure_ascii=False, indent=2) + "\n"
if args.output:
args.output.parent.mkdir(parents=True, exist_ok=True)
args.output.write_text(rendered, encoding="utf-8")
print(rendered, end="")
return 0 if report["status"] != "blocked" else 1
if __name__ == "__main__":
raise SystemExit(main())
scripts/western_timing_engine.py
+71 -3
View File
@@ -132,6 +132,27 @@ def _progressed_planets(progressed_jd: float) -> dict[str, dict[str, Any]]:
return planets
def _quotidian_progressed_angles(progressed_jd: float, birth: dict[str, Any]) -> dict[str, Any]:
local = _jd_to_local(progressed_jd, birth["timezone"])
progressed_birth = {
**birth,
"year": local.year,
"month": local.month,
"day": local.day,
"hour": local.hour,
"minute": local.minute,
"second": local.second,
}
chart = build_tropical_natal_chart(**progressed_birth)
return {
"status": "used",
"method": "secondary_quotidian_progressed_date_same_location",
"progressed_local_time": local.isoformat(),
"angles": chart["natal"]["angles"],
"boundary": "Quotidian progressed-date angles; Naibod and solar-arc angle variants are separate methods.",
}
def calculate_secondary_progressions(*, target_date: str, **birth: Any) -> dict[str, Any]:
"""Calculate progressed planets using one ephemeris day per tropical year."""
natal_chart = build_tropical_natal_chart(**birth)
@@ -140,6 +161,7 @@ def calculate_secondary_progressions(*, target_date: str, **birth: Any) -> dict[
elapsed_years = (target_jd - birth_jd) / 365.242189
progressed_jd = birth_jd + elapsed_years
planets = _progressed_planets(progressed_jd)
progressed_angles = _quotidian_progressed_angles(progressed_jd, birth)
natal_points = {
**natal_chart["natal"]["planets"],
"ascendant": natal_chart["natal"]["angles"]["ascendant"],
@@ -155,8 +177,9 @@ def calculate_secondary_progressions(*, target_date: str, **birth: Any) -> dict[
"progressed_julian_day_ut": round(progressed_jd, 8),
"natal_sun_longitude": natal_chart["natal"]["planets"]["sun"]["longitude"],
"progressed_planets": planets,
"progressed_angles": progressed_angles,
"aspects": _cross_aspects(planets, natal_points),
"boundary": "Progressed planets only. Progressed angles, lunar phases, stations, duration, and interpretation remain separate audited layers.",
"boundary": "Progressed planets plus explicitly selected quotidian angles. Lunar phases, stations, duration, and interpretation remain separate audited layers.",
}
@@ -391,11 +414,56 @@ def calculate_transit_duration_scan(*, start_date: str, end_date: str, max_days:
def calculate_parans_status(*, target_date: str | None = None, **birth: Any) -> dict[str, Any]:
if not target_date:
return {"technique": "parans", "status": "blocked", "reason": "target_date_required"}
start_jd, target_local = _target_jd(target_date, birth["timezone"])
geopos = (float(birth["longitude"]), float(birth["latitude"]), 0.0)
modes = {
"rise": swe.CALC_RISE,
"upper_culmination": swe.CALC_MTRANSIT,
"set": swe.CALC_SET,
}
events: list[dict[str, Any]] = []
for planet, planet_id in _PLANETS.items():
for angle, mode in modes.items():
try:
status, values = swe.rise_trans(start_jd, planet_id, mode, geopos, 0.0, 15.0, swe.FLG_SWIEPH)
except (swe.Error, TypeError, ValueError):
continue
if status != 0:
continue
event_local = _jd_to_local(float(values[0]), birth["timezone"])
if event_local.date() != target_local.date():
continue
events.append({
"planet": planet,
"angle": angle,
"julian_day_ut": round(float(values[0]), 8),
"local_time": event_local.isoformat(),
})
events.sort(key=lambda row: (row["julian_day_ut"], row["planet"], row["angle"]))
pairs: list[dict[str, Any]] = []
for index, first in enumerate(events):
for second in events[index + 1:]:
delta_minutes = (second["julian_day_ut"] - first["julian_day_ut"]) * 1440.0
if delta_minutes > 4.0:
break
if first["planet"] == second["planet"]:
continue
pairs.append({
"first": {key: first[key] for key in ("planet", "angle", "local_time")},
"second": {key: second[key] for key in ("planet", "angle", "local_time")},
"separation_minutes": round(delta_minutes, 4),
})
return {
"technique": "parans",
"status": "blocked",
"status": "used",
"target_date": target_date,
"reason": "Parans need a dedicated rising/setting/culminating engine and latitude-aware event solver; not yet implemented in this repository.",
"method": "Swiss Ephemeris rise_trans latitude-aware angular events",
"event_count": len(events),
"events": events,
"paran_pairs_within_4_minutes": pairs,
"boundary": "Geometric rise/culmination/set simultaneity only; no interpretation or predictive claim is inferred.",
}