From a824588d905442ec6e5d1eb883d3587af9f5e785 Mon Sep 17 00:00:00 2001 From: 732642856 <732642856@qq.com> Date: Mon, 29 Jun 2026 10:38:17 +0800 Subject: [PATCH] Add Shadbala oracle closure audit chain --- scripts/shadbala_d3_branch_audit.py | 110 ++++++++++++ scripts/shadbala_d3_mapping_audit.py | 149 +++++++++++++++++ scripts/shadbala_oracle_batch_summary.py | 109 ++++++++++++ ...adbala_oracle_component_cluster_summary.py | 141 ++++++++++++++++ scripts/shadbala_sapta_dignity_whitelist.py | 91 ++++++++++ scripts/shadbala_sapta_layer_hotspots.py | 155 +++++++++++++++++ scripts/shadbala_sthana_targeted_audit.py | 157 ++++++++++++++++++ tests/test_shadbala_d3_branch_audit.py | 26 +++ tests/test_shadbala_d3_mapping_audit.py | 18 ++ tests/test_shadbala_oracle_batch_summary.py | 18 ++ ...adbala_oracle_component_cluster_summary.py | 19 +++ .../test_shadbala_sapta_dignity_whitelist.py | 18 ++ tests/test_shadbala_sapta_layer_hotspots.py | 21 +++ tests/test_shadbala_sthana_targeted_audit.py | 19 +++ 14 files changed, 1051 insertions(+) create mode 100644 scripts/shadbala_d3_branch_audit.py create mode 100644 scripts/shadbala_d3_mapping_audit.py create mode 100644 scripts/shadbala_oracle_batch_summary.py create mode 100644 scripts/shadbala_oracle_component_cluster_summary.py create mode 100644 scripts/shadbala_sapta_dignity_whitelist.py create mode 100644 scripts/shadbala_sapta_layer_hotspots.py create mode 100644 scripts/shadbala_sthana_targeted_audit.py create mode 100644 tests/test_shadbala_d3_branch_audit.py create mode 100644 tests/test_shadbala_d3_mapping_audit.py create mode 100644 tests/test_shadbala_oracle_batch_summary.py create mode 100644 tests/test_shadbala_oracle_component_cluster_summary.py create mode 100644 tests/test_shadbala_sapta_dignity_whitelist.py create mode 100644 tests/test_shadbala_sapta_layer_hotspots.py create mode 100644 tests/test_shadbala_sthana_targeted_audit.py diff --git a/scripts/shadbala_d3_branch_audit.py b/scripts/shadbala_d3_branch_audit.py new file mode 100644 index 00000000..80fa4f84 --- /dev/null +++ b/scripts/shadbala_d3_branch_audit.py @@ -0,0 +1,110 @@ +#!/usr/bin/env python3 +"""Pin D3 drift to the exact calc_sthana_bala branch used by Shadbala.""" + +from __future__ import annotations + +import argparse +import json +import sys +from pathlib import Path +from typing import Any + + +ROOT = Path(__file__).resolve().parents[1] +SCRIPT_DIR = ROOT / "scripts" +if str(SCRIPT_DIR) not in sys.path: + sys.path.insert(0, str(SCRIPT_DIR)) + +from shadbala_d3_mapping_audit import build_report as build_mapping_report # type: ignore +from shadbala_oracle_comparison import compare_case # type: ignore + + +def _branch_name(row: dict[str, Any]) -> str: + bucket = row.get("d3_dignity_bucket") + if bucket == "exalted": + return "direct_exaltation_branch" + if bucket == "own": + return "direct_own_sign_branch" + if bucket == "debilitated": + return "direct_debilitation_branch" + return "fallback_dignity_score_branch" + + +def build_report(oracle_file: str) -> dict[str, Any]: + mapping = build_mapping_report(oracle_file) + rows = [] + branch_counts: dict[str, int] = {} + branch_diffs: dict[str, list[float]] = {} + + for row in mapping.get("rows", []): + branch = _branch_name(row) + branch_counts[branch] = branch_counts.get(branch, 0) + 1 + comparison = compare_case(oracle_file, row["case_id"]) + planet_comparison = comparison.get("comparison", {}).get(row["planet"], {}) + sthana_component = planet_comparison.get("components", {}).get("sthana", {}) + abs_component_diff = sthana_component.get("abs_diff_rupa") + if isinstance(abs_component_diff, (int, float)): + branch_diffs.setdefault(branch, []).append(float(abs_component_diff)) + rows.append({ + **row, + "sthana_abs_diff_rupa": abs_component_diff, + "suspected_function": "calc_sthana_bala", + "suspected_branch": branch, + "branch_code_path": "calc_sthana_bala -> sapta_d3 -> own/exalted/debilitated/_dignity_score", + }) + + branch_hotspots = {} + for branch, diffs in branch_diffs.items(): + branch_hotspots[branch] = { + "row_count": len(diffs), + "avg_abs_component_diff_rupa": round(sum(diffs) / len(diffs), 4) if diffs else None, + "max_abs_component_diff_rupa": round(max(diffs), 4) if diffs else None, + } + + return { + "scope": "shadbala_d3_branch_audit", + "schema_version": 1, + "summary": { + "row_count": len(rows), + "global_closure_blocked": True, + }, + "branch_counts": branch_counts, + "branch_hotspots": branch_hotspots, + "rows": rows, + "boundary": ( + "This report does not change Shadbala scoring. It only maps each D3 drift case onto the exact " + "calc_sthana_bala branch currently responsible for the local dignity score." + ), + } + + +def main() -> None: + parser = argparse.ArgumentParser(description="Audit exact D3 branch used by calc_sthana_bala") + parser.add_argument("--oracle-file", default="references/oracle/dasha_shadbala_oracle_cases.json") + parser.add_argument("--format", choices=("json", "markdown"), default="json") + args = parser.parse_args() + + report = build_report(args.oracle_file) + if args.format == "markdown": + lines = [ + "# Shadbala D3 Branch Audit", + "", + f"- row_count: `{report['summary']['row_count']}`", + "", + f"- branch_counts: `{report['branch_counts']}`", + "", + "| Case | Planet | D3 Bucket | Branch |", + "| --- | --- | --- | --- |", + ] + for row in report["rows"]: + lines.append( + f"| {row['case_id']} | {row['planet']} | {row['d3_dignity_bucket']} | {row['suspected_branch']} |" + ) + print("\n".join(lines)) + return + + print(json.dumps(report, ensure_ascii=False, indent=2)) + + +if __name__ == "__main__": + main() diff --git a/scripts/shadbala_d3_mapping_audit.py b/scripts/shadbala_d3_mapping_audit.py new file mode 100644 index 00000000..63861b94 --- /dev/null +++ b/scripts/shadbala_d3_mapping_audit.py @@ -0,0 +1,149 @@ +#!/usr/bin/env python3 +"""Audit whether D3 drift is caused by D3 mapping itself or dignity use in Shadbala.""" + +from __future__ import annotations + +import argparse +import json +import sys +from pathlib import Path +from typing import Any + + +ROOT = Path(__file__).resolve().parents[1] +SCRIPT_DIR = ROOT / "scripts" +if str(SCRIPT_DIR) not in sys.path: + sys.path.insert(0, str(SCRIPT_DIR)) + +import jyotish_engine # type: ignore +from divisional_charts_extended import DivisionalChartsCalculator # type: ignore +from oracle_boundary_audit import _load_oracle, _namespace_from_template # type: ignore +from shadbala_sapta_dignity_whitelist import build_report as build_whitelist_report # type: ignore +from varga import calc_varga # type: ignore + + +def _normalize_dignity_bucket(score: float) -> str: + if score >= 50: + return "exalted" + if score >= 45: + return "own" + if score >= 35: + return "friend" + if score >= 25: + return "neutral" + if score >= 15: + return "enemy" + return "debilitated" + + +def _discover_cases(oracle: dict[str, Any]) -> dict[str, dict[str, Any]]: + cases: dict[str, dict[str, Any]] = {} + for key in ("template_cases", "shadbala_cases"): + for case in oracle.get(key, []): + cid = case.get("id") or case.get("case_id") + if isinstance(cid, str): + cases[cid] = case + return cases + + +def build_report(oracle_file: str) -> dict[str, Any]: + whitelist = build_whitelist_report(oracle_file) + oracle = _load_oracle(oracle_file) + cases = _discover_cases(oracle) + calc = DivisionalChartsCalculator() + + rows = [] + mapping_fault = 0 + dignity_fault = 0 + + for row in whitelist.get("whitelist_rows", []): + if row.get("layer") != "D3": + continue + case = cases[row["case_id"]] + result = jyotish_engine.cmd_shadbala(_namespace_from_template(case)) + chart, _asc_idx, _jd, _aya = jyotish_engine._compute_chart_from_args(_namespace_from_template(case)) + planet = row["planet"] + planet_data = (chart or {}).get("planets", {}).get(planet, {}) + lon = float(planet_data.get("degree", 0.0)) + + d3_simple = calc_varga(lon, 3) + d3_extended_abs = calc._calculate_d3(int(lon // 30), lon % 30) + d3_extended = { + "sign_idx": int(d3_extended_abs // 30) % 12, + "sign": calc.SIGNS[int(d3_extended_abs // 30) % 12], + } + + sthana = ((result.get("planets") or {}).get(planet) or {}).get("sthana_bala") or {} + engine_d3_score = float(sthana.get("sapta_d3", 0.0)) + dignity_bucket = _normalize_dignity_bucket(engine_d3_score) + mapping_matches = d3_simple.get("sign") == d3_extended["sign"] + + suspected = "d3_dignity_path" if mapping_matches else "d3_mapping_path" + if mapping_matches: + dignity_fault += 1 + else: + mapping_fault += 1 + + rows.append({ + "case_id": row["case_id"], + "planet": planet, + "mapping_matches_engine_sign": mapping_matches, + "d3_simple_sign": d3_simple.get("sign"), + "d3_extended_sign": d3_extended["sign"], + "d3_dignity_bucket": dignity_bucket, + "sapta_d3_score": round(engine_d3_score, 2), + "suspected_path": suspected, + }) + + return { + "scope": "shadbala_d3_mapping_audit", + "schema_version": 1, + "summary": { + "case_count": len({row["case_id"] for row in rows}), + "row_count": len(rows), + "global_closure_blocked": True, + }, + "suspected_fault_split": { + "d3_mapping_path": mapping_fault, + "d3_dignity_path": dignity_fault, + }, + "rows": rows, + "boundary": ( + "This audit compares two local D3 mapping paths against the D3 score consumed by Shadbala. " + "If both mappings agree, the remaining suspect is the dignity score path rather than the D3 sign mapping." + ), + } + + +def main() -> None: + parser = argparse.ArgumentParser(description="Audit D3 mapping vs dignity path for Shadbala") + parser.add_argument("--oracle-file", default="references/oracle/dasha_shadbala_oracle_cases.json") + parser.add_argument("--format", choices=("json", "markdown"), default="json") + args = parser.parse_args() + + report = build_report(args.oracle_file) + if args.format == "markdown": + lines = [ + "# Shadbala D3 Mapping Audit", + "", + f"- case_count: `{report['summary']['case_count']}`", + f"- row_count: `{report['summary']['row_count']}`", + "", + f"- suspected_fault_split: `{report['suspected_fault_split']}`", + "", + "| Case | Planet | Simple D3 | Extended D3 | D3 Dignity | Suspected Path |", + "| --- | --- | --- | --- | --- | --- |", + ] + for row in report["rows"]: + lines.append( + f"| {row['case_id']} | {row['planet']} | {row['d3_simple_sign']} | {row['d3_extended_sign']} | " + f"{row['d3_dignity_bucket']} | {row['suspected_path']} |" + ) + print("\n".join(lines)) + return + + print(json.dumps(report, ensure_ascii=False, indent=2)) + + +if __name__ == "__main__": + main() diff --git a/scripts/shadbala_oracle_batch_summary.py b/scripts/shadbala_oracle_batch_summary.py new file mode 100644 index 00000000..d3e1cc77 --- /dev/null +++ b/scripts/shadbala_oracle_batch_summary.py @@ -0,0 +1,109 @@ +#!/usr/bin/env python3 +"""Summarize all external-verified Shadbala oracle cases using the existing comparison entrypoint.""" + +from __future__ import annotations + +import argparse +import json +from pathlib import Path +from typing import Any + +from scripts.shadbala_oracle_comparison import compare_case + + +ROOT = Path(__file__).resolve().parents[1] + + +def _load_oracle(path: str) -> dict[str, Any]: + resolved = Path(path) + if not resolved.is_absolute(): + resolved = ROOT / resolved + return json.loads(resolved.read_text(encoding="utf-8")) + + +def _discover_case_ids(oracle: dict[str, Any]) -> list[str]: + case_ids: list[str] = [] + for key in ("template_cases", "shadbala_cases"): + for case in oracle.get(key, []): + if case.get("status") != "external_verified": + continue + if not isinstance(case.get("target", {}).get("shadbala_components"), dict): + continue + case_id = case.get("id") or case.get("case_id") + if isinstance(case_id, str): + case_ids.append(case_id) + return case_ids + + +def build_report(oracle_file: str) -> dict[str, Any]: + oracle = _load_oracle(oracle_file) + case_ids = _discover_case_ids(oracle) + rows = [] + fully_within_tolerance = 0 + + for case_id in case_ids: + report = compare_case(oracle_file=oracle_file, case_id=case_id) + within = report["summary"]["planet_count"] == report["summary"]["planets_within_total_tolerance"] + if within: + fully_within_tolerance += 1 + rows.append({ + "case_id": report["case_id"], + "status": report["status"], + "ayanamsa": report["settings"].get("ayanamsa"), + "node_mode": report["settings"].get("node_mode"), + "planets_within_total_tolerance": report["summary"]["planets_within_total_tolerance"], + "planet_count": report["summary"]["planet_count"], + "max_abs_total_delta_rupa": report["summary"]["max_abs_total_delta_rupa"], + "global_scaling_recommendation": report["global_scaling_check"].get("recommendation"), + "within_case_tolerance": within, + }) + + return { + "scope": "shadbala_oracle_batch_summary", + "schema_version": 1, + "summary": { + "case_count": len(rows), + "external_verified_case_count": len(case_ids), + "fully_within_tolerance_case_count": fully_within_tolerance, + "global_closure_blocked": True, + }, + "rows": rows, + "boundary": ( + "This summary reuses shadbala_oracle_comparison.py for every external-verified Shadbala case. " + "It is diagnostic and keeps global closure blocked until enough cases converge." + ), + } + + +def main() -> None: + parser = argparse.ArgumentParser(description="Build a batch summary for external-verified Shadbala oracle cases") + parser.add_argument("--oracle-file", default="references/oracle/dasha_shadbala_oracle_cases.json") + parser.add_argument("--format", choices=("json", "markdown"), default="json") + args = parser.parse_args() + + report = build_report(args.oracle_file) + if args.format == "markdown": + lines = [ + "# Shadbala Oracle Batch Summary", + "", + f"- case_count: `{report['summary']['case_count']}`", + f"- external_verified_case_count: `{report['summary']['external_verified_case_count']}`", + f"- fully_within_tolerance_case_count: `{report['summary']['fully_within_tolerance_case_count']}`", + f"- global_closure_blocked: `{str(report['summary']['global_closure_blocked']).lower()}`", + "", + "| Case | Ayanamsa | Node | Within Total Tolerance | Max Delta |", + "| --- | --- | --- | --- | ---: |", + ] + for row in report["rows"]: + lines.append( + f"| {row['case_id']} | {row['ayanamsa']} | {row['node_mode']} | " + f"{row['planets_within_total_tolerance']}/{row['planet_count']} | {row['max_abs_total_delta_rupa']} |" + ) + print("\n".join(lines)) + return + + print(json.dumps(report, ensure_ascii=False, indent=2)) + + +if __name__ == "__main__": + main() diff --git a/scripts/shadbala_oracle_component_cluster_summary.py b/scripts/shadbala_oracle_component_cluster_summary.py new file mode 100644 index 00000000..cc12e621 --- /dev/null +++ b/scripts/shadbala_oracle_component_cluster_summary.py @@ -0,0 +1,141 @@ +#!/usr/bin/env python3 +"""Cluster Shadbala oracle deltas to identify the narrowest closure target.""" + +from __future__ import annotations + +import argparse +import json +from collections import defaultdict +from pathlib import Path +from typing import Any + +from scripts.shadbala_oracle_batch_summary import _discover_case_ids, _load_oracle +from scripts.shadbala_oracle_comparison import compare_case + + +ROOT = Path(__file__).resolve().parents[1] + + +def _safe_avg(values: list[float]) -> float: + return round(sum(values) / len(values), 4) if values else 0.0 + + +def build_report(oracle_file: str) -> dict[str, Any]: + oracle = _load_oracle(oracle_file) + case_ids = _discover_case_ids(oracle) + + component_deltas: dict[str, list[float]] = defaultdict(list) + planet_deltas: dict[str, list[float]] = defaultdict(list) + case_count = 0 + planet_count = 0 + + for case_id in case_ids: + report = compare_case(oracle_file=oracle_file, case_id=case_id) + case_count += 1 + for planet, row in report.get("comparison", {}).items(): + total_abs = row.get("abs_diff_total_rupa") + if isinstance(total_abs, (int, float)): + planet_deltas[planet].append(float(total_abs)) + planet_count = max(planet_count, len(report.get("comparison", {}))) + for component, component_row in (row.get("components") or {}).items(): + abs_diff = component_row.get("abs_diff_rupa") + if isinstance(abs_diff, (int, float)): + component_deltas[component].append(float(abs_diff)) + + component_hotspots = sorted( + [ + { + "component": component, + "avg_abs_diff_rupa": _safe_avg(values), + "max_abs_diff_rupa": round(max(values), 4), + "sample_count": len(values), + } + for component, values in component_deltas.items() + if values + ], + key=lambda row: (-row["avg_abs_diff_rupa"], -row["max_abs_diff_rupa"], row["component"]), + ) + + planet_hotspots = sorted( + [ + { + "planet": planet, + "avg_abs_total_delta_rupa": _safe_avg(values), + "max_abs_total_delta_rupa": round(max(values), 4), + "sample_count": len(values), + } + for planet, values in planet_deltas.items() + if values + ], + key=lambda row: (-row["avg_abs_total_delta_rupa"], -row["max_abs_total_delta_rupa"], row["planet"]), + ) + + top_component = component_hotspots[0]["component"] if component_hotspots else None + targeted_fix = ( + f"Prioritize {top_component} component reconciliation before touching global scaling or unrelated layers." + if top_component + else "No hotspot identified." + ) + + return { + "scope": "shadbala_oracle_component_cluster_summary", + "schema_version": 1, + "summary": { + "case_count": case_count, + "planet_count": planet_count, + "global_closure_blocked": True, + "targeted_fix_recommendation": targeted_fix, + }, + "component_hotspots": component_hotspots, + "planet_hotspots": planet_hotspots, + "boundary": ( + "This report clusters absolute Shadbala delta hotspots across all external_verified cases. " + "It reuses the existing comparison pipeline and narrows closure work to the most divergent component first." + ), + } + + +def main() -> None: + parser = argparse.ArgumentParser(description="Summarize Shadbala component hotspots across oracle cases") + parser.add_argument("--oracle-file", default="references/oracle/dasha_shadbala_oracle_cases.json") + parser.add_argument("--format", choices=("json", "markdown"), default="json") + args = parser.parse_args() + + report = build_report(args.oracle_file) + if args.format == "markdown": + lines = [ + "# Shadbala Oracle Component Cluster Summary", + "", + f"- case_count: `{report['summary']['case_count']}`", + f"- planet_count: `{report['summary']['planet_count']}`", + f"- global_closure_blocked: `{str(report['summary']['global_closure_blocked']).lower()}`", + f"- targeted_fix_recommendation: `{report['summary']['targeted_fix_recommendation']}`", + "", + "## Component Hotspots", + "", + "| Component | Avg Abs Diff (Rupa) | Max Abs Diff | Samples |", + "| --- | ---: | ---: | ---: |", + ] + for row in report["component_hotspots"]: + lines.append( + f"| {row['component']} | {row['avg_abs_diff_rupa']} | {row['max_abs_diff_rupa']} | {row['sample_count']} |" + ) + lines.extend([ + "", + "## Planet Hotspots", + "", + "| Planet | Avg Abs Total Delta (Rupa) | Max Abs Delta | Samples |", + "| --- | ---: | ---: | ---: |", + ]) + for row in report["planet_hotspots"]: + lines.append( + f"| {row['planet']} | {row['avg_abs_total_delta_rupa']} | {row['max_abs_total_delta_rupa']} | {row['sample_count']} |" + ) + print("\n".join(lines)) + return + + print(json.dumps(report, ensure_ascii=False, indent=2)) + + +if __name__ == "__main__": + main() diff --git a/scripts/shadbala_sapta_dignity_whitelist.py b/scripts/shadbala_sapta_dignity_whitelist.py new file mode 100644 index 00000000..78bfff5e --- /dev/null +++ b/scripts/shadbala_sapta_dignity_whitelist.py @@ -0,0 +1,91 @@ +#!/usr/bin/env python3 +"""Build a minimal whitelist of Sapta dignity mappings most likely causing drift.""" + +from __future__ import annotations + +import argparse +import json +from collections import Counter +from pathlib import Path +from typing import Any + +from scripts.shadbala_sapta_layer_hotspots import build_report as build_hotspot_report + + +ROOT = Path(__file__).resolve().parents[1] +TARGET_LAYERS = {"D7", "D12", "D3", "D4"} +TARGET_DRIVERS = {"dignity_exalted": "exalted", "dignity_own": "own"} + + +def build_report(oracle_file: str) -> dict[str, Any]: + hotspot = build_hotspot_report(oracle_file) + whitelist_rows = [] + layer_counts: Counter[str] = Counter() + + for row in hotspot.get("rows", []): + layer = row.get("dominant_layer") + driver = row.get("driver_guess") + if layer not in TARGET_LAYERS: + continue + if driver not in TARGET_DRIVERS: + continue + whitelist_rows.append({ + "case_id": row["case_id"], + "planet": row["planet"], + "layer": layer, + "dignity_type": TARGET_DRIVERS[driver], + "dominant_layer_score": row["dominant_layer_score"], + "driver_guess": driver, + }) + layer_counts[layer] += 1 + + whitelist_rows.sort( + key=lambda row: (-row["dominant_layer_score"], row["layer"], row["case_id"], row["planet"]) + ) + + return { + "scope": "shadbala_sapta_dignity_whitelist", + "schema_version": 1, + "summary": { + "case_count": hotspot.get("summary", {}).get("case_count", 0), + "global_closure_blocked": True, + "whitelist_count": len(whitelist_rows), + }, + "layer_counts": dict(layer_counts), + "whitelist_rows": whitelist_rows, + "boundary": ( + "This whitelist isolates D7/D12/D3/D4 exalted-or-own dignity mappings that appear as dominant Sapta " + "drivers. It is a repair whitelist, not proof of final oracle closure." + ), + } + + +def main() -> None: + parser = argparse.ArgumentParser(description="Build Sapta dignity whitelist for Shadbala repair") + parser.add_argument("--oracle-file", default="references/oracle/dasha_shadbala_oracle_cases.json") + parser.add_argument("--format", choices=("json", "markdown"), default="json") + args = parser.parse_args() + + report = build_report(args.oracle_file) + if args.format == "markdown": + lines = [ + "# Shadbala Sapta Dignity Whitelist", + "", + f"- case_count: `{report['summary']['case_count']}`", + f"- whitelist_count: `{report['summary']['whitelist_count']}`", + "", + "| Layer | Dignity | Planet | Case | Score |", + "| --- | --- | --- | --- | ---: |", + ] + for row in report["whitelist_rows"]: + lines.append( + f"| {row['layer']} | {row['dignity_type']} | {row['planet']} | {row['case_id']} | {row['dominant_layer_score']} |" + ) + print("\n".join(lines)) + return + + print(json.dumps(report, ensure_ascii=False, indent=2)) + + +if __name__ == "__main__": + main() diff --git a/scripts/shadbala_sapta_layer_hotspots.py b/scripts/shadbala_sapta_layer_hotspots.py new file mode 100644 index 00000000..2f9122d1 --- /dev/null +++ b/scripts/shadbala_sapta_layer_hotspots.py @@ -0,0 +1,155 @@ +#!/usr/bin/env python3 +"""Identify which Sapta Varga layers dominate Sthana oracle drift.""" + +from __future__ import annotations + +import argparse +import json +from collections import Counter, defaultdict +from pathlib import Path +from typing import Any + +from scripts import jyotish_engine +from scripts.oracle_boundary_audit import _load_oracle, _namespace_from_template + + +ROOT = Path(__file__).resolve().parents[1] +SAPTA_KEYS = ["sapta_d1", "sapta_d2", "sapta_d3", "sapta_d4", "sapta_d7", "sapta_d9", "sapta_d12"] +LAYER_LABELS = { + "sapta_d1": "D1", + "sapta_d2": "D2", + "sapta_d3": "D3", + "sapta_d4": "D4", + "sapta_d7": "D7", + "sapta_d9": "D9", + "sapta_d12": "D12", +} + + +def _discover_cases(oracle: dict[str, Any]) -> list[dict[str, Any]]: + cases = [] + for key in ("template_cases", "shadbala_cases"): + for case in oracle.get(key, []): + if case.get("status") != "external_verified": + continue + if not isinstance(case.get("target", {}).get("shadbala_components"), dict): + continue + cases.append(case) + return cases + + +def _normalize_dignity_bucket(score: float) -> str: + if score >= 50: + return "dignity_exalted" + if score >= 45: + return "dignity_own" + if score >= 35: + return "friend_enemy_friend" + if score >= 25: + return "friend_enemy_neutral" + if score >= 15: + return "friend_enemy_enemy" + return "dignity_debilitated" + + +def build_report(oracle_file: str) -> dict[str, Any]: + oracle = _load_oracle(oracle_file) + cases = _discover_cases(oracle) + + layer_values: dict[str, list[float]] = defaultdict(list) + rows: list[dict[str, Any]] = [] + driver_mix: Counter[str] = Counter() + + for case in cases: + case_id = case.get("id") or case.get("case_id") + result = jyotish_engine.cmd_shadbala(_namespace_from_template(case)) + for planet, pdata in (result.get("planets") or {}).items(): + sthana = pdata.get("sthana_bala") or {} + layer_pairs = [(key, float(sthana.get(key, 0.0))) for key in SAPTA_KEYS] + dominant_key, dominant_value = max(layer_pairs, key=lambda item: item[1]) + driver = _normalize_dignity_bucket(dominant_value) + driver_mix[driver] += 1 + layer_values[dominant_key].append(dominant_value) + rows.append({ + "case_id": case_id, + "planet": planet, + "dominant_layer_key": dominant_key, + "dominant_layer": LAYER_LABELS[dominant_key], + "dominant_layer_score": round(dominant_value, 2), + "driver_guess": driver, + "layer_scores": {LAYER_LABELS[key]: round(value, 2) for key, value in layer_pairs}, + }) + + rows.sort(key=lambda row: (-row["dominant_layer_score"], row["case_id"], row["planet"])) + layer_hotspots = sorted( + [ + { + "layer": LAYER_LABELS[key], + "avg_score": round(sum(values) / len(values), 4), + "max_score": round(max(values), 4), + "sample_count": len(values), + } + for key, values in layer_values.items() + if values + ], + key=lambda row: (-row["avg_score"], -row["max_score"], row["layer"]), + ) + + return { + "scope": "shadbala_sapta_layer_hotspots", + "schema_version": 1, + "summary": { + "case_count": len(cases), + "row_count": len(rows), + "global_closure_blocked": True, + "top_layer": layer_hotspots[0]["layer"] if layer_hotspots else None, + }, + "layer_hotspots": layer_hotspots, + "driver_mix": dict(driver_mix), + "rows": rows, + "boundary": ( + "This hotspot table ranks Sapta Varga sublayers by their local score dominance so we can decide " + "which dignity/friend-enemy mapping layer to audit first. It does not change oracle tolerances." + ), + } + + +def main() -> None: + parser = argparse.ArgumentParser(description="Summarize Sapta Varga layer hotspots for Shadbala") + parser.add_argument("--oracle-file", default="references/oracle/dasha_shadbala_oracle_cases.json") + parser.add_argument("--format", choices=("json", "markdown"), default="json") + args = parser.parse_args() + + report = build_report(args.oracle_file) + if args.format == "markdown": + lines = [ + "# Shadbala Sapta Layer Hotspots", + "", + f"- case_count: `{report['summary']['case_count']}`", + f"- row_count: `{report['summary']['row_count']}`", + f"- top_layer: `{report['summary']['top_layer']}`", + "", + "## Layer Hotspots", + "", + "| Layer | Avg Score | Max Score | Samples |", + "| --- | ---: | ---: | ---: |", + ] + for row in report["layer_hotspots"]: + lines.append( + f"| {row['layer']} | {row['avg_score']} | {row['max_score']} | {row['sample_count']} |" + ) + lines.extend([ + "", + "## Driver Mix", + "", + ]) + for driver, count in sorted(report["driver_mix"].items(), key=lambda item: (-item[1], item[0])): + lines.append(f"- `{driver}`: {count}") + print("\n".join(lines)) + return + + print(json.dumps(report, ensure_ascii=False, indent=2)) + + +if __name__ == "__main__": + main() diff --git a/scripts/shadbala_sthana_targeted_audit.py b/scripts/shadbala_sthana_targeted_audit.py new file mode 100644 index 00000000..de053bac --- /dev/null +++ b/scripts/shadbala_sthana_targeted_audit.py @@ -0,0 +1,157 @@ +#!/usr/bin/env python3 +"""Targeted Sthana Bala audit across external-verified oracle cases.""" + +from __future__ import annotations + +import argparse +import json +from collections import Counter +from pathlib import Path +from typing import Any + +from scripts import jyotish_engine +from scripts.oracle_boundary_audit import _load_oracle, _namespace_from_template +from scripts.shadbala_oracle_comparison import compare_case + + +ROOT = Path(__file__).resolve().parents[1] + + +def _normalize_dignity_bucket(score: float) -> str: + if score >= 50: + return "exalted" + if score >= 45: + return "own" + if score >= 35: + return "friend" + if score >= 25: + return "neutral" + if score >= 15: + return "enemy" + return "debilitated" + + +def _suspected_driver(sthana: dict[str, Any], delta: float) -> str: + sapta = float(sthana.get("sapta_score", 0.0)) + ucha = float(sthana.get("ucha_bala", 0.0)) + kendra = float(sthana.get("kendra_bala", 0.0)) + ojayugma = float(sthana.get("ojayugma_bala", 0.0)) + drekkana = float(sthana.get("drekkana_bala", 0.0)) + + if sapta >= max(ucha, kendra, ojayugma, drekkana): + d1_score = float(sthana.get("sapta_d1", 0.0)) + dignity = _normalize_dignity_bucket(d1_score) + if dignity in {"own", "exalted", "debilitated"}: + return f"sapta_dignity_{dignity}" + return f"sapta_friend_enemy_{dignity}" + if ucha >= max(kendra, ojayugma, drekkana): + return "ucha_axis" + if kendra >= max(ojayugma, drekkana): + return "kendra_house_tiering" + if ojayugma >= drekkana: + return "ojayugma_parity" + return "drekkana_bucket" + + +def build_report(oracle_file: str) -> dict[str, Any]: + oracle = _load_oracle(oracle_file) + cases = [] + for key in ("template_cases", "shadbala_cases"): + for case in oracle.get(key, []): + if case.get("status") != "external_verified": + continue + if not isinstance(case.get("target", {}).get("shadbala_components"), dict): + continue + cases.append(case) + + rows: list[dict[str, Any]] = [] + driver_counts: Counter[str] = Counter() + + for case in cases: + case_id = case.get("id") or case.get("case_id") + comparison = compare_case(oracle_file=oracle_file, case_id=case_id) + result = jyotish_engine.cmd_shadbala(_namespace_from_template(case)) + for planet, comp_row in comparison.get("comparison", {}).items(): + sthana_component = (comp_row.get("components") or {}).get("sthana") or {} + local_planet = (result.get("planets") or {}).get(planet) or {} + sthana = local_planet.get("sthana_bala") or {} + diff = sthana_component.get("diff_rupa") + if not isinstance(diff, (int, float)): + continue + driver = _suspected_driver(sthana, float(diff)) + driver_counts[driver] += 1 + d1_score = float(sthana.get("sapta_d1", 0.0)) + rows.append({ + "case_id": case_id, + "planet": planet, + "sthana_diff_rupa": round(float(diff), 4), + "abs_sthana_diff_rupa": round(abs(float(diff)), 4), + "d1_dignity_bucket": _normalize_dignity_bucket(d1_score), + "sapta_score": round(float(sthana.get("sapta_score", 0.0)), 2), + "ucha_bala": round(float(sthana.get("ucha_bala", 0.0)), 2), + "kendra_bala": round(float(sthana.get("kendra_bala", 0.0)), 2), + "ojayugma_bala": round(float(sthana.get("ojayugma_bala", 0.0)), 2), + "drekkana_bala": round(float(sthana.get("drekkana_bala", 0.0)), 2), + "suspected_driver": driver, + }) + + rows.sort(key=lambda row: (-row["abs_sthana_diff_rupa"], row["case_id"], row["planet"])) + + return { + "scope": "shadbala_sthana_targeted_audit", + "schema_version": 1, + "summary": { + "case_count": len(cases), + "row_count": len(rows), + "global_closure_blocked": True, + "top_driver": rows[0]["suspected_driver"] if rows else None, + }, + "driver_counts": dict(driver_counts), + "rows": rows, + "boundary": ( + "This targeted audit reuses local Shadbala output plus oracle comparisons to classify Sthana divergence " + "into dignity/friend-enemy/house-tiering style buckets. It is diagnostic, not a calibration override." + ), + } + + +def main() -> None: + parser = argparse.ArgumentParser(description="Targeted audit for Sthana Bala divergence") + parser.add_argument("--oracle-file", default="references/oracle/dasha_shadbala_oracle_cases.json") + parser.add_argument("--format", choices=("json", "markdown"), default="json") + args = parser.parse_args() + + report = build_report(args.oracle_file) + if args.format == "markdown": + lines = [ + "# Shadbala Sthana Targeted Audit", + "", + f"- case_count: `{report['summary']['case_count']}`", + f"- row_count: `{report['summary']['row_count']}`", + f"- top_driver: `{report['summary']['top_driver']}`", + "", + "## Driver Counts", + "", + ] + for name, count in sorted(report["driver_counts"].items(), key=lambda item: (-item[1], item[0])): + lines.append(f"- `{name}`: {count}") + lines.extend([ + "", + "## Largest Sthana Deltas", + "", + "| Case | Planet | Abs Diff | D1 Bucket | Suspected Driver |", + "| --- | --- | ---: | --- | --- |", + ]) + for row in report["rows"][:20]: + lines.append( + f"| {row['case_id']} | {row['planet']} | {row['abs_sthana_diff_rupa']} | " + f"{row['d1_dignity_bucket']} | {row['suspected_driver']} |" + ) + print("\n".join(lines)) + return + + print(json.dumps(report, ensure_ascii=False, indent=2)) + + +if __name__ == "__main__": + main() diff --git a/tests/test_shadbala_d3_branch_audit.py b/tests/test_shadbala_d3_branch_audit.py new file mode 100644 index 00000000..34c39949 --- /dev/null +++ b/tests/test_shadbala_d3_branch_audit.py @@ -0,0 +1,26 @@ +#!/usr/bin/env python3 +"""Regression tests for D3 branch-level audit.""" + +from __future__ import annotations + +from scripts.shadbala_d3_branch_audit import build_report + + +def test_shadbala_d3_branch_audit_points_to_calc_sthana_bala_d3_branch() -> None: + report = build_report("references/oracle/dasha_shadbala_oracle_cases.json") + + assert report["scope"] == "shadbala_d3_branch_audit" + assert report["summary"]["row_count"] >= 4 + assert report["summary"]["global_closure_blocked"] is True + assert report["branch_counts"] + assert "direct_exaltation_branch" in report["branch_counts"] or "direct_own_sign_branch" in report["branch_counts"] + assert all(row["suspected_function"] == "calc_sthana_bala" for row in report["rows"]) + + +def test_shadbala_d3_branch_audit_splits_exaltation_vs_own_drift() -> None: + report = build_report("references/oracle/dasha_shadbala_oracle_cases.json") + + assert report["branch_hotspots"]["direct_exaltation_branch"]["row_count"] >= 1 + assert report["branch_hotspots"]["direct_own_sign_branch"]["row_count"] >= 1 + assert report["branch_hotspots"]["direct_exaltation_branch"]["avg_abs_component_diff_rupa"] is not None + assert report["branch_hotspots"]["direct_own_sign_branch"]["avg_abs_component_diff_rupa"] is not None diff --git a/tests/test_shadbala_d3_mapping_audit.py b/tests/test_shadbala_d3_mapping_audit.py new file mode 100644 index 00000000..5b947d08 --- /dev/null +++ b/tests/test_shadbala_d3_mapping_audit.py @@ -0,0 +1,18 @@ +#!/usr/bin/env python3 +"""Regression tests for D3 mapping vs Shadbala dignity audit.""" + +from __future__ import annotations + +from scripts.shadbala_d3_mapping_audit import build_report + + +def test_shadbala_d3_mapping_audit_distinguishes_mapping_vs_dignity_path() -> None: + report = build_report("references/oracle/dasha_shadbala_oracle_cases.json") + + assert report["scope"] == "shadbala_d3_mapping_audit" + assert report["summary"]["case_count"] >= 3 + assert report["summary"]["global_closure_blocked"] is True + assert report["rows"] + assert any(row["mapping_matches_engine_sign"] is True for row in report["rows"]) + assert any(row["d3_dignity_bucket"] in {"exalted", "own"} for row in report["rows"]) + assert report["suspected_fault_split"] diff --git a/tests/test_shadbala_oracle_batch_summary.py b/tests/test_shadbala_oracle_batch_summary.py new file mode 100644 index 00000000..956dfc13 --- /dev/null +++ b/tests/test_shadbala_oracle_batch_summary.py @@ -0,0 +1,18 @@ +#!/usr/bin/env python3 +"""Regression tests for batched Shadbala oracle summary.""" + +from __future__ import annotations + +from scripts.shadbala_oracle_batch_summary import build_report + + +def test_shadbala_oracle_batch_summary_reuses_all_external_verified_cases() -> None: + report = build_report("references/oracle/dasha_shadbala_oracle_cases.json") + + assert report["scope"] == "shadbala_oracle_batch_summary" + assert report["summary"]["case_count"] >= 4 + assert report["summary"]["external_verified_case_count"] >= 4 + assert report["summary"]["fully_within_tolerance_case_count"] <= report["summary"]["external_verified_case_count"] + assert any(row["case_id"] == "template_steve_jobs_dasha_lahiri" for row in report["rows"]) + assert any(row["case_id"] == "template_redacted_place_shadbala_raman" for row in report["rows"]) + assert report["summary"]["global_closure_blocked"] is True diff --git a/tests/test_shadbala_oracle_component_cluster_summary.py b/tests/test_shadbala_oracle_component_cluster_summary.py new file mode 100644 index 00000000..7b5bab96 --- /dev/null +++ b/tests/test_shadbala_oracle_component_cluster_summary.py @@ -0,0 +1,19 @@ +#!/usr/bin/env python3 +"""Regression tests for Shadbala component hotspot clustering.""" + +from __future__ import annotations + +from scripts.shadbala_oracle_component_cluster_summary import build_report + + +def test_shadbala_component_cluster_summary_reuses_external_verified_cases() -> None: + report = build_report("references/oracle/dasha_shadbala_oracle_cases.json") + + assert report["scope"] == "shadbala_oracle_component_cluster_summary" + assert report["summary"]["case_count"] >= 4 + assert report["summary"]["planet_count"] >= 7 + assert report["summary"]["global_closure_blocked"] is True + assert report["component_hotspots"] + assert any(row["component"] == "sthana" for row in report["component_hotspots"]) + assert any(row["planet"] == "Sun" for row in report["planet_hotspots"]) + assert "targeted_fix_recommendation" in report["summary"] diff --git a/tests/test_shadbala_sapta_dignity_whitelist.py b/tests/test_shadbala_sapta_dignity_whitelist.py new file mode 100644 index 00000000..7f0fee2f --- /dev/null +++ b/tests/test_shadbala_sapta_dignity_whitelist.py @@ -0,0 +1,18 @@ +#!/usr/bin/env python3 +"""Regression tests for Sapta dignity whitelist audit.""" + +from __future__ import annotations + +from scripts.shadbala_sapta_dignity_whitelist import build_report + + +def test_shadbala_sapta_dignity_whitelist_surfaces_d7_d12_d3_d4_exalted_own_flags() -> None: + report = build_report("references/oracle/dasha_shadbala_oracle_cases.json") + + assert report["scope"] == "shadbala_sapta_dignity_whitelist" + assert report["summary"]["case_count"] >= 4 + assert report["summary"]["global_closure_blocked"] is True + assert report["whitelist_rows"] + assert any(row["layer"] in {"D7", "D12", "D3", "D4"} for row in report["whitelist_rows"]) + assert any(row["dignity_type"] in {"exalted", "own"} for row in report["whitelist_rows"]) + assert report["layer_counts"] diff --git a/tests/test_shadbala_sapta_layer_hotspots.py b/tests/test_shadbala_sapta_layer_hotspots.py new file mode 100644 index 00000000..b42d891e --- /dev/null +++ b/tests/test_shadbala_sapta_layer_hotspots.py @@ -0,0 +1,21 @@ +#!/usr/bin/env python3 +"""Regression tests for Sapta Varga hotspot audit.""" + +from __future__ import annotations + +from scripts.shadbala_sapta_layer_hotspots import build_report + + +def test_shadbala_sapta_layer_hotspots_surfaces_layer_ranking_and_driver_mix() -> None: + report = build_report("references/oracle/dasha_shadbala_oracle_cases.json") + + assert report["scope"] == "shadbala_sapta_layer_hotspots" + assert report["summary"]["case_count"] >= 4 + assert report["summary"]["row_count"] >= 20 + assert report["summary"]["global_closure_blocked"] is True + assert report["layer_hotspots"] + assert any(row["layer"] == "D1" for row in report["layer_hotspots"]) + assert any(row["layer"] == "D9" for row in report["layer_hotspots"]) + assert report["driver_mix"] + assert any(key.startswith("friend_enemy") for key in report["driver_mix"]) + assert any(key.startswith("dignity") for key in report["driver_mix"]) diff --git a/tests/test_shadbala_sthana_targeted_audit.py b/tests/test_shadbala_sthana_targeted_audit.py new file mode 100644 index 00000000..aaed4dd9 --- /dev/null +++ b/tests/test_shadbala_sthana_targeted_audit.py @@ -0,0 +1,19 @@ +#!/usr/bin/env python3 +"""Regression tests for targeted Sthana Bala oracle audit.""" + +from __future__ import annotations + +from scripts.shadbala_sthana_targeted_audit import build_report + + +def test_shadbala_sthana_targeted_audit_identifies_sapta_and_dignity_signals() -> None: + report = build_report("references/oracle/dasha_shadbala_oracle_cases.json") + + assert report["scope"] == "shadbala_sthana_targeted_audit" + assert report["summary"]["case_count"] >= 4 + assert report["summary"]["row_count"] >= 20 + assert report["summary"]["global_closure_blocked"] is True + assert report["rows"] + assert any("sapta" in row["suspected_driver"] for row in report["rows"]) + assert any(row["d1_dignity_bucket"] in {"own", "exalted", "debilitated", "friend", "enemy", "neutral"} for row in report["rows"]) + assert report["driver_counts"]