enhanced_base_recognizer.py 17 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485
  1. from langchain.tools import BaseTool
  2. from abc import abstractmethod
  3. from typing import Dict, Any
  4. import pandas as pd
  5. from datetime import datetime
  6. from pydantic import Field, PrivateAttr
  7. class EnhancedBaseRecognizer(BaseTool):
  8. """增强版异常识别器基类 - 提供统一的数据处理和异常记录格式"""
  9. name: str = Field(..., description="识别器名称")
  10. description: str = Field(..., description="识别器描述")
  11. display_name: str = Field("", description="显示名称")
  12. # 使用 PrivateAttr 定义不需要验证的私有属性
  13. _recognized_count: int = PrivateAttr(0)
  14. _csv_path: str = PrivateAttr(None) # 标准化后的csv文件路径
  15. _data: pd.DataFrame = PrivateAttr(None) # 加载的数据
  16. _config: Dict[str, Any] = PrivateAttr({}) # 配置参数
  17. def __init__(self, name: str, description: str, display_name: str = "",
  18. csv_path: str = None, config: Dict[str, Any] = None, **kwargs):
  19. """
  20. 初始化增强版识别器
  21. Args:
  22. name: 识别器名称
  23. description: 识别器描述
  24. display_name: 显示名称
  25. csv_path: CSV文件路径
  26. config: 配置参数字典
  27. """
  28. # 确保display_name有默认值
  29. if not display_name:
  30. display_name = name
  31. # 调用父类初始化
  32. super().__init__(
  33. name=name,
  34. description=description,
  35. **kwargs
  36. )
  37. # 设置属性
  38. self.display_name = display_name
  39. self._recognized_count = 0
  40. self._csv_path = csv_path
  41. self._config = config or {}
  42. self._data = None
  43. # ==================== 统一的数据处理方法 ====================
  44. def load_data(self, csv_path: str = None) -> pd.DataFrame:
  45. """
  46. 加载并标准化数据
  47. Args:
  48. csv_path: CSV文件路径,如果为None则使用初始化时的路径
  49. Returns:
  50. pd.DataFrame: 标准化后的数据
  51. Raises:
  52. ValueError: 如果没有提供数据路径
  53. FileNotFoundError: 如果文件不存在
  54. """
  55. data_path = csv_path or self._csv_path
  56. if not data_path:
  57. raise ValueError("未提供数据路径,请在初始化时设置csv_path或调用时传入")
  58. print(f"📥 {self.display_name} 正在加载数据: {data_path}")
  59. try:
  60. # 加载数据
  61. df = pd.read_csv(data_path)
  62. print(f"✅ 成功加载 {len(df)} 条交易记录")
  63. # 数据标准化处理
  64. df = self._standardize_data(df)
  65. # 缓存数据
  66. self._data = df
  67. return df
  68. except FileNotFoundError:
  69. raise FileNotFoundError(f"文件不存在: {data_path}")
  70. except Exception as e:
  71. raise Exception(f"数据加载失败: {str(e)}")
  72. # enhanced_base_recognizer.py 中的 _standardize_data 方法
  73. def _standardize_data(self, df: pd.DataFrame) -> pd.DataFrame:
  74. """
  75. 标准化数据格式
  76. Args:
  77. df: 原始数据
  78. Returns:
  79. pd.DataFrame: 标准化后的数据
  80. """
  81. df = df.copy()
  82. # 1. 确保关键字段存在
  83. required_fields = ['txId', 'txDate', 'txTime', 'txAmount', 'txDirection']
  84. missing_fields = [f for f in required_fields if f not in df.columns]
  85. if missing_fields:
  86. print(f"⚠️ 警告:缺少字段 {missing_fields},某些检查可能无法进行")
  87. # 2. 关键字段类型转换
  88. # txId: 统一转换为字符串,确保格式化时不会出错
  89. if 'txId' in df.columns:
  90. df['txId'] = df['txId'].astype(str).str.strip()
  91. print(f" ✅ txId 已转换为字符串类型,共 {len(df)} 条记录")
  92. # txAmount 和 txBalance: 转换为数值类型
  93. if 'txAmount' in df.columns:
  94. df['txAmount'] = pd.to_numeric(df['txAmount'], errors='coerce')
  95. invalid_amounts = df['txAmount'].isna().sum()
  96. if invalid_amounts > 0:
  97. print(f" ⚠️ 有 {invalid_amounts} 条记录的 txAmount 无法转换为数值")
  98. if 'txBalance' in df.columns:
  99. df['txBalance'] = pd.to_numeric(df['txBalance'], errors='coerce')
  100. invalid_balances = df['txBalance'].isna().sum()
  101. if invalid_balances > 0:
  102. print(f" ⚠️ 有 {invalid_balances} 条记录的 txBalance 无法转换为数值")
  103. # 3. 字符串字段清理
  104. if 'txDirection' in df.columns:
  105. df['txDirection'] = df['txDirection'].astype(str).str.strip()
  106. # 统计方向分布
  107. direction_counts = df['txDirection'].value_counts()
  108. print(f" 📊 交易方向分布: {dict(direction_counts)}")
  109. if 'txSummary' in df.columns:
  110. df['txSummary'] = df['txSummary'].astype(str).str.strip()
  111. if 'txCounterparty' in df.columns:
  112. df['txCounterparty'] = df['txCounterparty'].astype(str).str.strip()
  113. # 4. 创建统一的datetime字段(如果日期时间字段存在)
  114. if 'txDate' in df.columns and 'txTime' in df.columns:
  115. try:
  116. # 先确保是字符串
  117. df['txDate'] = df['txDate'].astype(str).str.strip()
  118. df['txTime'] = df['txTime'].astype(str).str.strip()
  119. # 组合成datetime
  120. datetime_str = df['txDate'] + ' ' + df['txTime']
  121. df['datetime'] = pd.to_datetime(datetime_str, errors='coerce')
  122. # 统计解析失败的数量
  123. failed_parse = df['datetime'].isna().sum()
  124. if failed_parse > 0:
  125. print(f" ⚠️ 有 {failed_parse} 条记录的时间解析失败")
  126. else:
  127. print(f" ✅ 所有 {len(df)} 条记录的时间解析成功")
  128. # 提取时间组件
  129. df['hour'] = df['datetime'].dt.hour
  130. df['minute'] = df['datetime'].dt.minute
  131. df['date_only'] = df['datetime'].dt.date
  132. df['day_of_week'] = df['datetime'].dt.dayofweek # 0=周一, 6=周日
  133. except Exception as e:
  134. print(f"⚠️ 时间解析失败: {e}")
  135. # 5. 数据质量检查
  136. print(f"📊 数据标准化完成统计:")
  137. print(f" 总记录数: {len(df)}")
  138. if 'txId' in df.columns:
  139. unique_ids = df['txId'].nunique()
  140. print(f" 唯一交易ID数: {unique_ids}")
  141. if unique_ids != len(df):
  142. print(f" ⚠️ 警告: 有 {len(df) - unique_ids} 条重复的交易ID")
  143. if 'datetime' in df.columns:
  144. date_range = df['datetime'].min(), df['datetime'].max()
  145. print(f" 时间范围: {date_range[0]} 到 {date_range[1]}")
  146. print(f" 总天数: {df['date_only'].nunique()}")
  147. return df
  148. # ==================== 统一的异常记录格式 ====================
  149. def format_anomaly_record(self, row: pd.Series, reason: str,
  150. severity: str = "medium", **additional_info) -> Dict[str, Any]:
  151. """
  152. 创建标准化的异常记录
  153. Args:
  154. row: 数据行(pandas Series)
  155. reason: 异常原因描述
  156. severity: 严重程度 (high/medium/low)
  157. **additional_info: 额外信息
  158. Returns:
  159. Dict[str, Any]: 标准化异常记录
  160. """
  161. # 基础字段
  162. record = {
  163. 'recognition_type': self.display_name,
  164. 'txId': str(row.get('txId', '')),
  165. 'txDate': str(row.get('txDate', '')),
  166. 'txTime': str(row.get('txTime', '')),
  167. 'txAmount': float(row.get('txAmount', 0)),
  168. 'txDirection': str(row.get('txDirection', '')),
  169. 'recognition_reason': reason,
  170. 'severity': severity,
  171. 'additional_info': additional_info,
  172. 'status': '待核查',
  173. 'recognizer_name': self.name
  174. }
  175. # 可选字段
  176. optional_fields = ['txBalance', 'txSummary', 'txCounterparty']
  177. for field in optional_fields:
  178. if field in row and pd.notna(row[field]):
  179. record[field] = row[field]
  180. # 添加datetime信息(如果已标准化)
  181. if 'datetime' in row and pd.notna(row['datetime']):
  182. record['datetime'] = row['datetime'].strftime("%Y-%m-%d %H:%M:%S")
  183. return record
  184. # ==================== 配置管理 ====================
  185. def get_config_value(self, key: str, default: Any = None) -> Any:
  186. """
  187. 获取配置值
  188. Args:
  189. key: 配置键
  190. default: 默认值
  191. Returns:
  192. 配置值或默认值
  193. """
  194. return self._config.get(key, default)
  195. # ==================== 工具方法 ====================
  196. def is_night_time(self, hour: int, start_hour: int = 2, end_hour: int = 5) -> bool:
  197. """判断是否为夜间时间(凌晨2-5点)"""
  198. return start_hour <= hour <= end_hour
  199. def is_integer_amount(self, amount: float, base_amount: float = 10000.0,
  200. tolerance: float = 0.01) -> bool:
  201. """判断是否为整数金额(基准金额的整数倍)"""
  202. if pd.isna(amount):
  203. return False
  204. return abs(amount % base_amount) < tolerance or abs(amount % base_amount - base_amount) < tolerance
  205. def calculate_time_difference(self, time1: datetime, time2: datetime,
  206. unit: str = 'hours') -> float:
  207. """计算两个时间点的时间差"""
  208. if pd.isna(time1) or pd.isna(time2):
  209. return float('inf')
  210. diff_seconds = abs((time2 - time1).total_seconds())
  211. if unit == 'hours':
  212. return diff_seconds / 3600
  213. elif unit == 'days':
  214. return diff_seconds / 86400
  215. elif unit == 'minutes':
  216. return diff_seconds / 60
  217. else:
  218. return diff_seconds
  219. # ==================== 抽象方法(子类必须实现) ====================
  220. @abstractmethod
  221. def recognize(self, **kwargs) -> Dict[str, Any]:
  222. """
  223. 执行异常识别(子类必须实现)
  224. Returns:
  225. Dict[str, Any]: 识别结果,必须包含:
  226. - recognition_type: 识别类型
  227. - identified_count: 识别的异常数量
  228. - identified_anomalies: 异常记录列表
  229. - recognition_status: 识别状态
  230. """
  231. pass
  232. # ==================== BaseTool接口实现 ====================
  233. def _run(self, csv_path: str = None, **kwargs) -> Dict[str, Any]:
  234. """
  235. 实现BaseTool的_run方法
  236. Args:
  237. csv_path: CSV文件路径
  238. **kwargs: 其他参数
  239. Returns:
  240. Dict[str, Any]: 识别结果
  241. """
  242. try:
  243. start_time = datetime.now()
  244. # 执行识别
  245. result = self.recognize(csv_path=csv_path, **kwargs)
  246. # 对结果进行标准化处理
  247. standardized_result = self._standardize_result(result)
  248. # 确保结果是字典(不是字符串)
  249. if isinstance(standardized_result, str):
  250. try:
  251. # 尝试解析字符串为字典
  252. import json
  253. standardized_result = json.loads(standardized_result)
  254. except json.JSONDecodeError:
  255. # 如果无法解析,包装成字典
  256. standardized_result = {
  257. 'recognition_type': self.display_name,
  258. 'identified_count': 0,
  259. 'identified_anomalies': [],
  260. 'recognition_status': '完成',
  261. 'raw_output': standardized_result
  262. }
  263. # 确保结果包含必要字段(原有逻辑)
  264. if 'recognition_type' not in standardized_result:
  265. standardized_result['recognition_type'] = self.display_name
  266. if 'identified_count' not in standardized_result:
  267. standardized_result['identified_count'] = len(standardized_result.get('identified_anomalies', []))
  268. if 'recognition_status' not in standardized_result:
  269. standardized_result['recognition_status'] = '完成'
  270. # 添加执行信息
  271. execution_time = (datetime.now() - start_time).total_seconds()
  272. standardized_result['execution_info'] = {
  273. 'recognizer_name': self.name,
  274. 'display_name': self.display_name,
  275. 'execution_time_seconds': execution_time,
  276. 'execution_time': start_time.strftime("%Y-%m-%d %H:%M:%S")
  277. }
  278. # 更新识别计数
  279. self._recognized_count = standardized_result.get('identified_count', 0)
  280. return standardized_result
  281. return result
  282. except Exception as e:
  283. error_msg = f"异常识别失败: {str(e)}"
  284. print(f"❌ {self.display_name} - {error_msg}")
  285. return {
  286. 'recognition_type': self.display_name,
  287. 'identified_count': 0,
  288. 'identified_anomalies': [],
  289. 'recognition_status': '失败',
  290. 'error': error_msg,
  291. 'execution_info': {
  292. 'recognizer_name': self.name,
  293. 'display_name': self.display_name,
  294. 'error': str(e)
  295. }
  296. }
  297. async def _arun(self, **kwargs):
  298. raise NotImplementedError("异步识别不支持")
  299. def _standardize_result(self, raw_result: Any) -> Dict[str, Any]:
  300. """
  301. 标准化工具返回结果
  302. Args:
  303. raw_result: 原始结果(可能是dict、str、list等)
  304. Returns:
  305. 标准化的字典结果
  306. """
  307. # 1. 如果已经是字典,直接返回
  308. if isinstance(raw_result, dict):
  309. return raw_result
  310. # 2. 如果是字符串
  311. elif isinstance(raw_result, str):
  312. # 尝试解析JSON
  313. try:
  314. import json
  315. parsed = json.loads(raw_result)
  316. if isinstance(parsed, dict):
  317. return parsed
  318. else:
  319. # JSON但不是字典(如list、str等)
  320. return {
  321. 'recognition_type': self.display_name,
  322. 'identified_count': 0,
  323. 'identified_anomalies': [],
  324. 'recognition_status': '完成',
  325. 'raw_output': raw_result[:500] # 截断长文本
  326. }
  327. except json.JSONDecodeError:
  328. # 不是JSON格式的字符串
  329. # 检查是否是Python字典的字符串表示(如"{'key': 'value'}")
  330. if raw_result.startswith('{') and raw_result.endswith('}'):
  331. try:
  332. # 尝试使用ast安全解析
  333. import ast
  334. parsed = ast.literal_eval(raw_result)
  335. if isinstance(parsed, dict):
  336. return parsed
  337. except (SyntaxError, ValueError):
  338. pass
  339. # 无法解析,包装成标准格式
  340. return {
  341. 'recognition_type': self.display_name,
  342. 'identified_count': 0,
  343. 'identified_anomalies': [],
  344. 'recognition_status': '完成',
  345. 'error': '工具返回了无法解析的字符串格式',
  346. 'raw_output_preview': raw_result[:200]
  347. }
  348. # 3. 其他类型(list、tuple、数字等)
  349. else:
  350. return {
  351. 'recognition_type': self.display_name,
  352. 'identified_count': 0,
  353. 'identified_anomalies': [],
  354. 'recognition_status': '完成',
  355. 'error': f'工具返回了非标准类型: {type(raw_result).__name__}',
  356. 'raw_output': str(raw_result)[:500]
  357. }
  358. # ==================== 其他方法 ====================
  359. def get_summary(self) -> Dict[str, Any]:
  360. """获取识别器摘要"""
  361. return {
  362. 'name': self.name,
  363. 'display_name': self.display_name,
  364. 'description': self.description,
  365. 'recognized_count': self._recognized_count,
  366. 'csv_path': self._csv_path,
  367. 'config': self._config
  368. }
  369. def get_data_summary(self) -> Dict[str, Any]:
  370. """获取数据摘要"""
  371. if self._data is None:
  372. return {}
  373. df = self._data
  374. summary = {
  375. 'total_records': len(df),
  376. 'date_range': None,
  377. 'amount_stats': None
  378. }
  379. if 'datetime' in df.columns and not df['datetime'].isna().all():
  380. summary['date_range'] = {
  381. 'start': df['datetime'].min().strftime("%Y-%m-%d"),
  382. 'end': df['datetime'].max().strftime("%Y-%m-%d")
  383. }
  384. if 'txAmount' in df.columns:
  385. summary['amount_stats'] = {
  386. 'mean': float(df['txAmount'].mean()),
  387. 'max': float(df['txAmount'].max()),
  388. 'min': float(df['txAmount'].min()),
  389. 'sum': float(df['txAmount'].sum()),
  390. 'std': float(df['txAmount'].std())
  391. }
  392. return summary