File size: 3,064 Bytes
19cd60d | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 | """Conservative company-name labels for transaction model training."""
from __future__ import annotations
import re
from typing import Optional
from pipeline.merchant_classifier import (
CURATED_TRANSACTION_MARKERS,
HANDLE_CATEGORY_MAP,
MERCHANT_ALIASES,
classify_upi_merchant,
extract_upi_handle,
get_merchant,
)
_PERSONAL_CATEGORIES = {
"personal_transfer",
"friends",
"family",
"staff_salary",
"rental",
"transfer",
"cash_withdrawal",
}
_GENERIC_NAMES = {
"",
"unknown upi counterparty",
"upi transfer",
"payment",
"transfer",
"unknown",
}
def _name_key(value: object) -> str:
return " ".join(re.sub(r"[^a-z0-9]+", " ", str(value or "").lower()).split())
_CANONICAL_COMPANIES = {
_name_key(company[0]): company[0]
for company in (
*MERCHANT_ALIASES.values(),
*CURATED_TRANSACTION_MARKERS.values(),
*HANDLE_CATEGORY_MAP.values(),
)
}
_CANONICAL_ALIASES = {
"indian cle": "Indian Clearing Corporation",
"iccl zerodha credit": "Indian Clearing Corporation",
"iccl zerod": "Indian Clearing Corporation",
"zerodha br": "Zerodha",
"zerodha deposit": "Zerodha",
}
def _clean_company_name(value: object) -> Optional[str]:
name = " ".join(str(value or "").split()).strip(" -/|")[:100]
if name.lower() in _GENERIC_NAMES:
return None
return name or None
def _canonical_company_name(value: object) -> Optional[str]:
cleaned = _clean_company_name(value)
if not cleaned:
return None
key = _name_key(cleaned)
if key.startswith("cred ") or key == "cred":
return "CRED"
return _CANONICAL_ALIASES.get(key) or _CANONICAL_COMPANIES.get(key)
def infer_company_name(
description: str,
*,
category: str = "",
explicit_name: object = None,
) -> Optional[str]:
"""Return a company only when merchant evidence is strong enough to label."""
if category in _PERSONAL_CATEGORIES:
return None
explicit = _canonical_company_name(explicit_name)
if explicit:
return explicit
handle = extract_upi_handle(description or "")
if handle:
try:
merchant = get_merchant(handle)
except Exception:
merchant = None
if (
merchant
and merchant.get("category") not in _PERSONAL_CATEGORIES | {"unclassified", "upi_spend"}
and float(merchant.get("confidence", 0.0)) >= 0.70
):
company = _canonical_company_name(merchant.get("display_name"))
if company:
return company
try:
evidence_description = "" if handle else (description or "")
inferred = classify_upi_merchant(handle or "", evidence_description, learn=False)
except Exception:
return None
if (
inferred.get("category") in _PERSONAL_CATEGORIES | {"unclassified"}
or float(inferred.get("confidence", 0.0)) < 0.70
):
return None
return _canonical_company_name(inferred.get("display_name"))
|