from pydantic import BaseModel, Field from typing import Dict, Any, Optional, Type, List, Tuple import pandas as pd from datetime import timedelta from .enhanced_base_recognizer import EnhancedBaseRecognizer class OverBookTransactionInput(BaseModel): """疑似过账流水识别工具输入""" csv_path: Optional[str] = Field( None, description="CSV文件路径(可选)。如果初始化时已提供csv_path,可以不用再次传入。" ) class Config: arbitrary_types_allowed = True class OverBookTransactionRecognizer(EnhancedBaseRecognizer): """ 疑似过账流水识别器 异常规则定义: 账户在接收大额资金入账后,7个自然日内即发生与该入账金额完全一致(或高度接近)的资金流出, 形成"入账-流出"的闭环资金流动,且缺乏合理商业背景、实际业务往来支撑或真实收付需求, 资金未发生实质性使用或流转,仅通过账户完成过渡划转,符合过账交易核心属性。 核心逻辑: 1. 筛选≥阈值金额的"收入"交易 2. 查找每笔大额收入后7天内的"支出"交易 3. 匹配金额(±容忍度范围内) 4. 分析交易背景合理性 5. 标记疑似过账的交易对 """ args_schema: Type[BaseModel] = OverBookTransactionInput # 配置参数 amount_threshold: float = Field( 100000.0, description="金额阈值(元),交易金额≥此值才进行检测" ) time_window_days: int = Field( 7, description="时间窗口(天),从收入发生日开始计算" ) amount_tolerance: float = Field( 0.01, description="金额容忍度,±此比例内视为金额匹配" ) min_stay_time_hours: int = Field( 1, description="最小停留时间(小时),避免即时进出被视为过账" ) # 合理性判断参数 enable_background_check: bool = Field( True, description="是否启用交易背景合理性检查" ) reasonable_background_keywords: List[str] = Field( [ "工资发放", "奖金发放", "绩效发放", "报销款", "货款", "租金收入", "投资款", "贷款", "还款", "采购付款", "支付货款", "缴税", "缴费", "消费", "日常支出", "生活支出", "业务往来", "贸易款" ], description="合理业务背景关键词列表" ) high_risk_keywords: List[str] = Field( [ "过账", "过渡", "走账", "倒账", "资金划转", "临时周转", "无实际业务", "过渡资金", "资金过桥", "代收代付", "代转", "垫资", "拆借", "内部往来" ], description="高风险关键词(过账特征)列表" ) # 交易对手分析 enable_counterparty_check: bool = Field( True, description="是否启用交易对手关联性检查" ) # 模式检测配置 detect_single_pair: bool = Field( True, description="检测单笔流入-流出对" ) detect_split_pattern: bool = Field( True, description="检测拆分过账(一笔流入多笔流出)" ) detect_merge_pattern: bool = Field( True, description="检测合并过账(多笔流入一笔流出)" ) # 严重程度配置 severity_level: str = Field( 'high', description="异常严重程度(high/medium/low)" ) def __init__(self, csv_path: str = None, config: Dict[str, Any] = None, **kwargs): """ 初始化疑似过账流水识别器 Args: csv_path: CSV文件路径 config: 配置参数 **kwargs: 其他参数 """ # 调用父类的 __init__ super().__init__( name="over_book_transaction_recognizer", description="识别疑似过账流水:大额资金短期内相同金额进出,缺乏真实业务背景。", display_name="疑似过账流水识别器", csv_path=csv_path, config=config, **kwargs ) # 从config获取配置,更新Field属性 overbook_config = self.get_config_value('over_book_transaction_recognition', {}) if overbook_config: config_mapping = { 'amount_threshold': 'amount_threshold', 'time_window_days': 'time_window_days', 'amount_tolerance': 'amount_tolerance', 'min_stay_time_hours': 'min_stay_time_hours', 'enable_background_check': 'enable_background_check', 'reasonable_background_keywords': 'reasonable_background_keywords', 'high_risk_keywords': 'high_risk_keywords', 'enable_counterparty_check': 'enable_counterparty_check', 'detect_single_pair': 'detect_single_pair', 'detect_split_pattern': 'detect_split_pattern', 'detect_merge_pattern': 'detect_merge_pattern', 'severity_level': 'severity_level' } for config_key, attr_name in config_mapping.items(): if config_key in overbook_config: setattr(self, attr_name, overbook_config[config_key]) print(f"✅ {self.display_name} 初始化完成") print(f" 金额阈值: ¥{self.amount_threshold:,.2f}") print(f" 时间窗口: {self.time_window_days}天") print(f" 金额容忍度: ±{self.amount_tolerance:.1%}") print(f" 最小停留时间: {self.min_stay_time_hours}小时") print(f" 背景检查: {'启用' if self.enable_background_check else '禁用'}") print(f" 对手方检查: {'启用' if self.enable_counterparty_check else '禁用'}") print(f" 检测模式: 单笔匹配/拆分/合并") print(f" 异常严重程度: {self.severity_level.upper()}") def _is_large_inflow(self, row: pd.Series) -> bool: """ 判断是否为需要检测的大额收入 Args: row: 交易记录 Returns: bool: 是否为大额收入 """ # 必须是收入方向 if row.get('txDirection') != '收入': return False # 金额必须达到阈值 amount = row.get('txAmount', 0) if pd.isna(amount) or amount < self.amount_threshold: return False return True def _find_matching_outflows(self, inflow: pd.Series, df: pd.DataFrame) -> List[pd.Series]: """ 查找匹配的流出交易 Args: inflow: 大额收入记录 df: 完整数据集 Returns: List[pd.Series]: 匹配的流出交易列表 """ if pd.isna(inflow.get('datetime')): return [] inflow_time = inflow['datetime'] inflow_amount = inflow['txAmount'] # 计算时间窗口 time_end = inflow_time + timedelta(days=self.time_window_days) # 筛选条件:时间窗口内、支出方向、金额匹配 mask = ( (df['datetime'] > inflow_time) & # 晚于流入时间 (df['datetime'] <= time_end) & # 在时间窗口内 (df['txDirection'] == '支出') & # 支出方向 (df['txId'] != inflow['txId']) # 排除同一笔交易 ) candidate_outflows = df[mask].copy() # 金额匹配检查 matching_outflows = [] for _, outflow in candidate_outflows.iterrows(): outflow_amount = outflow['txAmount'] # 检查金额是否匹配(考虑容忍度) amount_ratio = outflow_amount / inflow_amount if abs(amount_ratio - 1.0) <= self.amount_tolerance: matching_outflows.append(outflow) return matching_outflows def _analyze_background_reasonableness(self, inflow: pd.Series, outflow: pd.Series) -> Tuple[bool, str]: """ 分析交易背景合理性 Args: inflow: 流入交易记录 outflow: 流出交易记录 Returns: Tuple[bool, str]: (是否合理, 合理性描述) """ if not self.enable_background_check: return True, "背景检查已禁用" inflow_summary = str(inflow.get('txSummary', '')).lower() outflow_summary = str(outflow.get('txSummary', '')).lower() inflow_counterparty = str(inflow.get('txCounterparty', '')).lower() outflow_counterparty = str(outflow.get('txCounterparty', '')).lower() reasons = [] is_reasonable = True # 1. 检查高风险关键词 for keyword in self.high_risk_keywords: if keyword in inflow_summary or keyword in outflow_summary: reasons.append(f"包含高风险关键词: '{keyword}'") is_reasonable = False # 2. 检查合理背景关键词 has_reasonable_keyword = False for keyword in self.reasonable_background_keywords: if keyword in inflow_summary or keyword in outflow_summary: has_reasonable_keyword = True reasons.append(f"包含合理背景关键词: '{keyword}'") if has_reasonable_keyword: is_reasonable = True # 3. 检查交易对手关系(如果启用) if self.enable_counterparty_check: if inflow_counterparty == outflow_counterparty and inflow_counterparty not in ['', 'nan']: reasons.append(f"相同交易对手: {inflow_counterparty}") # 相同对手方可能是正常业务(如还款),也可能是过账嫌疑 if '还款' in inflow_summary or '还款' in outflow_summary: reasons.append("可能为正常还款业务") is_reasonable = True else: reasons.append("相同对手方资金来回流动,需关注") is_reasonable = False # 4. 检查停留时间(太短可能有问题) stay_time_hours = (outflow['datetime'] - inflow['datetime']).total_seconds() / 3600 if stay_time_hours < self.min_stay_time_hours: reasons.append(f"资金停留时间过短: {stay_time_hours:.1f}小时") is_reasonable = False # 5. 检查摘要信息完整性 if inflow_summary == '' or outflow_summary == '': reasons.append("交易摘要信息不完整") is_reasonable = False # 生成描述 if not reasons: description = "背景检查未发现明显异常" else: description = "; ".join(reasons) return is_reasonable, description def _format_over_book_reason(self, inflow: pd.Series, outflow: pd.Series, background_analysis: str, stay_days: float) -> str: """ 生成过账异常原因描述 Args: inflow: 流入交易记录 outflow: 流出交易记录 background_analysis: 背景分析结果 stay_days: 停留天数 Returns: str: 异常原因描述 """ inflow_amount = inflow['txAmount'] outflow_amount = outflow['txAmount'] amount_diff = abs(outflow_amount - inflow_amount) amount_diff_percent = (amount_diff / inflow_amount) * 100 reason_parts = [ f"疑似过账交易:收入¥{inflow_amount:,.2f}后{stay_days:.1f}天内支出¥{outflow_amount:,.2f}", f"金额匹配度:差异¥{amount_diff:,.2f}({amount_diff_percent:.2f}%)" ] if stay_days < 1: reason_parts.append(f"资金停留时间仅{stay_days * 24:.1f}小时") else: reason_parts.append(f"资金停留时间{stay_days:.1f}天") if background_analysis: reason_parts.append(f"背景分析:{background_analysis}") 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.amount_threshold:,.2f}收入 → {self.time_window_days}天内 → 匹配金额支出") # 检查必需字段 required_fields = ['txId', 'txDate', 'txTime', 'txAmount', 'txDirection', 'txSummary'] 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}' } # 确保数据按时间排序 if 'datetime' not in df.columns: return { 'recognition_type': self.display_name, 'identified_count': 0, 'identified_anomalies': [], 'recognition_status': '失败', 'error': '缺少datetime字段,无法进行时间序列分析' } df = df.sort_values('datetime').copy() # ============ 识别大额收入交易 ============ print(f"🔍 正在识别大额收入交易...") # 筛选大额收入 large_inflows_mask = df.apply(self._is_large_inflow, axis=1) large_inflows = df[large_inflows_mask].copy() if len(large_inflows) == 0: print(f"📊 未发现≥¥{self.amount_threshold:,.2f}的大额收入记录") return { 'recognition_type': self.display_name, 'identified_count': 0, 'identified_anomalies': [], 'recognition_status': '完成', 'recognition_parameters': { 'amount_threshold': self.amount_threshold, 'time_window_days': self.time_window_days, 'amount_tolerance': self.amount_tolerance, 'total_checked': len(df) }, 'statistics': { 'total_transactions': len(df), 'large_inflows_count': 0, 'max_transaction_amount': float(df['txAmount'].max()), 'avg_transaction_amount': float(df['txAmount'].mean()) } } print(f"📊 发现 {len(large_inflows)} 笔大额收入记录") print(f" 大额收入金额范围: ¥{large_inflows['txAmount'].min():,.2f} - ¥{large_inflows['txAmount'].max():,.2f}") # ============ 查找匹配的流出交易 ============ print(f"🔍 正在查找匹配的流出交易...") identified_anomalies = [] transaction_pairs = [] match_statistics = { 'total_pairs_found': 0, 'reasonable_pairs': 0, 'suspicious_pairs': 0 } for idx, inflow in large_inflows.iterrows(): inflow_id = str(inflow['txId']) inflow_amount = inflow['txAmount'] inflow_date = inflow['datetime'].strftime('%Y-%m-%d %H:%M:%S') print(f" 🔍 分析大额收入 {inflow_id}: ¥{inflow_amount:,.2f} ({inflow_date})") # 查找匹配的流出 matching_outflows = self._find_matching_outflows(inflow, df) if not matching_outflows: print(f" ✅ 未发现匹配的流出交易") continue print(f" 📊 发现 {len(matching_outflows)} 笔匹配流出") # 分析每对交易 for outflow in matching_outflows: outflow_id = str(outflow['txId']) outflow_amount = outflow['txAmount'] # 计算停留时间 stay_time = outflow['datetime'] - inflow['datetime'] stay_days = stay_time.total_seconds() / 86400 # 分析背景合理性 is_reasonable, background_analysis = self._analyze_background_reasonableness(inflow, outflow) # 记录交易对信息 pair_info = { 'inflow_id': inflow_id, 'outflow_id': outflow_id, 'inflow_amount': inflow_amount, 'outflow_amount': outflow_amount, 'amount_diff': abs(outflow_amount - inflow_amount), 'stay_days': stay_days, 'is_reasonable': is_reasonable, 'background_analysis': background_analysis } transaction_pairs.append(pair_info) match_statistics['total_pairs_found'] += 1 if is_reasonable: match_statistics['reasonable_pairs'] += 1 print(f" ✅ 交易对 {inflow_id}→{outflow_id}: 合理背景 ({background_analysis[:50]}...)") else: match_statistics['suspicious_pairs'] += 1 # 生成异常原因 reason = self._format_over_book_reason(inflow, outflow, background_analysis, stay_days) print(f" ❌ 发现疑似过账: {inflow_id}→{outflow_id}") print(f" 原因: {reason[:80]}...") # 创建异常记录(记录流出交易作为异常点) additional_info = { 'over_book_analysis': { 'inflow_transaction': { 'txId': inflow_id, 'txDate': inflow['txDate'], 'txTime': inflow['txTime'], 'txAmount': inflow_amount, 'txSummary': inflow.get('txSummary', ''), 'txCounterparty': inflow.get('txCounterparty', '') }, 'outflow_transaction': { 'txId': outflow_id, 'txDate': outflow['txDate'], 'txTime': outflow['txTime'], 'txAmount': outflow_amount, 'txSummary': outflow.get('txSummary', ''), 'txCounterparty': outflow.get('txCounterparty', '') }, 'pair_analysis': { 'stay_days': stay_days, 'stay_hours': stay_days * 24, 'amount_match_ratio': outflow_amount / inflow_amount, 'background_check_result': background_analysis, 'is_reasonable': is_reasonable, 'detection_parameters': { 'amount_threshold': self.amount_threshold, 'time_window_days': self.time_window_days, 'amount_tolerance': self.amount_tolerance } } } } # 使用流出交易作为异常记录主体 anomaly = self.format_anomaly_record( row=outflow, reason=reason, severity=self.severity_level, check_type='over_book_transaction', **additional_info ) identified_anomalies.append(anomaly) # ============ 结果统计 ============ print(f"✅ {self.display_name}检查完成") print(f" 检查结果:") print(f" 大额收入记录: {len(large_inflows)} 笔") print(f" 匹配交易对: {match_statistics['total_pairs_found']} 对") print(f" 合理交易对: {match_statistics['reasonable_pairs']} 对") print(f" 疑似过账对: {match_statistics['suspicious_pairs']} 对") print(f" 异常记录数: {len(identified_anomalies)} 条") # 显示详细信息 if match_statistics['suspicious_pairs'] > 0: print("📋 疑似过账交易详情:") for i, pair in enumerate(transaction_pairs[:5]): # 显示前5条 if not pair['is_reasonable']: print(f" {i + 1}. {pair['inflow_id']}→{pair['outflow_id']}: " f"¥{pair['inflow_amount']:,.2f}→¥{pair['outflow_amount']:,.2f} " f"({pair['stay_days']:.1f}天)") return { 'recognition_type': self.display_name, 'identified_count': len(identified_anomalies), 'identified_anomalies': identified_anomalies, 'recognition_status': '完成', 'recognition_parameters': { 'amount_threshold': self.amount_threshold, 'time_window_days': self.time_window_days, 'amount_tolerance': self.amount_tolerance, 'min_stay_time_hours': self.min_stay_time_hours, 'enable_background_check': self.enable_background_check, 'enable_counterparty_check': self.enable_counterparty_check, 'total_large_inflows': len(large_inflows) }, 'statistics': { 'total_transactions': len(df), 'large_inflows_count': len(large_inflows), 'large_inflows_amount_stats': { 'total': float(large_inflows['txAmount'].sum()), 'avg': float(large_inflows['txAmount'].mean()), 'max': float(large_inflows['txAmount'].max()), 'min': float(large_inflows['txAmount'].min()) } if len(large_inflows) > 0 else {}, 'match_statistics': match_statistics, 'transaction_pairs_count': len(transaction_pairs), 'suspicious_pairs_details': [ { 'inflow_id': p['inflow_id'], 'outflow_id': p['outflow_id'], 'inflow_amount': p['inflow_amount'], 'outflow_amount': p['outflow_amount'], 'stay_days': p['stay_days'], 'background_analysis': p['background_analysis'] } for p in transaction_pairs if not p['is_reasonable'] ][:10] # 只保留前10条详情 } } 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({ 'amount_threshold': self.amount_threshold, 'time_window_days': self.time_window_days, 'amount_tolerance': self.amount_tolerance, 'min_stay_time_hours': self.min_stay_time_hours, 'enable_background_check': self.enable_background_check, 'reasonable_keywords_count': len(self.reasonable_background_keywords), 'high_risk_keywords_count': len(self.high_risk_keywords), 'enable_counterparty_check': self.enable_counterparty_check, 'detect_patterns': { 'single_pair': self.detect_single_pair, 'split_pattern': self.detect_split_pattern, 'merge_pattern': self.detect_merge_pattern }, 'severity_level': self.severity_level, 'data_loaded': self._data is not None }) return summary def get_config_summary(self) -> Dict[str, Any]: """获取配置摘要""" return { "金额阈值": f"¥{self.amount_threshold:,.2f}", "时间窗口": f"{self.time_window_days}天", "金额容忍度": f"±{self.amount_tolerance:.1%}", "最小停留时间": f"{self.min_stay_time_hours}小时", "背景检查": "启用" if self.enable_background_check else "禁用", "合理关键词": f"{len(self.reasonable_background_keywords)}个", "高风险关键词": f"{len(self.high_risk_keywords)}个", "对手方检查": "启用" if self.enable_counterparty_check else "禁用", "检测逻辑": f"大额收入后{self.time_window_days}天内出现匹配金额支出,且缺乏合理业务背景", "业务规则描述": "资金短暂停留即流出,缺乏真实业务背景,疑似过账交易" }