from typing import Dict, Any, Optional, Type, List from datetime import datetime from pydantic import BaseModel, Field from .enhanced_base_recognizer import EnhancedBaseRecognizer class BalanceInfoMissingInput(BaseModel): """余额信息缺失识别工具输入""" csv_path: Optional[str] = Field( None, description="CSV文件路径(可选)。如果初始化时已提供csv_path,可以不用再次传入。" ) class BalanceInfoMissingRecognizer(EnhancedBaseRecognizer): """ 余额信息缺失异常识别器 异常规则定义: 银行流水若表头未列示余额信息,缺乏资金结余的关键展示项,会导致无法直观获取账户资金存量的核心数据, 不符合银行流水应完整呈现账户资金变动及结余情况的常规要求,属于信息完整性缺失的情形。 """ args_schema: Type[BaseModel] = BalanceInfoMissingInput # 需要检查的余额相关字段 balance_columns_to_check: List[str] = Field( ['txBalance'], description="需要检查的余额字段名称列表" ) # 严重程度配置 missing_severity: str = Field( 'high', description="余额信息缺失的严重程度(high/medium/low)" ) # 影响描述 impact_description: str = Field( "缺少余额信息将导致无法进行:1)余额连续性检查 2)资金存量分析 3)异常余额波动检测 4)交易真实性验证", description="余额信息缺失的影响描述" ) def __init__(self, csv_path: str = None, config: Dict[str, Any] = None, **kwargs): """ 初始化余额信息缺失识别器 Args: csv_path: CSV文件路径 config: 配置参数 **kwargs: 其他参数 """ super().__init__( name="balance_info_missing_recognizer", description="检查银行流水中是否缺少余额信息字段,识别信息完整性缺失异常。", display_name="余额信息缺失检查器", csv_path=csv_path, config=config, **kwargs ) # 从config获取配置 balance_missing_config = self.get_config_value('balance_missing_check', {}) if balance_missing_config: if 'balance_columns_to_check' in balance_missing_config: self.balance_columns_to_check = balance_missing_config['balance_columns_to_check'] if 'missing_severity' in balance_missing_config: self.missing_severity = balance_missing_config['missing_severity'] print(f"✅ {self.display_name} 初始化完成") print(f" 检查字段: {', '.join(self.balance_columns_to_check)}") print(f" 缺失严重度: {self.missing_severity}") def recognize(self, csv_path: str = None, **kwargs) -> Dict[str, Any]: """ 检查余额信息是否缺失 Args: csv_path: CSV文件路径 **kwargs: 其他参数 Returns: Dict[str, Any]: 识别结果 """ try: # 使用父类的load_data方法加载标准化数据 df = self.load_data(csv_path) print(f"🔍 {self.display_name}开始检查,共 {len(df)} 条记录") print(f" 检查规则: 银行流水应包含余额信息,缺失则属于信息完整性异常") # 检查所有可能的余额字段 columns_found = [] columns_missing = [] for col in self.balance_columns_to_check: if col in df.columns: columns_found.append(col) # 检查该列是否有有效数据(非空) non_null_count = df[col].notna().sum() print(f" ✓ 找到字段 '{col}',有效数据: {non_null_count}/{len(df)} 条") else: columns_missing.append(col) print(f" ✗ 缺少字段 '{col}'") # 判断是否有任何余额字段 has_any_balance_column = len(columns_found) > 0 # 如果没有找到任何余额字段,标记为异常 if not has_any_balance_column: print(f"❌ 严重异常: 未找到任何余额字段!") print(f" 缺少字段: {', '.join(columns_missing)}") print(f" 影响: {self.impact_description}") # 生成异常记录(系统级异常) anomaly = { 'txId': 'SYSTEM_BALANCE_INFO_MISSING', 'txDate': datetime.now().strftime('%Y-%m-%d'), 'txTime': datetime.now().strftime('%H:%M:%S'), 'datetime': datetime.now(), 'txAmount': 0.0, 'txDirection': '系统', 'recognition_reason': f"银行流水缺少余额信息字段,无法获取账户资金结余的核心数据。缺少的字段包括: {', '.join(columns_missing)}。", 'severity': self.missing_severity, 'status': '数据完整性异常', 'check_type': 'balance_info_missing', 'impact_analysis': self.impact_description, 'recommendation': '建议提供包含余额信息的完整银行流水,或至少包含以下字段之一: balance, txBalance, balance_after' } # 格式化异常记录 formatted_anomaly = self.format_anomaly_record( row=None, reason=anomaly['recognition_reason'], severity=anomaly['severity'], check_type=anomaly['check_type'], impact_analysis=anomaly['impact_analysis'], recommendation=anomaly['recommendation'], txId=anomaly['txId'], txDate=anomaly['txDate'], txTime=anomaly['txTime'], txAmount=anomaly['txAmount'], txDirection=anomaly['txDirection'] ) return { 'recognition_type': self.display_name, 'identified_count': 1, 'identified_anomalies': [formatted_anomaly], 'recognition_status': '完成', 'missing_columns_analysis': { 'has_any_balance_column': False, 'columns_found': columns_found, 'columns_missing': columns_missing, 'columns_checked': self.balance_columns_to_check, 'impact_level': '严重 - 无法进行余额相关分析' }, 'execution_info': { 'total_records': len(df), 'check_time': datetime.now().strftime('%Y-%m-%d %H:%M:%S'), 'data_file': csv_path or self.csv_path } } else: print(f"✅ 通过检查: 找到余额字段 {len(columns_found)} 个") print(f" 可用字段: {', '.join(columns_found)}") print(f" 缺失字段: {', '.join(columns_missing) if columns_missing else '无'}") # 分析找到的余额字段 balance_analysis = {} for col in columns_found: col_data = df[col] non_null_count = col_data.notna().sum() null_count = col_data.isna().sum() unique_count = col_data.nunique() balance_analysis[col] = { 'non_null_count': non_null_count, 'null_count': null_count, 'null_percentage': f"{(null_count / len(df) * 100):.1f}%", 'unique_values': unique_count, 'data_type': str(col_data.dtype) } return { 'recognition_type': self.display_name, 'identified_count': 0, 'identified_anomalies': [], 'recognition_status': '完成', 'missing_columns_analysis': { 'has_any_balance_column': True, 'columns_found': columns_found, 'columns_missing': columns_missing, 'columns_checked': self.balance_columns_to_check, 'impact_level': '正常 - 可进行余额相关分析', 'balance_analysis': balance_analysis }, 'execution_info': { 'total_records': len(df), 'check_time': datetime.now().strftime('%Y-%m-%d %H:%M:%S'), 'data_file': csv_path or self.csv_path } } except FileNotFoundError as e: error_result = { 'recognition_type': self.display_name, 'identified_count': 0, 'identified_anomalies': [], 'recognition_status': '失败', 'error': f'文件不存在: {str(e)}' } print(f"❌ 文件加载失败: {str(e)}") return error_result except Exception as e: import traceback traceback.print_exc() error_result = { 'recognition_type': self.display_name, 'identified_count': 0, 'identified_anomalies': [], 'recognition_status': '失败', 'error': f'检查失败: {str(e)}' } print(f"❌ 检查失败: {str(e)}") return error_result def get_summary(self) -> Dict[str, Any]: """获取识别器摘要""" summary = super().get_summary() summary.update({ 'balance_columns_to_check': self.balance_columns_to_check, 'missing_severity': self.missing_severity, 'check_type': '数据完整性检查', 'priority': 'high' }) return summary def get_config_summary(self) -> Dict[str, Any]: """获取配置摘要""" return { "检查字段": f"{len(self.balance_columns_to_check)}个: {', '.join(self.balance_columns_to_check[:3])}..." if len( self.balance_columns_to_check) > 3 else f"{len(self.balance_columns_to_check)}个: {', '.join(self.balance_columns_to_check)}", "缺失严重度": self.missing_severity.upper(), "检查优先级": "高", "异常规则": "银行流水缺少余额信息字段属于信息完整性缺失", "影响分析": "无法进行余额连续性、资金存量分析等关键检查" }