| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250 |
- 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(),
- "检查优先级": "高",
- "异常规则": "银行流水缺少余额信息字段属于信息完整性缺失",
- "影响分析": "无法进行余额连续性、资金存量分析等关键检查"
- }
|