from __future__ import annotations import re import unicodedata from collections import Counter from dataclasses import dataclass, field from difflib import SequenceMatcher from datetime import date from decimal import Decimal from sqlite3 import Connection from typing import Any CLASSIFICATION_VERSION = "household_classification_v2.3" USER_DECISION_VERSION = "household_user_decision_v1" USER_CONFIRMED_UNMATCHED_TRANSFER = "user_confirmed_unmatched_transfer" # Canonical names are resolved against active database rows. Aliases are lookup # compatibility only; this module never creates categories. CATEGORY_ALIASES: dict[str, tuple[str, ...]] = { "Essen & Haushalt": ("Essen + Haushalt",), "Auto Melanie": ("Mobilität",), "Gesundheit": (), "Haustiere": (), "Kleider und Schuhe": (), "Krankenkasse": (), "Versicherungen Wohnen": (), "Versicherungen Leben": (), "Steuern": (), "Spenden": (), "Wohnen": (), "Ferien/Reisen": (), "Elektronische Medien": (), "Kinder/Familie": (), "Hobbys & Selfcare": (), "Liegenschaftsunterhalt": (), "Einmalige Anschaffungen": (), "Sonstige Einnahmen": (), "Rückerstattungen": (), } # Public, conservative merchant/description taxonomy. Ambiguous marketplaces, # general department stores, payment processors and unknown TWINT counterparties # intentionally have no default category. MERCHANT_FAMILY_RULES: tuple[tuple[str, str, str, tuple[str, ...]], ...] = ( ("food_supermarkets", "Lebensmittel und Supermärkte", "Essen & Haushalt", ( "migros", "coop", "denner", "aldi", "lidl", "volg", "spar", "avec", "k kiosk", "prodega", "aligro", "ottos food", "topcc", )), ("food_restaurants", "Restaurants und Verpflegung", "Essen & Haushalt", ( "restaurant", "restaur", "cafe", "café", "bistro", "baeckerei", "bäckerei", "takeaway", "mcdonald", "burger king", "starbucks", "subway", "just eat", "justeat", "ubereats", "uber eats", "too good to go", "lieferando", "pizzeria", "pizza", "sushi", "kantine", "confiserie", "felfel", "gasthaus", "ristorante", "gran bar", "yavuz market", "nespresso", "holy cow", "marché restaurant", "marche restaurant", "marche restaurants", "family food fun", "metzgerei", "migrolino", "schmidt s märkte", "schmidt s maerkte", "supermercato", "bk cala", "restaurants und verpflegung", )), ("auto_fuel_parking", "Tanken, Parkieren und Fahrzeug", "Auto Melanie", ( "migrol", "agrola", "tamoil", "socar", "avia", "shell", "bp service", "tankstelle", "parkhaus", "parking", "easypark", "parkingpay", "parkwart", "autobahnvignette", "garage", "reifen", "car wash", "autowasch", "fahrzeug", "strassenverkehrsamt", "parkplatz", "volvo car", "bp", "esso", "parkdepot", "mobility hub parkservice", "tank rastanlage", "voegtlin meyer", "tanken parkieren und fahrzeug", )), ("health_pharmacy", "Apotheken und medizinische Anbieter", "Gesundheit", ( "apotheke", "pharmacie", "amavita", "sun store", "medbase", "arztpraxis", "arzt", "zahnarzt", "spital", "klinik", "hospital", "physio", "drogerie", "zur rose", "medikament", "optiker", "zahnarztzentrum", "dr med dent", "müller drogeriemarkt", "mueller drogeriemarkt", "rotpunkt apoth", "analytica med", "medizin labor", "kantonsspital", "universitätsspital", "universitaetsspital", "stiftung gesundheit", "thermalquelle", )), ("pets", "Tierarzt und Tierbedarf", "Haustiere", ( "qualipet", "fressnapf", "tierarzt", "tierklinik", "veterin", "zoofach", "petshop", "kleintierpraxis", "koi breeder", "zoo hasel", )), ("clothes_shoes", "Kleider und Schuhe", "Kleider und Schuhe", ( "zalando", "h m", "h&m", "c a", "c&a", "c+a", "zara", "mango", "about you", "peek cloppenburg", "ochsner shoes", "dosenbach", "deichmann", "schuh", "shoe", "calzedonia", "intimissimi", "bonprix", "jack jones", "only", "vero moda", "esprit", "uniqlo", "snipes", "foot locker", "claire s", "essentialbag", "skechers", "van graaf", "venum", "olymp digital", "schiesser", )), ("health_insurance", "Krankenkassen", "Krankenkasse", ( "krankenkasse", "krankenversicherung", "oekk", "helsana", "swica", "sanitas", "concordia", "atupri", "assura", "kpt krankenkasse", "css versicherung", )), ("housing_insurance", "Wohnversicherungen", "Versicherungen Wohnen", ( "hausratversicherung", "gebaeudeversicherung", "gebäudeversicherung", "die mobiliar", "axa versicherung", "helvetia versicherung", "allianz versicherung", "vaudoise", )), ("life_insurance", "Lebensversicherungen", "Versicherungen Leben", ( "lebensversicherung", "risikoversicherung", "todesfallversicherung", )), ("taxes", "Steuern", "Steuern", ( "steueramt", "steuerverwaltung", "steuerbezug", "kantonssteuer", "gemeindesteuer", "direkte bundessteuer", )), ("donations", "Spenden", "Spenden", ( "spende", "caritas", "unicef", "rotes kreuz", "red cross", "heilsarmee", "wwf schweiz", )), ("housing", "Wohnen", "Wohnen", ( "mietzins", "miete wohnung", "hypothekarzins", "hypothek", "hausverwaltung", "nebenkosten wohnung", )), ("travel_transport", "Reisen und öffentlicher Verkehr", "Ferien/Reisen", ( "sbb", "cff", "ffs", "swiss international", "easyjet", "ryanair", "booking com", "airbnb", "hotel", "hostel", "camping", "flughafen", "airport", "bahn ticket", "öv ticket", "europa park", "apartamentos", "kurhotel", )), ("digital_media", "Digitale Medien", "Elektronische Medien", ( "netflix", "spotify", "disney plus", "youtube premium", "apple com bill", "google play", "audible", "kindle", "deezer", "sky show", "microsoft 365", "adobe", "dropbox", "openai", "disney streaming", "paramount", "prime video", "tradingview", "babbel", "ex libris", "solar manager", "ifolor", "bitdefender", "comobitdefender", "norton", "sms", "sunrise", "serafe", )), ("children_family", "Kinder und Familie", "Kinder/Familie", ( "kita", "kindergarten", "spielgruppe", "schulmaterial", "baby", "kids", "toys", "spielwaren", )), ("selfcare_hobbies", "Hobbys und Selfcare", "Hobbys & Selfcare", ( "coiffeur", "coiffeuse", "hair", "kosmetik", "beauty", "fitness", "gym", "wellness", "massage", "hobby", "sportzentrum", "freizeit", "puregym", "sportcenter", "sport freizeitcenter", "sportanlagen", "aquabasilea", "sole uno", "decathlon", "alpha progression", "cinema", "casino", "legionarspfad", "kinderzoo", "the hall", "ticketcorner", "feuerwerk", "alpamare", "klettereldorado", "pathé", "pathe", "technorama", "zoo basel", "zoo zürich", "zoo zurich", "seilbahn weissenstein", )), ("property_maintenance", "Liegenschaftsunterhalt", "Liegenschaftsunterhalt", ( "sanitaer", "sanitär", "elektriker", "heizung service", "gartenbau", "malerarbeiten", "schreiner", "liegenschaftsunterhalt", )), ) AMBIGUOUS_MERCHANT_TOKENS = ( "amazon", "galaxus", "manor", "warenhaus", "paypal", "stripe", "twint", ) AMBIGUOUS_REVIEW_FAMILY_RULES: tuple[tuple[str, str, tuple[str, ...]], ...] = ( ("online_marketplaces", "Allgemeine Online-Marktplätze", ( "amazon", "digitec", "galaxus", "ricardo", "aliexpress", "paypal", "stripe", )), ("general_home_retailers", "Mehrsortiment Wohnen und Haushalt", ("ikea", "jumbo")), ("unknown_twint_counterparties", "Nicht eindeutig erkennbare TWINT-Gegenparteien", ("twint",)), ("cash_withdrawals", "Bargeldbezüge", ("geldbezug", "cash withdrawal")), ) def ambiguous_review_family(value: Any) -> dict[str, str] | None: merchant = normalize_merchant(value) padded = f" {merchant} " for family_id, label, tokens in AMBIGUOUS_REVIEW_FAMILY_RULES: if any(f" {normalize_merchant(token)} " in padded for token in tokens): return {"family_id": family_id, "label": label} return None def _decimal(value: Any) -> Decimal: try: return Decimal(str(value if value not in (None, "") else "0")) except (ArithmeticError, ValueError): return Decimal("0") def _normalized_text(value: Any) -> str: text = unicodedata.normalize("NFKD", str(value or "").casefold().strip()) text = "".join(char for char in text if not unicodedata.combining(char)) text = re.sub(r"[^\w&+]+", " ", text, flags=re.UNICODE) return " ".join(text.split()) def extract_counterparty(value: Any, source_type: str | None = None) -> str: """Derive a stable counterparty while preserving the full row description.""" text = _normalized_text(value) original = text origin_marker = "ursprunglicher auftraggeber" incoming = text.startswith(("zahlungseingang ", "gutschrift ", "rueckgutschrift ", "uebertrag von ")) if origin_marker in text: before, after = text.split(origin_marker, 1) text = after if incoming else before for prefix in ( r"^(?:belastung\s+e\s*banking|belastung|zahlungseingang|gutschrift|rueckgutschrift|e\s*banking|kartenzahlung|einkauf)\s+", r"^uebertrag\s+(?:auf|von)\s+", ): text = re.sub(prefix, "", text) text = text.split(" mitteilung ", 1)[0] text = re.split(r"\b(?:lohnzahlung|salary|payroll)\b", text, maxsplit=1)[0] text = re.split(r"\b(?:chf|eur|usd)\s+[-+]?\d", text, maxsplit=1)[0] text = re.sub(r"^ref nr\s+\d+\s+", "", text) processor = next( ( token for token in ("paypal", "twint", "stripe", "sumup", "worldline") if text == token or text.startswith(token + " ") ), None, ) if processor and text != processor: candidate = text[len(processor):].strip() candidate = re.sub( r"^(?:payment|zahlung|transaction|transaktion|merchant)\s+", "", candidate ) if candidate and not re.fullmatch( r"(?:(?:terminal|term|tid|txn|trx|ref)\s*)?[a-z0-9-]{3,}", candidate ): text = candidate else: text = processor text = re.sub(r"\bref nr\s+\d+\b", " ", text) text = re.split(r"\b(?:mitteilung|debit mastercard nr)\b", text, maxsplit=1)[0] text = re.sub(r"\b\d{2}\s+\d{2}\s+\d{4}\b", " ", text) text = re.sub(r"\b(?:terminal|term|tid|transaktion|transaction|txn|beleg|receipt|ref)\s*[-:#]?\s*[a-z0-9-]{3,}\b", " ", text) text = re.sub(r"\b(?:terminal|term|tid|transaktion|transaction|txn|trx|beleg|receipt)\s*(?:nr|no|id)?\s*[a-z0-9-]{3,}\b", " ", text) text = re.sub(r"\b(?:filiale|branch|store)\s*(?:nr|no)?\s*\d{1,6}\b", " ", text) text = re.sub(r"\b\d{6,}\b", " ", text) text = " ".join(text.split()) if text: return text for processor_name in ("paypal", "twint", "stripe", "sumup", "worldline"): if processor_name in original: return processor_name return "" def normalize_merchant(value: Any) -> str: return extract_counterparty(value) def _rule_token_matches(merchant: str, token: str) -> bool: normalized = normalize_merchant(token) if not normalized: return False if f" {normalized} " in f" {merchant} ": return True return False def merchant_family(value: Any) -> dict[str, str] | None: merchant = normalize_merchant(value) for family_id, label, category_name, tokens in MERCHANT_FAMILY_RULES: for token in tokens: if _rule_token_matches(merchant, token): return {"family_id": family_id, "label": label, "category_name": category_name} return None def _account_role(source_type: str) -> str: if source_type == "visa_credit_card": return "card_liability" if source_type in {"akb_bank", "raiffeisen_bank"}: return "bank_cash" return source_type or "unknown" def _category( conn: Connection, category_id: str | None, category_type: str | None = None, cache: ClassificationLookupCache | None = None, ) -> tuple[str | None, str | None]: if not category_id: return None, None if cache is not None: row = next( ( item for item in cache.category_rows() if str(item["category_id"]) == category_id and (category_type is None or str(item["category_type"]) == category_type) ), None, ) else: row = conn.execute( """SELECT category_id,name FROM budget_categories WHERE category_id=? AND is_active=1 AND (? IS NULL OR category_type=?)""", (category_id, category_type, category_type), ).fetchone() return (str(row["category_id"]), str(row["name"])) if row else (None, None) def resolve_category( conn: Connection, canonical_name: str, category_type: str | None = None, cache: ClassificationLookupCache | None = None, ) -> tuple[str | None, str | None]: names = (canonical_name, *CATEGORY_ALIASES.get(canonical_name, ())) if cache is not None: rows = [ row for row in cache.category_rows() if category_type is None or str(row["category_type"]) == category_type ] else: rows = conn.execute( """SELECT category_id,name FROM budget_categories WHERE is_active=1 AND (? IS NULL OR category_type=?)""", (category_type, category_type), ).fetchall() by_name = {str(row["name"]).casefold(): row for row in rows} # Canonical name always wins when both canonical and legacy alias exist. for name in names: row = by_name.get(name.casefold()) if row: return str(row["category_id"]), str(row["name"]) return None, None def _base(*, state: str, semantics: str, message: str) -> dict[str, Any]: return { "classification_version": CLASSIFICATION_VERSION, "category_id": None, "category_name": None, "user_state": state, "transaction_semantics": semantics, "budget_effect_chf": "0.00" if semantics in { "transfer", "credit_card_payment", USER_CONFIRMED_UNMATCHED_TRANSFER, "unmatched_neutral_transfer", } else None, "user_message": message, "origin": "unresolved", "learned_rule": False, "evidence_count": 0, "conflict_count": 0, "create_counterbooking": False, "internal_reason": "unresolved_exception", "evidence_transaction_ids": [], "merchant_family_id": None, "merchant_family_label": None, "income_kind": None, "positive_features": [], "negative_guardrails": [], } def _merchant_key(value: Any) -> tuple[str, str | None, str | None]: normalized = normalize_merchant(value) family = merchant_family(normalized) return ( f"merchant:{normalized}", family["family_id"] if family else None, family["label"] if family else None, ) @dataclass class ClassificationLookupCache: """Request-local immutable lookup snapshots for a multi-row Preview.""" conn: Connection categories: list[Any] | None = None history_index: dict[tuple[str, str, str], list[dict[str, str]]] | None = None direct_merchants: dict[str, tuple[str, str]] | None = None aliases_by_source: dict[str, list[Any]] = field(default_factory=dict) review_rules_by_source: dict[str, list[Any]] = field(default_factory=dict) similarity_merchants: list[Any] | None = None recurring_payments: list[Any] | None = None def category_rows(self) -> list[Any]: if self.categories is None: self.categories = self.conn.execute( "SELECT category_id,name,category_type FROM budget_categories WHERE is_active=1 ORDER BY rowid" ).fetchall() return self.categories def histories(self) -> dict[tuple[str, str, str], list[dict[str, str]]]: if self.history_index is None: rows = self.conn.execute( """SELECT t.budget_transaction_id,t.category_id,t.transaction_type, c.source_type,c.merchant,c.description FROM budget_transactions t JOIN budget_transaction_candidates c ON c.transaction_candidate_id=t.source_candidate_id WHERE t.status='confirmed' AND t.category_id IS NOT NULL ORDER BY t.transaction_date,t.budget_transaction_id""" ).fetchall() index: dict[tuple[str, str, str], list[dict[str, str]]] = {} for item in rows: merchant_key, _, _ = _merchant_key(item["merchant"] or item["description"]) key = ( merchant_key, str(item["source_type"] or ""), str(item["transaction_type"] or ""), ) index.setdefault(key, []).append( { "transaction_id": str(item["budget_transaction_id"]), "category_id": str(item["category_id"]), } ) self.history_index = index return self.history_index def merchant_defaults(self) -> dict[str, tuple[str, str]]: if self.direct_merchants is None: rows = self.conn.execute( """SELECT merchant_id,normalized_name,default_category_id FROM budget_merchants WHERE is_active=1 AND default_category_id IS NOT NULL ORDER BY merchant_id""" ).fetchall() defaults: dict[str, tuple[str, str]] = {} for row in rows: defaults.setdefault( str(row["normalized_name"]), (str(row["merchant_id"]), str(row["default_category_id"])), ) self.direct_merchants = defaults self.similarity_merchants = rows return self.direct_merchants def aliases(self, source_type: str) -> list[Any]: if source_type not in self.aliases_by_source: self.aliases_by_source[source_type] = self.conn.execute( """SELECT a.alias_id,a.pattern,a.match_type,m.default_category_id FROM budget_merchant_aliases a JOIN budget_merchants m ON m.merchant_id=a.merchant_id WHERE a.is_active=1 AND m.is_active=1 AND m.default_category_id IS NOT NULL AND (a.source_type IS NULL OR a.source_type='all' OR a.source_type=?) ORDER BY a.priority DESC,a.alias_id""", (source_type,), ).fetchall() return self.aliases_by_source[source_type] def review_rules(self, source_type: str) -> list[Any]: if source_type not in self.review_rules_by_source: self.review_rules_by_source[source_type] = self.conn.execute( """SELECT rule_id,merchant_contains,category_id FROM budget_review_rules WHERE is_active=1 AND category_id IS NOT NULL AND (source_type IS NULL OR source_type='' OR source_type=?) ORDER BY priority DESC,rule_id""", (source_type,), ).fetchall() return self.review_rules_by_source[source_type] def recurring(self) -> list[Any]: if self.recurring_payments is None: self.recurring_payments = self.conn.execute( """SELECT recurring_id,name,merchant_name,category_id,account_id,currency, expected_amount_text,tolerance_percent,frequency, expected_day_of_month,expected_month,date_tolerance_days FROM budget_recurring_payments WHERE status='active'""" ).fetchall() return self.recurring_payments def _confirmed_history( conn: Connection, merchant_value: Any, source_type: str, semantics: str, cache: ClassificationLookupCache | None = None, ) -> list[dict[str, str]]: wanted_key, _, _ = _merchant_key(merchant_value) if cache is not None: return list(cache.histories().get((wanted_key, source_type, semantics), [])) rows = conn.execute( """SELECT t.budget_transaction_id,t.category_id,t.transaction_type, c.source_type,c.merchant,c.description FROM budget_transactions t JOIN budget_transaction_candidates c ON c.transaction_candidate_id=t.source_candidate_id WHERE t.status='confirmed' AND t.category_id IS NOT NULL ORDER BY t.transaction_date,t.budget_transaction_id""" ).fetchall() result: list[dict[str, str]] = [] for item in rows: historical_semantics = str(item["transaction_type"] or "") historical_key, _, _ = _merchant_key(item["merchant"] or item["description"]) if ( historical_key == wanted_key and historical_semantics == semantics and str(item["source_type"] or "") == source_type ): result.append({ "transaction_id": str(item["budget_transaction_id"]), "category_id": str(item["category_id"]), }) return result def _exact_rule( conn: Connection, merchant: str, source_type: str, semantics: str, cache: ClassificationLookupCache | None = None, ) -> tuple[str | None, str | None, str | None]: if cache is not None: cached_direct = cache.merchant_defaults().get(merchant) direct = ( {"merchant_id": cached_direct[0], "default_category_id": cached_direct[1]} if cached_direct else None ) else: direct = conn.execute( """SELECT merchant_id,default_category_id FROM budget_merchants WHERE is_active=1 AND normalized_name=? AND default_category_id IS NOT NULL ORDER BY merchant_id LIMIT 1""", (merchant,), ).fetchone() if direct: category_id, name = _category( conn, str(direct["default_category_id"]), semantics, cache ) if category_id: return category_id, name, str(direct["merchant_id"]) aliases = cache.aliases(source_type) if cache is not None else conn.execute( """SELECT a.alias_id,a.pattern,a.match_type,m.default_category_id FROM budget_merchant_aliases a JOIN budget_merchants m ON m.merchant_id=a.merchant_id WHERE a.is_active=1 AND m.is_active=1 AND m.default_category_id IS NOT NULL AND (a.source_type IS NULL OR a.source_type='all' OR a.source_type=?) ORDER BY a.priority DESC,a.alias_id""", (source_type,), ).fetchall() for alias in aliases: pattern = normalize_merchant(alias["pattern"]) matched = merchant == pattern if str(alias["match_type"]) == "exact" else bool(pattern and f" {pattern} " in f" {merchant} ") if matched: category_id, name = _category( conn, str(alias["default_category_id"]), semantics, cache ) if category_id: return category_id, name, str(alias["alias_id"]) return None, None, None def _source_rule( conn: Connection, merchant: str, source_type: str, semantics: str, cache: ClassificationLookupCache | None = None, ) -> tuple[str | None, str | None, str | None]: rows = cache.review_rules(source_type) if cache is not None else conn.execute( """SELECT rule_id,merchant_contains,category_id FROM budget_review_rules WHERE is_active=1 AND category_id IS NOT NULL AND (source_type IS NULL OR source_type='' OR source_type=? ) ORDER BY priority DESC,rule_id""", (source_type,), ).fetchall() for rule in rows: token = normalize_merchant(rule["merchant_contains"]) if token and f" {token} " in f" {merchant} ": category_id, name = _category(conn, str(rule["category_id"]), semantics, cache) if category_id: return category_id, name, str(rule["rule_id"]) return None, None, None def _similarity_rule( conn: Connection, merchant: str, semantics: str, cache: ClassificationLookupCache | None = None, ) -> tuple[str | None, str | None, str | None]: if not merchant or any(token in merchant for token in AMBIGUOUS_MERCHANT_TOKENS): return None, None, None candidates = [] if cache is not None: cache.merchant_defaults() source_rows = cache.similarity_merchants or [] else: source_rows = conn.execute( """SELECT merchant_id,normalized_name,default_category_id FROM budget_merchants WHERE is_active=1 AND default_category_id IS NOT NULL ORDER BY merchant_id""" ) for row in source_rows: target = normalize_merchant(row["normalized_name"]) score = SequenceMatcher(None, merchant, target).ratio() if score >= 0.92: candidates.append((score, str(row["merchant_id"]), str(row["default_category_id"]))) candidates.sort(reverse=True) if not candidates or (len(candidates) > 1 and candidates[0][0] - candidates[1][0] < 0.08): return None, None, None category_id, name = _category(conn, candidates[0][2], semantics, cache) return (category_id, name, candidates[0][1]) if category_id else (None, None, None) def _recurring_rule( conn: Connection, description: str, row: dict[str, Any], semantics: str, cache: ClassificationLookupCache | None = None, ) -> tuple[str | None, str | None, str | None]: if semantics != "expense" or ambiguous_review_family(description): return None, None, None mapping = row.get("mapping") or {} account_id = str(mapping.get("budget_account_id") or "") amount = abs(_decimal(row.get("signed_amount") if row.get("signed_amount") not in (None, "") else row.get("amount"))) rows = cache.recurring() if cache is not None else conn.execute( """SELECT recurring_id,name,merchant_name,category_id,account_id,currency, expected_amount_text,tolerance_percent,frequency, expected_day_of_month,expected_month,date_tolerance_days FROM budget_recurring_payments WHERE status='active'""" ).fetchall() row_currency = str(row.get("currency") or "").upper() for recurring in rows: merchant = normalize_merchant(recurring["merchant_name"] or recurring["name"]) if not merchant or merchant != description: continue if not recurring["account_id"] or not account_id or str(recurring["account_id"]) != account_id: continue if not recurring["currency"] or not row_currency or str(recurring["currency"]).upper() != row_currency: continue expected = abs(_decimal(recurring["expected_amount_text"])) tolerance = abs(_decimal(recurring["tolerance_percent"] or "10")) / Decimal("100") if expected <= 0 or abs(amount - expected) > expected * tolerance: continue frequency = str(recurring["frequency"] or "") expected_day = recurring["expected_day_of_month"] if frequency not in {"monthly", "quarterly", "yearly"} or not expected_day: continue try: actual_date = date.fromisoformat(str(row.get("transaction_date") or "")) delta = abs(actual_date.day - int(expected_day)) delta = min(delta, 31 - delta) if delta > int(recurring["date_tolerance_days"] or 5): continue if frequency in {"quarterly", "yearly"}: expected_month = int(recurring["expected_month"] or 0) if expected_month not in range(1, 13): continue if frequency == "yearly" and actual_date.month != expected_month: continue if frequency == "quarterly" and (actual_date.month - expected_month) % 3 != 0: continue except (TypeError, ValueError): continue category_id, category_name = _category( conn, str(recurring["category_id"] or ""), "expense", cache ) if category_id: return category_id, category_name, str(recurring["recurring_id"]) return None, None, None def classify_household_row( conn: Connection, row: dict[str, Any], cache: ClassificationLookupCache | None = None, ) -> dict[str, Any]: source_type = str(row.get("source_type") or "") raw_merchant = row.get("merchant") or row.get("description") description = extract_counterparty(raw_merchant, source_type) full_description = _normalized_text(row.get("description") or raw_merchant) classification = str(row.get("classification") or "") disposition = str(row.get("disposition") or "") amount = Decimal(str(row.get("signed_amount") or "0")) semantics = "expense" if amount < 0 else "income" # 1. Safe transfer/card/refund/reversal semantics always win. if disposition == "transfer_confirmed" or row.get("pairing_class") == "safe": result = _base(state="special_case", semantics="transfer", message="Interner Transfer erkannt.") result.update(origin="safe_transfer", internal_reason="safe_own_transfer") return result if classification in {"credit_card_payment", "credit_card_payment_counterpost"} or "ihre zahlung danke" in description: result = _base( state="special_case", semantics="credit_card_payment", message="Kreditkartenrückzahlung. Die Gegenbuchung ist in den vorhandenen Importen noch nicht gefunden.", ) result.update(origin="card_payment_semantics", internal_reason="credit_card_payment_neutral") return result if classification in {"credit_card_refund", "credit_card_reversal"} or amount > 0 and source_type == "visa_credit_card": result = _base(state="special_case", semantics="refund", message="Rückerstattung oder Storno erkannt.") result.update(origin="refund_or_reversal", internal_reason=classification or "positive_card_refund") return result # 2. Explicit row-scoped user decision. The import fingerprint binds its # version, exact private row identity and decision value. if row.get("user_decision") == USER_CONFIRMED_UNMATCHED_TRANSFER: eligible = ( source_type in {"akb_bank", "raiffeisen_bank"} and disposition not in {"pending", "superseded_pending", "receipt_detail"} and bool(row.get("mapping")) and str(row.get("currency") or "") == "CHF" and row.get("unmatched_transfer_candidate") is True ) if eligible: result = _base( state="special_case", semantics=USER_CONFIRMED_UNMATCHED_TRANSFER, message=( "Nutzerbestätigter interner Transfer. Die Gegenbuchung ist in den vorhandenen " "Importen noch nicht gefunden." ), ) result.update( origin="explicit_user_decision", internal_reason="user_confirmed_unmatched_transfer", decision_version=USER_DECISION_VERSION, ) return result # 3. Clear bank-income semantics are resolved before merchant rules. A # positive amount alone is never enough to call an own-account transfer income. if source_type in {"akb_bank", "raiffeisen_bank"} and amount > 0: narrow_transfer_marker = any(token in full_description for token in ( "uebertrag eigenes konto", "ubertrag eigenes konto", "transfer eigenes konto", "own account transfer", "internal transfer", "kontouebertrag", )) structured_account_transfer = ( any(token in full_description for token in ("uebertrag", "ubertrag", "transfer")) and any(token in full_description for token in ("konto", "account", "depot", "wallet")) ) possible_transfer = narrow_transfer_marker or structured_account_transfer if possible_transfer: result = _base( state="decision_needed", semantics="income", message="Möglicher interner Transfer. Eine Gegenbuchung ist noch nicht eindeutig belegt.", ) result.update( origin="possible_internal_transfer", income_kind="possible_internal_transfer", internal_reason="incoming_transfer_requires_pair_or_user_decision", positive_features=["positive_bank_amount", "own_account_marker"], negative_guardrails=["no_safe_counterbooking"], ) return result if any(token in full_description for token in ( "ruckerstattung", "rueckerstattung", "refund", "storno", "reversal", )): category_id, category_name = resolve_category( conn, "Rückerstattungen", "income", cache ) result = _base( state="proposal_ready" if category_id else "decision_needed", semantics="income", message=("Bestehende Kategorie für die Rückerstattung vorgeschlagen." if category_id else "Rückerstattung erkannt; die ursprüngliche Kategorie ist noch zu bestätigen."), ) result.update( category_id=category_id, category_name=category_name, origin="refund_income_pattern", income_kind="refund", internal_reason="bank_refund_pattern", positive_features=["positive_bank_amount", "refund_marker"], negative_guardrails=["not_card_settlement", "not_own_account_marker"], ) return result if re.search(r"\b(?:lohn|gehalt|salary|payroll|lohnzahlung|gehaltszahlung)\b", full_description): category_id, category_name = resolve_category(conn, "Lohn", "income", cache) result = _base( state="proposal_ready" if category_id else "decision_needed", semantics="income", message=("Bestehende Lohnkategorie vorgeschlagen." if category_id else "Lohneingang erkannt; die passende bestehende Lohnkategorie ist noch zu wählen."), ) result.update( category_id=category_id, category_name=category_name, origin="clear_income_pattern", income_kind="salary", internal_reason="salary_income_pattern", positive_features=["positive_bank_amount", "salary_marker"], negative_guardrails=["not_card_settlement", "not_own_account_marker"], ) return result # 4. Bank interest reuses an active canonical income category. if source_type in {"akb_bank", "raiffeisen_bank"} and amount > 0 and any( token in full_description for token in ("zins", "interest") ): category_rows = cache.category_rows() if cache is not None else conn.execute( "SELECT category_id,name,category_type FROM budget_categories " "WHERE is_active=1 ORDER BY rowid" ).fetchall() category = None for canonical_name in ("zinsertrag", "kapitalertrag", "sonstige einnahmen"): category = next(( item for item in category_rows if str(item["category_type"]) == "income" and str(item["name"]).casefold() == canonical_name ), None) if category: break if category: result = _base(state="proposal_ready", semantics="income", message="Bestehende Kategorie für den Zinsertrag vorgeschlagen.") result.update( category_id=str(category["category_id"]), category_name=str(category["name"]), origin="bank_interest_existing_category", internal_reason="bank_interest_income_category_reused", income_kind="interest", positive_features=["positive_bank_amount", "interest_marker"], negative_guardrails=["not_card_settlement", "not_own_account_marker"], ) return result if source_type in {"akb_bank", "raiffeisen_bank"} and amount > 0: result = _base( state="decision_needed", semantics="income", message="Zahlungseingang erkannt; Herkunft und passende Kategorie sind noch zu bestätigen.", ) result.update( origin="unclassified_payment_incoming", income_kind="payment_incoming", internal_reason="positive_bank_payment_without_safe_income_pattern", positive_features=["positive_bank_amount"], negative_guardrails=["no_salary_interest_refund_or_safe_transfer_marker"], ) return result # 5. Active exact merchant/alias rule. category_id, category_name, rule_id = _exact_rule( conn, description, source_type, semantics, cache ) if category_id: result = _base(state="proposal_ready", semantics=semantics, message="Kategorie aus Ihrer Händlerregel vorgeschlagen.") result.update(category_id=category_id, category_name=category_name, origin="exact_merchant_rule", internal_reason="exact_merchant_rule", rule_id=rule_id) return result # 5. Consistent confirmed history, including the same stable family across sources. history = _confirmed_history(conn, raw_merchant, source_type, semantics, cache) category_counts = Counter(item["category_id"] for item in history) if len(category_counts) == 1: selected = next(iter(category_counts)) category_id, category_name = _category(conn, selected, semantics, cache) if category_id: evidence = [item["transaction_id"] for item in history] result = _base(state="proposal_ready", semantics=semantics, message="Kategorie aus ähnlichen bestätigten Buchungen vorgeschlagen.") result.update(category_id=category_id, category_name=category_name, origin="learned_merchant_history" if len(history) >= 2 else "confirmed_merchant_history", learned_rule=len(history) >= 2, evidence_count=len(history), evidence_transaction_ids=evidence, internal_reason="consistent_confirmed_merchant_history") return result if len(category_counts) > 1: result = _base(state="decision_needed", semantics=semantics, message="Bisherige Buchungen wurden unterschiedlich kategorisiert.") result.update(origin="conflicting_history", evidence_count=len(history), conflict_count=len(category_counts), internal_reason="conflicting_confirmed_categories", evidence_transaction_ids=[item["transaction_id"] for item in history]) return result # 6. Stable public merchant families resolve only to active canonical categories. family = merchant_family(raw_merchant) if family and semantics == "expense": category_id, category_name = resolve_category( conn, family["category_name"], "expense", cache ) if category_id: result = _base(state="proposal_ready", semantics="expense", message=f"Kategorie für {family['label']} vorgeschlagen.") result.update(category_id=category_id, category_name=category_name, origin="stable_merchant_family", internal_reason="stable_merchant_family", merchant_family_id=family["family_id"], merchant_family_label=family["label"]) return result # 7. Source-specific active rules. category_id, category_name, rule_id = _source_rule( conn, description, source_type, semantics, cache ) if category_id: result = _base(state="proposal_ready", semantics=semantics, message="Kategorie aus einer passenden Quellregel vorgeschlagen.") result.update(category_id=category_id, category_name=category_name, origin="source_rule", internal_reason="source_specific_rule", rule_id=rule_id) return result # 8. Active recurring pattern with exact counterparty, account, amount and cadence guardrails. category_id, category_name, recurring_id = _recurring_rule( conn, description, row, semantics, cache ) if category_id: result = _base(state="proposal_ready", semantics=semantics, message="Kategorie aus einem aktiven wiederkehrenden Muster vorgeschlagen.") result.update( category_id=category_id, category_name=category_name, origin="active_recurring_pattern", internal_reason="exact_recurring_pattern", rule_id=recurring_id, positive_features=["exact_counterparty", "active_recurring_rule", "amount_within_tolerance"], negative_guardrails=["not_marketplace", "not_generic_payment_provider", "account_scope_checked"], ) return result # 9. Controlled high-threshold similarity to active merchant rules. category_id, category_name, rule_id = _similarity_rule( conn, description, semantics, cache ) if category_id: result = _base(state="proposal_ready", semantics=semantics, message="Kategorie aus einem sehr ähnlichen bestätigten Händler vorgeschlagen.") result.update(category_id=category_id, category_name=category_name, origin="controlled_similarity", internal_reason="controlled_high_similarity", rule_id=rule_id) return result return _base(state="decision_needed", semantics=semantics, message="Bitte eine Kategorie auswählen.")