night_transaction_recognizer.py 19 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473
  1. from pydantic import BaseModel, Field
  2. from typing import Dict, Any, Optional, Type, List, Set
  3. import pandas as pd
  4. from .enhanced_base_recognizer import EnhancedBaseRecognizer
  5. class NightTransactionInput(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 NightTransactionRecognizer(EnhancedBaseRecognizer):
  14. """
  15. 夜间交易(2-5点)异常识别器
  16. 基于图2规则定义:
  17. 1. 时间范围:特指凌晨2点至5点期间发生的交易
  18. 2. 时段性质:偏离日常资金往来常规时段,通常为非经营、非生活活动的常规休息时段
  19. 3. 异常判定:
  20. - 频繁出现资金收付记录
  21. - 交易金额较大
  22. - 缺乏合理交易背景
  23. 4. 合理性例外:特定行业经营需求、紧急资金周转等
  24. 核心逻辑:2-5点交易 + (高频或大额) + (无合理背景) = 异常夜间交易
  25. """
  26. args_schema: Type[BaseModel] = NightTransactionInput
  27. # 配置参数
  28. night_start_hour: int = Field(
  29. 2,
  30. description="夜间检测开始小时(0-23),默认2点"
  31. )
  32. night_end_hour: int = Field(
  33. 5,
  34. description="夜间检测结束小时(0-23),默认5点"
  35. )
  36. frequency_threshold_per_hour: int = Field(
  37. 3,
  38. description="高频交易阈值,每小时超过此笔数视为高频"
  39. )
  40. large_amount_threshold: float = Field(
  41. 50000.0,
  42. description="大额交易阈值(元),超过此金额的夜间交易视为大额异常"
  43. )
  44. # 行业特征关键词(24小时营业或夜间经营行业)
  45. night_industry_keywords: List[str] = Field(
  46. [
  47. "酒店", "宾馆", "KTV", "酒吧", "夜总会", "网吧", "便利店",
  48. "医院", "急救", "急诊", "消防", "公安", "保安", "物流",
  49. "运输", "出租车", "网约车", "外卖", "配送"
  50. ],
  51. description="夜间经营行业关键词,用于识别可能有合理背景的交易"
  52. )
  53. # 紧急情况关键词
  54. emergency_keywords: List[str] = Field(
  55. [
  56. "急救", "急诊", "抢救", "紧急", "urgent", "emergency",
  57. "抢险", "救援", "救灾", "应急", "加急"
  58. ],
  59. description="紧急情况关键词,用于识别可能有合理背景的交易"
  60. )
  61. def __init__(self, csv_path: str = None, config: Dict[str, Any] = None, **kwargs):
  62. """
  63. 初始化夜间交易识别器
  64. Args:
  65. csv_path: CSV文件路径
  66. config: 配置参数
  67. **kwargs: 其他参数
  68. """
  69. # 调用父类的 __init__
  70. super().__init__(
  71. name="night_transaction_recognizer",
  72. description="识别银行流水中的夜间交易异常(2-5点),检测高频、大额等异常特征。",
  73. display_name="夜间交易异常识别",
  74. csv_path=csv_path,
  75. config=config,
  76. **kwargs
  77. )
  78. # 从config获取配置,更新Field属性
  79. night_config = self.get_config_value('night_transaction', {})
  80. if night_config:
  81. config_mapping = {
  82. 'night_start_hour': 'night_start_hour',
  83. 'night_end_hour': 'night_end_hour',
  84. 'frequency_threshold_per_hour': 'frequency_threshold_per_hour',
  85. 'large_amount_threshold': 'large_amount_threshold',
  86. 'night_industry_keywords': 'night_industry_keywords',
  87. 'emergency_keywords': 'emergency_keywords'
  88. }
  89. for config_key, attr_name in config_mapping.items():
  90. if config_key in night_config:
  91. setattr(self, attr_name, night_config[config_key])
  92. # 验证时间配置
  93. self._validate_time_config()
  94. print(f"✅ {self.display_name} 初始化完成")
  95. print(f" 夜间时段: {self.night_start_hour:02d}:00 - {self.night_end_hour:02d}:00")
  96. print(f" 高频阈值: {self.frequency_threshold_per_hour}笔/小时")
  97. print(f" 大额阈值: ¥{self.large_amount_threshold:,.2f}")
  98. print(f" 夜间行业关键词: {len(self.night_industry_keywords)}个")
  99. print(f" 紧急情况关键词: {len(self.emergency_keywords)}个")
  100. def _validate_time_config(self):
  101. """验证时间配置合理性"""
  102. if not (0 <= self.night_start_hour <= 23):
  103. raise ValueError(f"夜间开始小时必须在0-23之间: {self.night_start_hour}")
  104. if not (0 <= self.night_end_hour <= 23):
  105. raise ValueError(f"夜间结束小时必须在0-23之间: {self.night_end_hour}")
  106. # 确保开始时间早于结束时间
  107. if self.night_start_hour >= self.night_end_hour:
  108. print(f"⚠️ 注意:夜间开始时间({self.night_start_hour}:00) >= 结束时间({self.night_end_hour}:00),"
  109. f"将按跨午夜处理")
  110. def _is_in_night_period(self, hour: int) -> bool:
  111. """
  112. 判断小时数是否在夜间时段内
  113. Args:
  114. hour: 小时数(0-23)
  115. Returns:
  116. bool: 是否在夜间时段
  117. """
  118. if self.night_start_hour < self.night_end_hour:
  119. # 正常情况:开始时间 < 结束时间
  120. return self.night_start_hour <= hour < self.night_end_hour
  121. else:
  122. # 跨午夜情况:开始时间 >= 结束时间
  123. return hour >= self.night_start_hour or hour < self.night_end_hour
  124. def _has_reasonable_background(self, row: pd.Series) -> bool:
  125. """
  126. 判断交易是否有合理背景
  127. 根据图2规则,合理的背景包括:
  128. 1. 夜间特定行业经营需求
  129. 2. 紧急资金周转
  130. Args:
  131. row: 交易记录
  132. Returns:
  133. bool: 是否有合理背景
  134. """
  135. # 检查交易摘要中的关键词
  136. summary = str(row.get('txSummary', '')).lower()
  137. counterparty = str(row.get('txCounterparty', '')).lower()
  138. # 合并检查文本
  139. check_text = f"{summary} {counterparty}"
  140. # 1. 检查是否为夜间行业
  141. for keyword in self.night_industry_keywords:
  142. if keyword.lower() in check_text:
  143. return True
  144. # 2. 检查是否为紧急情况
  145. for keyword in self.emergency_keywords:
  146. if keyword.lower() in check_text:
  147. return True
  148. # 3. 可以扩展其他合理背景检查逻辑
  149. return False
  150. def _detect_high_frequency_transactions(self,
  151. night_transactions: pd.DataFrame,
  152. date_col: str = 'date_only') -> Set[str]:
  153. """
  154. 检测高频夜间交易
  155. Args:
  156. night_transactions: 夜间交易数据
  157. date_col: 日期列名
  158. Returns:
  159. Set[str]: 高频交易ID集合
  160. """
  161. high_freq_ids = set()
  162. if len(night_transactions) == 0:
  163. return high_freq_ids
  164. # 按日期和小时分组统计
  165. if 'hour' in night_transactions.columns and date_col in night_transactions.columns:
  166. # 统计每小时交易笔数
  167. hourly_counts = night_transactions.groupby([date_col, 'hour']).size()
  168. for (trans_date, trans_hour), count in hourly_counts.items():
  169. if count > self.frequency_threshold_per_hour:
  170. # 获取该小时的所有交易
  171. mask = (night_transactions[date_col] == trans_date) & \
  172. (night_transactions['hour'] == trans_hour)
  173. hour_transactions = night_transactions[mask]
  174. # 收集交易ID
  175. for tx_id in hour_transactions['txId'].unique():
  176. high_freq_ids.add(str(tx_id))
  177. print(f" ⚠️ {trans_date} {trans_hour:02d}:00-{trans_hour + 1:02d}:00: "
  178. f"{count}笔交易,超过阈值{self.frequency_threshold_per_hour}笔")
  179. return high_freq_ids
  180. def _detect_large_amount_transactions(self, night_transactions: pd.DataFrame) -> Set[str]:
  181. """
  182. 检测大额夜间交易
  183. Args:
  184. night_transactions: 夜间交易数据
  185. Returns:
  186. Set[str]: 大额交易ID集合
  187. """
  188. large_amount_ids = set()
  189. if len(night_transactions) == 0:
  190. return large_amount_ids
  191. # 筛选大额交易
  192. large_amount_mask = night_transactions['txAmount'].abs() >= self.large_amount_threshold
  193. large_amount_tx = night_transactions[large_amount_mask]
  194. for _, row in large_amount_tx.iterrows():
  195. tx_id = str(row['txId'])
  196. large_amount_ids.add(tx_id)
  197. if len(large_amount_ids) > 0:
  198. print(f" ⚠️ 发现 {len(large_amount_ids)} 笔大额夜间交易(≥¥{self.large_amount_threshold:,.2f})")
  199. return large_amount_ids
  200. def recognize(self, csv_path: str = None, **kwargs) -> Dict[str, Any]:
  201. """
  202. 识别夜间交易异常
  203. Args:
  204. csv_path: CSV文件路径
  205. **kwargs: 其他参数
  206. Returns:
  207. Dict[str, Any]: 识别结果
  208. """
  209. try:
  210. # 使用父类的load_data方法加载标准化数据
  211. df = self.load_data(csv_path)
  212. print(f"🔍 {self.display_name}开始检查,共 {len(df)} 条记录")
  213. print(f" 夜间时段: {self.night_start_hour:02d}:00 - {self.night_end_hour:02d}:00")
  214. print(f" 检测规则: 夜间交易 + (高频或大额) + (无合理背景) = 异常")
  215. # 检查必需字段
  216. required_fields = ['txId', 'datetime', 'txAmount', 'txDirection']
  217. missing_fields = [field for field in required_fields if field not in df.columns]
  218. if missing_fields:
  219. return {
  220. 'recognition_type': self.display_name,
  221. 'identified_count': 0,
  222. 'identified_anomalies': [],
  223. 'recognition_status': '失败',
  224. 'error': f'缺少必需字段: {missing_fields}'
  225. }
  226. # 确保datetime列已正确解析
  227. if not pd.api.types.is_datetime64_any_dtype(df['datetime']):
  228. df['datetime'] = pd.to_datetime(df['datetime'], errors='coerce')
  229. # 提取时间组件
  230. df['hour'] = df['datetime'].dt.hour
  231. df['minute'] = df['datetime'].dt.minute
  232. df['date_only'] = df['datetime'].dt.date
  233. # ============ 识别所有夜间交易 ============
  234. night_mask = df['hour'].apply(self._is_in_night_period)
  235. night_transactions = df[night_mask].copy()
  236. if len(night_transactions) == 0:
  237. print(f"✅ 未发现夜间交易({self.night_start_hour:02d}:00-{self.night_end_hour:02d}:00)")
  238. return {
  239. 'recognition_type': self.display_name,
  240. 'identified_count': 0,
  241. 'identified_anomalies': [],
  242. 'recognition_status': '完成',
  243. 'recognition_parameters': {
  244. 'night_period': f"{self.night_start_hour:02d}:00-{self.night_end_hour:02d}:00",
  245. 'frequency_threshold': f"{self.frequency_threshold_per_hour}笔/小时",
  246. 'large_amount_threshold': self.large_amount_threshold,
  247. 'total_checked': len(df),
  248. 'night_transactions_found': 0
  249. },
  250. 'statistics': {
  251. 'total_transactions': len(df),
  252. 'night_transaction_count': 0,
  253. 'night_transaction_ratio': 0.0
  254. }
  255. }
  256. print(f"📊 发现 {len(night_transactions)} 笔夜间交易")
  257. # ============ 合理性背景分析 ============
  258. reasonable_transactions = []
  259. for _, row in night_transactions.iterrows():
  260. if self._has_reasonable_background(row):
  261. reasonable_transactions.append(row['txId'])
  262. if reasonable_transactions:
  263. print(f"📊 其中 {len(reasonable_transactions)} 笔交易可能有合理背景(夜间行业/紧急情况)")
  264. # ============ 异常特征检测 ============
  265. identified_anomalies = []
  266. # 1. 检测高频交易
  267. high_freq_ids = self._detect_high_frequency_transactions(night_transactions)
  268. # 2. 检测大额交易
  269. large_amount_ids = self._detect_large_amount_transactions(night_transactions)
  270. # 3. 合并异常交易ID(排除有合理背景的)
  271. abnormal_ids = (high_freq_ids | large_amount_ids) - set(reasonable_transactions)
  272. # 4. 生成异常记录
  273. for tx_id in abnormal_ids:
  274. mask = night_transactions['txId'] == tx_id
  275. if mask.any():
  276. row = night_transactions[mask].iloc[0]
  277. # 判断异常类型
  278. is_high_freq = tx_id in high_freq_ids
  279. is_large_amount = tx_id in large_amount_ids
  280. # 生成异常原因
  281. reasons = []
  282. if is_high_freq:
  283. reasons.append("高频夜间交易")
  284. if is_large_amount:
  285. reasons.append(f"大额夜间交易(¥{row['txAmount']:,.2f}≥¥{self.large_amount_threshold:,.2f})")
  286. reason_str = ",".join(reasons)
  287. # 检查是否有合理背景但依然被标记为异常的原因
  288. additional_info = {
  289. 'hour': int(row['hour']),
  290. 'is_high_frequency': is_high_freq,
  291. 'is_large_amount': is_large_amount,
  292. 'amount': float(row['txAmount']),
  293. 'has_reasonable_background': str(row['txId']) in reasonable_transactions
  294. }
  295. anomaly = self.format_anomaly_record(
  296. row=row,
  297. reason=f"夜间{reason_str},缺乏合理交易背景",
  298. severity='high' if is_large_amount else 'medium',
  299. check_type='night_transaction_abnormal',
  300. **additional_info
  301. )
  302. identified_anomalies.append(anomaly)
  303. # ============ 结果统计和汇总 ============
  304. print(f"✅ {self.display_name}检查完成")
  305. print(f" 检查交易总数: {len(df)}")
  306. print(f" 夜间交易数: {len(night_transactions)}")
  307. print(f" 有合理背景: {len(reasonable_transactions)}")
  308. print(f" 高频异常: {len(high_freq_ids)}")
  309. print(f" 大额异常: {len(large_amount_ids)}")
  310. print(f" 最终异常数: {len(identified_anomalies)}")
  311. # 显示夜间交易时间分布
  312. if len(night_transactions) > 0:
  313. hour_distribution = night_transactions['hour'].value_counts().sort_index()
  314. print("📋 夜间交易时间分布:")
  315. for hour, count in hour_distribution.items():
  316. print(f" {hour:02d}:00-{hour + 1:02d}:00: {count}笔")
  317. # 显示前5笔异常交易详情
  318. if len(identified_anomalies) > 0:
  319. print("📋 异常夜间交易示例:")
  320. for i, anomaly in enumerate(identified_anomalies[:5], 1):
  321. time_str = f"{anomaly.get('txDate', '')} {anomaly.get('txTime', '')}"
  322. print(
  323. f" {i}. ID:{anomaly['txId']} | {time_str} | ¥{anomaly['txAmount']:,.2f} | {anomaly['recognition_reason']}")
  324. return {
  325. 'recognition_type': self.display_name,
  326. 'identified_count': len(identified_anomalies),
  327. 'identified_anomalies': identified_anomalies,
  328. 'recognition_status': '完成',
  329. 'recognition_parameters': {
  330. 'night_period': f"{self.night_start_hour:02d}:00-{self.night_end_hour:02d}:00",
  331. 'frequency_threshold_per_hour': self.frequency_threshold_per_hour,
  332. 'large_amount_threshold': self.large_amount_threshold,
  333. 'night_industry_keywords_count': len(self.night_industry_keywords),
  334. 'emergency_keywords_count': len(self.emergency_keywords),
  335. 'total_checked': len(df),
  336. 'night_transactions_found': len(night_transactions),
  337. 'reasonable_transactions': len(reasonable_transactions)
  338. },
  339. 'statistics': {
  340. 'total_transactions': len(df),
  341. 'night_transaction_count': len(night_transactions),
  342. 'night_transaction_ratio': len(night_transactions) / max(1, len(df)),
  343. 'reasonable_transaction_count': len(reasonable_transactions),
  344. 'high_frequency_count': len(high_freq_ids),
  345. 'large_amount_count': len(large_amount_ids),
  346. 'abnormal_night_transaction_count': len(identified_anomalies),
  347. 'hour_distribution': {
  348. str(hour): int(count)
  349. for hour, count in night_transactions['hour'].value_counts().items()
  350. } if len(night_transactions) > 0 else {}
  351. }
  352. }
  353. except FileNotFoundError as e:
  354. return {
  355. 'recognition_type': self.display_name,
  356. 'identified_count': 0,
  357. 'identified_anomalies': [],
  358. 'recognition_status': '失败',
  359. 'error': f'文件不存在: {str(e)}'
  360. }
  361. except Exception as e:
  362. import traceback
  363. traceback.print_exc()
  364. return {
  365. 'recognition_type': self.display_name,
  366. 'identified_count': 0,
  367. 'identified_anomalies': [],
  368. 'recognition_status': '失败',
  369. 'error': f'数据加载或处理失败: {str(e)}'
  370. }
  371. def get_summary(self) -> Dict[str, Any]:
  372. """获取识别器摘要"""
  373. summary = super().get_summary()
  374. summary.update({
  375. 'night_start_hour': self.night_start_hour,
  376. 'night_end_hour': self.night_end_hour,
  377. 'frequency_threshold_per_hour': self.frequency_threshold_per_hour,
  378. 'large_amount_threshold': self.large_amount_threshold,
  379. 'night_industry_keywords_count': len(self.night_industry_keywords),
  380. 'emergency_keywords_count': len(self.emergency_keywords),
  381. 'data_loaded': self._data is not None
  382. })
  383. return summary
  384. def get_config_summary(self) -> Dict[str, Any]:
  385. """获取配置摘要"""
  386. return {
  387. "夜间时段": f"{self.night_start_hour:02d}:00 - {self.night_end_hour:02d}:00",
  388. "高频阈值": f"{self.frequency_threshold_per_hour}笔/小时",
  389. "大额阈值": f"¥{self.large_amount_threshold:,.2f}",
  390. "夜间行业关键词": f"{len(self.night_industry_keywords)}个",
  391. "紧急情况关键词": f"{len(self.emergency_keywords)}个",
  392. "检测逻辑": "夜间交易 + (高频或大额) + (无合理背景) = 异常"
  393. }