astrologybazi-api / accuracy_validation.py
th
feat(bazi): add evidence-first dayun assessment
0d792fe
Raw History Blame Contribute Delete
14.6 kB
"""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())