Add Shadbala oracle closure audit chain

This commit is contained in:
732642856
2026-06-29 10:38:17 +08:00
parent 0f9b06d9d4
commit a824588d90
14 changed files with 1051 additions and 0 deletions
+110
View File
@@ -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()
+149
View File
@@ -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()
+109
View File
@@ -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()
@@ -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()
@@ -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()
+155
View File
@@ -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()
+157
View File
@@ -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()
+26
View File
@@ -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
+18
View File
@@ -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"]
@@ -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
@@ -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"]
@@ -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"]
@@ -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"])
@@ -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"]