"""Metrics — quality signals without golden dataset. Checks that work immediately: - Arithmetic: sum == price * qty (free signal of LLM/data errors) - JSON fix rate: how often _safe_json_parse has to repair LLM output """ from decimal import Decimal, InvalidOperation def check_arithmetic(ops: list) -> list[dict]: """Check sum == price * qty for ADD/UPDATE operations. Returns list of mismatches: [{action, name, price, qty, expected_sum, actual_sum, diff}] """ mismatches = [] for op in ops: action = op.get("action", "") if action not in ("ADD", "UPDATE"): continue row = op.get("new_row") or op.get("new_values") or {} price = _to_decimal(row.get("price")) qty = _to_decimal(row.get("qty")) actual_sum = _to_decimal(row.get("sum")) if price is None or qty is None or actual_sum is None: continue # can't check without all three expected = price * qty if expected != actual_sum: mismatches.append({ "action": action, "name": row.get("name", "")[:100], "price": float(price), "qty": float(qty), "expected_sum": float(expected), "actual_sum": float(actual_sum), "diff": float(actual_sum - expected), }) return mismatches class ClassifyMetrics: """Track _safe_json_parse fix rate per batch.""" def __init__(self): self.total = 0 self.fixes = 0 # how many times JSON needed repair def record(self, needed_fix: bool): self.total += 1 if needed_fix: self.fixes += 1 @property def fix_rate(self) -> float: return self.fixes / self.total if self.total else 0.0 def summary(self) -> dict: return { "total_classifications": self.total, "json_fixes": self.fixes, "json_fix_rate": round(self.fix_rate, 3), } def _to_decimal(val) -> Decimal | None: """Safe conversion to Decimal.""" if val is None or val == "": return None try: return Decimal(str(val)) except (InvalidOperation, ValueError): return None