| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539 |
- from pydantic import BaseModel, Field
- from typing import Dict, Any, Optional, Type, List
- import pandas as pd
- from .enhanced_base_recognizer import EnhancedBaseRecognizer
- class LargeAmountInput(BaseModel):
- """大额交易识别工具输入"""
- csv_path: Optional[str] = Field(
- None,
- description="CSV文件路径(可选)。如果初始化时已提供csv_path,可以不用再次传入。"
- )
- class Config:
- arbitrary_types_allowed = True
- class LargeAmountTransactionRecognizer(EnhancedBaseRecognizer):
- """
- 大额交易异常识别器
- 业务规则定义:
- 若交易对手方个人银行账户出现单次交易金额超过预设阈值(如 5 万元、20 万元等)的大额资金往来,
- 且该交易与账户日常交易规模、资金使用场景及个人经济活动特征不匹配,缺乏合理交易背景支撑,
- 可触发大额交易异常提示,需进一步核查该笔交易的真实性、合法性及资金来源与去向。
- """
- args_schema: Type[BaseModel] = LargeAmountInput
- # 配置参数
- amount_threshold: float = Field(
- 50000.0,
- description="大额交易阈值(元),单次交易金额超过此值视为大额交易"
- )
- # 历史分析参数
- history_days: int = Field(
- 90,
- description="历史分析天数,用于分析账户日常交易规模"
- )
- outlier_multiplier: float = Field(
- 3.0,
- description="异常倍数阈值,交易金额超过历史均值的多少倍视为异常"
- )
- # 背景分析参数
- enable_background_check: bool = Field(
- True,
- description="是否启用交易背景检查"
- )
- # 合理背景关键词(常见的大额合理交易场景)
- reasonable_background_keywords: List[str] = Field(
- [
- "工资", "奖金", "绩效", "年终奖", "报销", "货款", "租金",
- "购房款", "装修款", "学费", "医疗费", "保险", "理财",
- "投资款", "分红", "还款", "借款", "赠与", "遗产"
- ],
- description="合理的交易背景关键词,用于识别可能有合理背景的大额交易"
- )
- # 高风险关键词(可能需要关注的场景)
- high_risk_keywords: List[str] = Field(
- [
- "赌博", "赌资", "彩票", "博彩", "虚拟货币", "比特币",
- "地下钱庄", "洗钱", "套现", "非法", "不明", "无摘要"
- ],
- description="高风险关键词,出现这些词的大额交易需要重点关注"
- )
- def __init__(self, csv_path: str = None, config: Dict[str, Any] = None, **kwargs):
- """
- 初始化大额交易识别器
- Args:
- csv_path: CSV文件路径
- config: 配置参数
- **kwargs: 其他参数
- """
- # 调用父类的 __init__
- super().__init__(
- name="large_amount_recognizer",
- description="识别银行流水中的大额交易异常,检测单次交易金额超过阈值且与账户历史行为不匹配的交易。",
- display_name="大额交易异常识别",
- csv_path=csv_path,
- config=config,
- **kwargs
- )
- # 从config获取配置,更新Field属性
- large_amount_config = self.get_config_value('large_amount_recognition', {})
- if large_amount_config:
- config_mapping = {
- 'amount_threshold': 'amount_threshold',
- 'history_days': 'history_days',
- 'outlier_multiplier': 'outlier_multiplier',
- 'enable_background_check': 'enable_background_check',
- 'reasonable_background_keywords': 'reasonable_background_keywords',
- 'high_risk_keywords': 'high_risk_keywords'
- }
- for config_key, attr_name in config_mapping.items():
- if config_key in large_amount_config:
- setattr(self, attr_name, large_amount_config[config_key])
- print(f"✅ {self.display_name} 初始化完成")
- print(f" 金额阈值: ¥{self.amount_threshold:,.2f}")
- print(f" 历史分析天数: {self.history_days}天")
- print(f" 异常倍数阈值: {self.outlier_multiplier}倍")
- print(f" 背景检查: {'启用' if self.enable_background_check else '禁用'}")
- print(f" 合理背景关键词: {len(self.reasonable_background_keywords)}个")
- print(f" 高风险关键词: {len(self.high_risk_keywords)}个")
- def _analyze_account_history(self, df: pd.DataFrame, current_date: pd.Timestamp) -> Dict[str, Any]:
- """
- 分析账户历史交易特征
- Args:
- df: 交易数据
- current_date: 当前交易日期
- Returns:
- Dict[str, Any]: 账户历史交易特征
- """
- # 计算历史日期范围
- history_start = current_date - pd.Timedelta(days=self.history_days)
- # 筛选历史交易(当前日期之前的历史数据)
- history_df = df[df['datetime'] < current_date]
- history_df = history_df[history_df['datetime'] >= history_start]
- if len(history_df) == 0:
- return {
- 'has_history': False,
- 'message': f'无最近{self.history_days}天的历史交易数据'
- }
- # 计算历史交易特征
- history_features = {
- 'has_history': True,
- 'history_days': self.history_days,
- 'total_transactions': len(history_df),
- 'avg_amount': float(history_df['txAmount'].mean()) if len(history_df) > 0 else 0,
- 'max_amount': float(history_df['txAmount'].max()) if len(history_df) > 0 else 0,
- 'min_amount': float(history_df['txAmount'].min()) if len(history_df) > 0 else 0,
- 'std_amount': float(history_df['txAmount'].std()) if len(history_df) > 0 else 0,
- 'total_income': float(history_df[history_df['txDirection'] == '收入']['txAmount'].sum()),
- 'total_expense': float(history_df[history_df['txDirection'] == '支出']['txAmount'].sum()),
- 'income_count': len(history_df[history_df['txDirection'] == '收入']),
- 'expense_count': len(history_df[history_df['txDirection'] == '支出']),
- 'date_range': {
- 'start': history_df['datetime'].min().strftime('%Y-%m-%d'),
- 'end': history_df['datetime'].max().strftime('%Y-%m-%d')
- }
- }
- return history_features
- def _check_transaction_background(self, row: pd.Series) -> Dict[str, Any]:
- """
- 检查交易背景合理性
- Args:
- row: 交易记录
- Returns:
- Dict[str, Any]: 背景检查结果
- """
- background_result = {
- 'has_reasonable_background': False,
- 'has_high_risk_indicator': False,
- 'reasonable_keywords_found': [],
- 'high_risk_keywords_found': [],
- 'summary': '',
- 'counterparty': '',
- 'summary_text': ''
- }
- if not self.enable_background_check:
- return background_result
- # 获取交易摘要和对手方信息
- summary = str(row.get('txSummary', '')).lower()
- counterparty = str(row.get('txCounterparty', '')).lower()
- # 检查合理背景关键词
- reasonable_found = []
- for keyword in self.reasonable_background_keywords:
- if keyword in summary or keyword in counterparty:
- reasonable_found.append(keyword)
- # 检查高风险关键词
- high_risk_found = []
- for keyword in self.high_risk_keywords:
- if keyword in summary or keyword in counterparty:
- high_risk_found.append(keyword)
- # 判断是否有合理背景
- has_reasonable_background = len(reasonable_found) > 0
- has_high_risk = len(high_risk_found) > 0
- # 生成背景描述
- background_desc = []
- if reasonable_found:
- background_desc.append(f"合理背景: {', '.join(reasonable_found)}")
- if high_risk_found:
- background_desc.append(f"高风险关键词: {', '.join(high_risk_found)}")
- background_result.update({
- 'has_reasonable_background': has_reasonable_background,
- 'has_high_risk_indicator': has_high_risk,
- 'reasonable_keywords_found': reasonable_found,
- 'high_risk_keywords_found': high_risk_found,
- 'summary': '; '.join(background_desc) if background_desc else '无特殊背景信息',
- 'counterparty': counterparty,
- 'summary_text': summary
- })
- return background_result
- def _is_amount_outlier(self, amount: float, history_features: Dict[str, Any]) -> bool:
- """
- 判断交易金额是否为异常值(与历史行为不匹配)
- Args:
- amount: 当前交易金额
- history_features: 账户历史特征
- Returns:
- bool: 是否为异常值
- """
- if not history_features['has_history']:
- # 无历史数据,无法判断是否为异常值
- return False
- if history_features['avg_amount'] == 0:
- # 历史平均金额为0,无法判断
- return False
- # 判断是否超过历史平均值的异常倍数
- is_outlier = amount > (history_features['avg_amount'] * self.outlier_multiplier)
- return is_outlier
- 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}")
- print(f" 检测规则: 大额金额 + 与历史不匹配 + 缺乏合理背景 = 大额交易异常")
- # 检查必需字段
- required_fields = ['txId', 'datetime', 'txAmount', 'txDirection']
- 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}'
- }
- # 确保datetime列已正确解析
- if not pd.api.types.is_datetime64_any_dtype(df['datetime']):
- df['datetime'] = pd.to_datetime(df['datetime'], errors='coerce')
- # 按时间排序,便于历史分析
- df = df.sort_values('datetime')
- # 复制一份用于分析(避免修改原始数据)
- analysis_df = df.copy()
- # ============ 识别大额交易 ============
- # 根据业务规则:单次交易金额超过预设阈值
- large_amount_mask = analysis_df['txAmount'].abs() >= self.amount_threshold
- large_amount_transactions = analysis_df[large_amount_mask].copy()
- if len(large_amount_transactions) == 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,
- 'history_days': self.history_days,
- 'outlier_multiplier': self.outlier_multiplier,
- 'enable_background_check': self.enable_background_check,
- 'total_checked': len(df)
- },
- 'statistics': {
- 'total_transactions': len(df),
- 'large_amount_transactions': 0,
- 'max_amount': float(df['txAmount'].max()),
- 'min_amount': float(df['txAmount'].min()),
- 'avg_amount': float(df['txAmount'].mean())
- }
- }
- print(f"📊 发现 {len(large_amount_transactions)} 笔大额交易(≥¥{self.amount_threshold:,.2f})")
- # ============ 分析每笔大额交易 ============
- identified_anomalies = []
- analyzed_transactions = []
- for idx, row in large_amount_transactions.iterrows():
- tx_id = str(row['txId'])
- tx_date = row['datetime']
- tx_amount = float(row['txAmount'])
- print(f" 🔍 分析交易 {tx_id}: ¥{tx_amount:,.2f} ({row['txDirection']})")
- # 1. 分析账户历史特征
- history_features = self._analyze_account_history(analysis_df, tx_date)
- # 2. 检查交易背景
- background_check = self._check_transaction_background(row)
- # 3. 判断是否为异常值(与历史不匹配)
- is_amount_outlier = self._is_amount_outlier(abs(tx_amount), history_features)
- # 4. 综合判断是否为异常
- # 规则:大额 + (历史不匹配 或 缺乏合理背景) = 异常
- is_abnormal = True # 默认大额就是异常
- # 如果有合理背景,可以降低异常级别
- severity_level = 'high'
- if background_check['has_reasonable_background']:
- if not is_amount_outlier:
- # 有合理背景且金额不异常,可能不是异常
- is_abnormal = False
- print(f" ✅ 有合理背景且金额不异常,跳过")
- continue
- else:
- severity_level = 'medium'
- print(f" ⚠️ 有合理背景但金额异常")
- # 如果有高风险关键词,提高异常级别
- if background_check['has_high_risk_indicator']:
- severity_level = 'high'
- print(f" ⚠️ 发现高风险关键词")
- # 记录分析结果
- transaction_analysis = {
- 'tx_id': tx_id,
- 'date': tx_date.strftime('%Y-%m-%d'),
- 'time': tx_date.strftime('%H:%M:%S'),
- 'amount': tx_amount,
- 'direction': row['txDirection'],
- 'is_large_amount': True,
- 'is_amount_outlier': is_amount_outlier,
- 'history_features': history_features,
- 'background_check': background_check,
- 'is_abnormal': is_abnormal,
- 'severity_level': severity_level
- }
- analyzed_transactions.append(transaction_analysis)
- # 如果判断为异常,生成异常记录
- if is_abnormal:
- # 生成异常原因
- reasons = []
- reasons.append(f"大额交易(¥{tx_amount:,.2f}≥¥{self.amount_threshold:,.2f})")
- if is_amount_outlier:
- if history_features['has_history']:
- avg_amount = history_features['avg_amount']
- outlier_ratio = tx_amount / avg_amount if avg_amount > 0 else float('inf')
- reasons.append(f"金额异常(超出历史均值{outlier_ratio:.1f}倍)")
- if not background_check['has_reasonable_background']:
- reasons.append("缺乏合理交易背景")
- if background_check['has_high_risk_indicator']:
- reasons.append("存在高风险关键词")
- reason_str = ",".join(reasons)
- # 额外信息
- additional_info = {
- 'amount_analysis': {
- 'threshold': self.amount_threshold,
- 'is_outlier': is_amount_outlier,
- 'outlier_ratio': float(tx_amount / history_features['avg_amount']) if history_features[
- 'has_history'] and
- history_features[
- 'avg_amount'] > 0 else None,
- 'history_avg': history_features['avg_amount'] if history_features['has_history'] else None
- },
- 'background_analysis': background_check,
- 'history_analysis': history_features
- }
- anomaly = self.format_anomaly_record(
- row=row,
- reason=f"大额交易异常: {reason_str},需核查真实性、合法性及资金来源去向",
- severity=severity_level,
- check_type='large_amount_transaction',
- **additional_info
- )
- identified_anomalies.append(anomaly)
- print(f" ❌ 标记为异常: {reason_str}")
- else:
- print(f" ✅ 未标记为异常")
- # ============ 结果统计 ============
- print(f"✅ {self.display_name}检查完成")
- print(f" 检查交易总数: {len(df)}")
- print(f" 大额交易数: {len(large_amount_transactions)}")
- print(f" 异常交易数: {len(identified_anomalies)}")
- print(f" 通过检查数: {len(large_amount_transactions) - len(identified_anomalies)}")
- # 显示大额交易统计
- if len(large_amount_transactions) > 0:
- print("📋 大额交易统计:")
- total_large_amount = large_amount_transactions['txAmount'].sum()
- avg_large_amount = large_amount_transactions['txAmount'].mean()
- print(f" 总大额金额: ¥{total_large_amount:,.2f}")
- print(f" 平均大额金额: ¥{avg_large_amount:,.2f}")
- # 按方向统计
- income_large = large_amount_transactions[large_amount_transactions['txDirection'] == '收入']
- expense_large = large_amount_transactions[large_amount_transactions['txDirection'] == '支出']
- print(f" 大额收入: {len(income_large)}笔, ¥{income_large['txAmount'].sum():,.2f}")
- print(f" 大额支出: {len(expense_large)}笔, ¥{expense_large['txAmount'].sum():,.2f}")
- # 显示异常交易示例
- if len(identified_anomalies) > 0:
- print("📋 大额异常交易示例:")
- for i, anomaly in enumerate(identified_anomalies[:5], 1):
- time_str = f"{anomaly.get('txDate', '')} {anomaly.get('txTime', '')}"
- print(
- f" {i}. ID:{anomaly['txId']} | {time_str} | {anomaly['txDirection']} ¥{anomaly['txAmount']:,.2f} | {anomaly['recognition_reason'][:50]}...")
- return {
- 'recognition_type': self.display_name,
- 'identified_count': len(identified_anomalies),
- 'identified_anomalies': identified_anomalies,
- 'recognition_status': '完成',
- 'recognition_parameters': {
- 'amount_threshold': self.amount_threshold,
- 'history_days': self.history_days,
- 'outlier_multiplier': self.outlier_multiplier,
- 'enable_background_check': self.enable_background_check,
- 'total_checked': len(df),
- 'large_transactions_found': len(large_amount_transactions)
- },
- 'statistics': {
- 'total_transactions': len(df),
- 'large_amount_transactions': len(large_amount_transactions),
- 'abnormal_large_transactions': len(identified_anomalies),
- 'amount_statistics': {
- 'max_amount': float(df['txAmount'].max()),
- 'min_amount': float(df['txAmount'].min()),
- 'avg_amount': float(df['txAmount'].mean()),
- 'total_amount': float(df['txAmount'].sum()),
- 'large_amount_total': float(large_amount_transactions['txAmount'].sum()),
- 'large_amount_avg': float(large_amount_transactions['txAmount'].mean()) if len(
- large_amount_transactions) > 0 else 0
- },
- 'direction_distribution': {
- 'income_count': len(df[df['txDirection'] == '收入']),
- 'expense_count': len(df[df['txDirection'] == '支出']),
- 'large_income_count': len(
- large_amount_transactions[large_amount_transactions['txDirection'] == '收入']),
- 'large_expense_count': len(
- large_amount_transactions[large_amount_transactions['txDirection'] == '支出'])
- },
- 'background_analysis': {
- 'reasonable_background_count': sum(
- 1 for t in analyzed_transactions if t['background_check']['has_reasonable_background']),
- 'high_risk_count': sum(
- 1 for t in analyzed_transactions if t['background_check']['has_high_risk_indicator']),
- 'outlier_count': sum(1 for t in analyzed_transactions if t['is_amount_outlier'])
- }
- }
- }
- 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,
- 'history_days': self.history_days,
- 'outlier_multiplier': self.outlier_multiplier,
- 'enable_background_check': self.enable_background_check,
- 'reasonable_background_keywords_count': len(self.reasonable_background_keywords),
- 'high_risk_keywords_count': len(self.high_risk_keywords),
- '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.history_days}天",
- "异常倍数阈值": f"{self.outlier_multiplier}倍",
- "背景检查": "启用" if self.enable_background_check else "禁用",
- "检测逻辑": "大额金额 + 与历史不匹配 + 缺乏合理背景 = 大额交易异常",
- "业务规则描述": "单次交易金额超过阈值且与账户历史行为不匹配,缺乏合理背景"
- }
|