from pydantic import BaseModel, Field from typing import Dict, Any, Optional, Type import pandas as pd from .enhanced_base_recognizer import EnhancedBaseRecognizer class HighFrequencyInput(BaseModel): """高频交易识别工具输入""" csv_path: Optional[str] = Field( None, description="CSV文件路径(可选)。如果初始化时已提供csv_path,可以不用再次传入。" ) class Config: arbitrary_types_allowed = True class HighFrequencyTransactionRecognizer(EnhancedBaseRecognizer): """ 高频交易异常识别器 严格遵循业务规则定义: 银行流水高频交易(按小时维度)的定义为: 以"日期 + 小时"为统计单位,通过汇总该时间间隔内的交易笔数与交易金额, 识别出单小时交易笔数超过 10 笔的特定时段组合, 此类在短时间内(1 小时)集中发生、交易频次密集的资金收付行为, 即为银行流水高频交易,其核心特征是交易笔数在单位小时内达到预设阈值(10 笔), 体现出资金往来的集中性与活跃度异常。 """ args_schema: Type[BaseModel] = HighFrequencyInput # 配置参数 - 严格按照业务规则 frequency_threshold: int = Field( 10, description="高频交易阈值,每小时超过此笔数视为高频交易(业务规则要求10笔)" ) def __init__(self, csv_path: str = None, config: Dict[str, Any] = None, **kwargs): """ 初始化高频交易识别器 Args: csv_path: CSV文件路径 config: 配置参数 **kwargs: 其他参数 """ # 调用父类的 __init__ super().__init__( name="high_frequency_recognizer", description="识别银行流水中的高频交易异常,严格按照业务规则:单小时交易笔数超过10笔即为高频异常。", display_name="高频交易异常识别", csv_path=csv_path, config=config, **kwargs ) # 从config获取配置 high_freq_config = self.get_config_value('high_frequency', {}) if high_freq_config and 'frequency_threshold' in high_freq_config: self.frequency_threshold = high_freq_config['frequency_threshold'] print(f"✅ {self.display_name} 初始化完成") print(f" 严格遵循业务规则:单小时交易笔数 > {self.frequency_threshold}笔 = 高频异常") def _calculate_hourly_statistics(self, df: pd.DataFrame) -> pd.DataFrame: """ 计算每小时的交易统计数据 Args: df: 标准化后的交易数据 Returns: pd.DataFrame: 每小时统计结果 """ # 提取日期和小时 df['date_hour'] = df['datetime'].dt.strftime('%Y-%m-%d %H') df['date'] = df['datetime'].dt.date df['hour'] = df['datetime'].dt.hour # 按照业务规则:以"日期 + 小时"为统计单位 hour_stats = [] # 按日期+小时分组 for (date_val, hour_val), group in df.groupby(['date', 'hour']): transaction_count = len(group) amount_sum = group['txAmount'].sum() # 统计交易方向分布 direction_counts = group['txDirection'].value_counts().to_dict() hour_stats.append({ 'date': date_val, 'hour': hour_val, 'transaction_count': transaction_count, 'amount_sum': amount_sum, 'amount_avg': amount_sum / transaction_count if transaction_count > 0 else 0, 'in_count': direction_counts.get('收入', 0), 'out_count': direction_counts.get('支出', 0) }) return pd.DataFrame(hour_stats) def _identify_high_frequency_periods(self, hour_stats: pd.DataFrame) -> pd.DataFrame: """ 识别高频交易时段 Args: hour_stats: 每小时统计数据 Returns: pd.DataFrame: 标记了高频时段的结果 """ # 严格按照业务规则:单小时交易笔数超过10笔 hour_stats['is_high_frequency'] = hour_stats['transaction_count'] > self.frequency_threshold return hour_stats def _get_transactions_in_period(self, df: pd.DataFrame, date_val: pd.Timestamp.date, hour_val: int) -> pd.DataFrame: """ 获取指定时段内的所有交易 Args: df: 原始交易数据 date_val: 日期 hour_val: 小时 Returns: pd.DataFrame: 指定时段内的交易 """ return df[ (df['datetime'].dt.date == date_val) & (df['datetime'].dt.hour == hour_val) ].copy() 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" 业务规则: 以'日期 + 小时'为统计单位") print(f" 高频阈值: 单小时交易笔数 > {self.frequency_threshold}笔") # 检查必需字段 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') # 检查是否有无效的时间数据 invalid_times = df['datetime'].isna().sum() if invalid_times > 0: print(f"⚠️ 警告: 有 {invalid_times} 条记录的时间解析失败,将跳过这些记录") df = df[df['datetime'].notna()] if len(df) == 0: return { 'recognition_type': self.display_name, 'identified_count': 0, 'identified_anomalies': [], 'recognition_status': '完成', 'statistics': {'total_valid_transactions': 0} } # ============ 按照业务规则:以"日期 + 小时"为统计单位 ============ hour_stats = self._calculate_hourly_statistics(df) if len(hour_stats) == 0: return { 'recognition_type': self.display_name, 'identified_count': 0, 'identified_anomalies': [], 'recognition_status': '完成', 'statistics': { 'total_transactions': len(df), 'hour_periods': 0, 'high_frequency_periods': 0 } } print(f"📊 按'日期+小时'分组统计完成,共 {len(hour_stats)} 个时段") # ============ 识别高频时段 ============ hour_stats = self._identify_high_frequency_periods(hour_stats) # 高频时段统计 high_freq_periods = hour_stats[hour_stats['is_high_frequency']] print(f"📊 发现 {len(high_freq_periods)} 个高频时段(> {self.frequency_threshold}笔/小时)") # ============ 生成异常记录 ============ identified_anomalies = [] # 为每个高频时段创建异常记录 for _, period_row in high_freq_periods.iterrows(): date_val = period_row['date'] hour_val = period_row['hour'] # 获取该时段内的所有交易 period_transactions = self._get_transactions_in_period(df, date_val, hour_val) # 为每笔交易创建异常记录 for _, tx_row in period_transactions.iterrows(): # 生成异常原因 - 严格按照业务规则描述 reason = f"属于高频交易时段:{date_val} {hour_val:02d}:00-{hour_val + 1:02d}:00,该时段共{period_row['transaction_count']}笔交易,超过阈值{self.frequency_threshold}笔,体现资金往来的集中性与活跃度异常" # 额外信息 additional_info = { 'period_info': { 'date': date_val.strftime('%Y-%m-%d'), 'hour': int(hour_val), 'start_time': f"{hour_val:02d}:00", 'end_time': f"{hour_val + 1:02d}:00", 'transaction_count': int(period_row['transaction_count']), 'amount_sum': float(period_row['amount_sum']), 'amount_avg': float(period_row['amount_avg']), 'in_count': int(period_row['in_count']), 'out_count': int(period_row['out_count']) }, 'business_rule': { 'statistic_unit': "日期 + 小时", 'threshold': self.frequency_threshold, 'description': "单小时交易笔数超过阈值,体现资金往来的集中性与活跃度异常" } } anomaly = self.format_anomaly_record( row=tx_row, reason=reason, severity='high', # 高频交易通常视为高风险 check_type='high_frequency_transaction', **additional_info ) identified_anomalies.append(anomaly) # ============ 结果统计 ============ print(f"✅ {self.display_name}检查完成") print(f" 检查交易总数: {len(df)}") print(f" 统计时段总数: {len(hour_stats)}") print(f" 高频时段数: {len(high_freq_periods)}") print(f" 异常交易笔数: {len(identified_anomalies)}") # 显示高频时段详情 if len(high_freq_periods) > 0: print("📋 高频时段详情(按交易笔数排序):") sorted_periods = high_freq_periods.sort_values('transaction_count', ascending=False) for i, (_, row) in enumerate(sorted_periods.iterrows(), 1): time_range = f"{row['hour']:02d}:00-{row['hour'] + 1:02d}:00" print(f" {i}. {row['date']} {time_range}: {row['transaction_count']}笔交易") print(f" 收入: {row['in_count']}笔,支出: {row['out_count']}笔") print(f" 总金额: ¥{row['amount_sum']:,.2f},笔均: ¥{row['amount_avg']:,.2f}") # 显示整体统计 if len(hour_stats) > 0: max_transactions = hour_stats['transaction_count'].max() avg_transactions = hour_stats['transaction_count'].mean() print(f"📊 整体统计: 最高{max_transactions}笔/小时,平均{avg_transactions:.1f}笔/小时") return { 'recognition_type': self.display_name, 'identified_count': len(identified_anomalies), 'identified_anomalies': identified_anomalies, 'recognition_status': '完成', 'recognition_parameters': { 'frequency_threshold': self.frequency_threshold, 'statistic_unit': "日期 + 小时", 'business_rule': "单小时交易笔数超过阈值即视为高频交易异常" }, 'statistics': { 'total_transactions': len(df), 'total_periods': len(hour_stats), 'high_frequency_periods': len(high_freq_periods), 'period_statistics': { 'max_transactions_per_hour': int(hour_stats['transaction_count'].max()), 'min_transactions_per_hour': int(hour_stats['transaction_count'].min()), 'avg_transactions_per_hour': float(hour_stats['transaction_count'].mean()), 'max_amount_per_hour': float(hour_stats['amount_sum'].max()), 'min_amount_per_hour': float(hour_stats['amount_sum'].min()), 'avg_amount_per_hour': float(hour_stats['amount_sum'].mean()) }, 'high_frequency_details': [ { 'date': row['date'].strftime('%Y-%m-%d'), 'hour': int(row['hour']), 'transaction_count': int(row['transaction_count']), 'amount_sum': float(row['amount_sum']), 'in_count': int(row['in_count']), 'out_count': int(row['out_count']) } for _, row in high_freq_periods.iterrows() ] if len(high_freq_periods) > 0 else [] } } 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({ 'frequency_threshold': self.frequency_threshold, 'business_rule': '单小时交易笔数超过阈值即视为高频交易异常', 'data_loaded': self._data is not None }) return summary def get_config_summary(self) -> Dict[str, Any]: """获取配置摘要""" return { "高频交易阈值": f"{self.frequency_threshold}笔/小时", "统计单位": "日期 + 小时", "检测逻辑": "单小时交易笔数 > 阈值 = 高频交易异常", "业务规则描述": "识别短时间内集中发生、交易频次密集的资金收付行为" }