| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628 |
- 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}天内出现匹配金额支出,且缺乏合理业务背景",
- "业务规则描述": "资金短暂停留即流出,缺乏真实业务背景,疑似过账交易"
- }
|