"""Deterministic BaZi accuracy validation and shadow-engine comparison. This module deliberately does not participate in the production calculation path. It answers a narrower question: did the current chart contract drift, and where do the optional comparison engines disagree? A disagreement is reported, not silently resolved by majority vote. Run from this directory:: python accuracy_validation.py python accuracy_validation.py --strict-shadow Optional shadow dependencies are listed in ``requirements-accuracy.txt``. The Tyme engine shares the ShouXing lineage with lunar-python, so it is useful for policy/portability checks but is not an independent astronomical authority. """ from __future__ import annotations import argparse import importlib import json from dataclasses import dataclass from datetime import datetime from pathlib import Path from typing import Any, Dict, Iterable, List, Optional, Tuple from bazi_calculator import BaZiCalculator from lunar_python import Solar from rule_contract import validate_rule_contract ROOT = Path(__file__).resolve().parent MANIFEST_PATH = ROOT / "data" / "accuracy" / "golden_cases.json" # These are current product contracts, made explicit so a future migration has # to choose and document a new policy instead of changing it accidentally. CHART_DAY_BOUNDARY_POLICY = "late_zi_keeps_birth_day" CHART_SECT = 2 YUN_SECT = 1 @dataclass(frozen=True) class ValidationCase: case_id: str birth_date: str birth_time: str gender: str expected_pillars: Optional[List[str]] = None expected_lunar: Optional[Dict[str, Any]] = None expected_yun: Optional[Dict[str, Any]] = None def load_manifest(path: Path = MANIFEST_PATH) -> Dict[str, Any]: with path.open("r", encoding="utf-8") as handle: return json.load(handle) def _parts(case: ValidationCase) -> Tuple[int, int, int, int, int]: year, month, day = (int(part) for part in case.birth_date.split("-")) hour, minute = (int(part) for part in case.birth_time.split(":")) return year, month, day, hour, minute def _current_snapshot(case: ValidationCase) -> Dict[str, Any]: year, month, day, hour, minute = _parts(case) solar = Solar.fromYmdHms(year, month, day, hour, minute, 0) lunar = solar.getLunar() eight_char = lunar.getEightChar() gender_int = 1 if case.gender == "male" else 0 yun = eight_char.getYun(gender_int, YUN_SECT) start_solar = yun.getStartSolar() all_dayun = yun.getDaYun() return { "engine": "lunar-python", "chart_sect": CHART_SECT, "yun_sect": YUN_SECT, "day_boundary_policy": CHART_DAY_BOUNDARY_POLICY, "pillars": [ eight_char.getYear(), eight_char.getMonth(), eight_char.getDay(), eight_char.getTime(), ], "lunar": { "lunar_year": lunar.getYear(), "lunar_month": abs(lunar.getMonth()), "lunar_day": lunar.getDay(), "is_leap": lunar.getMonth() < 0, }, "yun": { "direction": "forward" if yun.isForward() else "backward", "start_solar": start_solar.toYmdHms(), "start_offset": [ yun.getStartYear(), yun.getStartMonth(), yun.getStartDay(), yun.getStartHour(), ], "first_cycles": [ [item.getGanZhi(), item.getStartAge(), item.getStartYear()] for item in all_dayun[1:4] ], }, } def _tyme_snapshot(case: ValidationCase) -> Dict[str, Any]: """Calculate the same product policy with Tyme4py when installed.""" from tyme4py.eightchar import ChildLimit from tyme4py.eightchar.provider.impl import ( LunarSect1ChildLimitProvider, LunarSect2EightCharProvider, ) from tyme4py.enums import Gender from tyme4py.lunar import LunarHour from tyme4py.solar import SolarTime year, month, day, hour, minute = _parts(case) solar_time = SolarTime(year, month, day, hour, minute, 0) old_lunar_provider = LunarHour.provider old_child_provider = ChildLimit.provider try: # Current production chart semantics: sect 2 (late Zi keeps the day). LunarHour.provider = LunarSect2EightCharProvider() # Current production Yun semantics: sect 1 (3 days = 1 year, etc.). ChildLimit.provider = LunarSect1ChildLimitProvider() eight_char = solar_time.get_lunar_hour().get_eight_char() child_limit = ChildLimit( solar_time, Gender.MAN if case.gender == "male" else Gender.WOMAN, ) first = child_limit.get_start_decade_fortune() first_cycles = [first.next(index) for index in range(3)] end_time = child_limit.get_end_time() return { "engine": "tyme4py", "chart_policy": "LunarSect2EightCharProvider", "yun_policy": "LunarSect1ChildLimitProvider", "pillars": [ eight_char.get_year().get_name(), eight_char.get_month().get_name(), eight_char.get_day().get_name(), eight_char.get_hour().get_name(), ], "yun": { "start_solar": ( f"{end_time.get_year():04d}-{end_time.get_month():02d}-" f"{end_time.get_day():02d} {end_time.get_hour():02d}:" f"{end_time.get_minute():02d}:{end_time.get_second():02d}" ), "start_offset": [ child_limit.get_year_count(), child_limit.get_month_count(), child_limit.get_day_count(), child_limit.get_hour_count(), ], "first_cycles": [ [ item.get_name(), item.get_start_age(), item.get_start_sixty_cycle_year().get_year(), ] for item in first_cycles ], }, } finally: LunarHour.provider = old_lunar_provider ChildLimit.provider = old_child_provider def _sxtwl_snapshot(case: ValidationCase) -> Dict[str, Any]: """Compare calendar/pillars with the optional sxtwl Python binding. sxtwl does not implement our Da Yun contract, so this intentionally only compares civil-calendar and four-pillar fields. It is a second implementation check, not a replacement for HKO/JPL reference data. """ sxtwl = importlib.import_module("sxtwl") year, month, day, hour, _minute = _parts(case) item = sxtwl.fromSolar(year, month, day) gan = "甲乙丙丁戊己庚辛壬癸" zhi = "子丑寅卯辰巳午未申酉戌亥" def gz_name(value: Any) -> str: return f"{gan[value.tg]}{zhi[value.dz]}" return { "engine": "sxtwl", "pillars": [ gz_name(item.getYearGZ(True)), gz_name(item.getMonthGZ()), gz_name(item.getDayGZ()), gz_name(item.getHourGZ(hour)), ], "lunar": { "lunar_year": item.getLunarYear(), "lunar_month": abs(item.getLunarMonth()), "lunar_day": item.getLunarDay(), "is_leap": item.isLunarLeap(), }, } def _compare(expected: Any, actual: Any, path: str, diffs: List[str]) -> None: if expected != actual: diffs.append(f"{path}: expected {expected!r}, got {actual!r}") def validate_calendar_case(raw: Dict[str, Any]) -> Dict[str, Any]: year, month, day = (int(part) for part in raw["solar"].split("-")) solar = Solar.fromYmd(year, month, day) lunar = solar.getLunar() actual = { "lunar_year": lunar.getYear(), "lunar_month": abs(lunar.getMonth()), "lunar_day": lunar.getDay(), "is_leap": lunar.getMonth() < 0, } diffs: List[str] = [] _compare(raw["expected"], actual, "calendar", diffs) return { "id": raw["id"], "solar": raw["solar"], "expected": raw["expected"], "actual": actual, "passed": not diffs, "diffs": diffs, } def _case_from_raw(raw: Dict[str, Any]) -> ValidationCase: return ValidationCase( case_id=raw["id"], birth_date=raw["birth_date"], birth_time=raw["birth_time"], gender=raw["gender"], expected_pillars=raw.get("expected_pillars"), expected_lunar=raw.get("expected_lunar"), expected_yun=raw.get("expected_yun"), ) def validate_bazi_case(raw: Dict[str, Any], include_shadows: bool = True) -> Dict[str, Any]: case = _case_from_raw(raw) current = _current_snapshot(case) diffs: List[str] = [] if case.expected_pillars is not None: _compare(case.expected_pillars, current["pillars"], "pillars", diffs) if case.expected_lunar is not None: _compare(case.expected_lunar, current["lunar"], "lunar", diffs) if case.expected_yun is not None: _compare(case.expected_yun, current["yun"], "yun", diffs) shadows: Dict[str, Any] = {} for name, loader in (("tyme4py", _tyme_snapshot), ("sxtwl", _sxtwl_snapshot)): if not include_shadows: continue try: shadow = loader(case) shadow_diffs: List[str] = [] _compare(current["pillars"], shadow["pillars"], "pillars", shadow_diffs) if name == "tyme4py": # Tyme exposes the 起运 instant and cycle sequence, but not a # separate direction field. Compare the common contract fields # instead of treating an omitted field as a disagreement. for key in ("start_solar", "start_offset", "first_cycles"): _compare(current["yun"][key], shadow["yun"][key], f"yun.{key}", shadow_diffs) else: _compare(current["lunar"], shadow["lunar"], "lunar", shadow_diffs) shadows[name] = { "available": True, "snapshot": shadow, "passed": not shadow_diffs, "diffs": shadow_diffs, } except ImportError as exc: shadows[name] = {"available": False, "reason": str(exc)} except Exception as exc: # report adapter failures instead of hiding them shadows[name] = { "available": True, "passed": False, "error": f"{type(exc).__name__}: {exc}", } return { "id": case.case_id, "birth_date": case.birth_date, "birth_time": case.birth_time, "gender": case.gender, "expected": { "pillars": case.expected_pillars, "lunar": case.expected_lunar, "yun": case.expected_yun, }, "current": current, "passed": not diffs, "diffs": diffs, "shadows": shadows, } def validate_boundary_policy() -> Dict[str, Any]: """Lock the current late-Zi contract around 23:00 and midnight.""" def snapshot(date: str, time: str) -> Dict[str, Any]: return _current_snapshot( ValidationCase(f"boundary-{date}-{time}", date, time, "male") ) before = snapshot("2024-06-15", "22:59") late = snapshot("2024-06-15", "23:00") late_end = snapshot("2024-06-15", "23:59") midnight = snapshot("2024-06-16", "00:00") midnight_end = snapshot("2024-06-16", "00:59") diffs: List[str] = [] _compare("亥", before["pillars"][3][1], "22:59 hour branch", diffs) _compare("子", late["pillars"][3][1], "23:00 hour branch", diffs) _compare(late["pillars"], late_end["pillars"], "23:00..23:59 stability", diffs) _compare(midnight["pillars"], midnight_end["pillars"], "00:00..00:59 stability", diffs) _compare(before["pillars"][2], late["pillars"][2], "late Zi day policy", diffs) if midnight["pillars"][2] == late["pillars"][2]: diffs.append("midnight must roll to a different day pillar") return { "policy": CHART_DAY_BOUNDARY_POLICY, "cases": { "22:59": before["pillars"], "23:00": late["pillars"], "23:59": late_end["pillars"], "00:00": midnight["pillars"], "00:59": midnight_end["pillars"], }, "passed": not diffs, "diffs": diffs, } def run_validation(include_shadows: bool = True) -> Dict[str, Any]: manifest = load_manifest() calendar_results = [validate_calendar_case(item) for item in manifest["calendar_cases"]] bazi_results = [ validate_bazi_case(item, include_shadows=include_shadows) for item in manifest["bazi_cases"] ] boundary_result = validate_boundary_policy() rule_contract = validate_rule_contract() return { "schema_version": 1, "manifest": str(MANIFEST_PATH.relative_to(ROOT)), "contract": { "chart_sect": CHART_SECT, "yun_sect": YUN_SECT, "day_boundary_policy": CHART_DAY_BOUNDARY_POLICY, }, "calendar": { "passed": all(item["passed"] for item in calendar_results), "cases": calendar_results, }, "bazi": { "passed": all(item["passed"] for item in bazi_results), "cases": bazi_results, }, "boundary": boundary_result, "rule_contract": rule_contract, } def _has_shadow_failure(report: Dict[str, Any]) -> bool: for case in report["bazi"]["cases"]: for shadow in case.get("shadows", {}).values(): if shadow.get("available") and not shadow.get("passed", False): return True return False def main() -> int: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument( "--strict-shadow", action="store_true", help="fail if an installed optional shadow engine disagrees", ) parser.add_argument( "--no-shadow", action="store_true", help="run only local and external golden cases", ) args = parser.parse_args() report = run_validation(include_shadows=not args.no_shadow) print(json.dumps(report, ensure_ascii=False, indent=2)) if ( not report["calendar"]["passed"] or not report["bazi"]["passed"] or not report["boundary"]["passed"] or not report["rule_contract"]["passed"] ): return 1 if args.strict_shadow and _has_shadow_failure(report): return 2 return 0 if __name__ == "__main__": raise SystemExit(main())