astrololo/tests/oracle/harness.py

"""Porównanie naszych domów z wyrocznią (Swiss Ephemeris) i raport zgodności.

ZASADA IZOLACJI: obu implementacjom podajemy TE SAME wejścia (RAMC, ε, φ) przez
`swe.houses_armc`. Gdybyśmy porównywali „nasz horoskop" z „horoskopem swissepha",
różnica w cuspie mogłaby wynikać z innego czasu gwiazdowego albo innego ε — czyli
z czegoś, co nie jest testowaną funkcją. Wtedy przy tysiącach przypadków utonęlibyśmy
w fałszywych alarmach. Czas gwiazdowy i ε testujemy OSOBNO (`compare_asc_mc`).

KRYTERIUM: nie procent zgodności, tylko liczba przypadków POWYŻEJ tolerancji plus
maksymalne odchylenie i MIEJSCE jego wystąpienia. Procent ukrywa kształt błędu:
„97%" nie odróżnia szumu zmiennoprzecinkowego od jednego rogu dziedziny, w którym
mylimy się o 30°.

LICENCJA: Swiss Ephemeris (AGPL) jest tu WYŁĄCZNIE wyrocznią testową. Nie wchodzi
do ścieżki produkcyjnej ani do zależności produktu — izolacja z LOG-27 zostaje.
"""
from __future__ import annotations

import random
from dataclasses import dataclass, field

import swisseph as swe

from domain import NARROWER_THAN_ORACLE, Case, in_domain, obliquity

# Litery systemów w Swiss Ephemeris.
SWE_CODE = {
    "whole_sign": b"W", "equal": b"E",
    "equal_mc": b"D",
    "whole_sign_aries": b"N", "porphyry": b"O",
    "placidus": b"P", "koch": b"K", "regiomontanus": b"R", "campanus": b"C",
    "morinus": b"M", "alcabitus": b"B", "vehlow": b"V", "topocentric": b"T",
}

TOLERANCE_DEG = 1.0 / 3600.0        # 1″ — uzgodniony próg


def angular_diff(a: float, b: float) -> float:
    """Najmniejsza różnica kątowa [0,180] — 359,9° i 0,1° dzieli 0,2°, nie 359,8°."""
    d = abs(a - b) % 360.0
    return min(d, 360.0 - d)


@dataclass
class Result:
    system: str
    checked: int = 0
    over_tolerance: int = 0
    skipped_out_of_domain: int = 0
    max_dev: float = 0.0
    max_where: str = ""
    domain_mismatch: list[str] = field(default_factory=list)
    deviations: list[float] = field(default_factory=list)

    @property
    def passed(self) -> bool:
        return self.over_tolerance == 0 and not self.domain_mismatch


def oracle_cusps(case: Case, eps: float, system: str) -> list[float] | None:
    """Cuspy z wyroczni albo None, gdy wyrocznia odmawia (poza dziedziną).

    pyswisseph zgłasza wyjątek dla Placidusa/Kocha powyżej koła podbiegunowego —
    to sygnał, że zestaw jest poza dziedziną, i sam w sobie jest użyteczny: pozwala
    sprawdzić, czy NASZA detekcja granicy pokrywa się z wyrocznią."""
    try:
        cusps, _ = swe.houses_armc(case.ramc % 360.0, case.lat, eps, SWE_CODE[system])
        return list(cusps)
    except Exception:
        return None


def compare_asc_mc(cases: list[Case]) -> Result:
    """Test funkcji WEJŚCIOWEJ: nasze Asc/MC vs wyrocznia, przy tym samym (RAMC, ε, φ).

    Osobno, bo błąd tutaj kaskadowałby na wszystkie systemy i wyglądał jak błąd domów.
    """
    from app.engine.houses import compute_asc, compute_mc

    r = Result(system="asc/mc")
    for c in cases:
        eps = obliquity(c.jd)
        try:
            _, ascmc = swe.houses_armc(c.ramc % 360.0, c.lat, eps, b"O")
        except Exception:
            r.skipped_out_of_domain += 1
            continue
        for ours, theirs, what in (
            (compute_asc(c.ramc, eps, c.lat), ascmc[0], "Asc"),
            (compute_mc(c.ramc, eps), ascmc[1], "MC"),
        ):
            d = angular_diff(ours, theirs)
            r.checked += 1
            r.deviations.append(d)
            if d > r.max_dev:
                r.max_dev, r.max_where = d, f"{what} @ {c.key()} [{c.label}]"
            if d > TOLERANCE_DEG:
                r.over_tolerance += 1
    return r


def compare_system(system: str, cases: list[Case]) -> Result:
    """Porównanie jednego systemu domów z wyrocznią na zadanym zestawie przypadków."""
    from app.engine.houses import cusps_for

    r = Result(system=system)
    for c in cases:
        eps = obliquity(c.jd)
        ours_in_domain = in_domain(system, c.lat, c.jd)
        theirs = oracle_cusps(c, eps, system)

        # Zgodność co do DZIEDZINY: jeśli wyrocznia odmawia, a my uważamy, że
        # liczymy — nasza detekcja granicy jest zepsuta (albo odwrotnie).
        # Węższa dziedzina po naszej stronie jest świadoma (patrz NARROWER_THAN_ORACLE),
        # więc nie jest rozjazdem. Odwrotność — my liczymy, wyrocznia odmawia — nadal jest.
        narrower_on_purpose = (system in NARROWER_THAN_ORACLE
                               and theirs is not None and not ours_in_domain)
        if not narrower_on_purpose and (theirs is None) != (not ours_in_domain):
            r.domain_mismatch.append(
                f"{c.key()} [{c.label}] — my:{'w dziedzinie' if ours_in_domain else 'poza'}, "
                f"wyrocznia:{'odmawia' if theirs is None else 'liczy'}")
            continue
        if theirs is None or not ours_in_domain:
            r.skipped_out_of_domain += 1
            continue

        try:
            ours = cusps_for(c.ramc, eps, c.lat, system)
        except Exception as e:                       # nasz kod NIE MOŻE się wywalić
            r.domain_mismatch.append(f"{c.key()} [{c.label}] — nasz kod rzucił: {e}")
            continue

        for i in range(12):
            d = angular_diff(ours[i], theirs[i])
            r.checked += 1
            r.deviations.append(d)
            if d > r.max_dev:
                r.max_dev, r.max_where = d, f"dom {i + 1} @ {c.key()} [{c.label}]"
            if d > TOLERANCE_DEG:
                r.over_tolerance += 1
    return r


def histogram(devs: list[float]) -> str:
    """Rozkład odchyleń — pokazuje KSZTAŁT zgodności, nie tylko sam maksimum."""
    if not devs:
        return "     (brak danych)"
    buckets = [
        ("= 0", lambda d: d == 0.0),
        ("< 1e-9°", lambda d: 0 < d < 1e-9),
        ("< 1e-6°", lambda d: 1e-9 <= d < 1e-6),
        ("< 1″", lambda d: 1e-6 <= d < TOLERANCE_DEG),
        ("≥ 1″ (BŁĄD)", lambda d: d >= TOLERANCE_DEG),
    ]
    out = []
    for name, pred in buckets:
        n = sum(1 for d in devs if pred(d))
        if n:
            out.append(f"     {name:14} {n:7}  {'█' * min(40, 1 + n * 40 // len(devs))}")
    return "\n".join(out)


# Ile punktów ekliptyki sprawdzamy na jeden zestaw (RAMC, ε, φ).
_PROBES_PER_CASE = 6


def compare_house_assignment(system: str, cases: list[Case]) -> Result:
    """Nasze przypisanie obiektu do domu vs swe_house_pos.

    To OSOBNA funkcja od liczenia cuspów i osobny rodzaj błędu: cuspy mogą się
    zgadzać co do zera, a planeta i tak wylądować w złym domu. Tak właśnie było —
    przy |φ| powyżej koła podbiegunowego systemy dzielące koła wielkie mają
    kolejność domów ODWRÓCONĄ, a nasze assign_house szło zawsze do przodu.

    WYROCZNIA MA TU GRANICĘ WIARYGODNOŚCI. Dla topocentrica swe_house_pos
    przestaje być zgodne z WŁASNYMI cuspami swissepha powyżej koła podbiegunowego
    (zmierzone: 100% zgodności do 62°, 83,9% przy 66°, ok. 50% przy 72°; dla
    regiomontanusa w tych samych punktach 100%). Skoro wyrocznia przeczy sama
    sobie, nie może rozstrzygać — nie porównujemy tam, zamiast dopisywać wyjątek
    „bo topocentric". Nasze przypisanie jest w tym obszarze sprawdzane inaczej:
    testem samospójności z cuspami (services/logic/tests/test_houses.py).
    """
    from app.engine.houses import assign_house, cusps_for, polar_circle

    r = Result(system=f"dom:{system}")
    rng = random.Random(20260806)
    for c in cases:
        eps = obliquity(c.jd)
        if system == "topocentric" and abs(c.lat) >= polar_circle(eps):
            r.skipped_out_of_domain += 1
            continue
        try:
            ours_cusps = cusps_for(c.ramc, eps, c.lat, system)
        except Exception:
            r.skipped_out_of_domain += 1
            continue
        for _ in range(_PROBES_PER_CASE):
            lon = rng.uniform(0.0, 360.0)
            try:
                theirs = int(swe.house_pos(c.ramc % 360.0, c.lat, eps,
                                           (lon, 0.0), SWE_CODE[system]))
            except Exception:
                r.skipped_out_of_domain += 1
                continue
            r.checked += 1
            ours = assign_house(lon, ours_cusps)
            if ours != theirs:
                r.over_tolerance += 1
                r.max_dev = max(r.max_dev, float(abs(ours - theirs)))
                if len(r.domain_mismatch) < 5:
                    r.domain_mismatch.append(
                        f"{c.key()} lon={lon:.4f} — nasz dom {ours}, wyrocznia {theirs}")
    return r


def format_report(results: list[Result], seed: int | None = None) -> str:
    lines = ["", "=" * 78,
             "ZGODNOŚĆ Z WYROCZNIĄ (Swiss Ephemeris) — domy astrologiczne",
             f"tolerancja: {TOLERANCE_DEG:.8f}° (1″)" + (f"   ziarno: {seed}" if seed is not None else ""),
             "=" * 78,
             f"{'SYSTEM':<18}{'SPRAWDZONYCH':>13}{'>TOL':>7}{'POZA DZIEDZ.':>14}{'MAX ODCH.':>14}  WYNIK"]
    for r in results:
        lines.append(f"{r.system:<18}{r.checked:>13}{r.over_tolerance:>7}"
                     f"{r.skipped_out_of_domain:>14}{r.max_dev:>14.9f}  "
                     f"{'OK' if r.passed else 'BŁĄD'}")
    for r in results:
        if r.max_dev > 0 or not r.passed:
            lines += ["", f"  [{r.system}] największe odchylenie: {r.max_dev:.9f}° — {r.max_where}"]
        if r.domain_mismatch:
            lines.append(f"  [{r.system}] NIEZGODNOŚĆ DZIEDZINY ({len(r.domain_mismatch)}):")
            lines += [f"     {m}" for m in r.domain_mismatch[:5]]
            if len(r.domain_mismatch) > 5:
                lines.append(f"     … i {len(r.domain_mismatch) - 5} więcej")
    lines += ["", "  Rozkład odchyleń (wszystkie systemy):", histogram(
        [d for r in results for d in r.deviations]), "=" * 78]
    return "\n".join(lines)