feat: add public oracle maturity audits
This commit is contained in:
732642856
@@ -0,0 +1,134 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Inter-chart linkage, dispositor chain, and motion-point audit."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import subprocess
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
SIGNS = ["Aries", "Taurus", "Gemini", "Cancer", "Leo", "Virgo", "Libra", "Scorpio", "Sagittarius", "Capricorn", "Aquarius", "Pisces"]
|
||||
SIGN_LORDS = {
|
||||
"Aries": "Mars", "Taurus": "Venus", "Gemini": "Mercury", "Cancer": "Moon",
|
||||
"Leo": "Sun", "Virgo": "Mercury", "Libra": "Venus", "Scorpio": "Mars",
|
||||
"Sagittarius": "Jupiter", "Capricorn": "Saturn", "Aquarius": "Saturn", "Pisces": "Jupiter",
|
||||
}
|
||||
|
||||
|
||||
def _run_json(command: list[str], timeout: int = 90) -> dict[str, Any]:
|
||||
completed = subprocess.run(command, cwd=ROOT, text=True, capture_output=True, timeout=timeout, check=False)
|
||||
if completed.returncode != 0:
|
||||
return {"status": "error", "stderr": (completed.stderr or completed.stdout).strip()[:500]}
|
||||
try:
|
||||
return json.loads(completed.stdout)
|
||||
except json.JSONDecodeError:
|
||||
return {"status": "invalid_json", "stdout_excerpt": completed.stdout[:500]}
|
||||
|
||||
|
||||
def _birth_args(args: argparse.Namespace) -> list[str]:
|
||||
out: list[str] = []
|
||||
for key in ("year", "month", "day", "hour", "minute", "second", "lat", "lon", "tz"):
|
||||
out.extend([f"--{key}", str(getattr(args, key))])
|
||||
return out
|
||||
|
||||
|
||||
def _house(sign: str, asc_sign: str) -> int:
|
||||
return (SIGNS.index(sign) - SIGNS.index(asc_sign)) % 12 + 1
|
||||
|
||||
|
||||
def _inter_chart_linkage(chart: dict[str, Any], varga: dict[str, Any], planets: list[str]) -> dict[str, Any]:
|
||||
d1_asc = chart["ascendant"]["sign"]
|
||||
charts = {"D1": {"ascendant": d1_asc, **chart["planets"]}}
|
||||
charts.update(varga.get("divisional_charts", {}))
|
||||
result: dict[str, Any] = {}
|
||||
for planet in planets:
|
||||
rows = {}
|
||||
for chart_name, payload in charts.items():
|
||||
asc = payload.get("ascendant")
|
||||
if isinstance(asc, dict):
|
||||
asc = asc.get("sign")
|
||||
item = payload.get(planet) if isinstance(payload, dict) else None
|
||||
if not item or not asc:
|
||||
continue
|
||||
sign = item["sign"]
|
||||
rows[chart_name] = {
|
||||
"sign": sign,
|
||||
"house_in_chart": _house(sign, asc),
|
||||
"lord_of_sign": SIGN_LORDS.get(sign),
|
||||
}
|
||||
result[planet] = rows
|
||||
return result
|
||||
|
||||
|
||||
def _dispositor_chain(chart: dict[str, Any], planet: str, max_depth: int) -> list[dict[str, Any]]:
|
||||
planets = chart["planets"]
|
||||
chain = []
|
||||
seen = set()
|
||||
current = planet
|
||||
for _ in range(max_depth):
|
||||
item = planets.get(current)
|
||||
if not item:
|
||||
break
|
||||
sign = item["sign"]
|
||||
lord = SIGN_LORDS[sign]
|
||||
chain.append({"planet": current, "sign": sign, "dispositor": lord, "dispositor_sign": planets.get(lord, {}).get("sign")})
|
||||
key = (current, lord)
|
||||
if key in seen or lord == current:
|
||||
break
|
||||
seen.add(key)
|
||||
current = lord
|
||||
return chain
|
||||
|
||||
|
||||
def build_report(args: argparse.Namespace) -> dict[str, Any]:
|
||||
birth = _birth_args(args)
|
||||
chart = _run_json([sys.executable, "scripts/jyotish_engine.py", "chart", *birth])
|
||||
varga = _run_json([sys.executable, "scripts/jyotish_engine.py", "varga", *birth, "--all"])
|
||||
sudarshana = _run_json([sys.executable, "scripts/jyotish_engine.py", "sudarshana", *birth, "--house", str(args.event_house)])
|
||||
tajika = _run_json([sys.executable, "scripts/jyotish_engine.py", "tajika", *birth, "--age", str(args.age), "--mode", "muntha"]) if args.age is not None else {"status": "blocked", "reason": "age_required"}
|
||||
planets = [p.strip() for p in args.planets.split(",") if p.strip()]
|
||||
return {
|
||||
"scope": "flying_star_audit",
|
||||
"event_house": args.event_house,
|
||||
"planets": planets,
|
||||
"inter_chart_linkage": _inter_chart_linkage(chart, varga, planets),
|
||||
"dispositor_chains": {planet: _dispositor_chain(chart, planet, args.max_depth) for planet in planets},
|
||||
"motion_points": {
|
||||
"bcp": {"status": "available_reference", "module": "scripts/bhrigu_pada_dasha.py", "note": "BCP/Bhrigu Pada exists; this audit records availability pending clean CLI integration."},
|
||||
"tajika_muntha": tajika,
|
||||
"sudarshana": sudarshana,
|
||||
},
|
||||
"boundaries": {
|
||||
"nadi_chain": "reference_only_not_full_machine_adjudicator",
|
||||
"ul_specific_chain": "pending_ul_output_adapter",
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser(description=__doc__)
|
||||
parser.add_argument("--year", type=int, required=True)
|
||||
parser.add_argument("--month", type=int, required=True)
|
||||
parser.add_argument("--day", type=int, required=True)
|
||||
parser.add_argument("--hour", type=int, required=True)
|
||||
parser.add_argument("--minute", type=int, required=True)
|
||||
parser.add_argument("--second", type=int, default=0)
|
||||
parser.add_argument("--lat", type=float, required=True)
|
||||
parser.add_argument("--lon", type=float, required=True)
|
||||
parser.add_argument("--tz", type=float, default=0)
|
||||
parser.add_argument("--age", type=int)
|
||||
parser.add_argument("--event-house", type=int, default=7)
|
||||
parser.add_argument("--planets", default="Venus,Saturn,Mars,Jupiter")
|
||||
parser.add_argument("--max-depth", type=int, default=12)
|
||||
args = parser.parse_args()
|
||||
print(json.dumps(build_report(args), ensure_ascii=False, indent=2, sort_keys=True))
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
@@ -0,0 +1,78 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Audit public non-standard marriage timing trigger cases."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
from collections import Counter
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
DEFAULT_CASES = ROOT / "references" / "non_standard_marriage_trigger_cases.json"
|
||||
|
||||
|
||||
def _resolve(path: str | None) -> Path:
|
||||
if not path:
|
||||
return DEFAULT_CASES
|
||||
candidate = Path(path)
|
||||
return candidate if candidate.is_absolute() else ROOT / candidate
|
||||
|
||||
|
||||
def audit_cases(path: str | None = None) -> dict[str, Any]:
|
||||
data = json.loads(_resolve(path).read_text(encoding="utf-8"))
|
||||
standard = set(data["standard_marriage_lords"])
|
||||
rows: list[dict[str, Any]] = []
|
||||
lord_counts: Counter[str] = Counter()
|
||||
link_counts: Counter[str] = Counter()
|
||||
|
||||
for case in data["cases"]:
|
||||
active = set(case.get("reported_dasha") or [])
|
||||
non_standard = sorted(active - standard)
|
||||
links = sorted(set(case.get("marriage_network_links") or []))
|
||||
lord_counts.update(non_standard)
|
||||
link_counts.update(links)
|
||||
rows.append(
|
||||
{
|
||||
"id": case["id"],
|
||||
"event_date": case["event_date"],
|
||||
"reported_dasha": case.get("reported_dasha", []),
|
||||
"classification": "non_standard_proxy" if non_standard and links else "standard_or_blocked",
|
||||
"non_standard_lords": non_standard,
|
||||
"marriage_network_links": links,
|
||||
"source": case["source"],
|
||||
"url": case["url"],
|
||||
}
|
||||
)
|
||||
|
||||
return {
|
||||
"scope": "non_standard_marriage_trigger_audit",
|
||||
"schema_version": 1,
|
||||
"case_count": len(data["cases"]),
|
||||
"standard_marriage_lords": sorted(standard),
|
||||
"summary": {
|
||||
"non_standard_proxy_cases": sum(row["classification"] == "non_standard_proxy" for row in rows),
|
||||
"top_non_standard_lords": lord_counts.most_common(),
|
||||
"top_marriage_network_links": link_counts.most_common(),
|
||||
},
|
||||
"rules": data["rules"],
|
||||
"rows": rows,
|
||||
"boundary": (
|
||||
"Use this only to prevent over-narrow Venus/Jupiter/Saturn timing claims. "
|
||||
"A non-standard lord still needs chart-specific D1/D9/UL/DK/A7, dasha subperiod and transit confirmation."
|
||||
),
|
||||
}
|
||||
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser(description=__doc__)
|
||||
parser.add_argument("--cases", default=None)
|
||||
parser.add_argument("--pretty", action="store_true")
|
||||
args = parser.parse_args()
|
||||
print(json.dumps(audit_cases(args.cases), ensure_ascii=False, indent=2 if args.pretty else None))
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
@@ -0,0 +1,152 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Replay local Jyotish engines against public birth-data seed cases."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import subprocess
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
DEFAULT_CASES = ROOT / "references" / "public_oracle_cases.json"
|
||||
|
||||
|
||||
def _birth_args(birth: dict[str, Any]) -> list[str]:
|
||||
args: list[str] = []
|
||||
for key in ("year", "month", "day", "hour", "minute", "second", "lat", "lon", "tz"):
|
||||
args.extend([f"--{key}", str(birth[key])])
|
||||
return args
|
||||
|
||||
|
||||
def _commands(case: dict[str, Any]) -> list[dict[str, Any]]:
|
||||
birth = _birth_args(case["birth"])
|
||||
return [
|
||||
{"id": "chart_d1", "command": ["chart", *birth]},
|
||||
{"id": "varga_d9", "command": ["varga", *birth, "--d9"]},
|
||||
{"id": "varga_full_core", "command": ["varga-full", *birth, "--divisions", "D2,D4,D7,D9,D10,D12,D16,D20,D24,D30,D60"]},
|
||||
{"id": "vimshottari_md_ad", "command": ["dasha", *birth, "--years", "45"]},
|
||||
{"id": "vimshottari_pd", "script": "scripts/vimshottari_subperiod_timeline.py", "args": [*birth, "--years", "45"]},
|
||||
{"id": "yoga", "command": ["yoga", *birth]},
|
||||
{"id": "shadbala", "command": ["shadbala", *birth]},
|
||||
{"id": "ashtakavarga", "command": ["ashtakavarga", *birth]},
|
||||
{"id": "kp", "command": ["kp", *birth]},
|
||||
{"id": "jaimini", "command": ["jaimini", *birth]},
|
||||
{"id": "narayana_dasha", "command": ["narayana-dasha", *birth]},
|
||||
{"id": "tajika", "command": ["tajika", *birth, "--age", "30"]},
|
||||
{"id": "solar_return", "command": ["solar-return", *birth, "--target-year", str(case["birth"]["year"] + 30)]},
|
||||
{"id": "muhurta", "command": ["muhurta", "--date", "2027-03-01", "--scan-days", "1"]},
|
||||
{"id": "transit", "command": ["transit", "--year", "2027", "--month", "3", "--day", "1", "--planet", "Jupiter,Saturn", "--tz", str(case["birth"]["tz"])]},
|
||||
{"id": "daily_transit_scan", "script": "scripts/daily_transit_window_scan.py", "args": ["--start", "2027-03-01", "--end", "2027-03-01", "--planets", "Jupiter,Saturn", "--tz", str(case["birth"]["tz"])]},
|
||||
{"id": "double_transit_pac", "command": ["double-transit-pac", *birth, "--date", "2027-03-01", "--house", "7"]},
|
||||
{"id": "vivah_saham", "command": ["vivah-saham", *birth, "--transit-date", "2027-03-01"]},
|
||||
{"id": "flying_star_audit", "script": "scripts/flying_star_audit.py", "args": [*birth, "--age", "30", "--event-house", "7"]},
|
||||
{"id": "bhava_chalit", "command": ["bhava-chalit", *birth]},
|
||||
{"id": "sudarshana", "command": ["sudarshana", *birth]},
|
||||
{"id": "aspects", "command": ["aspects", *birth]},
|
||||
{"id": "full_reading", "command": ["full-reading", *birth, "--today", "2027-03-01", "--transit-date", "2027-03-01"], "timeout": 90, "heavy": True},
|
||||
]
|
||||
|
||||
|
||||
def _run(item: dict[str, Any], timeout: int) -> dict[str, Any]:
|
||||
if "script" in item:
|
||||
command = [sys.executable, item["script"], *item["args"]]
|
||||
else:
|
||||
command = [sys.executable, "scripts/jyotish_engine.py", *item["command"]]
|
||||
completed = subprocess.run(command, cwd=ROOT, text=True, capture_output=True, timeout=item.get("timeout", timeout), check=False)
|
||||
status = "tested" if completed.returncode == 0 else "failed"
|
||||
payload_type = "unknown"
|
||||
if completed.returncode == 0:
|
||||
try:
|
||||
json.loads(completed.stdout)
|
||||
payload_type = "json"
|
||||
except json.JSONDecodeError:
|
||||
payload_type = "text"
|
||||
return {
|
||||
"engine": item["id"],
|
||||
"status": status,
|
||||
"returncode": completed.returncode,
|
||||
"payload_type": payload_type,
|
||||
"stderr_excerpt": (completed.stderr or "").strip()[:300],
|
||||
}
|
||||
|
||||
|
||||
def _oracle_rows(case: dict[str, Any]) -> list[dict[str, Any]]:
|
||||
rows: list[dict[str, Any]] = []
|
||||
oracles = case.get("expected_oracles", {})
|
||||
for provider, payload in oracles.items():
|
||||
if not isinstance(payload, dict):
|
||||
continue
|
||||
if "status" in payload:
|
||||
rows.append({"case_id": case["id"], "provider": provider, "target": "provider", "status": payload["status"]})
|
||||
continue
|
||||
for target, details in payload.items():
|
||||
if isinstance(details, dict):
|
||||
rows.append({
|
||||
"case_id": case["id"],
|
||||
"provider": provider,
|
||||
"target": target,
|
||||
"status": details.get("status", "pending_external_capture"),
|
||||
"artifact": details.get("artifact"),
|
||||
"packet": details.get("packet"),
|
||||
"expected": details.get("expected"),
|
||||
})
|
||||
return rows
|
||||
|
||||
|
||||
def build_report(args: argparse.Namespace) -> dict[str, Any]:
|
||||
data = json.loads(Path(args.cases).read_text(encoding="utf-8"))
|
||||
cases = data["cases"][:1] if args.quick else data["cases"]
|
||||
rows = []
|
||||
skipped = []
|
||||
oracle_rows = []
|
||||
for case in cases:
|
||||
oracle_rows.extend(_oracle_rows(case))
|
||||
for item in _commands(case):
|
||||
if item.get("heavy") and not args.include_heavy:
|
||||
skipped.append({"case_id": case["id"], "engine": item["id"], "status": "untested", "reason": "heavy; rerun with --include-heavy"})
|
||||
continue
|
||||
rows.append({"case_id": case["id"], **_run(item, args.timeout)})
|
||||
blocked = [
|
||||
{"engine": "VedAstro official full snapshot", "status": "blocked", "reason": "premium_key/budget/raw_response not closed"},
|
||||
{"engine": "PyJHora/JHora parity", "status": "blocked", "reason": "jhora dependency missing"},
|
||||
{"engine": "JHora desktop oracle", "status": "blocked", "reason": "manual desktop oracle required"},
|
||||
{"engine": "all_35_dasha_full_matrix", "status": "untested", "reason": "not mapped into public replay yet"},
|
||||
{"engine": "all_D1_to_D144_full_matrix", "status": "untested", "reason": "core subset replayed; exhaustive divisional sweep not yet run"},
|
||||
{"engine": "predictive_accuracy_claims", "status": "no_public_oracle", "reason": "public birth data alone cannot validate prediction accuracy"},
|
||||
]
|
||||
summary: dict[str, int] = {}
|
||||
for row in [*rows, *skipped, *blocked]:
|
||||
summary[row["status"]] = summary.get(row["status"], 0) + 1
|
||||
oracle_summary: dict[str, int] = {}
|
||||
for row in oracle_rows:
|
||||
oracle_summary[row["status"]] = oracle_summary.get(row["status"], 0) + 1
|
||||
return {
|
||||
"scope": "public_oracle_replay",
|
||||
"boundary": data["boundary"],
|
||||
"case_count": len(cases),
|
||||
"summary": summary,
|
||||
"oracle_summary": oracle_summary,
|
||||
"expected_oracles": oracle_rows,
|
||||
"rows": rows,
|
||||
"blocked_or_untested": [*skipped, *blocked],
|
||||
}
|
||||
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser(description=__doc__)
|
||||
parser.add_argument("--cases", default=str(DEFAULT_CASES))
|
||||
parser.add_argument("--quick", action="store_true")
|
||||
parser.add_argument("--include-heavy", action="store_true")
|
||||
parser.add_argument("--timeout", type=int, default=45)
|
||||
args = parser.parse_args()
|
||||
report = build_report(args)
|
||||
print(json.dumps(report, ensure_ascii=False, indent=2, sort_keys=True))
|
||||
return 1 if report["summary"].get("failed") else 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
@@ -0,0 +1,43 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Report traditional Jyotish maturity gaps beyond technique-name coverage."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
DEFAULT_MATRIX = ROOT / "references" / "traditional_maturity_gap_matrix.json"
|
||||
|
||||
|
||||
def build_report(matrix_path: str) -> dict[str, Any]:
|
||||
matrix = json.loads(Path(matrix_path).read_text(encoding="utf-8"))
|
||||
priorities = sorted(matrix["priorities"], key=lambda item: item.get("rank", 999))
|
||||
counts: dict[str, int] = {}
|
||||
for item in priorities:
|
||||
counts[item["priority"]] = counts.get(item["priority"], 0) + 1
|
||||
return {
|
||||
"scope": matrix["scope"],
|
||||
"boundary": matrix["boundary"],
|
||||
"execution_phases": matrix.get("execution_phases", []),
|
||||
"summary": counts,
|
||||
"ordered_priorities": [{"rank": item.get("rank"), "id": item["id"], "priority": item["priority"]} for item in priorities],
|
||||
"p0": [item for item in priorities if item["priority"] == "P0"],
|
||||
"p1": [item for item in priorities if item["priority"] == "P1"],
|
||||
"p2": [item for item in priorities if item["priority"] == "P2"],
|
||||
}
|
||||
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser(description=__doc__)
|
||||
parser.add_argument("--matrix", default=str(DEFAULT_MATRIX))
|
||||
args = parser.parse_args()
|
||||
print(json.dumps(build_report(args.matrix), ensure_ascii=False, indent=2, sort_keys=True))
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
Reference in New Issue
Block a user