from langchain.tools import BaseTool from abc import abstractmethod from typing import Dict, Any import pandas as pd from datetime import datetime from pydantic import Field, PrivateAttr class EnhancedBaseRecognizer(BaseTool): """增强版异常识别器基类 - 提供统一的数据处理和异常记录格式""" name: str = Field(..., description="识别器名称") description: str = Field(..., description="识别器描述") display_name: str = Field("", description="显示名称") # 使用 PrivateAttr 定义不需要验证的私有属性 _recognized_count: int = PrivateAttr(0) _csv_path: str = PrivateAttr(None) # 标准化后的csv文件路径 _data: pd.DataFrame = PrivateAttr(None) # 加载的数据 _config: Dict[str, Any] = PrivateAttr({}) # 配置参数 def __init__(self, name: str, description: str, display_name: str = "", csv_path: str = None, config: Dict[str, Any] = None, **kwargs): """ 初始化增强版识别器 Args: name: 识别器名称 description: 识别器描述 display_name: 显示名称 csv_path: CSV文件路径 config: 配置参数字典 """ # 确保display_name有默认值 if not display_name: display_name = name # 调用父类初始化 super().__init__( name=name, description=description, **kwargs ) # 设置属性 self.display_name = display_name self._recognized_count = 0 self._csv_path = csv_path self._config = config or {} self._data = None # ==================== 统一的数据处理方法 ==================== def load_data(self, csv_path: str = None) -> pd.DataFrame: """ 加载并标准化数据 Args: csv_path: CSV文件路径,如果为None则使用初始化时的路径 Returns: pd.DataFrame: 标准化后的数据 Raises: ValueError: 如果没有提供数据路径 FileNotFoundError: 如果文件不存在 """ data_path = csv_path or self._csv_path if not data_path: raise ValueError("未提供数据路径,请在初始化时设置csv_path或调用时传入") print(f"📥 {self.display_name} 正在加载数据: {data_path}") try: # 加载数据 df = pd.read_csv(data_path) print(f"✅ 成功加载 {len(df)} 条交易记录") # 数据标准化处理 df = self._standardize_data(df) # 缓存数据 self._data = df return df except FileNotFoundError: raise FileNotFoundError(f"文件不存在: {data_path}") except Exception as e: raise Exception(f"数据加载失败: {str(e)}") # enhanced_base_recognizer.py 中的 _standardize_data 方法 def _standardize_data(self, df: pd.DataFrame) -> pd.DataFrame: """ 标准化数据格式 Args: df: 原始数据 Returns: pd.DataFrame: 标准化后的数据 """ df = df.copy() # 1. 确保关键字段存在 required_fields = ['txId', 'txDate', 'txTime', 'txAmount', 'txDirection'] missing_fields = [f for f in required_fields if f not in df.columns] if missing_fields: print(f"⚠️ 警告:缺少字段 {missing_fields},某些检查可能无法进行") # 2. 关键字段类型转换 # txId: 统一转换为字符串,确保格式化时不会出错 if 'txId' in df.columns: df['txId'] = df['txId'].astype(str).str.strip() print(f" ✅ txId 已转换为字符串类型,共 {len(df)} 条记录") # txAmount 和 txBalance: 转换为数值类型 if 'txAmount' in df.columns: df['txAmount'] = pd.to_numeric(df['txAmount'], errors='coerce') invalid_amounts = df['txAmount'].isna().sum() if invalid_amounts > 0: print(f" ⚠️ 有 {invalid_amounts} 条记录的 txAmount 无法转换为数值") if 'txBalance' in df.columns: df['txBalance'] = pd.to_numeric(df['txBalance'], errors='coerce') invalid_balances = df['txBalance'].isna().sum() if invalid_balances > 0: print(f" ⚠️ 有 {invalid_balances} 条记录的 txBalance 无法转换为数值") # 3. 字符串字段清理 if 'txDirection' in df.columns: df['txDirection'] = df['txDirection'].astype(str).str.strip() # 统计方向分布 direction_counts = df['txDirection'].value_counts() print(f" 📊 交易方向分布: {dict(direction_counts)}") if 'txSummary' in df.columns: df['txSummary'] = df['txSummary'].astype(str).str.strip() if 'txCounterparty' in df.columns: df['txCounterparty'] = df['txCounterparty'].astype(str).str.strip() # 4. 创建统一的datetime字段(如果日期时间字段存在) if 'txDate' in df.columns and 'txTime' in df.columns: try: # 先确保是字符串 df['txDate'] = df['txDate'].astype(str).str.strip() df['txTime'] = df['txTime'].astype(str).str.strip() # 组合成datetime datetime_str = df['txDate'] + ' ' + df['txTime'] df['datetime'] = pd.to_datetime(datetime_str, errors='coerce') # 统计解析失败的数量 failed_parse = df['datetime'].isna().sum() if failed_parse > 0: print(f" ⚠️ 有 {failed_parse} 条记录的时间解析失败") else: print(f" ✅ 所有 {len(df)} 条记录的时间解析成功") # 提取时间组件 df['hour'] = df['datetime'].dt.hour df['minute'] = df['datetime'].dt.minute df['date_only'] = df['datetime'].dt.date df['day_of_week'] = df['datetime'].dt.dayofweek # 0=周一, 6=周日 except Exception as e: print(f"⚠️ 时间解析失败: {e}") # 5. 数据质量检查 print(f"📊 数据标准化完成统计:") print(f" 总记录数: {len(df)}") if 'txId' in df.columns: unique_ids = df['txId'].nunique() print(f" 唯一交易ID数: {unique_ids}") if unique_ids != len(df): print(f" ⚠️ 警告: 有 {len(df) - unique_ids} 条重复的交易ID") if 'datetime' in df.columns: date_range = df['datetime'].min(), df['datetime'].max() print(f" 时间范围: {date_range[0]} 到 {date_range[1]}") print(f" 总天数: {df['date_only'].nunique()}") return df # ==================== 统一的异常记录格式 ==================== def format_anomaly_record(self, row: pd.Series, reason: str, severity: str = "medium", **additional_info) -> Dict[str, Any]: """ 创建标准化的异常记录 Args: row: 数据行(pandas Series) reason: 异常原因描述 severity: 严重程度 (high/medium/low) **additional_info: 额外信息 Returns: Dict[str, Any]: 标准化异常记录 """ # 基础字段 record = { 'recognition_type': self.display_name, 'txId': str(row.get('txId', '')), 'txDate': str(row.get('txDate', '')), 'txTime': str(row.get('txTime', '')), 'txAmount': float(row.get('txAmount', 0)), 'txDirection': str(row.get('txDirection', '')), 'recognition_reason': reason, 'severity': severity, 'additional_info': additional_info, 'status': '待核查', 'recognizer_name': self.name } # 可选字段 optional_fields = ['txBalance', 'txSummary', 'txCounterparty'] for field in optional_fields: if field in row and pd.notna(row[field]): record[field] = row[field] # 添加datetime信息(如果已标准化) if 'datetime' in row and pd.notna(row['datetime']): record['datetime'] = row['datetime'].strftime("%Y-%m-%d %H:%M:%S") return record # ==================== 配置管理 ==================== def get_config_value(self, key: str, default: Any = None) -> Any: """ 获取配置值 Args: key: 配置键 default: 默认值 Returns: 配置值或默认值 """ return self._config.get(key, default) # ==================== 工具方法 ==================== def is_night_time(self, hour: int, start_hour: int = 2, end_hour: int = 5) -> bool: """判断是否为夜间时间(凌晨2-5点)""" return start_hour <= hour <= end_hour def is_integer_amount(self, amount: float, base_amount: float = 10000.0, tolerance: float = 0.01) -> bool: """判断是否为整数金额(基准金额的整数倍)""" if pd.isna(amount): return False return abs(amount % base_amount) < tolerance or abs(amount % base_amount - base_amount) < tolerance def calculate_time_difference(self, time1: datetime, time2: datetime, unit: str = 'hours') -> float: """计算两个时间点的时间差""" if pd.isna(time1) or pd.isna(time2): return float('inf') diff_seconds = abs((time2 - time1).total_seconds()) if unit == 'hours': return diff_seconds / 3600 elif unit == 'days': return diff_seconds / 86400 elif unit == 'minutes': return diff_seconds / 60 else: return diff_seconds # ==================== 抽象方法(子类必须实现) ==================== @abstractmethod def recognize(self, **kwargs) -> Dict[str, Any]: """ 执行异常识别(子类必须实现) Returns: Dict[str, Any]: 识别结果,必须包含: - recognition_type: 识别类型 - identified_count: 识别的异常数量 - identified_anomalies: 异常记录列表 - recognition_status: 识别状态 """ pass # ==================== BaseTool接口实现 ==================== def _run(self, csv_path: str = None, **kwargs) -> Dict[str, Any]: """ 实现BaseTool的_run方法 Args: csv_path: CSV文件路径 **kwargs: 其他参数 Returns: Dict[str, Any]: 识别结果 """ try: start_time = datetime.now() # 执行识别 result = self.recognize(csv_path=csv_path, **kwargs) # 对结果进行标准化处理 standardized_result = self._standardize_result(result) # 确保结果是字典(不是字符串) if isinstance(standardized_result, str): try: # 尝试解析字符串为字典 import json standardized_result = json.loads(standardized_result) except json.JSONDecodeError: # 如果无法解析,包装成字典 standardized_result = { 'recognition_type': self.display_name, 'identified_count': 0, 'identified_anomalies': [], 'recognition_status': '完成', 'raw_output': standardized_result } # 确保结果包含必要字段(原有逻辑) if 'recognition_type' not in standardized_result: standardized_result['recognition_type'] = self.display_name if 'identified_count' not in standardized_result: standardized_result['identified_count'] = len(standardized_result.get('identified_anomalies', [])) if 'recognition_status' not in standardized_result: standardized_result['recognition_status'] = '完成' # 添加执行信息 execution_time = (datetime.now() - start_time).total_seconds() standardized_result['execution_info'] = { 'recognizer_name': self.name, 'display_name': self.display_name, 'execution_time_seconds': execution_time, 'execution_time': start_time.strftime("%Y-%m-%d %H:%M:%S") } # 更新识别计数 self._recognized_count = standardized_result.get('identified_count', 0) return standardized_result return result except Exception as e: error_msg = f"异常识别失败: {str(e)}" print(f"❌ {self.display_name} - {error_msg}") return { 'recognition_type': self.display_name, 'identified_count': 0, 'identified_anomalies': [], 'recognition_status': '失败', 'error': error_msg, 'execution_info': { 'recognizer_name': self.name, 'display_name': self.display_name, 'error': str(e) } } async def _arun(self, **kwargs): raise NotImplementedError("异步识别不支持") def _standardize_result(self, raw_result: Any) -> Dict[str, Any]: """ 标准化工具返回结果 Args: raw_result: 原始结果(可能是dict、str、list等) Returns: 标准化的字典结果 """ # 1. 如果已经是字典,直接返回 if isinstance(raw_result, dict): return raw_result # 2. 如果是字符串 elif isinstance(raw_result, str): # 尝试解析JSON try: import json parsed = json.loads(raw_result) if isinstance(parsed, dict): return parsed else: # JSON但不是字典(如list、str等) return { 'recognition_type': self.display_name, 'identified_count': 0, 'identified_anomalies': [], 'recognition_status': '完成', 'raw_output': raw_result[:500] # 截断长文本 } except json.JSONDecodeError: # 不是JSON格式的字符串 # 检查是否是Python字典的字符串表示(如"{'key': 'value'}") if raw_result.startswith('{') and raw_result.endswith('}'): try: # 尝试使用ast安全解析 import ast parsed = ast.literal_eval(raw_result) if isinstance(parsed, dict): return parsed except (SyntaxError, ValueError): pass # 无法解析,包装成标准格式 return { 'recognition_type': self.display_name, 'identified_count': 0, 'identified_anomalies': [], 'recognition_status': '完成', 'error': '工具返回了无法解析的字符串格式', 'raw_output_preview': raw_result[:200] } # 3. 其他类型(list、tuple、数字等) else: return { 'recognition_type': self.display_name, 'identified_count': 0, 'identified_anomalies': [], 'recognition_status': '完成', 'error': f'工具返回了非标准类型: {type(raw_result).__name__}', 'raw_output': str(raw_result)[:500] } # ==================== 其他方法 ==================== def get_summary(self) -> Dict[str, Any]: """获取识别器摘要""" return { 'name': self.name, 'display_name': self.display_name, 'description': self.description, 'recognized_count': self._recognized_count, 'csv_path': self._csv_path, 'config': self._config } def get_data_summary(self) -> Dict[str, Any]: """获取数据摘要""" if self._data is None: return {} df = self._data summary = { 'total_records': len(df), 'date_range': None, 'amount_stats': None } if 'datetime' in df.columns and not df['datetime'].isna().all(): summary['date_range'] = { 'start': df['datetime'].min().strftime("%Y-%m-%d"), 'end': df['datetime'].max().strftime("%Y-%m-%d") } if 'txAmount' in df.columns: summary['amount_stats'] = { 'mean': float(df['txAmount'].mean()), 'max': float(df['txAmount'].max()), 'min': float(df['txAmount'].min()), 'sum': float(df['txAmount'].sum()), 'std': float(df['txAmount'].std()) } return summary