lib/rrf.py

"""
lib/rrf.py — V6 Adaptive Reciprocal Rank Fusion

Replaces the fixed-weight fusion (0.40/0.35/0.25) with principled RRF.

RRF formula:
  score(d) = SUM_r  1 / (k + rank_r(d))

Where k is typically 60, but we adapt it based on:
  - JD complexity (more skills -> higher k to reduce dominance)
  - Score distribution (skewed distributions -> lower k to differentiate)

Adaptive K selection:
  - Compute score entropy for each ranking
  - Higher entropy = more uniform distribution = keep k at default
  - Lower entropy = dominated by few candidates = lower k to differentiate
"""
from __future__ import annotations
import math
from collections import Counter
from typing import Sequence


def _score_entropy(scores: list[float]) -> float:
    """Compute normalized entropy of a score distribution."""
    if not scores or max(scores) == min(scores):
        return 1.0  # uniform

    # Normalize to probabilities
    total = sum(scores)
    if total <= 0:
        return 1.0
    probs = [s / total for s in scores]

    entropy = -sum(p * math.log2(p) for p in probs if p > 0)
    max_entropy = math.log2(len(probs))
    return entropy / max_entropy if max_entropy > 0 else 1.0


def adaptive_k(sparse_scores: list[float],
               skill_scores: list[float],
               behaviour_scores: list[float],
               base_k: int = 60) -> int:
    """
    Adaptively choose k for RRF based on score distributions.

    When scores are very concentrated (low entropy), use a lower k
    to help differentiate candidates. When uniform (high entropy),
    keep default k.
    """
    # Compute entropy for each ranking
    sparse_ent = _score_entropy(sparse_scores)
    skill_ent = _score_entropy(skill_scores)
    beh_ent = _score_entropy(behaviour_scores)

    avg_ent = (sparse_ent + skill_ent + beh_ent) / 3.0

    # Low entropy -> reduce k to amplify rank differences
    # High entropy -> keep k at default
    k = int(base_k * (0.5 + 0.5 * avg_ent))
    return max(10, min(100, k))


def reciprocal_rank_fusion(
    rankings: list[Sequence[str]],
    k: int = 60,
) -> dict[str, float]:
    """
    Fuse multiple rankings using Reciprocal Rank Fusion.

    Args:
        rankings: list of ranked candidate ID lists (best first)
        k: RRF constant (higher = more forgiving of rank differences)

    Returns:
        dict mapping candidate_id -> fused RRF score
    """
    scores: dict[str, float] = {}

    for ranking in rankings:
        for rank_pos, cid in enumerate(ranking):
            rrf_score = 1.0 / (k + rank_pos + 1)
            scores[cid] = scores.get(cid, 0.0) + rrf_score

    return scores


def adaptive_rrf(
    sparse_ranking: Sequence[tuple[str, float]],
    skill_ranking: Sequence[tuple[str, float]],
    behaviour_ranking: Sequence[tuple[str, float]],
    base_k: int = 60,
) -> dict[str, float]:
    """
    Adaptive RRF that chooses k based on score distributions.

    Args:
        sparse_ranking: list of (candidate_id, sparse_score), sorted best first
        skill_ranking: list of (candidate_id, skill_score), sorted best first
        behaviour_ranking: list of (candidate_id, behaviour_score), sorted best first
        base_k: base RRF constant

    Returns:
        dict mapping candidate_id -> adaptive RRF score
    """
    # Extract just the IDs in rank order
    sparse_ids = [cid for cid, _ in sparse_ranking]
    skill_ids = [cid for cid, _ in skill_ranking]
    behaviour_ids = [cid for cid, _ in behaviour_ranking]

    # Extract scores for entropy computation
    sparse_scores = [s for _, s in sparse_ranking]
    skill_scores = [s for _, s in skill_ranking]
    behaviour_scores = [s for _, s in behaviour_ranking]

    # Choose adaptive k
    k = adaptive_k(sparse_scores, skill_scores, behaviour_scores, base_k)

    # Apply RRF
    return reciprocal_rank_fusion([sparse_ids, skill_ids, behaviour_ids], k=k)


def rrf_score_for_candidates(
    candidate_ids: list[str],
    sparse_scores: dict[str, float],
    skill_scores: dict[str, float],
    behaviour_scores: dict[str, float],
) -> dict[str, float]:
    """
    Compute adaptive RRF scores for a set of candidates.

    Convenience function that handles ranking creation internally.
    """
    # Sort by each signal to create rankings
    sparse_ranked = sorted(
        [(cid, sparse_scores.get(cid, 0)) for cid in candidate_ids],
        key=lambda x: x[1], reverse=True,
    )
    skill_ranked = sorted(
        [(cid, skill_scores.get(cid, 0)) for cid in candidate_ids],
        key=lambda x: x[1], reverse=True,
    )
    behaviour_ranked = sorted(
        [(cid, behaviour_scores.get(cid, 0)) for cid in candidate_ids],
        key=lambda x: x[1], reverse=True,
    )

    return adaptive_rrf(sparse_ranked, skill_ranked, behaviour_ranked)