large_amount_transaction_recognizer.py 24 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539
  1. from pydantic import BaseModel, Field
  2. from typing import Dict, Any, Optional, Type, List
  3. import pandas as pd
  4. from .enhanced_base_recognizer import EnhancedBaseRecognizer
  5. class LargeAmountInput(BaseModel):
  6. """大额交易识别工具输入"""
  7. csv_path: Optional[str] = Field(
  8. None,
  9. description="CSV文件路径(可选)。如果初始化时已提供csv_path,可以不用再次传入。"
  10. )
  11. class Config:
  12. arbitrary_types_allowed = True
  13. class LargeAmountTransactionRecognizer(EnhancedBaseRecognizer):
  14. """
  15. 大额交易异常识别器
  16. 业务规则定义:
  17. 若交易对手方个人银行账户出现单次交易金额超过预设阈值(如 5 万元、20 万元等)的大额资金往来,
  18. 且该交易与账户日常交易规模、资金使用场景及个人经济活动特征不匹配,缺乏合理交易背景支撑,
  19. 可触发大额交易异常提示,需进一步核查该笔交易的真实性、合法性及资金来源与去向。
  20. """
  21. args_schema: Type[BaseModel] = LargeAmountInput
  22. # 配置参数
  23. amount_threshold: float = Field(
  24. 50000.0,
  25. description="大额交易阈值(元),单次交易金额超过此值视为大额交易"
  26. )
  27. # 历史分析参数
  28. history_days: int = Field(
  29. 90,
  30. description="历史分析天数,用于分析账户日常交易规模"
  31. )
  32. outlier_multiplier: float = Field(
  33. 3.0,
  34. description="异常倍数阈值,交易金额超过历史均值的多少倍视为异常"
  35. )
  36. # 背景分析参数
  37. enable_background_check: bool = Field(
  38. True,
  39. description="是否启用交易背景检查"
  40. )
  41. # 合理背景关键词(常见的大额合理交易场景)
  42. reasonable_background_keywords: List[str] = Field(
  43. [
  44. "工资", "奖金", "绩效", "年终奖", "报销", "货款", "租金",
  45. "购房款", "装修款", "学费", "医疗费", "保险", "理财",
  46. "投资款", "分红", "还款", "借款", "赠与", "遗产"
  47. ],
  48. description="合理的交易背景关键词,用于识别可能有合理背景的大额交易"
  49. )
  50. # 高风险关键词(可能需要关注的场景)
  51. high_risk_keywords: List[str] = Field(
  52. [
  53. "赌博", "赌资", "彩票", "博彩", "虚拟货币", "比特币",
  54. "地下钱庄", "洗钱", "套现", "非法", "不明", "无摘要"
  55. ],
  56. description="高风险关键词,出现这些词的大额交易需要重点关注"
  57. )
  58. def __init__(self, csv_path: str = None, config: Dict[str, Any] = None, **kwargs):
  59. """
  60. 初始化大额交易识别器
  61. Args:
  62. csv_path: CSV文件路径
  63. config: 配置参数
  64. **kwargs: 其他参数
  65. """
  66. # 调用父类的 __init__
  67. super().__init__(
  68. name="large_amount_recognizer",
  69. description="识别银行流水中的大额交易异常,检测单次交易金额超过阈值且与账户历史行为不匹配的交易。",
  70. display_name="大额交易异常识别",
  71. csv_path=csv_path,
  72. config=config,
  73. **kwargs
  74. )
  75. # 从config获取配置,更新Field属性
  76. large_amount_config = self.get_config_value('large_amount_recognition', {})
  77. if large_amount_config:
  78. config_mapping = {
  79. 'amount_threshold': 'amount_threshold',
  80. 'history_days': 'history_days',
  81. 'outlier_multiplier': 'outlier_multiplier',
  82. 'enable_background_check': 'enable_background_check',
  83. 'reasonable_background_keywords': 'reasonable_background_keywords',
  84. 'high_risk_keywords': 'high_risk_keywords'
  85. }
  86. for config_key, attr_name in config_mapping.items():
  87. if config_key in large_amount_config:
  88. setattr(self, attr_name, large_amount_config[config_key])
  89. print(f"✅ {self.display_name} 初始化完成")
  90. print(f" 金额阈值: ¥{self.amount_threshold:,.2f}")
  91. print(f" 历史分析天数: {self.history_days}天")
  92. print(f" 异常倍数阈值: {self.outlier_multiplier}倍")
  93. print(f" 背景检查: {'启用' if self.enable_background_check else '禁用'}")
  94. print(f" 合理背景关键词: {len(self.reasonable_background_keywords)}个")
  95. print(f" 高风险关键词: {len(self.high_risk_keywords)}个")
  96. def _analyze_account_history(self, df: pd.DataFrame, current_date: pd.Timestamp) -> Dict[str, Any]:
  97. """
  98. 分析账户历史交易特征
  99. Args:
  100. df: 交易数据
  101. current_date: 当前交易日期
  102. Returns:
  103. Dict[str, Any]: 账户历史交易特征
  104. """
  105. # 计算历史日期范围
  106. history_start = current_date - pd.Timedelta(days=self.history_days)
  107. # 筛选历史交易(当前日期之前的历史数据)
  108. history_df = df[df['datetime'] < current_date]
  109. history_df = history_df[history_df['datetime'] >= history_start]
  110. if len(history_df) == 0:
  111. return {
  112. 'has_history': False,
  113. 'message': f'无最近{self.history_days}天的历史交易数据'
  114. }
  115. # 计算历史交易特征
  116. history_features = {
  117. 'has_history': True,
  118. 'history_days': self.history_days,
  119. 'total_transactions': len(history_df),
  120. 'avg_amount': float(history_df['txAmount'].mean()) if len(history_df) > 0 else 0,
  121. 'max_amount': float(history_df['txAmount'].max()) if len(history_df) > 0 else 0,
  122. 'min_amount': float(history_df['txAmount'].min()) if len(history_df) > 0 else 0,
  123. 'std_amount': float(history_df['txAmount'].std()) if len(history_df) > 0 else 0,
  124. 'total_income': float(history_df[history_df['txDirection'] == '收入']['txAmount'].sum()),
  125. 'total_expense': float(history_df[history_df['txDirection'] == '支出']['txAmount'].sum()),
  126. 'income_count': len(history_df[history_df['txDirection'] == '收入']),
  127. 'expense_count': len(history_df[history_df['txDirection'] == '支出']),
  128. 'date_range': {
  129. 'start': history_df['datetime'].min().strftime('%Y-%m-%d'),
  130. 'end': history_df['datetime'].max().strftime('%Y-%m-%d')
  131. }
  132. }
  133. return history_features
  134. def _check_transaction_background(self, row: pd.Series) -> Dict[str, Any]:
  135. """
  136. 检查交易背景合理性
  137. Args:
  138. row: 交易记录
  139. Returns:
  140. Dict[str, Any]: 背景检查结果
  141. """
  142. background_result = {
  143. 'has_reasonable_background': False,
  144. 'has_high_risk_indicator': False,
  145. 'reasonable_keywords_found': [],
  146. 'high_risk_keywords_found': [],
  147. 'summary': '',
  148. 'counterparty': '',
  149. 'summary_text': ''
  150. }
  151. if not self.enable_background_check:
  152. return background_result
  153. # 获取交易摘要和对手方信息
  154. summary = str(row.get('txSummary', '')).lower()
  155. counterparty = str(row.get('txCounterparty', '')).lower()
  156. # 检查合理背景关键词
  157. reasonable_found = []
  158. for keyword in self.reasonable_background_keywords:
  159. if keyword in summary or keyword in counterparty:
  160. reasonable_found.append(keyword)
  161. # 检查高风险关键词
  162. high_risk_found = []
  163. for keyword in self.high_risk_keywords:
  164. if keyword in summary or keyword in counterparty:
  165. high_risk_found.append(keyword)
  166. # 判断是否有合理背景
  167. has_reasonable_background = len(reasonable_found) > 0
  168. has_high_risk = len(high_risk_found) > 0
  169. # 生成背景描述
  170. background_desc = []
  171. if reasonable_found:
  172. background_desc.append(f"合理背景: {', '.join(reasonable_found)}")
  173. if high_risk_found:
  174. background_desc.append(f"高风险关键词: {', '.join(high_risk_found)}")
  175. background_result.update({
  176. 'has_reasonable_background': has_reasonable_background,
  177. 'has_high_risk_indicator': has_high_risk,
  178. 'reasonable_keywords_found': reasonable_found,
  179. 'high_risk_keywords_found': high_risk_found,
  180. 'summary': '; '.join(background_desc) if background_desc else '无特殊背景信息',
  181. 'counterparty': counterparty,
  182. 'summary_text': summary
  183. })
  184. return background_result
  185. def _is_amount_outlier(self, amount: float, history_features: Dict[str, Any]) -> bool:
  186. """
  187. 判断交易金额是否为异常值(与历史行为不匹配)
  188. Args:
  189. amount: 当前交易金额
  190. history_features: 账户历史特征
  191. Returns:
  192. bool: 是否为异常值
  193. """
  194. if not history_features['has_history']:
  195. # 无历史数据,无法判断是否为异常值
  196. return False
  197. if history_features['avg_amount'] == 0:
  198. # 历史平均金额为0,无法判断
  199. return False
  200. # 判断是否超过历史平均值的异常倍数
  201. is_outlier = amount > (history_features['avg_amount'] * self.outlier_multiplier)
  202. return is_outlier
  203. def recognize(self, csv_path: str = None, **kwargs) -> Dict[str, Any]:
  204. """
  205. 识别大额交易异常
  206. Args:
  207. csv_path: CSV文件路径
  208. **kwargs: 其他参数
  209. Returns:
  210. Dict[str, Any]: 识别结果
  211. """
  212. try:
  213. # 使用父类的load_data方法加载标准化数据
  214. df = self.load_data(csv_path)
  215. print(f"🔍 {self.display_name}开始检查,共 {len(df)} 条记录")
  216. print(f" 大额阈值: ¥{self.amount_threshold:,.2f}")
  217. print(f" 检测规则: 大额金额 + 与历史不匹配 + 缺乏合理背景 = 大额交易异常")
  218. # 检查必需字段
  219. required_fields = ['txId', 'datetime', 'txAmount', 'txDirection']
  220. missing_fields = [field for field in required_fields if field not in df.columns]
  221. if missing_fields:
  222. return {
  223. 'recognition_type': self.display_name,
  224. 'identified_count': 0,
  225. 'identified_anomalies': [],
  226. 'recognition_status': '失败',
  227. 'error': f'缺少必需字段: {missing_fields}'
  228. }
  229. # 确保datetime列已正确解析
  230. if not pd.api.types.is_datetime64_any_dtype(df['datetime']):
  231. df['datetime'] = pd.to_datetime(df['datetime'], errors='coerce')
  232. # 按时间排序,便于历史分析
  233. df = df.sort_values('datetime')
  234. # 复制一份用于分析(避免修改原始数据)
  235. analysis_df = df.copy()
  236. # ============ 识别大额交易 ============
  237. # 根据业务规则:单次交易金额超过预设阈值
  238. large_amount_mask = analysis_df['txAmount'].abs() >= self.amount_threshold
  239. large_amount_transactions = analysis_df[large_amount_mask].copy()
  240. if len(large_amount_transactions) == 0:
  241. print(f"📊 未发现大额交易(≥¥{self.amount_threshold:,.2f})")
  242. return {
  243. 'recognition_type': self.display_name,
  244. 'identified_count': 0,
  245. 'identified_anomalies': [],
  246. 'recognition_status': '完成',
  247. 'recognition_parameters': {
  248. 'amount_threshold': self.amount_threshold,
  249. 'history_days': self.history_days,
  250. 'outlier_multiplier': self.outlier_multiplier,
  251. 'enable_background_check': self.enable_background_check,
  252. 'total_checked': len(df)
  253. },
  254. 'statistics': {
  255. 'total_transactions': len(df),
  256. 'large_amount_transactions': 0,
  257. 'max_amount': float(df['txAmount'].max()),
  258. 'min_amount': float(df['txAmount'].min()),
  259. 'avg_amount': float(df['txAmount'].mean())
  260. }
  261. }
  262. print(f"📊 发现 {len(large_amount_transactions)} 笔大额交易(≥¥{self.amount_threshold:,.2f})")
  263. # ============ 分析每笔大额交易 ============
  264. identified_anomalies = []
  265. analyzed_transactions = []
  266. for idx, row in large_amount_transactions.iterrows():
  267. tx_id = str(row['txId'])
  268. tx_date = row['datetime']
  269. tx_amount = float(row['txAmount'])
  270. print(f" 🔍 分析交易 {tx_id}: ¥{tx_amount:,.2f} ({row['txDirection']})")
  271. # 1. 分析账户历史特征
  272. history_features = self._analyze_account_history(analysis_df, tx_date)
  273. # 2. 检查交易背景
  274. background_check = self._check_transaction_background(row)
  275. # 3. 判断是否为异常值(与历史不匹配)
  276. is_amount_outlier = self._is_amount_outlier(abs(tx_amount), history_features)
  277. # 4. 综合判断是否为异常
  278. # 规则:大额 + (历史不匹配 或 缺乏合理背景) = 异常
  279. is_abnormal = True # 默认大额就是异常
  280. # 如果有合理背景,可以降低异常级别
  281. severity_level = 'high'
  282. if background_check['has_reasonable_background']:
  283. if not is_amount_outlier:
  284. # 有合理背景且金额不异常,可能不是异常
  285. is_abnormal = False
  286. print(f" ✅ 有合理背景且金额不异常,跳过")
  287. continue
  288. else:
  289. severity_level = 'medium'
  290. print(f" ⚠️ 有合理背景但金额异常")
  291. # 如果有高风险关键词,提高异常级别
  292. if background_check['has_high_risk_indicator']:
  293. severity_level = 'high'
  294. print(f" ⚠️ 发现高风险关键词")
  295. # 记录分析结果
  296. transaction_analysis = {
  297. 'tx_id': tx_id,
  298. 'date': tx_date.strftime('%Y-%m-%d'),
  299. 'time': tx_date.strftime('%H:%M:%S'),
  300. 'amount': tx_amount,
  301. 'direction': row['txDirection'],
  302. 'is_large_amount': True,
  303. 'is_amount_outlier': is_amount_outlier,
  304. 'history_features': history_features,
  305. 'background_check': background_check,
  306. 'is_abnormal': is_abnormal,
  307. 'severity_level': severity_level
  308. }
  309. analyzed_transactions.append(transaction_analysis)
  310. # 如果判断为异常,生成异常记录
  311. if is_abnormal:
  312. # 生成异常原因
  313. reasons = []
  314. reasons.append(f"大额交易(¥{tx_amount:,.2f}≥¥{self.amount_threshold:,.2f})")
  315. if is_amount_outlier:
  316. if history_features['has_history']:
  317. avg_amount = history_features['avg_amount']
  318. outlier_ratio = tx_amount / avg_amount if avg_amount > 0 else float('inf')
  319. reasons.append(f"金额异常(超出历史均值{outlier_ratio:.1f}倍)")
  320. if not background_check['has_reasonable_background']:
  321. reasons.append("缺乏合理交易背景")
  322. if background_check['has_high_risk_indicator']:
  323. reasons.append("存在高风险关键词")
  324. reason_str = ",".join(reasons)
  325. # 额外信息
  326. additional_info = {
  327. 'amount_analysis': {
  328. 'threshold': self.amount_threshold,
  329. 'is_outlier': is_amount_outlier,
  330. 'outlier_ratio': float(tx_amount / history_features['avg_amount']) if history_features[
  331. 'has_history'] and
  332. history_features[
  333. 'avg_amount'] > 0 else None,
  334. 'history_avg': history_features['avg_amount'] if history_features['has_history'] else None
  335. },
  336. 'background_analysis': background_check,
  337. 'history_analysis': history_features
  338. }
  339. anomaly = self.format_anomaly_record(
  340. row=row,
  341. reason=f"大额交易异常: {reason_str},需核查真实性、合法性及资金来源去向",
  342. severity=severity_level,
  343. check_type='large_amount_transaction',
  344. **additional_info
  345. )
  346. identified_anomalies.append(anomaly)
  347. print(f" ❌ 标记为异常: {reason_str}")
  348. else:
  349. print(f" ✅ 未标记为异常")
  350. # ============ 结果统计 ============
  351. print(f"✅ {self.display_name}检查完成")
  352. print(f" 检查交易总数: {len(df)}")
  353. print(f" 大额交易数: {len(large_amount_transactions)}")
  354. print(f" 异常交易数: {len(identified_anomalies)}")
  355. print(f" 通过检查数: {len(large_amount_transactions) - len(identified_anomalies)}")
  356. # 显示大额交易统计
  357. if len(large_amount_transactions) > 0:
  358. print("📋 大额交易统计:")
  359. total_large_amount = large_amount_transactions['txAmount'].sum()
  360. avg_large_amount = large_amount_transactions['txAmount'].mean()
  361. print(f" 总大额金额: ¥{total_large_amount:,.2f}")
  362. print(f" 平均大额金额: ¥{avg_large_amount:,.2f}")
  363. # 按方向统计
  364. income_large = large_amount_transactions[large_amount_transactions['txDirection'] == '收入']
  365. expense_large = large_amount_transactions[large_amount_transactions['txDirection'] == '支出']
  366. print(f" 大额收入: {len(income_large)}笔, ¥{income_large['txAmount'].sum():,.2f}")
  367. print(f" 大额支出: {len(expense_large)}笔, ¥{expense_large['txAmount'].sum():,.2f}")
  368. # 显示异常交易示例
  369. if len(identified_anomalies) > 0:
  370. print("📋 大额异常交易示例:")
  371. for i, anomaly in enumerate(identified_anomalies[:5], 1):
  372. time_str = f"{anomaly.get('txDate', '')} {anomaly.get('txTime', '')}"
  373. print(
  374. f" {i}. ID:{anomaly['txId']} | {time_str} | {anomaly['txDirection']} ¥{anomaly['txAmount']:,.2f} | {anomaly['recognition_reason'][:50]}...")
  375. return {
  376. 'recognition_type': self.display_name,
  377. 'identified_count': len(identified_anomalies),
  378. 'identified_anomalies': identified_anomalies,
  379. 'recognition_status': '完成',
  380. 'recognition_parameters': {
  381. 'amount_threshold': self.amount_threshold,
  382. 'history_days': self.history_days,
  383. 'outlier_multiplier': self.outlier_multiplier,
  384. 'enable_background_check': self.enable_background_check,
  385. 'total_checked': len(df),
  386. 'large_transactions_found': len(large_amount_transactions)
  387. },
  388. 'statistics': {
  389. 'total_transactions': len(df),
  390. 'large_amount_transactions': len(large_amount_transactions),
  391. 'abnormal_large_transactions': len(identified_anomalies),
  392. 'amount_statistics': {
  393. 'max_amount': float(df['txAmount'].max()),
  394. 'min_amount': float(df['txAmount'].min()),
  395. 'avg_amount': float(df['txAmount'].mean()),
  396. 'total_amount': float(df['txAmount'].sum()),
  397. 'large_amount_total': float(large_amount_transactions['txAmount'].sum()),
  398. 'large_amount_avg': float(large_amount_transactions['txAmount'].mean()) if len(
  399. large_amount_transactions) > 0 else 0
  400. },
  401. 'direction_distribution': {
  402. 'income_count': len(df[df['txDirection'] == '收入']),
  403. 'expense_count': len(df[df['txDirection'] == '支出']),
  404. 'large_income_count': len(
  405. large_amount_transactions[large_amount_transactions['txDirection'] == '收入']),
  406. 'large_expense_count': len(
  407. large_amount_transactions[large_amount_transactions['txDirection'] == '支出'])
  408. },
  409. 'background_analysis': {
  410. 'reasonable_background_count': sum(
  411. 1 for t in analyzed_transactions if t['background_check']['has_reasonable_background']),
  412. 'high_risk_count': sum(
  413. 1 for t in analyzed_transactions if t['background_check']['has_high_risk_indicator']),
  414. 'outlier_count': sum(1 for t in analyzed_transactions if t['is_amount_outlier'])
  415. }
  416. }
  417. }
  418. except FileNotFoundError as e:
  419. return {
  420. 'recognition_type': self.display_name,
  421. 'identified_count': 0,
  422. 'identified_anomalies': [],
  423. 'recognition_status': '失败',
  424. 'error': f'文件不存在: {str(e)}'
  425. }
  426. except Exception as e:
  427. import traceback
  428. traceback.print_exc()
  429. return {
  430. 'recognition_type': self.display_name,
  431. 'identified_count': 0,
  432. 'identified_anomalies': [],
  433. 'recognition_status': '失败',
  434. 'error': f'数据加载或处理失败: {str(e)}'
  435. }
  436. def get_summary(self) -> Dict[str, Any]:
  437. """获取识别器摘要"""
  438. summary = super().get_summary()
  439. summary.update({
  440. 'amount_threshold': self.amount_threshold,
  441. 'history_days': self.history_days,
  442. 'outlier_multiplier': self.outlier_multiplier,
  443. 'enable_background_check': self.enable_background_check,
  444. 'reasonable_background_keywords_count': len(self.reasonable_background_keywords),
  445. 'high_risk_keywords_count': len(self.high_risk_keywords),
  446. 'data_loaded': self._data is not None
  447. })
  448. return summary
  449. def get_config_summary(self) -> Dict[str, Any]:
  450. """获取配置摘要"""
  451. return {
  452. "大额阈值": f"¥{self.amount_threshold:,.2f}",
  453. "历史分析天数": f"{self.history_days}天",
  454. "异常倍数阈值": f"{self.outlier_multiplier}倍",
  455. "背景检查": "启用" if self.enable_background_check else "禁用",
  456. "检测逻辑": "大额金额 + 与历史不匹配 + 缺乏合理背景 = 大额交易异常",
  457. "业务规则描述": "单次交易金额超过阈值且与账户历史行为不匹配,缺乏合理背景"
  458. }