picarones/core/inter_engine.py

"""Métriques inter-moteurs (Sprint 35 — Étape 2 du plan d'évolution).

Deux familles de mesures qui répondent à des questions différentes mais
liées :

1. **Divergence taxonomique** (`kl_divergence`, `jensen_shannon_divergence`,
   `taxonomy_divergence_matrix`) — *à quel point les moteurs font-ils des
   erreurs de natures différentes ?*  Une divergence élevée signale des
   moteurs spécialisés sur des classes d'erreurs distinctes (visual vs
   abréviation vs casse) et donc des candidats pour un voting ensemble.

2. **Complémentarité** (`oracle_token_recall`, `complementarity_gap`,
   `pairwise_disagreement_rate`) — *quel CER serait atteignable si on
   combinait les moteurs ?*  La borne inférieure du CER atteignable par
   un voting majoritaire token-level est ``1 - oracle_token_recall``.
   Si elle est très inférieure au CER du meilleur moteur seul, l'effort
   d'un pipeline d'ensemble se justifie.  Sinon non.

Convention de typage
--------------------
Toutes les fonctions sont enregistrables dans le registre Sprint 34 si
on les wrappe par un adaptateur ``(input_types=(TEXT, TEXT))``.  Pour
limiter le bruit, on ne les enregistre **pas** automatiquement : ce sont
des métriques d'agrégation (multi-moteurs ou multi-documents) qui ne
correspondent pas au modèle « une jonction = une métrique » du runner.
Elles sont consommées par les détecteurs narratifs et le rapport HTML.

Note sur l'oracle
-----------------
La métrique ``oracle_token_recall`` retournée ici utilise un alignement
bag-of-words pondéré par multiplicité.  Ce n'est **pas** une vraie
borne atteignable par voting majoritaire séquentiel — c'est une borne
supérieure (proxy optimiste).  La vraie borne demanderait un
alignement séquentiel des hypothèses, ce qui est plus coûteux.  Pour
le diagnostic « ensemble vaut-il le coup ? », le proxy suffit
largement ; on documente clairement la limite dans le glossaire et le
rapport.
"""

from __future__ import annotations

import logging
import math
from collections import Counter

logger = logging.getLogger(__name__)


# ──────────────────────────────────────────────────────────────────────────
# Divergence taxonomique (KL / Jensen-Shannon)
# ──────────────────────────────────────────────────────────────────────────


def _smoothed_distribution(
    distribution: dict[str, float],
    keys: list[str],
    epsilon: float = 1e-12,
) -> list[float]:
    """Aligne une distribution sur l'ordre de ``keys`` et lisse les zéros.

    Le lissage évite ``log(0)`` dans la KL.  ``epsilon`` est volontairement
    minuscule pour ne pas modifier le résultat de manière sensible.
    """
    smoothed = [max(distribution.get(k, 0.0), epsilon) for k in keys]
    total = sum(smoothed)
    return [v / total for v in smoothed]


def kl_divergence(p: dict[str, float], q: dict[str, float]) -> float:
    """KL-divergence ``D(P||Q)`` en bits, sur l'union des clés.

    Les distributions n'ont pas besoin de partager exactement les mêmes
    clés ; les clés manquantes sont lissées à ``epsilon`` puis
    renormalisées.

    Returns
    -------
    float
        ``D(P||Q) ≥ 0``.  Vaut 0 si et seulement si P == Q.  N'est pas
        symétrique : ``kl(p, q) != kl(q, p)`` en général.
    """
    keys = sorted(set(p.keys()) | set(q.keys()))
    if not keys:
        return 0.0
    p_vec = _smoothed_distribution(p, keys)
    q_vec = _smoothed_distribution(q, keys)
    return sum(pi * math.log2(pi / qi) for pi, qi in zip(p_vec, q_vec))


def jensen_shannon_divergence(
    p: dict[str, float],
    q: dict[str, float],
) -> float:
    """JS-divergence symétrique en bits, bornée dans ``[0, 1]``.

    ``JS(P, Q) = ½ D(P||M) + ½ D(Q||M)`` avec ``M = (P + Q) / 2``.
    Symétrique et bornée — préférable à la KL pour construire une
    matrice triangulaire de divergences entre moteurs.
    """
    keys = sorted(set(p.keys()) | set(q.keys()))
    if not keys:
        return 0.0
    p_vec = _smoothed_distribution(p, keys)
    q_vec = _smoothed_distribution(q, keys)
    m_vec = [(pi + qi) / 2.0 for pi, qi in zip(p_vec, q_vec)]

    def _kl(a: list[float], b: list[float]) -> float:
        return sum(ai * math.log2(ai / bi) for ai, bi in zip(a, b) if ai > 0)

    js = 0.5 * _kl(p_vec, m_vec) + 0.5 * _kl(q_vec, m_vec)
    # Borne théorique : JS ∈ [0, 1] en bits.  Clamp pour absorber les
    # erreurs d'arrondi flottant.
    return max(0.0, min(1.0, js))


def taxonomy_divergence_matrix(
    distributions: dict[str, dict[str, float]],
    metric: str = "js",
) -> dict[str, dict[str, float]]:
    """Construit la matrice de divergence triangulaire entre moteurs.

    Parameters
    ----------
    distributions:
        ``{engine_name: {error_class: probability}}``.  Chaque
        distribution doit sommer à environ 1 (pas de validation stricte
        — les distributions taxonomiques de Picarones sont déjà
        normalisées par ``aggregate_taxonomy``).
    metric:
        ``"js"`` (défaut, symétrique) ou ``"kl"`` (asymétrique).

    Returns
    -------
    dict[str, dict[str, float]]
        Matrice ``{engine_a: {engine_b: divergence}}`` symétrique pour
        ``js``, asymétrique pour ``kl``.  La diagonale vaut 0.
    """
    if metric not in ("js", "kl"):
        raise ValueError(f"metric doit être 'js' ou 'kl' — reçu {metric!r}")
    fn = jensen_shannon_divergence if metric == "js" else kl_divergence

    engines = sorted(distributions.keys())
    matrix: dict[str, dict[str, float]] = {a: {} for a in engines}
    for a in engines:
        for b in engines:
            if a == b:
                matrix[a][b] = 0.0
            elif metric == "js" and b in matrix and a in matrix[b]:
                # Symétrique : recopie pour éviter de recalculer
                matrix[a][b] = matrix[b][a]
            else:
                matrix[a][b] = fn(distributions[a], distributions[b])
    return matrix


# ──────────────────────────────────────────────────────────────────────────
# Complémentarité (oracle token recall)
# ──────────────────────────────────────────────────────────────────────────


def _word_multiset(text: str) -> Counter[str]:
    """Décomposition en multiset de tokens (séparateur whitespace)."""
    return Counter(tok for tok in text.split() if tok)


def oracle_token_recall(
    reference: str,
    hypotheses: dict[str, str],
) -> float:
    """Borne supérieure (proxy bag-of-words) du token-recall atteignable
    par un voting majoritaire entre tous les moteurs fournis.

    Pour chaque token de la référence (avec sa multiplicité), on
    considère qu'il est "préservé" par l'ensemble si au moins un moteur
    en produit une occurrence non encore comptée.  Le score est le ratio
    d'occurrences GT préservées sur le total.

    Parameters
    ----------
    reference:
        Texte GT.
    hypotheses:
        ``{engine_name: hypothesis_text}``.

    Returns
    -------
    float
        Ratio dans ``[0, 1]``.  ``1.0`` = chaque token GT est présent
        dans au moins une hypothèse à hauteur de sa multiplicité.

    Note
    ----
    Cette borne est **optimiste** (supérieure à la vraie borne par
    voting séquentiel) car elle ignore l'ordre d'apparition.  Pour le
    diagnostic « un voting vaut-il l'effort ? » le proxy suffit ; pour
    une vraie borne il faudrait un alignement séquentiel.
    """
    ref_counter = _word_multiset(reference)
    if not ref_counter or not hypotheses:
        return 1.0 if not ref_counter else 0.0

    hyp_counters = [_word_multiset(h) for h in hypotheses.values()]
    total_ref = sum(ref_counter.values())
    preserved = 0
    for token, gt_count in ref_counter.items():
        # Pour chaque moteur, le nombre d'occurrences disponibles, plafonné
        # à la multiplicité GT.  L'oracle prend le max sur les moteurs.
        best = max((min(gt_count, hc.get(token, 0)) for hc in hyp_counters), default=0)
        preserved += best
    return preserved / total_ref


def complementarity_gap(
    reference: str,
    hypotheses: dict[str, str],
) -> dict[str, float]:
    """Compare l'oracle au meilleur moteur seul.

    Returns
    -------
    dict
        ``{
            "oracle_recall": float,        # bag-of-words recall de l'oracle
            "best_single_recall": float,   # meilleur recall token d'un moteur seul
            "best_engine": str,            # nom du moteur correspondant
            "absolute_gap": float,         # oracle - best_single (toujours ≥ 0)
            "relative_gap": float,         # absolute_gap / (1 - best_single + ε)
                                           # = fraction des erreurs encore évitables
                                           # par un ensemble
        }``
    """
    ref_counter = _word_multiset(reference)
    total = sum(ref_counter.values())
    if not total:
        return {
            "oracle_recall": 1.0,
            "best_single_recall": 1.0,
            "best_engine": "",
            "absolute_gap": 0.0,
            "relative_gap": 0.0,
        }

    def _single_recall(hyp_text: str) -> float:
        hc = _word_multiset(hyp_text)
        preserved = sum(min(gt, hc.get(tok, 0)) for tok, gt in ref_counter.items())
        return preserved / total

    if not hypotheses:
        return {
            "oracle_recall": 0.0,
            "best_single_recall": 0.0,
            "best_engine": "",
            "absolute_gap": 0.0,
            "relative_gap": 0.0,
        }

    per_engine = {name: _single_recall(h) for name, h in hypotheses.items()}
    best_engine, best_recall = max(per_engine.items(), key=lambda kv: kv[1])
    oracle = oracle_token_recall(reference, hypotheses)

    absolute_gap = max(0.0, oracle - best_recall)
    # relative_gap : fraction des erreurs du meilleur moteur que l'ensemble
    # serait théoriquement capable de récupérer (∈ [0, 1])
    headroom = max(1.0 - best_recall, 1e-12)
    relative_gap = min(1.0, absolute_gap / headroom)

    return {
        "oracle_recall": oracle,
        "best_single_recall": best_recall,
        "best_engine": best_engine,
        "absolute_gap": absolute_gap,
        "relative_gap": relative_gap,
    }


def pairwise_disagreement_rate(
    reference: str,
    hyp_a: str,
    hyp_b: str,
) -> float:
    """Fraction de tokens GT pour lesquels A et B sont en désaccord.

    Un désaccord = (l'un préserve le token, l'autre non) OU
    (les deux le ratent mais avec des substitutions différentes — non
    capturé ici, on reste sur la version simple présence/absence).

    Returns
    -------
    float
        Ratio dans ``[0, 1]``.  ``0`` = A et B font les mêmes choix
        (pas de gain d'ensemble).  ``1`` = A et B sont toujours en
        désaccord (gain d'ensemble maximal).
    """
    ref_counter = _word_multiset(reference)
    if not ref_counter:
        return 0.0
    a = _word_multiset(hyp_a)
    b = _word_multiset(hyp_b)
    total = sum(ref_counter.values())
    disagree = 0
    for tok, gt_count in ref_counter.items():
        a_pres = min(gt_count, a.get(tok, 0))
        b_pres = min(gt_count, b.get(tok, 0))
        # Compte les positions où A et B donnent une réponse différente
        disagree += abs(a_pres - b_pres)
    return disagree / total


# ──────────────────────────────────────────────────────────────────────────
# Agrégation au niveau benchmark (Sprint 36)
# ──────────────────────────────────────────────────────────────────────────


def compute_inter_engine_analysis(
    *,
    per_engine_outputs: dict[str, dict[str, str]],
    ground_truths: dict[str, str],
    taxonomy_distributions: dict[str, dict[str, float]] | None = None,
    divergence_metric: str = "js",
) -> dict:
    """Agrège les métriques inter-moteurs sur l'ensemble du corpus.

    Parameters
    ----------
    per_engine_outputs:
        ``{engine_name: {doc_id: hypothesis_text}}``.  Une entrée par
        moteur, avec une hypothèse par document.  Les documents absents
        d'un moteur (échecs, timeouts) sont simplement ignorés pour ce
        moteur — l'oracle est calculé sur les moteurs qui ont produit
        une sortie pour le doc.
    ground_truths:
        ``{doc_id: ground_truth_text}``.  La GT est la même pour tous
        les moteurs ; on la passe une seule fois.
    taxonomy_distributions:
        ``{engine_name: {error_class: probability}}`` — typiquement
        ``EngineReport.aggregated_taxonomy["class_distribution"]``.  Si
        ``None`` ou vide, la divergence taxonomique n'est pas calculée.
    divergence_metric:
        ``"js"`` (défaut, symétrique) ou ``"kl"``.

    Returns
    -------
    dict
        Structure stable consommable par les détecteurs narratifs et le
        rapport HTML :
        ``{
            "complementarity": {
                "oracle_recall": float,
                "best_single_recall": float,
                "best_engine": str,
                "absolute_gap": float,
                "relative_gap": float,
                "doc_count": int,
                "per_doc": [{doc_id, oracle, best, gap}, ...]   # max 50 docs
            },
            "taxonomy_divergence": {
                "metric": "js"|"kl",
                "matrix": {engine_a: {engine_b: divergence}},
                "max_pair": [engine_a, engine_b, value]   # paire la plus divergente
            } | None,
            "engines": [...],   # liste des moteurs analysés (ordre stable)
        }``
    """
    engines = sorted(per_engine_outputs.keys())
    result: dict = {"engines": engines}

    # ── Complémentarité agrégée doc par doc ──────────────────────────────
    if not engines:
        result["complementarity"] = None
    else:
        total_oracle_preserved = 0
        total_ref_tokens = 0
        per_engine_preserved: dict[str, int] = {name: 0 for name in engines}
        per_doc_records: list[dict] = []

        for doc_id, gt in ground_truths.items():
            ref_counter = _word_multiset(gt)
            ref_total = sum(ref_counter.values())
            if not ref_total:
                continue
            total_ref_tokens += ref_total

            doc_hyps: dict[str, str] = {}
            for name in engines:
                hyp = per_engine_outputs.get(name, {}).get(doc_id)
                if hyp is not None:
                    doc_hyps[name] = hyp

            if not doc_hyps:
                continue

            hyp_counters = {n: _word_multiset(h) for n, h in doc_hyps.items()}

            doc_oracle = 0
            doc_best_per_engine: dict[str, int] = {n: 0 for n in doc_hyps}
            for tok, gt_count in ref_counter.items():
                # Oracle : meilleur des moteurs sur ce token
                best_for_token = 0
                for name, hc in hyp_counters.items():
                    preserved = min(gt_count, hc.get(tok, 0))
                    doc_best_per_engine[name] += preserved
                    if preserved > best_for_token:
                        best_for_token = preserved
                doc_oracle += best_for_token

            total_oracle_preserved += doc_oracle
            for name, count in doc_best_per_engine.items():
                per_engine_preserved[name] += count

            doc_best = max(doc_best_per_engine.values()) if doc_best_per_engine else 0
            per_doc_records.append({
                "doc_id": doc_id,
                "oracle_recall": doc_oracle / ref_total,
                "best_single_recall": doc_best / ref_total,
                "absolute_gap": (doc_oracle - doc_best) / ref_total,
            })

        if total_ref_tokens == 0:
            result["complementarity"] = None
        else:
            oracle_recall = total_oracle_preserved / total_ref_tokens
            recalls = {
                name: per_engine_preserved[name] / total_ref_tokens
                for name in engines
            }
            best_engine, best_recall = max(recalls.items(), key=lambda kv: kv[1])
            absolute_gap = max(0.0, oracle_recall - best_recall)
            headroom = max(1.0 - best_recall, 1e-12)
            relative_gap = min(1.0, absolute_gap / headroom)

            # Garder les ``per_doc_records`` les plus instructifs : tri par
            # gap absolu décroissant, top 50.  Les détecteurs narratifs
            # n'en consomment que quelques-uns.
            per_doc_records.sort(key=lambda r: r["absolute_gap"], reverse=True)
            per_doc_top = per_doc_records[:50]

            result["complementarity"] = {
                "oracle_recall": oracle_recall,
                "best_single_recall": best_recall,
                "best_engine": best_engine,
                "absolute_gap": absolute_gap,
                "relative_gap": relative_gap,
                "doc_count": len(per_doc_records),
                "per_engine_recall": recalls,
                "per_doc": per_doc_top,
            }

    # ── Divergence taxonomique ─────────────────────────────────────────
    if not taxonomy_distributions:
        result["taxonomy_divergence"] = None
    else:
        matrix = taxonomy_divergence_matrix(
            taxonomy_distributions,
            metric=divergence_metric,
        )
        # Cherche la paire la plus divergente (utile pour la synthèse
        # narrative qui veut nommer les deux moteurs candidats à
        # l'ensemble).
        max_pair: tuple[str, str, float] = ("", "", 0.0)
        names = sorted(matrix.keys())
        for i, a in enumerate(names):
            for b in names[i + 1:]:
                v = matrix[a][b]
                if v > max_pair[2]:
                    max_pair = (a, b, v)

        result["taxonomy_divergence"] = {
            "metric": divergence_metric,
            "matrix": matrix,
            "max_pair": list(max_pair) if max_pair[2] > 0 else None,
        }

    return result


__all__ = [
    "kl_divergence",
    "jensen_shannon_divergence",
    "taxonomy_divergence_matrix",
    "oracle_token_recall",
    "complementarity_gap",
    "pairwise_disagreement_rate",
    "compute_inter_engine_analysis",
]