| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509 |
- 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%} = 异常",
- "业务规则描述": "结息金额对应的有效利率显著低于基准利率,需核查结息真实性"
- }
|