| """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}" |
| ) |
|
|
