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