"""Decode token IDs back to human-readable transaction descriptions. Handles the reserved-token offset (0=MASK, 1=OOV, 2=NULL, 3+=values) and maps each feature's raw value to readable text. """ from __future__ import annotations from dataclasses import dataclass import numpy as np from src.data.generator import AMOUNT_RANGE_LABELS from src.data.schema import SchemaConfig, VALUES_START from src.demo.merchant_catalog import DemoMerchantCatalog, MCC_NAMES DOW_NAMES: list[str] = ["Mon", "Tue", "Wed", "Thu", "Fri", "Sat", "Sun"] ENTRY_MODE_DISPLAY: dict[str, str] = { "card_present": "Card Present", "card_not_present": "Online", "contactless": "Contactless", "chip": "Chip", "manual_key": "Manual", } COUNTRY_NAMES: list[str] = [ "US", "UK", "CA", "DE", "FR", "JP", "AU", "BR", "IN", "MX", "IT", "ES", "NL", "CH", "SE", "NO", "DK", "FI", "KR", "SG", "HK", "TW", "NZ", "IE", "BE", "AT", "PT", "PL", "CZ", "GR", "IL", "AE", "SA", "TH", "MY", "PH", "ID", "VN", "CL", "CO", "AR", "ZA", "NG", "EG", "KE", "TR", "RU", "UA", "RO", "HU", ] AVS_DISPLAY: dict[str, str] = { "full_match": "AVS Match", "zip_match": "ZIP Match", "address_match": "Addr Match", "no_match": "AVS No Match", "not_checked": "AVS N/A", } CVV_DISPLAY: dict[str, str] = { "match": "CVV Match", "no_match": "CVV No Match", "not_provided": "No CVV", } @dataclass class DecodedTransaction: """A single transaction with all features decoded to display strings.""" index: int hour: str dow: str days_since_last: str is_recurring: str mcc: str merchant_name: str merchant_category: str customer_merchant_count: str entry_mode: str amount_range: str card_product: str country: str avs: str cvv: str device_hash: str customer_tenure: str class TransactionDecoder: """Converts raw token_ids (T, F) into human-readable transactions.""" def __init__(self, schema: SchemaConfig, merchant_catalog: DemoMerchantCatalog) -> None: self.schema = schema self.merchants = merchant_catalog self._feature_names = schema.feature_names() def _decode_value(self, feature_idx: int, token_id: int) -> str: """Decode a single token_id for a given feature index.""" feat = self.schema.features[feature_idx] if token_id == 0: return "[MASK]" if token_id == 1: return "[OOV]" if token_id == 2: return "[NULL]" value = token_id - VALUES_START if feat.name == "hour": if 0 <= value <= 23: h = value % 12 or 12 ampm = "AM" if value < 12 else "PM" return f"{h} {ampm}" return f"H{value}" if feat.name == "dow": return DOW_NAMES[value] if 0 <= value < 7 else f"D{value}" if feat.name == "days_since_last": if value == 0: return "Same day" if value <= 5: return f"{value}d ago" bucket_size = 365 / 30 approx = int(value * bucket_size) return f"~{approx}d ago" if feat.name == "is_recurring": return "Recurring" if value == 1 else "One-time" if feat.name == "mcc": return MCC_NAMES.get(value, f"MCC-{value}") if feat.name == "merchant_id": info = self.merchants.get(value) return info.name if feat.name == "customer_merchant_count": if value == 0: return "1st visit" if value < 5: return f"{value + 1} visits" bucket_size = 500 / 20 approx = int(value * bucket_size) return f"~{approx} visits" if feat.name == "entry_mode": if feat.values and value in feat.values: raw = feat.values[value] return ENTRY_MODE_DISPLAY.get(raw, raw) return f"Entry-{value}" if feat.name == "amount": range_idx = value // 16 return AMOUNT_RANGE_LABELS.get(range_idx, f"${value}") if feat.name == "card_product": if feat.values and value in feat.values: raw = feat.values[value] return raw.replace("_", " ").title() return f"Card-{value}" if feat.name == "country": return COUNTRY_NAMES[value] if 0 <= value < len(COUNTRY_NAMES) else f"Country-{value}" if feat.name == "avs": if feat.values and value in feat.values: raw = feat.values[value] return AVS_DISPLAY.get(raw, raw) return f"AVS-{value}" if feat.name == "cvv": if feat.values and value in feat.values: raw = feat.values[value] return CVV_DISPLAY.get(raw, raw) return f"CVV-{value}" if feat.name == "device_hash": return f"Device-{value:04d}" if feat.name == "customer_tenure": months = value * 12 if months < 12: return f"<1 year" return f"~{months // 12}yr" return str(value) def decode_sequence(self, token_ids: np.ndarray) -> list[DecodedTransaction]: """Decode a full (T, F) sequence into readable transactions. Returns transactions in reverse chronological order (most recent first). """ num_tx = token_ids.shape[0] txns: list[DecodedTransaction] = [] for t in range(num_tx - 1, -1, -1): row = token_ids[t] merchant_val = int(row[5]) - VALUES_START merchant_info = self.merchants.get(max(0, merchant_val)) txn = DecodedTransaction( index=t, hour=self._decode_value(0, int(row[0])), dow=self._decode_value(1, int(row[1])), days_since_last=self._decode_value(2, int(row[2])), is_recurring=self._decode_value(3, int(row[3])), mcc=self._decode_value(4, int(row[4])), merchant_name=self._decode_value(5, int(row[5])), merchant_category=merchant_info.category, customer_merchant_count=self._decode_value(6, int(row[6])), entry_mode=self._decode_value(7, int(row[7])), amount_range=self._decode_value(8, int(row[8])), card_product=self._decode_value(9, int(row[9])), country=self._decode_value(10, int(row[10])), avs=self._decode_value(11, int(row[11])), cvv=self._decode_value(12, int(row[12])), device_hash=self._decode_value(13, int(row[13])), customer_tenure=self._decode_value(14, int(row[14])), ) txns.append(txn) return txns def summarize_customer( self, token_ids: np.ndarray, is_fraud: bool, ) -> str: """One-line behavioral summary derived from actual token data.""" num_tx = token_ids.shape[0] merchant_ids = set() mcc_counts: dict[str, int] = {} for t in range(num_tx): mid = int(token_ids[t, 5]) - VALUES_START merchant_ids.add(mid) mcc_val = int(token_ids[t, 4]) - VALUES_START cat = MCC_NAMES.get(mcc_val, f"Cat-{mcc_val}") mcc_counts[cat] = mcc_counts.get(cat, 0) + 1 top_cats = sorted(mcc_counts.items(), key=lambda x: -x[1])[:3] cat_str = ", ".join(c[0] for c in top_cats) label = "FRAUD" if is_fraud else "Legitimate" return ( f"{num_tx} transactions | {len(merchant_ids)} unique merchants | " f"Top categories: {cat_str} | Label: {label}" )