Blame for accuracy_validation.py

"""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())