balance_info_missing_recognizer.py 11 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250
  1. from typing import Dict, Any, Optional, Type, List
  2. from datetime import datetime
  3. from pydantic import BaseModel, Field
  4. from .enhanced_base_recognizer import EnhancedBaseRecognizer
  5. class BalanceInfoMissingInput(BaseModel):
  6. """余额信息缺失识别工具输入"""
  7. csv_path: Optional[str] = Field(
  8. None,
  9. description="CSV文件路径(可选)。如果初始化时已提供csv_path,可以不用再次传入。"
  10. )
  11. class BalanceInfoMissingRecognizer(EnhancedBaseRecognizer):
  12. """
  13. 余额信息缺失异常识别器
  14. 异常规则定义:
  15. 银行流水若表头未列示余额信息,缺乏资金结余的关键展示项,会导致无法直观获取账户资金存量的核心数据,
  16. 不符合银行流水应完整呈现账户资金变动及结余情况的常规要求,属于信息完整性缺失的情形。
  17. """
  18. args_schema: Type[BaseModel] = BalanceInfoMissingInput
  19. # 需要检查的余额相关字段
  20. balance_columns_to_check: List[str] = Field(
  21. ['txBalance'],
  22. description="需要检查的余额字段名称列表"
  23. )
  24. # 严重程度配置
  25. missing_severity: str = Field(
  26. 'high',
  27. description="余额信息缺失的严重程度(high/medium/low)"
  28. )
  29. # 影响描述
  30. impact_description: str = Field(
  31. "缺少余额信息将导致无法进行:1)余额连续性检查 2)资金存量分析 3)异常余额波动检测 4)交易真实性验证",
  32. description="余额信息缺失的影响描述"
  33. )
  34. def __init__(self, csv_path: str = None, config: Dict[str, Any] = None, **kwargs):
  35. """
  36. 初始化余额信息缺失识别器
  37. Args:
  38. csv_path: CSV文件路径
  39. config: 配置参数
  40. **kwargs: 其他参数
  41. """
  42. super().__init__(
  43. name="balance_info_missing_recognizer",
  44. description="检查银行流水中是否缺少余额信息字段,识别信息完整性缺失异常。",
  45. display_name="余额信息缺失检查器",
  46. csv_path=csv_path,
  47. config=config,
  48. **kwargs
  49. )
  50. # 从config获取配置
  51. balance_missing_config = self.get_config_value('balance_missing_check', {})
  52. if balance_missing_config:
  53. if 'balance_columns_to_check' in balance_missing_config:
  54. self.balance_columns_to_check = balance_missing_config['balance_columns_to_check']
  55. if 'missing_severity' in balance_missing_config:
  56. self.missing_severity = balance_missing_config['missing_severity']
  57. print(f"✅ {self.display_name} 初始化完成")
  58. print(f" 检查字段: {', '.join(self.balance_columns_to_check)}")
  59. print(f" 缺失严重度: {self.missing_severity}")
  60. def recognize(self, csv_path: str = None, **kwargs) -> Dict[str, Any]:
  61. """
  62. 检查余额信息是否缺失
  63. Args:
  64. csv_path: CSV文件路径
  65. **kwargs: 其他参数
  66. Returns:
  67. Dict[str, Any]: 识别结果
  68. """
  69. try:
  70. # 使用父类的load_data方法加载标准化数据
  71. df = self.load_data(csv_path)
  72. print(f"🔍 {self.display_name}开始检查,共 {len(df)} 条记录")
  73. print(f" 检查规则: 银行流水应包含余额信息,缺失则属于信息完整性异常")
  74. # 检查所有可能的余额字段
  75. columns_found = []
  76. columns_missing = []
  77. for col in self.balance_columns_to_check:
  78. if col in df.columns:
  79. columns_found.append(col)
  80. # 检查该列是否有有效数据(非空)
  81. non_null_count = df[col].notna().sum()
  82. print(f" ✓ 找到字段 '{col}',有效数据: {non_null_count}/{len(df)} 条")
  83. else:
  84. columns_missing.append(col)
  85. print(f" ✗ 缺少字段 '{col}'")
  86. # 判断是否有任何余额字段
  87. has_any_balance_column = len(columns_found) > 0
  88. # 如果没有找到任何余额字段,标记为异常
  89. if not has_any_balance_column:
  90. print(f"❌ 严重异常: 未找到任何余额字段!")
  91. print(f" 缺少字段: {', '.join(columns_missing)}")
  92. print(f" 影响: {self.impact_description}")
  93. # 生成异常记录(系统级异常)
  94. anomaly = {
  95. 'txId': 'SYSTEM_BALANCE_INFO_MISSING',
  96. 'txDate': datetime.now().strftime('%Y-%m-%d'),
  97. 'txTime': datetime.now().strftime('%H:%M:%S'),
  98. 'datetime': datetime.now(),
  99. 'txAmount': 0.0,
  100. 'txDirection': '系统',
  101. 'recognition_reason': f"银行流水缺少余额信息字段,无法获取账户资金结余的核心数据。缺少的字段包括: {', '.join(columns_missing)}。",
  102. 'severity': self.missing_severity,
  103. 'status': '数据完整性异常',
  104. 'check_type': 'balance_info_missing',
  105. 'impact_analysis': self.impact_description,
  106. 'recommendation': '建议提供包含余额信息的完整银行流水,或至少包含以下字段之一: balance, txBalance, balance_after'
  107. }
  108. # 格式化异常记录
  109. formatted_anomaly = self.format_anomaly_record(
  110. row=None,
  111. reason=anomaly['recognition_reason'],
  112. severity=anomaly['severity'],
  113. check_type=anomaly['check_type'],
  114. impact_analysis=anomaly['impact_analysis'],
  115. recommendation=anomaly['recommendation'],
  116. txId=anomaly['txId'],
  117. txDate=anomaly['txDate'],
  118. txTime=anomaly['txTime'],
  119. txAmount=anomaly['txAmount'],
  120. txDirection=anomaly['txDirection']
  121. )
  122. return {
  123. 'recognition_type': self.display_name,
  124. 'identified_count': 1,
  125. 'identified_anomalies': [formatted_anomaly],
  126. 'recognition_status': '完成',
  127. 'missing_columns_analysis': {
  128. 'has_any_balance_column': False,
  129. 'columns_found': columns_found,
  130. 'columns_missing': columns_missing,
  131. 'columns_checked': self.balance_columns_to_check,
  132. 'impact_level': '严重 - 无法进行余额相关分析'
  133. },
  134. 'execution_info': {
  135. 'total_records': len(df),
  136. 'check_time': datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
  137. 'data_file': csv_path or self.csv_path
  138. }
  139. }
  140. else:
  141. print(f"✅ 通过检查: 找到余额字段 {len(columns_found)} 个")
  142. print(f" 可用字段: {', '.join(columns_found)}")
  143. print(f" 缺失字段: {', '.join(columns_missing) if columns_missing else '无'}")
  144. # 分析找到的余额字段
  145. balance_analysis = {}
  146. for col in columns_found:
  147. col_data = df[col]
  148. non_null_count = col_data.notna().sum()
  149. null_count = col_data.isna().sum()
  150. unique_count = col_data.nunique()
  151. balance_analysis[col] = {
  152. 'non_null_count': non_null_count,
  153. 'null_count': null_count,
  154. 'null_percentage': f"{(null_count / len(df) * 100):.1f}%",
  155. 'unique_values': unique_count,
  156. 'data_type': str(col_data.dtype)
  157. }
  158. return {
  159. 'recognition_type': self.display_name,
  160. 'identified_count': 0,
  161. 'identified_anomalies': [],
  162. 'recognition_status': '完成',
  163. 'missing_columns_analysis': {
  164. 'has_any_balance_column': True,
  165. 'columns_found': columns_found,
  166. 'columns_missing': columns_missing,
  167. 'columns_checked': self.balance_columns_to_check,
  168. 'impact_level': '正常 - 可进行余额相关分析',
  169. 'balance_analysis': balance_analysis
  170. },
  171. 'execution_info': {
  172. 'total_records': len(df),
  173. 'check_time': datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
  174. 'data_file': csv_path or self.csv_path
  175. }
  176. }
  177. except FileNotFoundError as e:
  178. error_result = {
  179. 'recognition_type': self.display_name,
  180. 'identified_count': 0,
  181. 'identified_anomalies': [],
  182. 'recognition_status': '失败',
  183. 'error': f'文件不存在: {str(e)}'
  184. }
  185. print(f"❌ 文件加载失败: {str(e)}")
  186. return error_result
  187. except Exception as e:
  188. import traceback
  189. traceback.print_exc()
  190. error_result = {
  191. 'recognition_type': self.display_name,
  192. 'identified_count': 0,
  193. 'identified_anomalies': [],
  194. 'recognition_status': '失败',
  195. 'error': f'检查失败: {str(e)}'
  196. }
  197. print(f"❌ 检查失败: {str(e)}")
  198. return error_result
  199. def get_summary(self) -> Dict[str, Any]:
  200. """获取识别器摘要"""
  201. summary = super().get_summary()
  202. summary.update({
  203. 'balance_columns_to_check': self.balance_columns_to_check,
  204. 'missing_severity': self.missing_severity,
  205. 'check_type': '数据完整性检查',
  206. 'priority': 'high'
  207. })
  208. return summary
  209. def get_config_summary(self) -> Dict[str, Any]:
  210. """获取配置摘要"""
  211. return {
  212. "检查字段": f"{len(self.balance_columns_to_check)}个: {', '.join(self.balance_columns_to_check[:3])}..." if len(
  213. self.balance_columns_to_check) > 3 else f"{len(self.balance_columns_to_check)}个: {', '.join(self.balance_columns_to_check)}",
  214. "缺失严重度": self.missing_severity.upper(),
  215. "检查优先级": "高",
  216. "异常规则": "银行流水缺少余额信息字段属于信息完整性缺失",
  217. "影响分析": "无法进行余额连续性、资金存量分析等关键检查"
  218. }