Jyotisha/scripts/rectification/varga_style_calibration_report.py

"""Offline hit-rate for yearless personality probes.

Reads an anonymized JSON export. Prints n and hit rate grouped by D9 / D10 /
nakshatra. Never prints a per-case row.
"""

from __future__ import annotations

import argparse
import json
import sys
from collections import defaultdict
from pathlib import Path
from typing import Any, Iterable, Mapping

GROUPS = ("d9", "d10", "nakshatra")
HOSPITAL = "hospital_record"
MAX_UNCERTAINTY_MINUTES = 2


def _text(value: object) -> str:
    return value.strip() if isinstance(value, str) else ""


def _int(value: object) -> int | None:
    if isinstance(value, bool) or value is None:
        return None
    if isinstance(value, int) and not isinstance(value, bool):
        return value
    if isinstance(value, float) and value.is_integer():
        return int(value)
    if isinstance(value, str) and value.strip().lstrip("-").isdigit():
        return int(value.strip())
    return None


def infer_group(row: Mapping[str, Any]) -> str | None:
    explicit = _text(row.get("group")).lower()
    if explicit in GROUPS:
        return explicit
    key = _text(row.get("semantic_key") or row.get("semanticKey")).lower()
    if key.startswith("varga.d9.") or ".d9." in key:
        return "d9"
    if key.startswith("varga.d10.") or ".d10." in key:
        return "d10"
    if "nakshatra" in key:
        return "nakshatra"
    return None


def _sign(value: object) -> str:
    return _text(value)


def selected_sign(row: Mapping[str, Any]) -> str | None:
    answer = _text(row.get("answer_class") or row.get("answerClass"))
    if not answer or answer in {"unsure", "no"}:
        return None
    options = row.get("options")
    if not isinstance(options, list):
        return None
    for option in options:
        if not isinstance(option, Mapping):
            continue
        option_class = _text(option.get("answer_class") or option.get("answerClass"))
        if option_class != answer:
            continue
        sign = _sign(option.get("sign"))
        return sign or None
    return None


def recorded_sign(row: Mapping[str, Any]) -> str | None:
    sign = _sign(row.get("recorded_sign") or row.get("recordedSign"))
    return sign or None


def hospital_case(row: Mapping[str, Any]) -> bool:
    if _text(row.get("birth_time_source") or row.get("birthTimeSource")) != HOSPITAL:
        return False
    before = _int(row.get("uncertainty_before_minutes") or row.get("uncertaintyBeforeMinutes"))
    after = _int(row.get("uncertainty_after_minutes") or row.get("uncertaintyAfterMinutes"))
    if before is None or after is None:
        return False
    return before <= MAX_UNCERTAINTY_MINUTES and after <= MAX_UNCERTAINTY_MINUTES


def summarize(payload: Mapping[str, Any] | Iterable[Any]) -> dict[str, dict[str, float | int | None]]:
    cases = payload.get("cases") if isinstance(payload, Mapping) else payload
    if not isinstance(cases, list):
        cases = []
    hits: dict[str, int] = defaultdict(int)
    total: dict[str, int] = defaultdict(int)
    for case in cases:
        if not isinstance(case, Mapping) or not hospital_case(case):
            continue
        answers = case.get("answers")
        if not isinstance(answers, list):
            continue
        for answer in answers:
            if not isinstance(answer, Mapping):
                continue
            group = infer_group(answer)
            chosen = selected_sign(answer)
            recorded = recorded_sign(answer)
            if group is None or not chosen or not recorded:
                continue
            total[group] += 1
            if chosen == recorded:
                hits[group] += 1
    out: dict[str, dict[str, float | int | None]] = {}
    for group in GROUPS:
        n = total[group]
        rate = round(hits[group] / n, 3) if n else None
        out[group] = {"n": n, "hit_rate": rate}
    return out


def load_payload(path: Path) -> Any:
    return json.loads(path.read_text(encoding="utf-8"))


def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(description="Yearless personality probe hit-rate (hospital records only).")
    parser.add_argument("path", type=Path, help="Anonymized JSON export")
    args = parser.parse_args(argv)
    payload = load_payload(args.path)
    report = summarize(payload)
    json.dump(report, sys.stdout, ensure_ascii=False)
    sys.stdout.write("\n")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())