from pydantic import BaseModel, Field from typing import Dict, Any, Optional, Type, List import pandas as pd from .enhanced_base_recognizer import EnhancedBaseRecognizer class LargeAmountInput(BaseModel): """大额交易识别工具输入""" csv_path: Optional[str] = Field( None, description="CSV文件路径(可选)。如果初始化时已提供csv_path,可以不用再次传入。" ) class Config: arbitrary_types_allowed = True class LargeAmountTransactionRecognizer(EnhancedBaseRecognizer): """ 大额交易异常识别器 业务规则定义: 若交易对手方个人银行账户出现单次交易金额超过预设阈值(如 5 万元、20 万元等)的大额资金往来, 且该交易与账户日常交易规模、资金使用场景及个人经济活动特征不匹配,缺乏合理交易背景支撑, 可触发大额交易异常提示,需进一步核查该笔交易的真实性、合法性及资金来源与去向。 """ args_schema: Type[BaseModel] = LargeAmountInput # 配置参数 amount_threshold: float = Field( 50000.0, description="大额交易阈值(元),单次交易金额超过此值视为大额交易" ) # 历史分析参数 history_days: int = Field( 90, description="历史分析天数,用于分析账户日常交易规模" ) outlier_multiplier: float = Field( 3.0, description="异常倍数阈值,交易金额超过历史均值的多少倍视为异常" ) # 背景分析参数 enable_background_check: bool = Field( True, description="是否启用交易背景检查" ) # 合理背景关键词(常见的大额合理交易场景) reasonable_background_keywords: List[str] = Field( [ "工资", "奖金", "绩效", "年终奖", "报销", "货款", "租金", "购房款", "装修款", "学费", "医疗费", "保险", "理财", "投资款", "分红", "还款", "借款", "赠与", "遗产" ], description="合理的交易背景关键词,用于识别可能有合理背景的大额交易" ) # 高风险关键词(可能需要关注的场景) high_risk_keywords: List[str] = Field( [ "赌博", "赌资", "彩票", "博彩", "虚拟货币", "比特币", "地下钱庄", "洗钱", "套现", "非法", "不明", "无摘要" ], description="高风险关键词,出现这些词的大额交易需要重点关注" ) def __init__(self, csv_path: str = None, config: Dict[str, Any] = None, **kwargs): """ 初始化大额交易识别器 Args: csv_path: CSV文件路径 config: 配置参数 **kwargs: 其他参数 """ # 调用父类的 __init__ super().__init__( name="large_amount_recognizer", description="识别银行流水中的大额交易异常,检测单次交易金额超过阈值且与账户历史行为不匹配的交易。", display_name="大额交易异常识别", csv_path=csv_path, config=config, **kwargs ) # 从config获取配置,更新Field属性 large_amount_config = self.get_config_value('large_amount_recognition', {}) if large_amount_config: config_mapping = { 'amount_threshold': 'amount_threshold', 'history_days': 'history_days', 'outlier_multiplier': 'outlier_multiplier', 'enable_background_check': 'enable_background_check', 'reasonable_background_keywords': 'reasonable_background_keywords', 'high_risk_keywords': 'high_risk_keywords' } for config_key, attr_name in config_mapping.items(): if config_key in large_amount_config: setattr(self, attr_name, large_amount_config[config_key]) print(f"✅ {self.display_name} 初始化完成") print(f" 金额阈值: ¥{self.amount_threshold:,.2f}") print(f" 历史分析天数: {self.history_days}天") print(f" 异常倍数阈值: {self.outlier_multiplier}倍") print(f" 背景检查: {'启用' if self.enable_background_check else '禁用'}") print(f" 合理背景关键词: {len(self.reasonable_background_keywords)}个") print(f" 高风险关键词: {len(self.high_risk_keywords)}个") def _analyze_account_history(self, df: pd.DataFrame, current_date: pd.Timestamp) -> Dict[str, Any]: """ 分析账户历史交易特征 Args: df: 交易数据 current_date: 当前交易日期 Returns: Dict[str, Any]: 账户历史交易特征 """ # 计算历史日期范围 history_start = current_date - pd.Timedelta(days=self.history_days) # 筛选历史交易(当前日期之前的历史数据) history_df = df[df['datetime'] < current_date] history_df = history_df[history_df['datetime'] >= history_start] if len(history_df) == 0: return { 'has_history': False, 'message': f'无最近{self.history_days}天的历史交易数据' } # 计算历史交易特征 history_features = { 'has_history': True, 'history_days': self.history_days, 'total_transactions': len(history_df), 'avg_amount': float(history_df['txAmount'].mean()) if len(history_df) > 0 else 0, 'max_amount': float(history_df['txAmount'].max()) if len(history_df) > 0 else 0, 'min_amount': float(history_df['txAmount'].min()) if len(history_df) > 0 else 0, 'std_amount': float(history_df['txAmount'].std()) if len(history_df) > 0 else 0, 'total_income': float(history_df[history_df['txDirection'] == '收入']['txAmount'].sum()), 'total_expense': float(history_df[history_df['txDirection'] == '支出']['txAmount'].sum()), 'income_count': len(history_df[history_df['txDirection'] == '收入']), 'expense_count': len(history_df[history_df['txDirection'] == '支出']), 'date_range': { 'start': history_df['datetime'].min().strftime('%Y-%m-%d'), 'end': history_df['datetime'].max().strftime('%Y-%m-%d') } } return history_features def _check_transaction_background(self, row: pd.Series) -> Dict[str, Any]: """ 检查交易背景合理性 Args: row: 交易记录 Returns: Dict[str, Any]: 背景检查结果 """ background_result = { 'has_reasonable_background': False, 'has_high_risk_indicator': False, 'reasonable_keywords_found': [], 'high_risk_keywords_found': [], 'summary': '', 'counterparty': '', 'summary_text': '' } if not self.enable_background_check: return background_result # 获取交易摘要和对手方信息 summary = str(row.get('txSummary', '')).lower() counterparty = str(row.get('txCounterparty', '')).lower() # 检查合理背景关键词 reasonable_found = [] for keyword in self.reasonable_background_keywords: if keyword in summary or keyword in counterparty: reasonable_found.append(keyword) # 检查高风险关键词 high_risk_found = [] for keyword in self.high_risk_keywords: if keyword in summary or keyword in counterparty: high_risk_found.append(keyword) # 判断是否有合理背景 has_reasonable_background = len(reasonable_found) > 0 has_high_risk = len(high_risk_found) > 0 # 生成背景描述 background_desc = [] if reasonable_found: background_desc.append(f"合理背景: {', '.join(reasonable_found)}") if high_risk_found: background_desc.append(f"高风险关键词: {', '.join(high_risk_found)}") background_result.update({ 'has_reasonable_background': has_reasonable_background, 'has_high_risk_indicator': has_high_risk, 'reasonable_keywords_found': reasonable_found, 'high_risk_keywords_found': high_risk_found, 'summary': '; '.join(background_desc) if background_desc else '无特殊背景信息', 'counterparty': counterparty, 'summary_text': summary }) return background_result def _is_amount_outlier(self, amount: float, history_features: Dict[str, Any]) -> bool: """ 判断交易金额是否为异常值(与历史行为不匹配) Args: amount: 当前交易金额 history_features: 账户历史特征 Returns: bool: 是否为异常值 """ if not history_features['has_history']: # 无历史数据,无法判断是否为异常值 return False if history_features['avg_amount'] == 0: # 历史平均金额为0,无法判断 return False # 判断是否超过历史平均值的异常倍数 is_outlier = amount > (history_features['avg_amount'] * self.outlier_multiplier) return is_outlier def recognize(self, csv_path: str = None, **kwargs) -> Dict[str, Any]: """ 识别大额交易异常 Args: csv_path: CSV文件路径 **kwargs: 其他参数 Returns: Dict[str, Any]: 识别结果 """ try: # 使用父类的load_data方法加载标准化数据 df = self.load_data(csv_path) print(f"🔍 {self.display_name}开始检查,共 {len(df)} 条记录") print(f" 大额阈值: ¥{self.amount_threshold:,.2f}") print(f" 检测规则: 大额金额 + 与历史不匹配 + 缺乏合理背景 = 大额交易异常") # 检查必需字段 required_fields = ['txId', 'datetime', 'txAmount', 'txDirection'] missing_fields = [field for field in required_fields if field not in df.columns] if missing_fields: return { 'recognition_type': self.display_name, 'identified_count': 0, 'identified_anomalies': [], 'recognition_status': '失败', 'error': f'缺少必需字段: {missing_fields}' } # 确保datetime列已正确解析 if not pd.api.types.is_datetime64_any_dtype(df['datetime']): df['datetime'] = pd.to_datetime(df['datetime'], errors='coerce') # 按时间排序,便于历史分析 df = df.sort_values('datetime') # 复制一份用于分析(避免修改原始数据) analysis_df = df.copy() # ============ 识别大额交易 ============ # 根据业务规则:单次交易金额超过预设阈值 large_amount_mask = analysis_df['txAmount'].abs() >= self.amount_threshold large_amount_transactions = analysis_df[large_amount_mask].copy() if len(large_amount_transactions) == 0: print(f"📊 未发现大额交易(≥¥{self.amount_threshold:,.2f})") return { 'recognition_type': self.display_name, 'identified_count': 0, 'identified_anomalies': [], 'recognition_status': '完成', 'recognition_parameters': { 'amount_threshold': self.amount_threshold, 'history_days': self.history_days, 'outlier_multiplier': self.outlier_multiplier, 'enable_background_check': self.enable_background_check, 'total_checked': len(df) }, 'statistics': { 'total_transactions': len(df), 'large_amount_transactions': 0, 'max_amount': float(df['txAmount'].max()), 'min_amount': float(df['txAmount'].min()), 'avg_amount': float(df['txAmount'].mean()) } } print(f"📊 发现 {len(large_amount_transactions)} 笔大额交易(≥¥{self.amount_threshold:,.2f})") # ============ 分析每笔大额交易 ============ identified_anomalies = [] analyzed_transactions = [] for idx, row in large_amount_transactions.iterrows(): tx_id = str(row['txId']) tx_date = row['datetime'] tx_amount = float(row['txAmount']) print(f" 🔍 分析交易 {tx_id}: ¥{tx_amount:,.2f} ({row['txDirection']})") # 1. 分析账户历史特征 history_features = self._analyze_account_history(analysis_df, tx_date) # 2. 检查交易背景 background_check = self._check_transaction_background(row) # 3. 判断是否为异常值(与历史不匹配) is_amount_outlier = self._is_amount_outlier(abs(tx_amount), history_features) # 4. 综合判断是否为异常 # 规则:大额 + (历史不匹配 或 缺乏合理背景) = 异常 is_abnormal = True # 默认大额就是异常 # 如果有合理背景,可以降低异常级别 severity_level = 'high' if background_check['has_reasonable_background']: if not is_amount_outlier: # 有合理背景且金额不异常,可能不是异常 is_abnormal = False print(f" ✅ 有合理背景且金额不异常,跳过") continue else: severity_level = 'medium' print(f" ⚠️ 有合理背景但金额异常") # 如果有高风险关键词,提高异常级别 if background_check['has_high_risk_indicator']: severity_level = 'high' print(f" ⚠️ 发现高风险关键词") # 记录分析结果 transaction_analysis = { 'tx_id': tx_id, 'date': tx_date.strftime('%Y-%m-%d'), 'time': tx_date.strftime('%H:%M:%S'), 'amount': tx_amount, 'direction': row['txDirection'], 'is_large_amount': True, 'is_amount_outlier': is_amount_outlier, 'history_features': history_features, 'background_check': background_check, 'is_abnormal': is_abnormal, 'severity_level': severity_level } analyzed_transactions.append(transaction_analysis) # 如果判断为异常,生成异常记录 if is_abnormal: # 生成异常原因 reasons = [] reasons.append(f"大额交易(¥{tx_amount:,.2f}≥¥{self.amount_threshold:,.2f})") if is_amount_outlier: if history_features['has_history']: avg_amount = history_features['avg_amount'] outlier_ratio = tx_amount / avg_amount if avg_amount > 0 else float('inf') reasons.append(f"金额异常(超出历史均值{outlier_ratio:.1f}倍)") if not background_check['has_reasonable_background']: reasons.append("缺乏合理交易背景") if background_check['has_high_risk_indicator']: reasons.append("存在高风险关键词") reason_str = ",".join(reasons) # 额外信息 additional_info = { 'amount_analysis': { 'threshold': self.amount_threshold, 'is_outlier': is_amount_outlier, 'outlier_ratio': float(tx_amount / history_features['avg_amount']) if history_features[ 'has_history'] and history_features[ 'avg_amount'] > 0 else None, 'history_avg': history_features['avg_amount'] if history_features['has_history'] else None }, 'background_analysis': background_check, 'history_analysis': history_features } anomaly = self.format_anomaly_record( row=row, reason=f"大额交易异常: {reason_str},需核查真实性、合法性及资金来源去向", severity=severity_level, check_type='large_amount_transaction', **additional_info ) identified_anomalies.append(anomaly) print(f" ❌ 标记为异常: {reason_str}") else: print(f" ✅ 未标记为异常") # ============ 结果统计 ============ print(f"✅ {self.display_name}检查完成") print(f" 检查交易总数: {len(df)}") print(f" 大额交易数: {len(large_amount_transactions)}") print(f" 异常交易数: {len(identified_anomalies)}") print(f" 通过检查数: {len(large_amount_transactions) - len(identified_anomalies)}") # 显示大额交易统计 if len(large_amount_transactions) > 0: print("📋 大额交易统计:") total_large_amount = large_amount_transactions['txAmount'].sum() avg_large_amount = large_amount_transactions['txAmount'].mean() print(f" 总大额金额: ¥{total_large_amount:,.2f}") print(f" 平均大额金额: ¥{avg_large_amount:,.2f}") # 按方向统计 income_large = large_amount_transactions[large_amount_transactions['txDirection'] == '收入'] expense_large = large_amount_transactions[large_amount_transactions['txDirection'] == '支出'] print(f" 大额收入: {len(income_large)}笔, ¥{income_large['txAmount'].sum():,.2f}") print(f" 大额支出: {len(expense_large)}笔, ¥{expense_large['txAmount'].sum():,.2f}") # 显示异常交易示例 if len(identified_anomalies) > 0: print("📋 大额异常交易示例:") for i, anomaly in enumerate(identified_anomalies[:5], 1): time_str = f"{anomaly.get('txDate', '')} {anomaly.get('txTime', '')}" print( f" {i}. ID:{anomaly['txId']} | {time_str} | {anomaly['txDirection']} ¥{anomaly['txAmount']:,.2f} | {anomaly['recognition_reason'][:50]}...") return { 'recognition_type': self.display_name, 'identified_count': len(identified_anomalies), 'identified_anomalies': identified_anomalies, 'recognition_status': '完成', 'recognition_parameters': { 'amount_threshold': self.amount_threshold, 'history_days': self.history_days, 'outlier_multiplier': self.outlier_multiplier, 'enable_background_check': self.enable_background_check, 'total_checked': len(df), 'large_transactions_found': len(large_amount_transactions) }, 'statistics': { 'total_transactions': len(df), 'large_amount_transactions': len(large_amount_transactions), 'abnormal_large_transactions': len(identified_anomalies), 'amount_statistics': { 'max_amount': float(df['txAmount'].max()), 'min_amount': float(df['txAmount'].min()), 'avg_amount': float(df['txAmount'].mean()), 'total_amount': float(df['txAmount'].sum()), 'large_amount_total': float(large_amount_transactions['txAmount'].sum()), 'large_amount_avg': float(large_amount_transactions['txAmount'].mean()) if len( large_amount_transactions) > 0 else 0 }, 'direction_distribution': { 'income_count': len(df[df['txDirection'] == '收入']), 'expense_count': len(df[df['txDirection'] == '支出']), 'large_income_count': len( large_amount_transactions[large_amount_transactions['txDirection'] == '收入']), 'large_expense_count': len( large_amount_transactions[large_amount_transactions['txDirection'] == '支出']) }, 'background_analysis': { 'reasonable_background_count': sum( 1 for t in analyzed_transactions if t['background_check']['has_reasonable_background']), 'high_risk_count': sum( 1 for t in analyzed_transactions if t['background_check']['has_high_risk_indicator']), 'outlier_count': sum(1 for t in analyzed_transactions if t['is_amount_outlier']) } } } except FileNotFoundError as e: return { 'recognition_type': self.display_name, 'identified_count': 0, 'identified_anomalies': [], 'recognition_status': '失败', 'error': f'文件不存在: {str(e)}' } except Exception as e: import traceback traceback.print_exc() return { 'recognition_type': self.display_name, 'identified_count': 0, 'identified_anomalies': [], 'recognition_status': '失败', 'error': f'数据加载或处理失败: {str(e)}' } def get_summary(self) -> Dict[str, Any]: """获取识别器摘要""" summary = super().get_summary() summary.update({ 'amount_threshold': self.amount_threshold, 'history_days': self.history_days, 'outlier_multiplier': self.outlier_multiplier, 'enable_background_check': self.enable_background_check, 'reasonable_background_keywords_count': len(self.reasonable_background_keywords), 'high_risk_keywords_count': len(self.high_risk_keywords), 'data_loaded': self._data is not None }) return summary def get_config_summary(self) -> Dict[str, Any]: """获取配置摘要""" return { "大额阈值": f"¥{self.amount_threshold:,.2f}", "历史分析天数": f"{self.history_days}天", "异常倍数阈值": f"{self.outlier_multiplier}倍", "背景检查": "启用" if self.enable_background_check else "禁用", "检测逻辑": "大额金额 + 与历史不匹配 + 缺乏合理背景 = 大额交易异常", "业务规则描述": "单次交易金额超过阈值且与账户历史行为不匹配,缺乏合理背景" }