from pydantic import BaseModel, Field from typing import Dict, Any, Optional, Type, List, Set import pandas as pd from .enhanced_base_recognizer import EnhancedBaseRecognizer class NightTransactionInput(BaseModel): """夜间交易识别工具输入""" csv_path: Optional[str] = Field( None, description="CSV文件路径(可选)。如果初始化时已提供csv_path,可以不用再次传入。" ) class Config: arbitrary_types_allowed = True class NightTransactionRecognizer(EnhancedBaseRecognizer): """ 夜间交易(2-5点)异常识别器 基于图2规则定义: 1. 时间范围:特指凌晨2点至5点期间发生的交易 2. 时段性质:偏离日常资金往来常规时段,通常为非经营、非生活活动的常规休息时段 3. 异常判定: - 频繁出现资金收付记录 - 交易金额较大 - 缺乏合理交易背景 4. 合理性例外:特定行业经营需求、紧急资金周转等 核心逻辑:2-5点交易 + (高频或大额) + (无合理背景) = 异常夜间交易 """ args_schema: Type[BaseModel] = NightTransactionInput # 配置参数 night_start_hour: int = Field( 2, description="夜间检测开始小时(0-23),默认2点" ) night_end_hour: int = Field( 5, description="夜间检测结束小时(0-23),默认5点" ) frequency_threshold_per_hour: int = Field( 3, description="高频交易阈值,每小时超过此笔数视为高频" ) large_amount_threshold: float = Field( 50000.0, description="大额交易阈值(元),超过此金额的夜间交易视为大额异常" ) # 行业特征关键词(24小时营业或夜间经营行业) night_industry_keywords: List[str] = Field( [ "酒店", "宾馆", "KTV", "酒吧", "夜总会", "网吧", "便利店", "医院", "急救", "急诊", "消防", "公安", "保安", "物流", "运输", "出租车", "网约车", "外卖", "配送" ], description="夜间经营行业关键词,用于识别可能有合理背景的交易" ) # 紧急情况关键词 emergency_keywords: List[str] = Field( [ "急救", "急诊", "抢救", "紧急", "urgent", "emergency", "抢险", "救援", "救灾", "应急", "加急" ], description="紧急情况关键词,用于识别可能有合理背景的交易" ) def __init__(self, csv_path: str = None, config: Dict[str, Any] = None, **kwargs): """ 初始化夜间交易识别器 Args: csv_path: CSV文件路径 config: 配置参数 **kwargs: 其他参数 """ # 调用父类的 __init__ super().__init__( name="night_transaction_recognizer", description="识别银行流水中的夜间交易异常(2-5点),检测高频、大额等异常特征。", display_name="夜间交易异常识别", csv_path=csv_path, config=config, **kwargs ) # 从config获取配置,更新Field属性 night_config = self.get_config_value('night_transaction', {}) if night_config: config_mapping = { 'night_start_hour': 'night_start_hour', 'night_end_hour': 'night_end_hour', 'frequency_threshold_per_hour': 'frequency_threshold_per_hour', 'large_amount_threshold': 'large_amount_threshold', 'night_industry_keywords': 'night_industry_keywords', 'emergency_keywords': 'emergency_keywords' } for config_key, attr_name in config_mapping.items(): if config_key in night_config: setattr(self, attr_name, night_config[config_key]) # 验证时间配置 self._validate_time_config() print(f"✅ {self.display_name} 初始化完成") print(f" 夜间时段: {self.night_start_hour:02d}:00 - {self.night_end_hour:02d}:00") print(f" 高频阈值: {self.frequency_threshold_per_hour}笔/小时") print(f" 大额阈值: ¥{self.large_amount_threshold:,.2f}") print(f" 夜间行业关键词: {len(self.night_industry_keywords)}个") print(f" 紧急情况关键词: {len(self.emergency_keywords)}个") def _validate_time_config(self): """验证时间配置合理性""" if not (0 <= self.night_start_hour <= 23): raise ValueError(f"夜间开始小时必须在0-23之间: {self.night_start_hour}") if not (0 <= self.night_end_hour <= 23): raise ValueError(f"夜间结束小时必须在0-23之间: {self.night_end_hour}") # 确保开始时间早于结束时间 if self.night_start_hour >= self.night_end_hour: print(f"⚠️ 注意:夜间开始时间({self.night_start_hour}:00) >= 结束时间({self.night_end_hour}:00)," f"将按跨午夜处理") def _is_in_night_period(self, hour: int) -> bool: """ 判断小时数是否在夜间时段内 Args: hour: 小时数(0-23) Returns: bool: 是否在夜间时段 """ if self.night_start_hour < self.night_end_hour: # 正常情况:开始时间 < 结束时间 return self.night_start_hour <= hour < self.night_end_hour else: # 跨午夜情况:开始时间 >= 结束时间 return hour >= self.night_start_hour or hour < self.night_end_hour def _has_reasonable_background(self, row: pd.Series) -> bool: """ 判断交易是否有合理背景 根据图2规则,合理的背景包括: 1. 夜间特定行业经营需求 2. 紧急资金周转 Args: row: 交易记录 Returns: bool: 是否有合理背景 """ # 检查交易摘要中的关键词 summary = str(row.get('txSummary', '')).lower() counterparty = str(row.get('txCounterparty', '')).lower() # 合并检查文本 check_text = f"{summary} {counterparty}" # 1. 检查是否为夜间行业 for keyword in self.night_industry_keywords: if keyword.lower() in check_text: return True # 2. 检查是否为紧急情况 for keyword in self.emergency_keywords: if keyword.lower() in check_text: return True # 3. 可以扩展其他合理背景检查逻辑 return False def _detect_high_frequency_transactions(self, night_transactions: pd.DataFrame, date_col: str = 'date_only') -> Set[str]: """ 检测高频夜间交易 Args: night_transactions: 夜间交易数据 date_col: 日期列名 Returns: Set[str]: 高频交易ID集合 """ high_freq_ids = set() if len(night_transactions) == 0: return high_freq_ids # 按日期和小时分组统计 if 'hour' in night_transactions.columns and date_col in night_transactions.columns: # 统计每小时交易笔数 hourly_counts = night_transactions.groupby([date_col, 'hour']).size() for (trans_date, trans_hour), count in hourly_counts.items(): if count > self.frequency_threshold_per_hour: # 获取该小时的所有交易 mask = (night_transactions[date_col] == trans_date) & \ (night_transactions['hour'] == trans_hour) hour_transactions = night_transactions[mask] # 收集交易ID for tx_id in hour_transactions['txId'].unique(): high_freq_ids.add(str(tx_id)) print(f" ⚠️ {trans_date} {trans_hour:02d}:00-{trans_hour + 1:02d}:00: " f"{count}笔交易,超过阈值{self.frequency_threshold_per_hour}笔") return high_freq_ids def _detect_large_amount_transactions(self, night_transactions: pd.DataFrame) -> Set[str]: """ 检测大额夜间交易 Args: night_transactions: 夜间交易数据 Returns: Set[str]: 大额交易ID集合 """ large_amount_ids = set() if len(night_transactions) == 0: return large_amount_ids # 筛选大额交易 large_amount_mask = night_transactions['txAmount'].abs() >= self.large_amount_threshold large_amount_tx = night_transactions[large_amount_mask] for _, row in large_amount_tx.iterrows(): tx_id = str(row['txId']) large_amount_ids.add(tx_id) if len(large_amount_ids) > 0: print(f" ⚠️ 发现 {len(large_amount_ids)} 笔大额夜间交易(≥¥{self.large_amount_threshold:,.2f})") return large_amount_ids 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.night_start_hour:02d}:00 - {self.night_end_hour:02d}:00") 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['hour'] = df['datetime'].dt.hour df['minute'] = df['datetime'].dt.minute df['date_only'] = df['datetime'].dt.date # ============ 识别所有夜间交易 ============ night_mask = df['hour'].apply(self._is_in_night_period) night_transactions = df[night_mask].copy() if len(night_transactions) == 0: print(f"✅ 未发现夜间交易({self.night_start_hour:02d}:00-{self.night_end_hour:02d}:00)") return { 'recognition_type': self.display_name, 'identified_count': 0, 'identified_anomalies': [], 'recognition_status': '完成', 'recognition_parameters': { 'night_period': f"{self.night_start_hour:02d}:00-{self.night_end_hour:02d}:00", 'frequency_threshold': f"{self.frequency_threshold_per_hour}笔/小时", 'large_amount_threshold': self.large_amount_threshold, 'total_checked': len(df), 'night_transactions_found': 0 }, 'statistics': { 'total_transactions': len(df), 'night_transaction_count': 0, 'night_transaction_ratio': 0.0 } } print(f"📊 发现 {len(night_transactions)} 笔夜间交易") # ============ 合理性背景分析 ============ reasonable_transactions = [] for _, row in night_transactions.iterrows(): if self._has_reasonable_background(row): reasonable_transactions.append(row['txId']) if reasonable_transactions: print(f"📊 其中 {len(reasonable_transactions)} 笔交易可能有合理背景(夜间行业/紧急情况)") # ============ 异常特征检测 ============ identified_anomalies = [] # 1. 检测高频交易 high_freq_ids = self._detect_high_frequency_transactions(night_transactions) # 2. 检测大额交易 large_amount_ids = self._detect_large_amount_transactions(night_transactions) # 3. 合并异常交易ID(排除有合理背景的) abnormal_ids = (high_freq_ids | large_amount_ids) - set(reasonable_transactions) # 4. 生成异常记录 for tx_id in abnormal_ids: mask = night_transactions['txId'] == tx_id if mask.any(): row = night_transactions[mask].iloc[0] # 判断异常类型 is_high_freq = tx_id in high_freq_ids is_large_amount = tx_id in large_amount_ids # 生成异常原因 reasons = [] if is_high_freq: reasons.append("高频夜间交易") if is_large_amount: reasons.append(f"大额夜间交易(¥{row['txAmount']:,.2f}≥¥{self.large_amount_threshold:,.2f})") reason_str = ",".join(reasons) # 检查是否有合理背景但依然被标记为异常的原因 additional_info = { 'hour': int(row['hour']), 'is_high_frequency': is_high_freq, 'is_large_amount': is_large_amount, 'amount': float(row['txAmount']), 'has_reasonable_background': str(row['txId']) in reasonable_transactions } anomaly = self.format_anomaly_record( row=row, reason=f"夜间{reason_str},缺乏合理交易背景", severity='high' if is_large_amount else 'medium', check_type='night_transaction_abnormal', **additional_info ) identified_anomalies.append(anomaly) # ============ 结果统计和汇总 ============ print(f"✅ {self.display_name}检查完成") print(f" 检查交易总数: {len(df)}") print(f" 夜间交易数: {len(night_transactions)}") print(f" 有合理背景: {len(reasonable_transactions)}") print(f" 高频异常: {len(high_freq_ids)}") print(f" 大额异常: {len(large_amount_ids)}") print(f" 最终异常数: {len(identified_anomalies)}") # 显示夜间交易时间分布 if len(night_transactions) > 0: hour_distribution = night_transactions['hour'].value_counts().sort_index() print("📋 夜间交易时间分布:") for hour, count in hour_distribution.items(): print(f" {hour:02d}:00-{hour + 1:02d}:00: {count}笔") # 显示前5笔异常交易详情 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['txAmount']:,.2f} | {anomaly['recognition_reason']}") return { 'recognition_type': self.display_name, 'identified_count': len(identified_anomalies), 'identified_anomalies': identified_anomalies, 'recognition_status': '完成', 'recognition_parameters': { 'night_period': f"{self.night_start_hour:02d}:00-{self.night_end_hour:02d}:00", 'frequency_threshold_per_hour': self.frequency_threshold_per_hour, 'large_amount_threshold': self.large_amount_threshold, 'night_industry_keywords_count': len(self.night_industry_keywords), 'emergency_keywords_count': len(self.emergency_keywords), 'total_checked': len(df), 'night_transactions_found': len(night_transactions), 'reasonable_transactions': len(reasonable_transactions) }, 'statistics': { 'total_transactions': len(df), 'night_transaction_count': len(night_transactions), 'night_transaction_ratio': len(night_transactions) / max(1, len(df)), 'reasonable_transaction_count': len(reasonable_transactions), 'high_frequency_count': len(high_freq_ids), 'large_amount_count': len(large_amount_ids), 'abnormal_night_transaction_count': len(identified_anomalies), 'hour_distribution': { str(hour): int(count) for hour, count in night_transactions['hour'].value_counts().items() } if len(night_transactions) > 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({ 'night_start_hour': self.night_start_hour, 'night_end_hour': self.night_end_hour, 'frequency_threshold_per_hour': self.frequency_threshold_per_hour, 'large_amount_threshold': self.large_amount_threshold, 'night_industry_keywords_count': len(self.night_industry_keywords), 'emergency_keywords_count': len(self.emergency_keywords), 'data_loaded': self._data is not None }) return summary def get_config_summary(self) -> Dict[str, Any]: """获取配置摘要""" return { "夜间时段": f"{self.night_start_hour:02d}:00 - {self.night_end_hour:02d}:00", "高频阈值": f"{self.frequency_threshold_per_hour}笔/小时", "大额阈值": f"¥{self.large_amount_threshold:,.2f}", "夜间行业关键词": f"{len(self.night_industry_keywords)}个", "紧急情况关键词": f"{len(self.emergency_keywords)}个", "检测逻辑": "夜间交易 + (高频或大额) + (无合理背景) = 异常" }