"""Identity ranking for catalog scores. Separates raw matcher scores from accept/unknown/ambiguous/reject so a zero-score crop cannot become a SKU. Thresholds are conservative defaults for the 21-SKU closed set; they are not calibrated accuracy claims. """ from __future__ import annotations from typing import Any, Dict, List, Optional, Sequence, Tuple IDENTITY_ACCEPTED = "accepted" IDENTITY_AMBIGUOUS = "ambiguous" IDENTITY_UNKNOWN = "unknown" IDENTITY_REJECTED = "rejected" # ponytail: closed-set defaults. Recalibrate when held-out scores exist. DEFAULT_MIN_SCORE = 1.0 DEFAULT_MIN_MARGIN = 0.15 DEFAULT_MIN_CONFIDENCE = 0.55 def relative_confidence(top_score: float, runner_up_score: float) -> float: """Existing matcher formula: top / (top + 0.5 * runner).""" return round(min(1.0, top_score / max(1.0, (top_score + runner_up_score * 0.5))), 4) def decide_identity( scores: Dict[str, float], *, min_score: float = DEFAULT_MIN_SCORE, min_margin: float = DEFAULT_MIN_MARGIN, min_confidence: float = DEFAULT_MIN_CONFIDENCE, details: Optional[Dict[str, Any]] = None, method: str = "hybrid", ) -> Dict[str, Any]: """Rank catalog scores into accepted / ambiguous / unknown / rejected. Zero or missing evidence is rejected. A weak but unique top score is unknown. Two close scores are ambiguous. Only a unique, above-threshold score is accepted. """ if not scores: return { "status": IDENTITY_REJECTED, "method": method, "best_sku": None, "score": 0.0, "confidence": 0.0, "identity_level": IDENTITY_REJECTED, "ranked_candidates": [], "details": {}, "message": "No templates available", } sorted_candidates: List[Tuple[str, float]] = sorted( scores.items(), key=lambda x: x[1], reverse=True ) top_sku, top_score = sorted_candidates[0] runner_up_sku, runner_up_score = ( sorted_candidates[1] if len(sorted_candidates) > 1 else (None, 0.0) ) confidence = relative_confidence(top_score, runner_up_score) ranked = [{"sku_id": s, "score": round(sc, 2)} for s, sc in sorted_candidates[:5]] top_details = (details or {}).get(top_sku, {}) if top_score <= 0.0: level = IDENTITY_REJECTED message = "No matching evidence" best = None elif ( runner_up_sku is not None and top_score > 0.0 and (top_score - runner_up_score) < min_margin * max(top_score, 1.0) ): level = IDENTITY_AMBIGUOUS message = f"Close scores: {top_sku} vs {runner_up_sku}" best = None elif top_score < min_score or confidence < min_confidence: level = IDENTITY_UNKNOWN message = "Top score below accept threshold" best = None else: level = IDENTITY_ACCEPTED message = "Accepted unique catalog match" best = top_sku return { "status": "success" if level == IDENTITY_ACCEPTED else level, "method": method, "best_sku": best, "score": round(top_score, 2), "confidence": confidence, "identity_level": level, "ranked_candidates": ranked, "details": top_details, "message": message, } def identity_from_ranked( ranked_candidates: Sequence[Dict[str, Any]], **kwargs: Any, ) -> Dict[str, Any]: scores = {str(row["sku_id"]): float(row["score"]) for row in ranked_candidates} return decide_identity(scores, **kwargs)