From 5578ba6e2975276622002df5820f360c5aec469c Mon Sep 17 00:00:00 2001 From: 732642856 <732642856@qq.com> Date: Fri, 26 Jun 2026 17:14:19 +0800 Subject: [PATCH] feat(oracle): compare verified template evidence --- README.md | 9 +++ scripts/oracle_boundary_audit.py | 100 ++++++++++++++++++++++++++++ tests/test_oracle_boundary_audit.py | 62 +++++++++++++++++ 3 files changed, 171 insertions(+) diff --git a/README.md b/README.md index 399feb7f..8a75bdf2 100644 --- a/README.md +++ b/README.md @@ -194,6 +194,15 @@ python3 scripts/oracle_evidence_validator.py \ 该验证器输出 `external_oracle_evidence_validation`，会检查 `evidence_packet` 必填元数据、`target_placeholders` 是否已填、是否覆盖 `target_fields`、是否包含外部 artifact，以及是否错误使用本仓库本地引擎输出。当前 draft 队列会保持 `valid_packets: 0` / `ready_for_calibration: 0`；只有状态为 `external_verified` 且证据完整的包才会进入可复核状态。 +证据包通过 validator 之后，再运行边界差异审计，比较本地引擎与外部 Dasha/Shadbala 目标值： + +```bash +python3 scripts/oracle_boundary_audit.py \ + --oracle-file references/oracle/dasha_shadbala_oracle_cases.json +``` + +审计报告中的 `template_comparisons` 会列出 external-verified template 的 Dasha 起点差异、Shadbala 七曜分量/总分差异，并继续保持 `production_tuning_recommended: false`，防止用单个样本调生产常数。 + `full-reading` 也会输出 `ai_prompt_pack`：这是给网页/app、skill 或后端 AI 代理使用的结构化 Prompt/RAG 上下文包。它不会硬编码断语，而是携带 D1/D9/Dasha/Shadbala/Ashtakavarga 的证据快照、推荐检索文档和边界提示，要求大模型基于计算证据交叉验证，避免单一配置下结论。 ### Prerequisites diff --git a/scripts/oracle_boundary_audit.py b/scripts/oracle_boundary_audit.py index f18a39d7..a13e9422 100644 --- a/scripts/oracle_boundary_audit.py +++ b/scripts/oracle_boundary_audit.py @@ -232,6 +232,99 @@ def _audit_template_case(case: dict[str, Any]) -> dict[str, Any]: } +def _namespace_from_template(case: dict[str, Any]) -> argparse.Namespace: + birth = dict(case.get("birth", {})) + settings = case.get("settings", {}) + if "node_mode" not in birth and settings.get("node_mode"): + birth["node_mode"] = settings["node_mode"] + return _namespace_from_birth( + birth, + ayanamsa=settings.get("ayanamsa", "lahiri"), + moon_lon=None, + nakshatra=None, + pada=None, + birthdate=None, + today=None, + years=120, + table=False, + ) + + +def _first_dasha_start(result: dict[str, Any]) -> str | None: + timeline = result.get("timeline") + if isinstance(timeline, list) and timeline: + return timeline[0].get("start") + return None + + +def _date_delta_days(engine_date: str | None, target_date: str | None) -> int | None: + if not engine_date or not target_date: + return None + from datetime import date + + try: + engine_parts = [int(part) for part in engine_date.split("-")[:3]] + target_parts = [int(part) for part in target_date.split("-")[:3]] + return (date(*engine_parts) - date(*target_parts)).days + except Exception: + return None + + +def _template_shadbala_comparison(case: dict[str, Any]) -> dict[str, Any]: + target_components = case.get("target", {}).get("shadbala_components") + if not isinstance(target_components, dict): + return {"status": "missing_external_shadbala_components", "planets": {}} + + result = engine.cmd_shadbala(_namespace_from_template(case)) + if "error" in result: + return {"status": "engine_error", "error": result["error"], "planets": {}} + + rows: dict[str, Any] = {} + for planet in SHADBALA_PLANETS: + external_row = target_components.get(planet, {}) + engine_row = result.get("planets", {}).get(planet, {}) + engine_components = _component_totals(engine_row) if engine_row else {} + external_total = external_row.get("total_rupa") if isinstance(external_row, dict) else None + engine_total = round(float(engine_row.get("total_rupas", 0.0)), 4) if engine_row else None + rows[planet] = { + "engine_total_rupa": engine_total, + "external_total_rupa": external_total, + "total_rupa_delta": ( + round(engine_total - float(external_total), 4) + if engine_total is not None and isinstance(external_total, (int, float)) + else None + ), + "engine_components": engine_components, + "external_components": external_row, + } + return {"status": "compared", "planets": rows} + + +def _audit_external_verified_template_case(case: dict[str, Any]) -> dict[str, Any]: + target = case.get("target", {}) + dasha_result = engine.cmd_dasha(_namespace_from_template(case)) + engine_start = _first_dasha_start(dasha_result) if "error" not in dasha_result else None + target_start = target.get("vimshottari_start_date") + return { + "case_id": case.get("id") or case.get("case_id"), + "status": case.get("status"), + "source": case.get("source"), + "metadata": case.get("evidence_packet", {}).get("metadata", {}), + "dasha": { + "status": "compared" if engine_start and target_start else "missing_dasha_target", + "engine_start_date": engine_start, + "target_start_date": target_start, + "date_delta_days": _date_delta_days(engine_start, target_start), + }, + "shadbala": _template_shadbala_comparison(case), + "calibration_decision": "do_not_tune_single_template", + "finding": ( + "External-verified template rows are comparison evidence. Production constants require " + "a multi-source sample matrix and must not be tuned to a single packet." + ), + } + + def _status_counts(rows: list[dict[str, Any]]) -> dict[str, int]: counts: dict[str, int] = {} for row in rows: @@ -242,6 +335,11 @@ def _status_counts(rows: list[dict[str, Any]]) -> dict[str, int]: def build_report(oracle: dict[str, Any]) -> dict[str, Any]: template_rows = [_audit_template_case(case) for case in oracle.get("template_cases", [])] + template_comparisons = [ + _audit_external_verified_template_case(case) + for case in oracle.get("template_cases", []) + if case.get("status") in ORACLE_TEMPLATE_READY_STATUSES and not _missing_target_fields(case.get("target", {})) + ] dasha_rows = [_audit_dasha_case(case) for case in oracle.get("dasha_cases", [])] longitude_rows = [_audit_longitude_case(case) for case in oracle.get("longitude_cases", [])] shadbala_rows = [_audit_shadbala_case(case) for case in oracle.get("shadbala_cases", [])] @@ -251,6 +349,7 @@ def build_report(oracle: dict[str, Any]) -> dict[str, Any]: "summary": { "template_cases": len(template_rows), "template_status_counts": _status_counts(template_rows), + "external_verified_template_cases": len(template_comparisons), "dasha_cases": len(dasha_rows), "longitude_cases": len(longitude_rows), "shadbala_cases": len(shadbala_rows), @@ -262,6 +361,7 @@ def build_report(oracle: dict[str, Any]) -> dict[str, Any]: ], }, "template_cases": template_rows, + "template_comparisons": template_comparisons, "dasha_cases": dasha_rows, "longitude_cases": longitude_rows, "shadbala_cases": shadbala_rows, diff --git a/tests/test_oracle_boundary_audit.py b/tests/test_oracle_boundary_audit.py index 41f1f9ce..5abf06e7 100644 --- a/tests/test_oracle_boundary_audit.py +++ b/tests/test_oracle_boundary_audit.py @@ -72,3 +72,65 @@ def test_oracle_boundary_audit_reports_dasha_and_shadbala_boundaries() -> None: assert template["status"] == "template_only" assert template["ready_for_calibration"] is False assert template["missing_target_fields"] + + +def test_oracle_boundary_audit_compares_external_verified_template_rows(tmp_path: Path) -> None: + oracle = json.loads((ROOT / "references/oracle/dasha_shadbala_oracle_cases.json").read_text(encoding="utf-8")) + case = oracle["template_cases"][1] + case["status"] = "external_verified" + case["target"]["vimshottari_start_date"] = "1951-11-01" + case["target"]["shadbala_components"] = { + planet: { + "sthana": 1.0, + "dig": 2.0, + "kala": 3.0, + "chesta": 4.0, + "naisargika": 5.0, + "drik": 6.0, + "total_rupa": 21.0, + } + for planet in ["Sun", "Moon", "Mars", "Mercury", "Jupiter", "Venus", "Saturn"] + } + case["evidence_packet"] = { + "capture_id": "external_template_steve_jobs_dasha_lahiri", + "status": "external_verified", + "metadata": { + "tool_name": "JHora", + "tool_version_or_url": "manual-jhora-8.0", + "capture_date": "2026-06-26", + "source_artifact": "references/oracle/artifacts/steve_jobs_jhora_redacted.png", + "ayanamsa": "lahiri", + "node_mode": "true", + "timezone": "UTC-08:00", + "operator_note": "Typed from redacted external JHora screenshot.", + }, + } + oracle_path = tmp_path / "oracle.json" + oracle_path.write_text(json.dumps(oracle, ensure_ascii=False), encoding="utf-8") + + completed = subprocess.run( + [ + sys.executable, + "scripts/oracle_boundary_audit.py", + "--oracle-file", + str(oracle_path), + ], + cwd=ROOT, + text=True, + capture_output=True, + timeout=60, + check=False, + ) + + assert completed.returncode == 0, completed.stderr or completed.stdout + report = json.loads(completed.stdout) + assert report["summary"]["external_verified_template_cases"] == 1 + assert report["summary"]["production_tuning_recommended"] is False + comparison = report["template_comparisons"][0] + assert comparison["case_id"] == "template_steve_jobs_dasha_lahiri" + assert comparison["status"] == "external_verified" + assert comparison["dasha"]["target_start_date"] == "1951-11-01" + assert comparison["dasha"]["date_delta_days"] is not None + assert comparison["shadbala"]["planets"]["Sun"]["external_total_rupa"] == 21.0 + assert comparison["shadbala"]["planets"]["Sun"]["total_rupa_delta"] is not None + assert comparison["calibration_decision"] == "do_not_tune_single_template"