from pydantic import BaseModel, Field from typing import Dict, Any, Optional, Type import pandas as pd from datetime import timedelta from .enhanced_base_recognizer import EnhancedBaseRecognizer class OccasionalHighIntegerTransactionInput(BaseModel): """偶发高额整数交易识别工具输入""" csv_path: Optional[str] = Field( None, description="CSV文件路径(可选)。如果初始化时已提供csv_path,可以不用再次传入。" ) class Config: arbitrary_types_allowed = True class OccasionalHighIntegerTransactionRecognizer(EnhancedBaseRecognizer): """ 偶发高额整数交易异常识别器 异常规则定义: 银行流水核查中,若存在金额为10,000元整数倍的交易,且该类交易金额超过整体流水平均交易金额的5倍 (构成极端异常值),同时此类极端异常交易呈现偶发且高频次发生的特征,可判定为金额维度存在 异常交易情形,需进一步核查交易真实性。 """ args_schema: Type[BaseModel] = OccasionalHighIntegerTransactionInput # 整数倍基数 integer_multiple: float = Field( 10000.0, description="整数倍基数(元),检查是否为该金额的整数倍" ) # 异常倍数阈值 outlier_multiplier: float = Field( 5.0, description="异常倍数阈值,交易金额超过整体平均交易金额的多少倍视为极端异常值" ) # 频率分析参数 frequency_window_days: int = Field( 7, description="频率分析时间窗口(天),用于判断是否为高频发生" ) min_occurrences_for_high_frequency: int = Field( 3, description="高频最小发生次数,在时间窗口内达到此次数视为高频" ) # 偶发性判断参数 gap_std_threshold: float = Field( 2.0, description="时间间隔标准差阈值,大于此值视为时间不规律(偶发)" ) # 严重程度配置 severity_level: str = Field( 'high', description="异常严重程度(high/medium/low)" ) def __init__(self, csv_path: str = None, config: Dict[str, Any] = None, **kwargs): """ 初始化偶发高额整数交易识别器 Args: csv_path: CSV文件路径 config: 配置参数 **kwargs: 其他参数 """ # 调用父类的 __init__ super().__init__( name="occasional_high_integer_recognizer", description="识别银行流水中偶发的高额整数倍交易异常,检测金额为整数倍、超过平均金额5倍且呈现偶发高频特征的交易。", display_name="偶发高额整数交易异常识别", csv_path=csv_path, config=config, **kwargs ) # 从config获取配置,更新Field属性 integer_config = self.get_config_value('occasional_integer_transaction', {}) if integer_config: config_mapping = { 'integer_multiple': 'integer_multiple', 'outlier_multiplier': 'outlier_multiplier', 'frequency_window_days': 'frequency_window_days', 'min_occurrences_for_high_frequency': 'min_occurrences_for_high_frequency', 'gap_std_threshold': 'gap_std_threshold', 'severity_level': 'severity_level' } for config_key, attr_name in config_mapping.items(): if config_key in integer_config: setattr(self, attr_name, integer_config[config_key]) print(f"✅ {self.display_name} 初始化完成") print(f" 整数倍基数: ¥{self.integer_multiple:,.0f}") print(f" 异常倍数阈值: {self.outlier_multiplier}倍") print(f" 频率分析窗口: {self.frequency_window_days}天") print(f" 高频最小次数: {self.min_occurrences_for_high_frequency}次") print(f" 偶发判断阈值: 间隔标准差>{self.gap_std_threshold}") print(f" 严重程度: {self.severity_level}") def _is_integer_multiple(self, amount: float, tolerance: float = 0.01) -> bool: """ 判断金额是否为整数倍 Args: amount: 交易金额 tolerance: 容差(元),解决浮点数精度问题 Returns: bool: 是否为整数倍 """ if pd.isna(amount): return False # 计算余数 remainder = abs(amount % self.integer_multiple) # 考虑浮点数精度,有两种情况: # 1. 余数接近于0(如10000 % 10000 = 0) # 2. 余数接近于整数倍基数(如10000 % 10000 = 0,但浮点误差可能为0.000001) return remainder < tolerance or abs(remainder - self.integer_multiple) < tolerance def _analyze_frequency_pattern(self, transactions: pd.DataFrame) -> Dict[str, Any]: """ 分析交易频率模式 Args: transactions: 交易数据 Returns: Dict[str, Any]: 频率分析结果 """ if len(transactions) < 2: return { 'is_occasional': False, 'is_high_frequency': False, 'total_count': len(transactions), 'gap_std': 0.0, 'time_analysis': '数据不足,无法分析频率模式' } # 确保按时间排序 sorted_transactions = transactions.sort_values('datetime') # 1. 计算时间间隔(天数) time_diffs = sorted_transactions['datetime'].diff().dt.total_seconds() / 86400.0 time_diffs = time_diffs.dropna() if len(time_diffs) == 0: return { 'is_occasional': False, 'is_high_frequency': False, 'total_count': len(transactions), 'gap_std': 0.0, 'time_analysis': '时间间隔数据不足' } # 2. 判断是否偶发(时间间隔不规律) gap_std = time_diffs.std() is_occasional = gap_std > self.gap_std_threshold # 3. 判断是否高频(在一定时间内多次发生) # 按天统计发生次数 date_counts = sorted_transactions['datetime'].dt.date.value_counts() # 检查是否有连续发生的情况 dates = sorted(sorted_transactions['datetime'].dt.date.unique()) date_diffs = [(dates[i + 1] - dates[i]).days for i in range(len(dates) - 1)] # 判断是否有在时间窗口内多次发生 is_high_frequency = False if len(transactions) >= self.min_occurrences_for_high_frequency: # 检查是否有在frequency_window_days内达到min_occurrences_for_high_frequency次 sliding_window_counts = [] for i in range(len(dates)): window_start = dates[i] window_end = window_start + timedelta(days=self.frequency_window_days) count_in_window = sum(1 for d in dates if window_start <= d <= window_end) sliding_window_counts.append(count_in_window) max_in_window = max(sliding_window_counts) if sliding_window_counts else 0 is_high_frequency = max_in_window >= self.min_occurrences_for_high_frequency return { 'is_occasional': is_occasional, 'is_high_frequency': is_high_frequency, 'total_count': len(transactions), 'gap_std': float(gap_std), 'gap_mean': float(time_diffs.mean()), 'date_counts': date_counts.to_dict(), 'unique_dates': len(date_counts), 'date_range': { 'start': dates[0].strftime('%Y-%m-%d') if dates else '', 'end': dates[-1].strftime('%Y-%m-%d') if dates else '', 'total_days': (dates[-1] - dates[0]).days + 1 if len(dates) > 1 else 1 }, 'time_analysis': f"时间间隔标准差: {gap_std:.2f}天,最大窗口内次数: {max_in_window if 'max_in_window' in locals() else 0}次" } def _generate_anomaly_reason(self, row: pd.Series, avg_amount: float, frequency_info: Dict[str, Any]) -> str: """ 生成异常原因描述 Args: row: 交易记录 avg_amount: 整体平均交易金额 frequency_info: 频率分析结果 Returns: str: 异常原因描述 """ reasons = [] # 整数倍特征 multiple = row['txAmount'] / self.integer_multiple reasons.append(f"金额为¥{self.integer_multiple:,.0f}的{multiple:.0f}倍整数") # 极端异常值特征 if avg_amount > 0: outlier_ratio = abs(row['txAmount']) / avg_amount reasons.append(f"金额超出整体平均{outlier_ratio:.1f}倍") # 频率特征 if frequency_info['is_occasional']: reasons.append(f"交易时间不规律(间隔标准差{frequency_info['gap_std']:.1f}天)") if frequency_info['is_high_frequency']: reasons.append(f"高频发生({frequency_info['total_count']}次)") return f"偶发高额整数交易异常: {','.join(reasons)},需核查交易真实性" 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.integer_multiple:,.0f}元) + 极端异常值({self.outlier_multiplier}倍) + 偶发高频 = 异常") # 检查必需字段 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') # 计算整体平均交易金额(绝对值) avg_amount = df['txAmount'].abs().mean() print(f"📊 整体平均交易金额: ¥{avg_amount:,.2f}") print(f" 极端异常值阈值: ¥{avg_amount * self.outlier_multiplier:,.2f}") # ============ 第一步:筛选整数倍交易 ============ integer_mask = df['txAmount'].apply(lambda x: self._is_integer_multiple(abs(x))) integer_transactions = df[integer_mask].copy() if len(integer_transactions) == 0: print(f"📊 未发现{self.integer_multiple:,.0f}元整数倍交易") return { 'recognition_type': self.display_name, 'identified_count': 0, 'identified_anomalies': [], 'recognition_status': '完成', 'statistics': { 'total_transactions': len(df), 'integer_transactions': 0, 'avg_amount': float(avg_amount), 'integer_multiple': self.integer_multiple } } print(f"📊 发现 {len(integer_transactions)} 笔{self.integer_multiple:,.0f}元整数倍交易") # ============ 第二步:筛选极端异常值 ============ outlier_threshold = avg_amount * self.outlier_multiplier outlier_mask = integer_transactions['txAmount'].abs() > outlier_threshold outlier_transactions = integer_transactions[outlier_mask].copy() if len(outlier_transactions) == 0: print(f"📊 未发现极端异常值整数倍交易(金额≤¥{outlier_threshold:,.2f})") return { 'recognition_type': self.display_name, 'identified_count': 0, 'identified_anomalies': [], 'recognition_status': '完成', 'statistics': { 'total_transactions': len(df), 'integer_transactions': len(integer_transactions), 'outlier_transactions': 0, 'avg_amount': float(avg_amount), 'outlier_threshold': float(outlier_threshold) } } print(f"📊 发现 {len(outlier_transactions)} 笔极端异常值整数倍交易(金额>¥{outlier_threshold:,.2f})") # ============ 第三步:分析频率模式 ============ frequency_info = self._analyze_frequency_pattern(outlier_transactions) print(f"📈 频率分析结果:") print(f" 总次数: {frequency_info['total_count']}") print(f" 是否偶发: {'是' if frequency_info['is_occasional'] else '否'} (标准差={frequency_info['gap_std']:.2f}天)") print(f" 是否高频: {'是' if frequency_info['is_high_frequency'] else '否'}") print(f" 涉及天数: {frequency_info['unique_dates']}天") # ============ 第四步:识别异常交易 ============ identified_anomalies = [] # 只有同时满足偶发且高频才标记为异常 if frequency_info['is_occasional'] and frequency_info['is_high_frequency']: print(f"⚠️ 检测到偶发且高频的高额整数交易,开始标记异常...") for idx, row in outlier_transactions.iterrows(): # 生成异常原因 reason = self._generate_anomaly_reason(row, avg_amount, frequency_info) # 额外信息 additional_info = { 'frequency_analysis': frequency_info, 'amount_analysis': { 'integer_multiple': self.integer_multiple, 'outlier_multiplier': self.outlier_multiplier, 'avg_amount': avg_amount, 'outlier_threshold': outlier_threshold, 'outlier_ratio': abs(row['txAmount']) / avg_amount if avg_amount > 0 else 0 }, 'integer_analysis': { 'multiple_count': row['txAmount'] / self.integer_multiple, 'is_integer_multiple': True } } # 生成异常记录 anomaly = self.format_anomaly_record( row=row, reason=reason, severity=self.severity_level, check_type='occasional_high_integer_transaction', **additional_info ) identified_anomalies.append(anomaly) print(f" ❌ 标记异常: ID={row['txId']}, ¥{row['txAmount']:,.2f}, {reason[:60]}...") else: print(f"✅ 未检测到偶发高频模式,不标记异常") print(f" 偶发性: {frequency_info['is_occasional']}") print(f" 高频性: {frequency_info['is_high_frequency']}") # ============ 第五步:结果统计 ============ print(f"✅ {self.display_name}检查完成") print(f" 检查交易总数: {len(df)}") print(f" 整数倍交易数: {len(integer_transactions)}") print(f" 极端异常值数: {len(outlier_transactions)}") print(f" 异常交易数: {len(identified_anomalies)}") # 显示整数倍交易统计 if len(integer_transactions) > 0: print("📋 整数倍交易统计:") total_integer_amount = integer_transactions['txAmount'].sum() avg_integer_amount = integer_transactions['txAmount'].mean() print(f" 总整数倍金额: ¥{total_integer_amount:,.2f}") print(f" 平均整数倍金额: ¥{avg_integer_amount:,.2f}") # 倍数分布 integer_transactions['multiple'] = (integer_transactions['txAmount'] / self.integer_multiple).round() multiple_counts = integer_transactions['multiple'].value_counts().sort_index() print(f" 倍数分布: {dict(multiple_counts.head(10))}" + (", ..." if len(multiple_counts) > 10 else "")) # 按方向统计 income_integer = integer_transactions[integer_transactions['txDirection'] == '收入'] expense_integer = integer_transactions[integer_transactions['txDirection'] == '支出'] print(f" 整数倍收入: {len(income_integer)}笔, ¥{income_integer['txAmount'].sum():,.2f}") print(f" 整数倍支出: {len(expense_integer)}笔, ¥{expense_integer['txAmount'].sum():,.2f}") return { 'recognition_type': self.display_name, 'identified_count': len(identified_anomalies), 'identified_anomalies': identified_anomalies, 'recognition_status': '完成', 'recognition_parameters': { 'integer_multiple': self.integer_multiple, 'outlier_multiplier': self.outlier_multiplier, 'frequency_window_days': self.frequency_window_days, 'min_occurrences_for_high_frequency': self.min_occurrences_for_high_frequency, 'gap_std_threshold': self.gap_std_threshold, 'severity_level': self.severity_level, 'avg_amount': float(avg_amount) }, 'statistics': { 'total_transactions': len(df), 'integer_transactions': len(integer_transactions), 'outlier_transactions': len(outlier_transactions), 'anomalous_transactions': len(identified_anomalies), 'frequency_analysis': frequency_info, 'amount_statistics': { 'overall_avg': float(avg_amount), 'overall_max': float(df['txAmount'].max()), 'overall_min': float(df['txAmount'].min()), 'integer_avg': float(integer_transactions['txAmount'].mean()) if len( integer_transactions) > 0 else 0, 'integer_total': float(integer_transactions['txAmount'].sum()) if len( integer_transactions) > 0 else 0, 'outlier_avg': float(outlier_transactions['txAmount'].mean()) if len( outlier_transactions) > 0 else 0 }, 'direction_distribution': { 'integer_income': len(integer_transactions[integer_transactions['txDirection'] == '收入']), 'integer_expense': len(integer_transactions[integer_transactions['txDirection'] == '支出']), 'outlier_income': len(outlier_transactions[outlier_transactions['txDirection'] == '收入']), 'outlier_expense': len(outlier_transactions[outlier_transactions['txDirection'] == '支出']) } } } 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({ 'integer_multiple': self.integer_multiple, 'outlier_multiplier': self.outlier_multiplier, 'frequency_window_days': self.frequency_window_days, 'min_occurrences_for_high_frequency': self.min_occurrences_for_high_frequency, 'gap_std_threshold': self.gap_std_threshold, 'severity_level': self.severity_level, 'data_loaded': self._data is not None }) return summary def get_config_summary(self) -> Dict[str, Any]: """获取配置摘要""" return { "整数倍基数": f"¥{self.integer_multiple:,.0f}元", "异常倍数阈值": f"{self.outlier_multiplier}倍", "频率分析窗口": f"{self.frequency_window_days}天", "高频最小次数": f"{self.min_occurrences_for_high_frequency}次", "偶发判断阈值": f"标准差>{self.gap_std_threshold}天", "检测逻辑": f"整数倍(¥{self.integer_multiple:,.0f}) + 极端异常值({self.outlier_multiplier}倍) + 偶发高频 = 异常", "严重程度": self.severity_level.upper(), "业务规则描述": "金额为整数倍、超过平均金额5倍且呈现偶发高频特征的交易需核查真实性" }