"""Transit (self-transfer) detection rules. Identifies transactions that are internal transfers between the victim's own accounts (e.g. bank -> Alipay -> WeChat) and should NOT be counted as fraud loss. """ from datetime import datetime from app.models.transaction import TransactionRecord SELF_KEYWORDS = ["本人", "自己", "余额", "充值", "提现", "银行卡转入", "银行卡充值"] FEE_TRANSIT_WINDOW_SECONDS = 120 FEE_TOLERANCE_RATIO = 0.02 def is_self_transfer(tx: TransactionRecord, known_self_accounts: list[str]) -> bool: """Check if a transaction is an inter-account transfer by the victim.""" counterparty = (tx.counterparty_name or "").lower() remark = (tx.remark or "").lower() # Rule 1: counterparty matches known self accounts for acct in known_self_accounts: if acct and acct.lower() in counterparty: return True # Rule 2: counterparty contains self-transfer keywords for kw in SELF_KEYWORDS: if kw in counterparty or kw in remark: return True # Rule 3: counterparty references another payment app owned by victim app_keywords = ["支付宝", "微信", "银行卡", "数字钱包"] victim_patterns = [f"{app}-" for app in app_keywords] + app_keywords for pat in victim_patterns: if pat in counterparty: if tx.direction.value == "out": return True return False def is_fee_tolerant_transit_pair(tx_a: TransactionRecord, tx_b: TransactionRecord) -> bool: """Two-way transfer pattern: opposite direction, close time, similar amount.""" if tx_a.direction.value == tx_b.direction.value: return False amount_a = float(tx_a.amount or 0) amount_b = float(tx_b.amount or 0) if amount_a <= 0 or amount_b <= 0: return False time_a = tx_a.trade_time time_b = tx_b.trade_time if not isinstance(time_a, datetime) or not isinstance(time_b, datetime): return False if abs((time_a - time_b).total_seconds()) > FEE_TRANSIT_WINDOW_SECONDS: return False amount_base = max(amount_a, amount_b) if amount_base <= 0: return False diff_ratio = abs(amount_a - amount_b) / amount_base return diff_ratio <= FEE_TOLERANCE_RATIO