251 lines
8.2 KiB
Python
251 lines
8.2 KiB
Python
"""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)
|