"""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) # Cosine scale (gallery). Same-product pairs were measured above 0.6 and # different products below 0.5. These are not a new calibration. COSINE_ACCEPT = 0.60 COSINE_AMBIGUOUS = 0.50 COSINE_MARGIN = 0.04 VOTE_CAP = 3 def _finite(value: Any) -> Optional[float]: try: number = float(value) except (TypeError, ValueError): return None if number != number or number in (float("inf"), float("-inf")): return None return number def aggregate_votes( observations: Sequence[Dict[str, float]], *, cap: int = VOTE_CAP, ) -> Dict[str, float]: """Mean of the newest observations, at most `cap`. Older frames are dropped, not averaged in. An empty observation still counts, so a later miss can outvote an earlier weak hit. Inlier and cosine scores must not be mixed in one list. """ window = list(observations)[-max(1, int(cap)):] if not window: return {} totals: Dict[str, float] = {} for scores in window: for sku_id, score in (scores or {}).items(): number = _finite(score) if number is None: continue totals[str(sku_id)] = totals.get(str(sku_id), 0.0) + number count = len(window) return {sku_id: total / count for sku_id, total in totals.items()} def _cosine_decision( sorted_candidates: List[Tuple[str, float]], *, method: str, details: Optional[Dict[str, Any]], accept: float, ambiguous: float, margin_min: float, ) -> Dict[str, Any]: top_sku, top_score = sorted_candidates[0] runner_score = sorted_candidates[1][1] if len(sorted_candidates) > 1 else 0.0 margin = top_score - runner_score if top_score >= accept and margin >= margin_min: level = IDENTITY_ACCEPTED elif top_score >= ambiguous: level = IDENTITY_AMBIGUOUS else: level = IDENTITY_REJECTED return { "status": level, "method": method, "best_sku": top_sku if level == IDENTITY_ACCEPTED else None, "score": float(top_score), "confidence": float(top_score), "identity_level": level, "ranked_candidates": [ {"sku_id": sku, "score": float(value)} for sku, value in sorted_candidates[:3] ], "details": details or {}, "message": None, } def decision_for_match( match: Any, *, method: str = "gallery", margin: float = COSINE_MARGIN, ) -> Dict[str, Any]: """Rank one gallery match. Rejection is the gallery's, not a second threshold. `match()` already applied the raw 0.55 line or the head's 0.40 line. Re-ranking those scores against COSINE_ACCEPT (0.60) would refuse a head match the gallery accepted. The only extra gate here is the published cosine margin. """ similarity = _finite(getattr(match, "similarity", 0.0)) or 0.0 if match is None or getattr(match, "rejected", False) or not getattr(match, "sku_id", None): return { "status": IDENTITY_REJECTED, "method": method, "best_sku": None, "score": similarity, "confidence": similarity, "identity_level": IDENTITY_REJECTED, "ranked_candidates": [], "details": {}, "message": "no enrolled product is close enough", } runner = getattr(match, "runner_up", None) runner_score = _finite(getattr(match, "runner_up_similarity", 0.0)) or 0.0 gap = similarity - runner_score close = bool(runner) and gap < margin level = IDENTITY_AMBIGUOUS if close else IDENTITY_ACCEPTED ranked = [{"sku_id": str(match.sku_id), "score": similarity}] if runner: ranked.append({"sku_id": str(runner), "score": runner_score}) return { "status": level, "method": method, "best_sku": None if close else str(match.sku_id), "score": similarity, "confidence": similarity, "identity_level": level, "ranked_candidates": ranked, "details": {}, "message": None, } 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", scale: str = "inlier", ) -> 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 scale == "cosine": if min_score == DEFAULT_MIN_SCORE: min_score = COSINE_ACCEPT if min_margin == DEFAULT_MIN_MARGIN: min_margin = COSINE_MARGIN if min_confidence == DEFAULT_MIN_CONFIDENCE: min_confidence = COSINE_AMBIGUOUS cleaned: Dict[str, float] = {} for sku_id, score in (scores or {}).items(): number = _finite(score) if number is None: continue cleaned[str(sku_id)] = number if not cleaned: 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( cleaned.items(), key=lambda item: item[1], reverse=True ) if scale == "cosine": return _cosine_decision( sorted_candidates, method=method, details=details, accept=min_score, ambiguous=min_confidence, margin_min=min_margin, ) 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)