from pydantic import BaseModel, Field from typing import Dict, Any, Optional, Type import pandas as pd from itertools import permutations from .enhanced_base_recognizer import EnhancedBaseRecognizer class BalanceRecognitionInput(BaseModel): """余额识别工具输入""" csv_path: Optional[str] = Field( None, description="CSV文件路径(可选)。如果初始化时已提供csv_path,可以不用再次传入。" ) class Config: arbitrary_types_allowed = True class BalanceContinuityRecognizer(EnhancedBaseRecognizer): """余额连续性异常识别器(带智能排序)""" args_schema: Type[BaseModel] = BalanceRecognitionInput # 配置参数 balance_tolerance: float = Field( 0.01, description="余额计算容差,允许的余额差异阈值" ) enable_smart_sorting: bool = Field( True, description="是否启用智能排序处理时间相同的交易" ) max_permutation_search: int = Field( 6, # 3! = 6, 4! = 24, 设置为6可以处理最多3笔时间相同的交易 description="最大排列搜索数,防止组合爆炸" ) def __init__(self, csv_path: str = None, config: Dict[str, Any] = None, **kwargs): """ 初始化余额连续性异常识别器 Args: csv_path: CSV文件路径 config: 配置参数 **kwargs: 其他参数 """ # 调用父类的 __init__ super().__init__( name="balance_continuity_recognizer", description="识别银行流水中的余额连续性异常,检查每笔交易后的余额计算是否正确。" "支持智能排序处理时间相同的交易。", display_name="余额连续性异常识别", csv_path=csv_path, config=config, **kwargs ) # 从config获取配置,更新Field属性 balance_config = self.get_config_value('balance_recognition', {}) if balance_config: if 'balance_tolerance' in balance_config: self.balance_tolerance = balance_config['balance_tolerance'] if 'enable_smart_sorting' in balance_config: self.enable_smart_sorting = balance_config['enable_smart_sorting'] if 'max_permutation_search' in balance_config: self.max_permutation_search = balance_config['max_permutation_search'] def recognize(self, csv_path: str = None, **kwargs) -> Dict[str, Any]: """识别余额连续性异常(带智能排序)""" try: # 使用父类的load_data方法加载标准化数据 df = self.load_data(csv_path) print(f"🔍 {self.display_name}开始检查余额连续性,共 {len(df)} 条记录") print(f" 余额容差: {self.balance_tolerance}") print(f" 智能排序: {'启用' if self.enable_smart_sorting else '禁用'}") # 检查必需字段 required_fields = ['txId', 'txDate', 'txTime', 'txAmount', 'txDirection', 'txBalance'] 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}' } # ============ 智能排序处理 ============ if self.enable_smart_sorting and 'datetime' in df.columns: df = self._apply_smart_sorting(df) else: # 简单排序:按时间,时间相同按ID if 'datetime' in df.columns: df = df.sort_values(['datetime', 'txId']) else: df = df.sort_values(['txDate', 'txTime', 'txId']) print("📋 排序后的交易顺序:") for i, (_, row) in enumerate(df.head(10).iterrows(), 1): time_str = row['datetime'].strftime( "%Y-%m-%d %H:%M:%S") if 'datetime' in row else f"{row['txDate']} {row['txTime']}" dir_symbol = "→" if row['txDirection'] == '支出' else "←" # 现在 txId 已经是字符串,可以直接使用 # 但为了代码清晰,可以明确标注 tx_id = row['txId'] # 已经是字符串 print( f" {i:2d}. ID:{tx_id:>4s} | {time_str} | {dir_symbol} {row['txAmount']:8.2f} | 余额:{row['txBalance']:8.2f}") # ============ 开始余额连续性检查 ============ identified_anomalies = [] prev_balance = None valid_transactions = 0 for idx, row in df.iterrows(): tx_id = row['txId'] current_balance = row['txBalance'] valid_transactions += 1 # 检查1:余额是否为空 if pd.isna(current_balance): anomaly = self.format_anomaly_record( row=row, reason='余额字段为空', severity='high', check_type='missing_balance', previous_balance=prev_balance ) identified_anomalies.append(anomaly) print(f" ❌ 交易ID {tx_id}: 余额字段为空") continue # 检查2:余额连续性(如果不是第一条记录) if prev_balance is not None: amount = row['txAmount'] direction = str(row['txDirection']).strip() # 计算预期余额 if direction == '收入': expected_balance = prev_balance + amount elif direction == '支出': expected_balance = prev_balance - amount else: # 未知方向,跳过检查 print(f" ⚠️ 交易ID {tx_id}: 未知的交易方向 '{direction}',跳过余额检查") prev_balance = current_balance continue # 检查余额是否连续(允许小误差) if pd.isna(expected_balance): # 预期余额计算异常 anomaly = self.format_anomaly_record( row=row, reason=f'预期余额计算异常,可能金额字段有问题: amount={amount}', severity='high', check_type='calculation_error', previous_balance=prev_balance, expected_balance=expected_balance, actual_balance=current_balance ) identified_anomalies.append(anomaly) print(f" ❌ 交易ID {tx_id}: 预期余额计算异常") else: balance_diff = abs(expected_balance - current_balance) if balance_diff > self.balance_tolerance: anomaly = self.format_anomaly_record( row=row, reason=f'余额计算不连续,预期{expected_balance:.2f},实际{current_balance:.2f},差异{balance_diff:.2f}', severity='high', check_type='balance_discontinuity', previous_balance=prev_balance, expected_balance=expected_balance, actual_balance=current_balance, balance_difference=balance_diff, tolerance_threshold=self.balance_tolerance ) identified_anomalies.append(anomaly) print(f" ❌ 交易ID {tx_id}: 余额不连续,差异 {balance_diff:.2f}") else: # 余额连续,正常情况 pass prev_balance = current_balance print(f"✅ {self.display_name}检查完成") print(f" 检查交易数: {valid_transactions}") print(f" 发现异常数: {len(identified_anomalies)}") # 统计不同类型异常的数量 missing_balance_count = len([a for a in identified_anomalies if a.get('additional_info', {}).get('check_type') == 'missing_balance']) discontinuity_count = len([a for a in identified_anomalies if a.get('additional_info', {}).get('check_type') == 'balance_discontinuity']) calculation_error_count = len([a for a in identified_anomalies if a.get('additional_info', {}).get('check_type') == 'calculation_error']) return { 'recognition_type': self.display_name, 'identified_count': len(identified_anomalies), 'identified_anomalies': identified_anomalies, 'recognition_status': '完成', 'recognition_parameters': { 'balance_tolerance': self.balance_tolerance, 'enable_smart_sorting': self.enable_smart_sorting, 'checked_transactions': valid_transactions, 'data_source': csv_path or self._csv_path }, 'statistics': { 'missing_balance_count': missing_balance_count, 'discontinuity_count': discontinuity_count, 'calculation_error_count': calculation_error_count, 'first_valid_balance': float(df['txBalance'].iloc[0]) if len(df) > 0 and not pd.isna( df['txBalance'].iloc[0]) else None, 'last_valid_balance': float(df['txBalance'].iloc[-1]) if len(df) > 0 and not pd.isna( df['txBalance'].iloc[-1]) else None, 'total_transactions': len(df), 'valid_balance_count': df['txBalance'].notna().sum(), 'avg_balance': float(df['txBalance'].mean()) if df['txBalance'].notna().any() else None } } 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 _apply_smart_sorting(self, df: pd.DataFrame) -> pd.DataFrame: """ 应用智能排序算法 处理步骤: 1. 按时间分组 2. 对每个时间组内的交易进行智能排序 3. 合并所有组 """ print("🧠 应用智能排序算法...") # 首先按时间排序,得到时间组 df = df.sort_values('datetime') # 找出所有时间相同的交易组 time_groups = list(df.groupby('datetime')) if len(time_groups) == len(df): print(" ✅ 所有交易时间都不同,无需智能排序") return df # 处理每个时间组 sorted_groups = [] prev_group_last_balance = None for i, (time_val, group) in enumerate(time_groups): group_size = len(group) if group_size == 1: # 单笔交易,直接加入 sorted_groups.append(group) if not group['txBalance'].isna().iloc[0]: prev_group_last_balance = group['txBalance'].iloc[0] continue # 多笔交易时间相同,需要智能排序 print(f" 🔍 时间组 {i + 1}/{len(time_groups)}: {time_val},共 {group_size} 笔交易") # 获取前一组的最后一笔余额(如果有) prev_balance = prev_group_last_balance # 智能排序这个组 sorted_group = self._smart_sort_time_group(group, prev_balance) sorted_groups.append(sorted_group) # 更新前一组的最后一笔余额 if not sorted_group['txBalance'].isna().iloc[-1]: prev_group_last_balance = sorted_group['txBalance'].iloc[-1] # 合并所有组 result_df = pd.concat(sorted_groups, ignore_index=True) print(f" ✅ 智能排序完成,处理了 {len(time_groups)} 个时间组") return result_df def _smart_sort_time_group(self, group: pd.DataFrame, prev_balance: float = None) -> pd.DataFrame: """ 智能排序一个时间组内的交易 策略: 1. 如果组内交易数 <= 3,尝试所有排列 2. 如果更多,使用启发式算法 """ group_size = len(group) if group_size == 0: return group # 显示组内交易详情 print(f" 组内交易详情:") for idx, (_, row) in enumerate(group.iterrows(), 1): dir_symbol = "→" if row['txDirection'] == '支出' else "←" balance_info = f"余额:{row['txBalance']:.2f}" if not pd.isna(row['txBalance']) else "余额:None" print(f" {idx}. ID:{row['txId']} {dir_symbol} {row['txAmount']:.2f} {balance_info}") # 策略1:少量交易,尝试所有排列 if group_size <= 3: return self._try_all_permutations(group, prev_balance) # 策略2:较多交易,使用启发式算法 else: return self._heuristic_sort(group, prev_balance) def _try_all_permutations(self, group: pd.DataFrame, prev_balance: float = None) -> pd.DataFrame: """ 尝试所有可能的排列,选择最优的 适用于少量交易(<=3笔) """ group_size = len(group) print(f" 尝试 {group_size} 笔交易的所有排列 ({self._factorial(group_size)} 种可能)...") # 如果是2笔交易,特殊处理(常见情况) if group_size == 2: return self._optimize_two_transactions(group, prev_balance) # 生成所有排列 best_order = None best_score = float('-inf') # 限制最大尝试数,防止组合爆炸 max_tries = min(self.max_permutation_search, self._factorial(group_size)) permutations_tried = 0 for perm_indices in permutations(range(group_size)): if permutations_tried >= max_tries: break perm_group = group.iloc[list(perm_indices)].reset_index(drop=True) score = self._evaluate_order_quality(perm_group, prev_balance) if score > best_score: best_score = score best_order = perm_group permutations_tried += 1 if best_order is not None: print(f" 找到最优排列,质量评分: {best_score:.2f}") if best_score < 0.5: print(f" ⚠️ 警告:最优排列质量评分较低 ({best_score:.2f})") # 显示最优顺序 print(f" 最优顺序:") for idx, (_, row) in enumerate(best_order.iterrows(), 1): dir_symbol = "→" if row['txDirection'] == '支出' else "←" print(f" {idx}. ID:{row['txId']} {dir_symbol} {row['txAmount']:.2f}") return best_order return group def _optimize_two_transactions(self, group: pd.DataFrame, prev_balance: float = None) -> pd.DataFrame: """ 优化两笔时间相同交易的顺序 这是最常见的情况,专门优化 """ if len(group) != 2: return group row1, row2 = group.iloc[0], group.iloc[1] # 计算两种顺序的质量评分 order1 = pd.DataFrame([row1, row2]) # 原始顺序 order2 = pd.DataFrame([row2, row1]) # 反转顺序 score1 = self._evaluate_order_quality(order1, prev_balance) score2 = self._evaluate_order_quality(order2, prev_balance) print(f" 顺序1 (ID {row1['txId']}→{row2['txId']}): 评分 {score1:.2f}") print(f" 顺序2 (ID {row2['txId']}→{row1['txId']}): 评分 {score2:.2f}") if score2 > score1: print(f" ✅ 选择顺序2: ID {row2['txId']} → ID {row1['txId']}") return order2 else: print(f" ✅ 选择顺序1: ID {row1['txId']} → ID {row2['txId']}") return order1 def _heuristic_sort(self, group: pd.DataFrame, prev_balance: float = None) -> pd.DataFrame: """ 启发式排序算法 适用于较多交易(>3笔) 启发式规则: 1. 先处理支出,后处理收入(常见模式) 2. 金额大的优先 3. 余额连续性验证 """ print(f" 使用启发式排序 ({len(group)} 笔交易)...") group = group.copy() # 启发式1:按交易方向排序 group['sort_direction'] = group['txDirection'].map({'支出': 0, '收入': 1}) # 启发式2:按金额排序(支出从大到小,收入从小到大) def get_amount_sort_key(row): if row['txDirection'] == '支出': return -row['txAmount'] # 支出金额大的优先 else: return row['txAmount'] # 收入金额小的优先 group['sort_amount'] = group.apply(get_amount_sort_key, axis=1) # 排序 sorted_group = group.sort_values(['sort_direction', 'sort_amount', 'txId']).drop( ['sort_direction', 'sort_amount'], axis=1) # 验证排序质量 score = self._evaluate_order_quality(sorted_group, prev_balance) print(f" 启发式排序质量评分: {score:.2f}") if score < 0.3: print(f" ⚠️ 启发式排序质量较低,考虑使用原始顺序") return group.drop(['sort_direction', 'sort_amount'], axis=1) return sorted_group def _evaluate_order_quality(self, ordered_group: pd.DataFrame, start_balance: float = None) -> float: """ 评估排序质量 基于余额连续性计算质量评分 返回0-1之间的分数,越高越好 """ if len(ordered_group) == 0: return 0.0 current_balance = start_balance total_score = 0.0 valid_checks = 0 for _, row in ordered_group.iterrows(): if pd.isna(row.get('txBalance')): # 缺失余额,无法评估 continue if current_balance is not None: # 计算预期余额 expected = self._calculate_expected_balance(current_balance, row) if expected is not None: diff = abs(expected - row['txBalance']) if diff <= self.balance_tolerance: total_score += 1.0 # 完美匹配 elif diff <= self.balance_tolerance * 10: # 允许10倍容差 total_score += 0.5 # 部分匹配 else: total_score -= 0.5 # 严重不匹配 valid_checks += 1 # 更新当前余额 current_balance = row['txBalance'] # 归一化分数 if valid_checks > 0: # 基础分数是余额连续性分数 continuity_score = total_score / valid_checks # 额外加分:如果整个组的总金额与余额变化匹配 if start_balance is not None and not ordered_group['txBalance'].isna().all(): final_balance = ordered_group['txBalance'].iloc[-1] total_change = sum( row['txAmount'] if row['txDirection'] == '收入' else -row['txAmount'] for _, row in ordered_group.iterrows() ) expected_final = start_balance + total_change final_diff = abs(expected_final - final_balance) if final_diff <= self.balance_tolerance: continuity_score += 0.2 # 额外加分 elif final_diff <= self.balance_tolerance * 10: continuity_score += 0.1 # 确保分数在0-1之间 return max(0.0, min(1.0, continuity_score)) return 0.5 # 没有足够信息,返回中性分数 def _calculate_expected_balance(self, prev_balance: float, row: pd.Series) -> float: """计算预期余额""" if pd.isna(prev_balance): return None amount = row['txAmount'] direction = row['txDirection'] if direction == '收入': return prev_balance + amount elif direction == '支出': return prev_balance - amount else: return None def _factorial(self, n: int) -> int: """计算阶乘(用于评估排列数)""" result = 1 for i in range(2, n + 1): result *= i return result # ==================== 其他方法 ==================== def _format_result_for_llm(self, result: Dict[str, Any]) -> str: """将识别结果格式化为适合LLM理解的字符串""" # ... 保持原有实现不变 pass def get_summary(self) -> Dict[str, Any]: """获取识别器摘要""" summary = super().get_summary() summary.update({ 'balance_tolerance': self.balance_tolerance, 'enable_smart_sorting': self.enable_smart_sorting, 'max_permutation_search': self.max_permutation_search, 'data_loaded': self._data is not None }) return summary