fix(rectification): downgrade yearless personality probes to tie-breakers (BUG-629)

Ask dated dasha probes first; D9/D10 and nakshatra wait until that pool is empty, score at half weight, and never eliminate. Skill 10.0.21.

Co-authored-by: Cursor <cursoragent@cursor.com>

scripts/rectification/varga_style_calibration_report.py

@@ -0,0 +1,135 @@
+"""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())