balance_recognizer.py 22 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550
  1. from pydantic import BaseModel, Field
  2. from typing import Dict, Any, Optional, Type
  3. import pandas as pd
  4. from itertools import permutations
  5. from .enhanced_base_recognizer import EnhancedBaseRecognizer
  6. class BalanceRecognitionInput(BaseModel):
  7. """余额识别工具输入"""
  8. csv_path: Optional[str] = Field(
  9. None,
  10. description="CSV文件路径(可选)。如果初始化时已提供csv_path,可以不用再次传入。"
  11. )
  12. class Config:
  13. arbitrary_types_allowed = True
  14. class BalanceContinuityRecognizer(EnhancedBaseRecognizer):
  15. """余额连续性异常识别器(带智能排序)"""
  16. args_schema: Type[BaseModel] = BalanceRecognitionInput
  17. # 配置参数
  18. balance_tolerance: float = Field(
  19. 0.01,
  20. description="余额计算容差,允许的余额差异阈值"
  21. )
  22. enable_smart_sorting: bool = Field(
  23. True,
  24. description="是否启用智能排序处理时间相同的交易"
  25. )
  26. max_permutation_search: int = Field(
  27. 6, # 3! = 6, 4! = 24, 设置为6可以处理最多3笔时间相同的交易
  28. description="最大排列搜索数,防止组合爆炸"
  29. )
  30. def __init__(self, csv_path: str = None, config: Dict[str, Any] = None, **kwargs):
  31. """
  32. 初始化余额连续性异常识别器
  33. Args:
  34. csv_path: CSV文件路径
  35. config: 配置参数
  36. **kwargs: 其他参数
  37. """
  38. # 调用父类的 __init__
  39. super().__init__(
  40. name="balance_continuity_recognizer",
  41. description="识别银行流水中的余额连续性异常,检查每笔交易后的余额计算是否正确。"
  42. "支持智能排序处理时间相同的交易。",
  43. display_name="余额连续性异常识别",
  44. csv_path=csv_path,
  45. config=config,
  46. **kwargs
  47. )
  48. # 从config获取配置,更新Field属性
  49. balance_config = self.get_config_value('balance_recognition', {})
  50. if balance_config:
  51. if 'balance_tolerance' in balance_config:
  52. self.balance_tolerance = balance_config['balance_tolerance']
  53. if 'enable_smart_sorting' in balance_config:
  54. self.enable_smart_sorting = balance_config['enable_smart_sorting']
  55. if 'max_permutation_search' in balance_config:
  56. self.max_permutation_search = balance_config['max_permutation_search']
  57. def recognize(self, csv_path: str = None, **kwargs) -> Dict[str, Any]:
  58. """识别余额连续性异常(带智能排序)"""
  59. try:
  60. # 使用父类的load_data方法加载标准化数据
  61. df = self.load_data(csv_path)
  62. print(f"🔍 {self.display_name}开始检查余额连续性,共 {len(df)} 条记录")
  63. print(f" 余额容差: {self.balance_tolerance}")
  64. print(f" 智能排序: {'启用' if self.enable_smart_sorting else '禁用'}")
  65. # 检查必需字段
  66. required_fields = ['txId', 'txDate', 'txTime', 'txAmount', 'txDirection', 'txBalance']
  67. missing_fields = [field for field in required_fields if field not in df.columns]
  68. if missing_fields:
  69. return {
  70. 'recognition_type': self.display_name,
  71. 'identified_count': 0,
  72. 'identified_anomalies': [],
  73. 'recognition_status': '失败',
  74. 'error': f'缺少必需字段: {missing_fields}'
  75. }
  76. # ============ 智能排序处理 ============
  77. if self.enable_smart_sorting and 'datetime' in df.columns:
  78. df = self._apply_smart_sorting(df)
  79. else:
  80. # 简单排序:按时间,时间相同按ID
  81. if 'datetime' in df.columns:
  82. df = df.sort_values(['datetime', 'txId'])
  83. else:
  84. df = df.sort_values(['txDate', 'txTime', 'txId'])
  85. print("📋 排序后的交易顺序:")
  86. for i, (_, row) in enumerate(df.head(10).iterrows(), 1):
  87. time_str = row['datetime'].strftime(
  88. "%Y-%m-%d %H:%M:%S") if 'datetime' in row else f"{row['txDate']} {row['txTime']}"
  89. dir_symbol = "→" if row['txDirection'] == '支出' else "←"
  90. # 现在 txId 已经是字符串,可以直接使用
  91. # 但为了代码清晰,可以明确标注
  92. tx_id = row['txId'] # 已经是字符串
  93. print(
  94. f" {i:2d}. ID:{tx_id:>4s} | {time_str} | {dir_symbol} {row['txAmount']:8.2f} | 余额:{row['txBalance']:8.2f}")
  95. # ============ 开始余额连续性检查 ============
  96. identified_anomalies = []
  97. prev_balance = None
  98. valid_transactions = 0
  99. for idx, row in df.iterrows():
  100. tx_id = row['txId']
  101. current_balance = row['txBalance']
  102. valid_transactions += 1
  103. # 检查1:余额是否为空
  104. if pd.isna(current_balance):
  105. anomaly = self.format_anomaly_record(
  106. row=row,
  107. reason='余额字段为空',
  108. severity='high',
  109. check_type='missing_balance',
  110. previous_balance=prev_balance
  111. )
  112. identified_anomalies.append(anomaly)
  113. print(f" ❌ 交易ID {tx_id}: 余额字段为空")
  114. continue
  115. # 检查2:余额连续性(如果不是第一条记录)
  116. if prev_balance is not None:
  117. amount = row['txAmount']
  118. direction = str(row['txDirection']).strip()
  119. # 计算预期余额
  120. if direction == '收入':
  121. expected_balance = prev_balance + amount
  122. elif direction == '支出':
  123. expected_balance = prev_balance - amount
  124. else:
  125. # 未知方向,跳过检查
  126. print(f" ⚠️ 交易ID {tx_id}: 未知的交易方向 '{direction}',跳过余额检查")
  127. prev_balance = current_balance
  128. continue
  129. # 检查余额是否连续(允许小误差)
  130. if pd.isna(expected_balance):
  131. # 预期余额计算异常
  132. anomaly = self.format_anomaly_record(
  133. row=row,
  134. reason=f'预期余额计算异常,可能金额字段有问题: amount={amount}',
  135. severity='high',
  136. check_type='calculation_error',
  137. previous_balance=prev_balance,
  138. expected_balance=expected_balance,
  139. actual_balance=current_balance
  140. )
  141. identified_anomalies.append(anomaly)
  142. print(f" ❌ 交易ID {tx_id}: 预期余额计算异常")
  143. else:
  144. balance_diff = abs(expected_balance - current_balance)
  145. if balance_diff > self.balance_tolerance:
  146. anomaly = self.format_anomaly_record(
  147. row=row,
  148. reason=f'余额计算不连续,预期{expected_balance:.2f},实际{current_balance:.2f},差异{balance_diff:.2f}',
  149. severity='high',
  150. check_type='balance_discontinuity',
  151. previous_balance=prev_balance,
  152. expected_balance=expected_balance,
  153. actual_balance=current_balance,
  154. balance_difference=balance_diff,
  155. tolerance_threshold=self.balance_tolerance
  156. )
  157. identified_anomalies.append(anomaly)
  158. print(f" ❌ 交易ID {tx_id}: 余额不连续,差异 {balance_diff:.2f}")
  159. else:
  160. # 余额连续,正常情况
  161. pass
  162. prev_balance = current_balance
  163. print(f"✅ {self.display_name}检查完成")
  164. print(f" 检查交易数: {valid_transactions}")
  165. print(f" 发现异常数: {len(identified_anomalies)}")
  166. # 统计不同类型异常的数量
  167. missing_balance_count = len([a for a in identified_anomalies
  168. if a.get('additional_info', {}).get('check_type') == 'missing_balance'])
  169. discontinuity_count = len([a for a in identified_anomalies
  170. if a.get('additional_info', {}).get('check_type') == 'balance_discontinuity'])
  171. calculation_error_count = len([a for a in identified_anomalies
  172. if a.get('additional_info', {}).get('check_type') == 'calculation_error'])
  173. return {
  174. 'recognition_type': self.display_name,
  175. 'identified_count': len(identified_anomalies),
  176. 'identified_anomalies': identified_anomalies,
  177. 'recognition_status': '完成',
  178. 'recognition_parameters': {
  179. 'balance_tolerance': self.balance_tolerance,
  180. 'enable_smart_sorting': self.enable_smart_sorting,
  181. 'checked_transactions': valid_transactions,
  182. 'data_source': csv_path or self._csv_path
  183. },
  184. 'statistics': {
  185. 'missing_balance_count': missing_balance_count,
  186. 'discontinuity_count': discontinuity_count,
  187. 'calculation_error_count': calculation_error_count,
  188. 'first_valid_balance': float(df['txBalance'].iloc[0]) if len(df) > 0 and not pd.isna(
  189. df['txBalance'].iloc[0]) else None,
  190. 'last_valid_balance': float(df['txBalance'].iloc[-1]) if len(df) > 0 and not pd.isna(
  191. df['txBalance'].iloc[-1]) else None,
  192. 'total_transactions': len(df),
  193. 'valid_balance_count': df['txBalance'].notna().sum(),
  194. 'avg_balance': float(df['txBalance'].mean()) if df['txBalance'].notna().any() else None
  195. }
  196. }
  197. except FileNotFoundError as e:
  198. return {
  199. 'recognition_type': self.display_name,
  200. 'identified_count': 0,
  201. 'identified_anomalies': [],
  202. 'recognition_status': '失败',
  203. 'error': f'文件不存在: {str(e)}'
  204. }
  205. except Exception as e:
  206. import traceback
  207. traceback.print_exc()
  208. return {
  209. 'recognition_type': self.display_name,
  210. 'identified_count': 0,
  211. 'identified_anomalies': [],
  212. 'recognition_status': '失败',
  213. 'error': f'数据加载或处理失败: {str(e)}'
  214. }
  215. # ==================== 智能排序核心算法 ====================
  216. def _apply_smart_sorting(self, df: pd.DataFrame) -> pd.DataFrame:
  217. """
  218. 应用智能排序算法
  219. 处理步骤:
  220. 1. 按时间分组
  221. 2. 对每个时间组内的交易进行智能排序
  222. 3. 合并所有组
  223. """
  224. print("🧠 应用智能排序算法...")
  225. # 首先按时间排序,得到时间组
  226. df = df.sort_values('datetime')
  227. # 找出所有时间相同的交易组
  228. time_groups = list(df.groupby('datetime'))
  229. if len(time_groups) == len(df):
  230. print(" ✅ 所有交易时间都不同,无需智能排序")
  231. return df
  232. # 处理每个时间组
  233. sorted_groups = []
  234. prev_group_last_balance = None
  235. for i, (time_val, group) in enumerate(time_groups):
  236. group_size = len(group)
  237. if group_size == 1:
  238. # 单笔交易,直接加入
  239. sorted_groups.append(group)
  240. if not group['txBalance'].isna().iloc[0]:
  241. prev_group_last_balance = group['txBalance'].iloc[0]
  242. continue
  243. # 多笔交易时间相同,需要智能排序
  244. print(f" 🔍 时间组 {i + 1}/{len(time_groups)}: {time_val},共 {group_size} 笔交易")
  245. # 获取前一组的最后一笔余额(如果有)
  246. prev_balance = prev_group_last_balance
  247. # 智能排序这个组
  248. sorted_group = self._smart_sort_time_group(group, prev_balance)
  249. sorted_groups.append(sorted_group)
  250. # 更新前一组的最后一笔余额
  251. if not sorted_group['txBalance'].isna().iloc[-1]:
  252. prev_group_last_balance = sorted_group['txBalance'].iloc[-1]
  253. # 合并所有组
  254. result_df = pd.concat(sorted_groups, ignore_index=True)
  255. print(f" ✅ 智能排序完成,处理了 {len(time_groups)} 个时间组")
  256. return result_df
  257. def _smart_sort_time_group(self, group: pd.DataFrame, prev_balance: float = None) -> pd.DataFrame:
  258. """
  259. 智能排序一个时间组内的交易
  260. 策略:
  261. 1. 如果组内交易数 <= 3,尝试所有排列
  262. 2. 如果更多,使用启发式算法
  263. """
  264. group_size = len(group)
  265. if group_size == 0:
  266. return group
  267. # 显示组内交易详情
  268. print(f" 组内交易详情:")
  269. for idx, (_, row) in enumerate(group.iterrows(), 1):
  270. dir_symbol = "→" if row['txDirection'] == '支出' else "←"
  271. balance_info = f"余额:{row['txBalance']:.2f}" if not pd.isna(row['txBalance']) else "余额:None"
  272. print(f" {idx}. ID:{row['txId']} {dir_symbol} {row['txAmount']:.2f} {balance_info}")
  273. # 策略1:少量交易,尝试所有排列
  274. if group_size <= 3:
  275. return self._try_all_permutations(group, prev_balance)
  276. # 策略2:较多交易,使用启发式算法
  277. else:
  278. return self._heuristic_sort(group, prev_balance)
  279. def _try_all_permutations(self, group: pd.DataFrame, prev_balance: float = None) -> pd.DataFrame:
  280. """
  281. 尝试所有可能的排列,选择最优的
  282. 适用于少量交易(<=3笔)
  283. """
  284. group_size = len(group)
  285. print(f" 尝试 {group_size} 笔交易的所有排列 ({self._factorial(group_size)} 种可能)...")
  286. # 如果是2笔交易,特殊处理(常见情况)
  287. if group_size == 2:
  288. return self._optimize_two_transactions(group, prev_balance)
  289. # 生成所有排列
  290. best_order = None
  291. best_score = float('-inf')
  292. # 限制最大尝试数,防止组合爆炸
  293. max_tries = min(self.max_permutation_search, self._factorial(group_size))
  294. permutations_tried = 0
  295. for perm_indices in permutations(range(group_size)):
  296. if permutations_tried >= max_tries:
  297. break
  298. perm_group = group.iloc[list(perm_indices)].reset_index(drop=True)
  299. score = self._evaluate_order_quality(perm_group, prev_balance)
  300. if score > best_score:
  301. best_score = score
  302. best_order = perm_group
  303. permutations_tried += 1
  304. if best_order is not None:
  305. print(f" 找到最优排列,质量评分: {best_score:.2f}")
  306. if best_score < 0.5:
  307. print(f" ⚠️ 警告:最优排列质量评分较低 ({best_score:.2f})")
  308. # 显示最优顺序
  309. print(f" 最优顺序:")
  310. for idx, (_, row) in enumerate(best_order.iterrows(), 1):
  311. dir_symbol = "→" if row['txDirection'] == '支出' else "←"
  312. print(f" {idx}. ID:{row['txId']} {dir_symbol} {row['txAmount']:.2f}")
  313. return best_order
  314. return group
  315. def _optimize_two_transactions(self, group: pd.DataFrame, prev_balance: float = None) -> pd.DataFrame:
  316. """
  317. 优化两笔时间相同交易的顺序
  318. 这是最常见的情况,专门优化
  319. """
  320. if len(group) != 2:
  321. return group
  322. row1, row2 = group.iloc[0], group.iloc[1]
  323. # 计算两种顺序的质量评分
  324. order1 = pd.DataFrame([row1, row2]) # 原始顺序
  325. order2 = pd.DataFrame([row2, row1]) # 反转顺序
  326. score1 = self._evaluate_order_quality(order1, prev_balance)
  327. score2 = self._evaluate_order_quality(order2, prev_balance)
  328. print(f" 顺序1 (ID {row1['txId']}→{row2['txId']}): 评分 {score1:.2f}")
  329. print(f" 顺序2 (ID {row2['txId']}→{row1['txId']}): 评分 {score2:.2f}")
  330. if score2 > score1:
  331. print(f" ✅ 选择顺序2: ID {row2['txId']} → ID {row1['txId']}")
  332. return order2
  333. else:
  334. print(f" ✅ 选择顺序1: ID {row1['txId']} → ID {row2['txId']}")
  335. return order1
  336. def _heuristic_sort(self, group: pd.DataFrame, prev_balance: float = None) -> pd.DataFrame:
  337. """
  338. 启发式排序算法
  339. 适用于较多交易(>3笔)
  340. 启发式规则:
  341. 1. 先处理支出,后处理收入(常见模式)
  342. 2. 金额大的优先
  343. 3. 余额连续性验证
  344. """
  345. print(f" 使用启发式排序 ({len(group)} 笔交易)...")
  346. group = group.copy()
  347. # 启发式1:按交易方向排序
  348. group['sort_direction'] = group['txDirection'].map({'支出': 0, '收入': 1})
  349. # 启发式2:按金额排序(支出从大到小,收入从小到大)
  350. def get_amount_sort_key(row):
  351. if row['txDirection'] == '支出':
  352. return -row['txAmount'] # 支出金额大的优先
  353. else:
  354. return row['txAmount'] # 收入金额小的优先
  355. group['sort_amount'] = group.apply(get_amount_sort_key, axis=1)
  356. # 排序
  357. sorted_group = group.sort_values(['sort_direction', 'sort_amount', 'txId']).drop(
  358. ['sort_direction', 'sort_amount'], axis=1)
  359. # 验证排序质量
  360. score = self._evaluate_order_quality(sorted_group, prev_balance)
  361. print(f" 启发式排序质量评分: {score:.2f}")
  362. if score < 0.3:
  363. print(f" ⚠️ 启发式排序质量较低,考虑使用原始顺序")
  364. return group.drop(['sort_direction', 'sort_amount'], axis=1)
  365. return sorted_group
  366. def _evaluate_order_quality(self, ordered_group: pd.DataFrame, start_balance: float = None) -> float:
  367. """
  368. 评估排序质量
  369. 基于余额连续性计算质量评分
  370. 返回0-1之间的分数,越高越好
  371. """
  372. if len(ordered_group) == 0:
  373. return 0.0
  374. current_balance = start_balance
  375. total_score = 0.0
  376. valid_checks = 0
  377. for _, row in ordered_group.iterrows():
  378. if pd.isna(row.get('txBalance')):
  379. # 缺失余额,无法评估
  380. continue
  381. if current_balance is not None:
  382. # 计算预期余额
  383. expected = self._calculate_expected_balance(current_balance, row)
  384. if expected is not None:
  385. diff = abs(expected - row['txBalance'])
  386. if diff <= self.balance_tolerance:
  387. total_score += 1.0 # 完美匹配
  388. elif diff <= self.balance_tolerance * 10: # 允许10倍容差
  389. total_score += 0.5 # 部分匹配
  390. else:
  391. total_score -= 0.5 # 严重不匹配
  392. valid_checks += 1
  393. # 更新当前余额
  394. current_balance = row['txBalance']
  395. # 归一化分数
  396. if valid_checks > 0:
  397. # 基础分数是余额连续性分数
  398. continuity_score = total_score / valid_checks
  399. # 额外加分:如果整个组的总金额与余额变化匹配
  400. if start_balance is not None and not ordered_group['txBalance'].isna().all():
  401. final_balance = ordered_group['txBalance'].iloc[-1]
  402. total_change = sum(
  403. row['txAmount'] if row['txDirection'] == '收入' else -row['txAmount']
  404. for _, row in ordered_group.iterrows()
  405. )
  406. expected_final = start_balance + total_change
  407. final_diff = abs(expected_final - final_balance)
  408. if final_diff <= self.balance_tolerance:
  409. continuity_score += 0.2 # 额外加分
  410. elif final_diff <= self.balance_tolerance * 10:
  411. continuity_score += 0.1
  412. # 确保分数在0-1之间
  413. return max(0.0, min(1.0, continuity_score))
  414. return 0.5 # 没有足够信息,返回中性分数
  415. def _calculate_expected_balance(self, prev_balance: float, row: pd.Series) -> float:
  416. """计算预期余额"""
  417. if pd.isna(prev_balance):
  418. return None
  419. amount = row['txAmount']
  420. direction = row['txDirection']
  421. if direction == '收入':
  422. return prev_balance + amount
  423. elif direction == '支出':
  424. return prev_balance - amount
  425. else:
  426. return None
  427. def _factorial(self, n: int) -> int:
  428. """计算阶乘(用于评估排列数)"""
  429. result = 1
  430. for i in range(2, n + 1):
  431. result *= i
  432. return result
  433. # ==================== 其他方法 ====================
  434. def _format_result_for_llm(self, result: Dict[str, Any]) -> str:
  435. """将识别结果格式化为适合LLM理解的字符串"""
  436. # ... 保持原有实现不变
  437. pass
  438. def get_summary(self) -> Dict[str, Any]:
  439. """获取识别器摘要"""
  440. summary = super().get_summary()
  441. summary.update({
  442. 'balance_tolerance': self.balance_tolerance,
  443. 'enable_smart_sorting': self.enable_smart_sorting,
  444. 'max_permutation_search': self.max_permutation_search,
  445. 'data_loaded': self._data is not None
  446. })
  447. return summary