| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485 |
- 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
|