from pydantic import BaseModel, Field from typing import Dict, Any, Optional, Type, List import pandas as pd import re from .enhanced_base_recognizer import EnhancedBaseRecognizer class LowInterestRateInput(BaseModel): """低利率结息记录识别工具输入""" csv_path: Optional[str] = Field( None, description="CSV文件路径(可选)。如果初始化时已提供csv_path,可以不用再次传入。" ) class Config: arbitrary_types_allowed = True class LowInterestRateRecognizer(EnhancedBaseRecognizer): """ 低利率结息记录识别器 异常规则定义: 银行流水结息核查中,若实际结息金额对应的有效利率显著低于同档期银行公布的 活期存款基准利率,或低于账户所属银行同期执行的活期存款利率标准, 且无合理利率下浮依据,可判定为结息记录存在异常,需进一步核查。 核心逻辑: 1. 识别结息记录(txSummary包含关键词,txDirection为收入) 2. 估算年化利率:结息金额 ÷ 结息时点余额 × 年化系数 3. 对比配置的基准利率和阈值 4. 标记利率异常低的结息记录 """ args_schema: Type[BaseModel] = LowInterestRateInput # 配置参数 base_interest_rate: float = Field( 0.0035, description="基准活期存款利率(如0.35%应输入为0.0035)" ) threshold_ratio: float = Field( 0.5, description="异常阈值比例,实际利率低于基准利率的比例时视为异常" ) # 结息识别关键词 interest_keywords: List[str] = Field( ['结息', '利息', '存款利息'], description="识别结息记录的关键词列表" ) # 计息参数 assumed_interest_days: int = Field( 90, description="假设计息天数(通常为90天,按季结息)" ) annual_days: int = Field( 360, description="年计息天数(银行常用360天)" ) # 计算所需的最小余额 min_balance_for_calc: float = Field( 100.0, description="计算利率所需的最小余额(元),余额低于此值可能无法准确计算" ) # 严重程度配置 severity_level: str = Field( 'medium', description="异常严重程度(high/medium/low)" ) # 模糊匹配设置 enable_fuzzy_match: bool = Field( True, description="是否启用模糊匹配(处理不规范的txSummary字段)" ) def __init__(self, csv_path: str = None, config: Dict[str, Any] = None, **kwargs): """ 初始化低利率结息记录识别器 Args: csv_path: CSV文件路径 config: 配置参数 **kwargs: 其他参数 """ # 调用父类的 __init__ super().__init__( name="low_interest_rate_recognizer", description="识别银行流水中利率异常低的结息记录,检测结息金额对应的有效利率是否显著低于基准利率。", display_name="低利率结息记录识别器", csv_path=csv_path, config=config, **kwargs ) # 从config获取配置,更新Field属性 interest_config = self.get_config_value('interest_rate_check', {}) if interest_config: config_mapping = { 'base_interest_rate': 'base_interest_rate', 'threshold_ratio': 'threshold_ratio', 'interest_keywords': 'interest_keywords', 'assumed_interest_days': 'assumed_interest_days', 'annual_days': 'annual_days', 'min_balance_for_calc': 'min_balance_for_calc', 'severity_level': 'severity_level', 'enable_fuzzy_match': 'enable_fuzzy_match' } for config_key, attr_name in config_mapping.items(): if config_key in interest_config: setattr(self, attr_name, interest_config[config_key]) print(f"✅ {self.display_name} 初始化完成") print(f" 基准利率: {self.base_interest_rate:.4%}") print(f" 异常阈值: 低于基准的{self.threshold_ratio:.0%}") print(f" 识别关键词: {', '.join(self.interest_keywords)}") print(f" 假设计息天数: {self.assumed_interest_days}天") print(f" 年计息天数: {self.annual_days}天") print(f" 最小计算余额: ¥{self.min_balance_for_calc:,.2f}") print(f" 模糊匹配: {'启用' if self.enable_fuzzy_match else '禁用'}") def _is_interest_record(self, summary: str, direction: str) -> bool: """ 判断是否为结息记录 Args: summary: 交易摘要 direction: 交易方向 Returns: bool: 是否为结息记录 """ # 交易方向必须是收入 if direction != '收入': return False # 检查摘要是否包含结息关键词 summary_lower = str(summary).lower() if self.enable_fuzzy_match: # 模糊匹配:检查是否包含任何关键词 for keyword in self.interest_keywords: if keyword in summary_lower: return True return False else: # 精确匹配:使用正则表达式 pattern = '|'.join(self.interest_keywords) return bool(re.search(pattern, summary_lower)) def _estimate_annual_interest_rate(self, interest_amount: float, balance_at_interest: float) -> Optional[float]: """ 估算年化利率 公式: 年化利率 = (结息金额 ÷ 结息时点余额) × (年计息天数 ÷ 假设计息天数) Args: interest_amount: 结息金额(元) balance_at_interest: 结息时点余额(元) Returns: Optional[float]: 估算的年化利率,如果无法计算则返回None """ # 检查输入有效性 if pd.isna(interest_amount) or pd.isna(balance_at_interest): return None if balance_at_interest <= 0: return None if abs(interest_amount) < 0.01: # 结息金额过小 return None if balance_at_interest < self.min_balance_for_calc: return None try: # 计算日利率 daily_rate = interest_amount / balance_at_interest # 年化利率 annual_rate = daily_rate * (self.annual_days / self.assumed_interest_days) return annual_rate except ZeroDivisionError: return None except Exception: return None def _is_abnormal_interest(self, annual_rate: float) -> bool: """ 判断结息利率是否异常 Args: annual_rate: 估算的年化利率 Returns: bool: 是否异常 """ if annual_rate is None: return False # 判断是否低于阈值 threshold_rate = self.base_interest_rate * self.threshold_ratio return annual_rate < threshold_rate def _generate_interest_reason(self, row: pd.Series, annual_rate: float) -> str: """ 生成异常原因描述 Args: row: 交易记录 annual_rate: 估算的年化利率 Returns: str: 异常原因描述 """ interest_amount = row['txAmount'] balance = row.get('txBalance', 0) threshold_rate = self.base_interest_rate * self.threshold_ratio reason_parts = [] # 利率对比 if annual_rate is not None: rate_diff_percent = (self.base_interest_rate - annual_rate) / self.base_interest_rate * 100 reason_parts.append( f"估算年化利率{annual_rate:.4%},显著低于基准利率{self.base_interest_rate:.4%}" f"(低{rate_diff_percent:.1f}%,低于阈值{threshold_rate:.4%})" ) # 金额信息 reason_parts.append(f"结息金额¥{interest_amount:,.2f},结息时点余额¥{balance:,.2f}") # 补充信息 if annual_rate is not None and annual_rate < 0.0001: # 利率极低 reason_parts.append("利率极低,可能存在异常") return "结息利率异常: " + ",".join(reason_parts) 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" 检查规则: 结息记录的实际利率 < 基准利率({self.base_interest_rate:.4%}) × 阈值({self.threshold_ratio:.0%})") # 检查必需字段 required_fields = ['txId', 'txSummary', 'txDirection', 'txAmount'] 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}' } # 检查余额字段(可选但重要) has_balance_field = 'txBalance' in df.columns if not has_balance_field: print(f"⚠️ 警告: 缺少txBalance字段,将无法准确计算利率") print(f" 建议: 确保数据包含余额信息以进行准确的利率分析") # ============ 识别结息记录 ============ print(f"🔍 正在识别结息记录...") # 筛选可能的结息记录 interest_mask = df.apply( lambda row: self._is_interest_record(row['txSummary'], row['txDirection']), axis=1 ) interest_transactions = df[interest_mask].copy() if len(interest_transactions) == 0: print(f"📊 未发现结息记录") print(f" 检查的关键词: {', '.join(self.interest_keywords)}") print(f" 检查的交易方向: 收入") return { 'recognition_type': self.display_name, 'identified_count': 0, 'identified_anomalies': [], 'recognition_status': '完成', 'recognition_parameters': { 'base_interest_rate': self.base_interest_rate, 'threshold_ratio': self.threshold_ratio, 'interest_keywords': self.interest_keywords, 'assumed_interest_days': self.assumed_interest_days, 'total_checked': len(df) }, 'statistics': { 'total_transactions': len(df), 'interest_transactions': 0, 'has_balance_field': has_balance_field } } print(f"📊 发现 {len(interest_transactions)} 笔结息记录") # ============ 分析结息记录 ============ print(f"🔍 正在分析结息记录利率...") identified_anomalies = [] interest_analyses = [] for idx, row in interest_transactions.iterrows(): tx_id = str(row['txId']) interest_amount = float(row['txAmount']) balance = float(row.get('txBalance', 0)) if has_balance_field else 0 print(f" 🔍 分析结息记录 {tx_id}: ¥{interest_amount:,.2f}") # 1. 估算年化利率 annual_rate = None if has_balance_field and balance >= self.min_balance_for_calc: annual_rate = self._estimate_annual_interest_rate(interest_amount, balance) else: print(f" ⚠️ 无法计算利率: 余额不足或缺少余额字段") # 2. 判断是否异常 is_abnormal = False if annual_rate is not None: is_abnormal = self._is_abnormal_interest(annual_rate) # 记录分析结果 analysis = { 'tx_id': tx_id, 'interest_amount': interest_amount, 'balance_at_interest': balance if has_balance_field else None, 'estimated_annual_rate': annual_rate, 'is_abnormal': is_abnormal, 'can_calculate_rate': annual_rate is not None } interest_analyses.append(analysis) # 3. 如果异常,生成异常记录 if is_abnormal: # 生成异常原因 reason = self._generate_interest_reason(row, annual_rate) print(f" ❌ 发现利率异常: {reason[:80]}...") # 额外信息 additional_info = { 'interest_analysis': { 'estimated_annual_rate': annual_rate, 'base_interest_rate': self.base_interest_rate, 'threshold_rate': self.base_interest_rate * self.threshold_ratio, 'interest_amount': interest_amount, 'balance_at_interest': balance, 'rate_calculation': { 'assumed_interest_days': self.assumed_interest_days, 'annual_days': self.annual_days, 'min_balance_for_calc': self.min_balance_for_calc } } } anomaly = self.format_anomaly_record( row=row, reason=reason, severity=self.severity_level, check_type='low_interest_rate', **additional_info ) identified_anomalies.append(anomaly) elif annual_rate is not None: print(f" ✅ 利率正常: {annual_rate:.4%} ≥ 阈值{self.base_interest_rate * self.threshold_ratio:.4%}") else: print(f" ⚠️ 无法判断: 缺少余额数据或余额不足") # ============ 结果统计 ============ print(f"✅ {self.display_name}检查完成") print(f" 检查结果:") print(f" 结息记录总数: {len(interest_transactions)}") print(f" 可计算利率记录: {sum(1 for a in interest_analyses if a['can_calculate_rate'])}") print(f" 利率异常记录: {len(identified_anomalies)}") # 显示结息统计 if len(interest_transactions) > 0: print("📋 结息记录统计:") total_interest = interest_transactions['txAmount'].sum() avg_interest = interest_transactions['txAmount'].mean() print(f" 总结息金额: ¥{total_interest:,.2f}") print(f" 平均结息金额: ¥{avg_interest:,.2f}") # 显示利率分布 valid_rates = [a['estimated_annual_rate'] for a in interest_analyses if a['estimated_annual_rate'] is not None] if valid_rates: avg_rate = sum(valid_rates) / len(valid_rates) min_rate = min(valid_rates) max_rate = max(valid_rates) print(f" 平均估算利率: {avg_rate:.4%}") print(f" 最低估算利率: {min_rate:.4%}") print(f" 最高估算利率: {max_rate:.4%}") return { 'recognition_type': self.display_name, 'identified_count': len(identified_anomalies), 'identified_anomalies': identified_anomalies, 'recognition_status': '完成', 'recognition_parameters': { 'base_interest_rate': self.base_interest_rate, 'threshold_ratio': self.threshold_ratio, 'interest_keywords': self.interest_keywords, 'assumed_interest_days': self.assumed_interest_days, 'annual_days': self.annual_days, 'min_balance_for_calc': self.min_balance_for_calc, 'has_balance_field': has_balance_field, 'total_interest_records': len(interest_transactions) }, 'statistics': { 'total_transactions': len(df), 'interest_transactions': len(interest_transactions), 'abnormal_interest_transactions': len(identified_anomalies), 'interest_amount_statistics': { 'total_interest': float(interest_transactions['txAmount'].sum()), 'avg_interest': float(interest_transactions['txAmount'].mean()), 'max_interest': float(interest_transactions['txAmount'].max()), 'min_interest': float(interest_transactions['txAmount'].min()) } if len(interest_transactions) > 0 else {}, 'rate_analysis': { 'valid_rate_count': sum(1 for a in interest_analyses if a['can_calculate_rate']), 'abnormal_rate_count': len(identified_anomalies), 'rate_summary': { 'avg_rate': float(sum(a['estimated_annual_rate'] for a in interest_analyses if a['estimated_annual_rate'] is not None) / max(1, sum(1 for a in interest_analyses if a['estimated_annual_rate'] is not None))) } if any(a['estimated_annual_rate'] is not None for a in interest_analyses) else {} } } } 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 get_summary(self) -> Dict[str, Any]: """获取识别器摘要""" summary = super().get_summary() summary.update({ 'base_interest_rate': self.base_interest_rate, 'threshold_ratio': self.threshold_ratio, 'interest_keywords_count': len(self.interest_keywords), 'assumed_interest_days': self.assumed_interest_days, 'annual_days': self.annual_days, 'min_balance_for_calc': self.min_balance_for_calc, 'severity_level': self.severity_level, 'enable_fuzzy_match': self.enable_fuzzy_match, 'data_loaded': self._data is not None }) return summary def get_config_summary(self) -> Dict[str, Any]: """获取配置摘要""" return { "基准利率": f"{self.base_interest_rate:.4%}", "异常阈值": f"低于基准的{self.threshold_ratio:.0%}", "识别关键词": f"{len(self.interest_keywords)}个: {', '.join(self.interest_keywords[:3])}..." if len( self.interest_keywords) > 3 else f"{len(self.interest_keywords)}个: {', '.join(self.interest_keywords)}", "计息参数": f"{self.assumed_interest_days}天/季,{self.annual_days}天/年", "最小计算余额": f"¥{self.min_balance_for_calc:,.2f}", "检测逻辑": f"结息记录的实际利率 < {self.base_interest_rate:.4%} × {self.threshold_ratio:.0%} = 异常", "业务规则描述": "结息金额对应的有效利率显著低于基准利率,需核查结息真实性" }