occasional_high_integer_transaction_recognizer.py 22 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492
  1. from pydantic import BaseModel, Field
  2. from typing import Dict, Any, Optional, Type
  3. import pandas as pd
  4. from datetime import timedelta
  5. from .enhanced_base_recognizer import EnhancedBaseRecognizer
  6. class OccasionalHighIntegerTransactionInput(BaseModel):
  7. """偶发高额整数交易识别工具输入"""
  8. csv_path: Optional[str] = Field(
  9. None,
  10. description="CSV文件路径(可选)。如果初始化时已提供csv_path,可以不用再次传入。"
  11. )
  12. class Config:
  13. arbitrary_types_allowed = True
  14. class OccasionalHighIntegerTransactionRecognizer(EnhancedBaseRecognizer):
  15. """
  16. 偶发高额整数交易异常识别器
  17. 异常规则定义:
  18. 银行流水核查中,若存在金额为10,000元整数倍的交易,且该类交易金额超过整体流水平均交易金额的5倍
  19. (构成极端异常值),同时此类极端异常交易呈现偶发且高频次发生的特征,可判定为金额维度存在
  20. 异常交易情形,需进一步核查交易真实性。
  21. """
  22. args_schema: Type[BaseModel] = OccasionalHighIntegerTransactionInput
  23. # 整数倍基数
  24. integer_multiple: float = Field(
  25. 10000.0,
  26. description="整数倍基数(元),检查是否为该金额的整数倍"
  27. )
  28. # 异常倍数阈值
  29. outlier_multiplier: float = Field(
  30. 5.0,
  31. description="异常倍数阈值,交易金额超过整体平均交易金额的多少倍视为极端异常值"
  32. )
  33. # 频率分析参数
  34. frequency_window_days: int = Field(
  35. 7,
  36. description="频率分析时间窗口(天),用于判断是否为高频发生"
  37. )
  38. min_occurrences_for_high_frequency: int = Field(
  39. 3,
  40. description="高频最小发生次数,在时间窗口内达到此次数视为高频"
  41. )
  42. # 偶发性判断参数
  43. gap_std_threshold: float = Field(
  44. 2.0,
  45. description="时间间隔标准差阈值,大于此值视为时间不规律(偶发)"
  46. )
  47. # 严重程度配置
  48. severity_level: str = Field(
  49. 'high',
  50. description="异常严重程度(high/medium/low)"
  51. )
  52. def __init__(self, csv_path: str = None, config: Dict[str, Any] = None, **kwargs):
  53. """
  54. 初始化偶发高额整数交易识别器
  55. Args:
  56. csv_path: CSV文件路径
  57. config: 配置参数
  58. **kwargs: 其他参数
  59. """
  60. # 调用父类的 __init__
  61. super().__init__(
  62. name="occasional_high_integer_recognizer",
  63. description="识别银行流水中偶发的高额整数倍交易异常,检测金额为整数倍、超过平均金额5倍且呈现偶发高频特征的交易。",
  64. display_name="偶发高额整数交易异常识别",
  65. csv_path=csv_path,
  66. config=config,
  67. **kwargs
  68. )
  69. # 从config获取配置,更新Field属性
  70. integer_config = self.get_config_value('occasional_integer_transaction', {})
  71. if integer_config:
  72. config_mapping = {
  73. 'integer_multiple': 'integer_multiple',
  74. 'outlier_multiplier': 'outlier_multiplier',
  75. 'frequency_window_days': 'frequency_window_days',
  76. 'min_occurrences_for_high_frequency': 'min_occurrences_for_high_frequency',
  77. 'gap_std_threshold': 'gap_std_threshold',
  78. 'severity_level': 'severity_level'
  79. }
  80. for config_key, attr_name in config_mapping.items():
  81. if config_key in integer_config:
  82. setattr(self, attr_name, integer_config[config_key])
  83. print(f"✅ {self.display_name} 初始化完成")
  84. print(f" 整数倍基数: ¥{self.integer_multiple:,.0f}")
  85. print(f" 异常倍数阈值: {self.outlier_multiplier}倍")
  86. print(f" 频率分析窗口: {self.frequency_window_days}天")
  87. print(f" 高频最小次数: {self.min_occurrences_for_high_frequency}次")
  88. print(f" 偶发判断阈值: 间隔标准差>{self.gap_std_threshold}")
  89. print(f" 严重程度: {self.severity_level}")
  90. def _is_integer_multiple(self, amount: float, tolerance: float = 0.01) -> bool:
  91. """
  92. 判断金额是否为整数倍
  93. Args:
  94. amount: 交易金额
  95. tolerance: 容差(元),解决浮点数精度问题
  96. Returns:
  97. bool: 是否为整数倍
  98. """
  99. if pd.isna(amount):
  100. return False
  101. # 计算余数
  102. remainder = abs(amount % self.integer_multiple)
  103. # 考虑浮点数精度,有两种情况:
  104. # 1. 余数接近于0(如10000 % 10000 = 0)
  105. # 2. 余数接近于整数倍基数(如10000 % 10000 = 0,但浮点误差可能为0.000001)
  106. return remainder < tolerance or abs(remainder - self.integer_multiple) < tolerance
  107. def _analyze_frequency_pattern(self, transactions: pd.DataFrame) -> Dict[str, Any]:
  108. """
  109. 分析交易频率模式
  110. Args:
  111. transactions: 交易数据
  112. Returns:
  113. Dict[str, Any]: 频率分析结果
  114. """
  115. if len(transactions) < 2:
  116. return {
  117. 'is_occasional': False,
  118. 'is_high_frequency': False,
  119. 'total_count': len(transactions),
  120. 'gap_std': 0.0,
  121. 'time_analysis': '数据不足,无法分析频率模式'
  122. }
  123. # 确保按时间排序
  124. sorted_transactions = transactions.sort_values('datetime')
  125. # 1. 计算时间间隔(天数)
  126. time_diffs = sorted_transactions['datetime'].diff().dt.total_seconds() / 86400.0
  127. time_diffs = time_diffs.dropna()
  128. if len(time_diffs) == 0:
  129. return {
  130. 'is_occasional': False,
  131. 'is_high_frequency': False,
  132. 'total_count': len(transactions),
  133. 'gap_std': 0.0,
  134. 'time_analysis': '时间间隔数据不足'
  135. }
  136. # 2. 判断是否偶发(时间间隔不规律)
  137. gap_std = time_diffs.std()
  138. is_occasional = gap_std > self.gap_std_threshold
  139. # 3. 判断是否高频(在一定时间内多次发生)
  140. # 按天统计发生次数
  141. date_counts = sorted_transactions['datetime'].dt.date.value_counts()
  142. # 检查是否有连续发生的情况
  143. dates = sorted(sorted_transactions['datetime'].dt.date.unique())
  144. date_diffs = [(dates[i + 1] - dates[i]).days for i in range(len(dates) - 1)]
  145. # 判断是否有在时间窗口内多次发生
  146. is_high_frequency = False
  147. if len(transactions) >= self.min_occurrences_for_high_frequency:
  148. # 检查是否有在frequency_window_days内达到min_occurrences_for_high_frequency次
  149. sliding_window_counts = []
  150. for i in range(len(dates)):
  151. window_start = dates[i]
  152. window_end = window_start + timedelta(days=self.frequency_window_days)
  153. count_in_window = sum(1 for d in dates if window_start <= d <= window_end)
  154. sliding_window_counts.append(count_in_window)
  155. max_in_window = max(sliding_window_counts) if sliding_window_counts else 0
  156. is_high_frequency = max_in_window >= self.min_occurrences_for_high_frequency
  157. return {
  158. 'is_occasional': is_occasional,
  159. 'is_high_frequency': is_high_frequency,
  160. 'total_count': len(transactions),
  161. 'gap_std': float(gap_std),
  162. 'gap_mean': float(time_diffs.mean()),
  163. 'date_counts': date_counts.to_dict(),
  164. 'unique_dates': len(date_counts),
  165. 'date_range': {
  166. 'start': dates[0].strftime('%Y-%m-%d') if dates else '',
  167. 'end': dates[-1].strftime('%Y-%m-%d') if dates else '',
  168. 'total_days': (dates[-1] - dates[0]).days + 1 if len(dates) > 1 else 1
  169. },
  170. 'time_analysis': f"时间间隔标准差: {gap_std:.2f}天,最大窗口内次数: {max_in_window if 'max_in_window' in locals() else 0}次"
  171. }
  172. def _generate_anomaly_reason(self, row: pd.Series, avg_amount: float,
  173. frequency_info: Dict[str, Any]) -> str:
  174. """
  175. 生成异常原因描述
  176. Args:
  177. row: 交易记录
  178. avg_amount: 整体平均交易金额
  179. frequency_info: 频率分析结果
  180. Returns:
  181. str: 异常原因描述
  182. """
  183. reasons = []
  184. # 整数倍特征
  185. multiple = row['txAmount'] / self.integer_multiple
  186. reasons.append(f"金额为¥{self.integer_multiple:,.0f}的{multiple:.0f}倍整数")
  187. # 极端异常值特征
  188. if avg_amount > 0:
  189. outlier_ratio = abs(row['txAmount']) / avg_amount
  190. reasons.append(f"金额超出整体平均{outlier_ratio:.1f}倍")
  191. # 频率特征
  192. if frequency_info['is_occasional']:
  193. reasons.append(f"交易时间不规律(间隔标准差{frequency_info['gap_std']:.1f}天)")
  194. if frequency_info['is_high_frequency']:
  195. reasons.append(f"高频发生({frequency_info['total_count']}次)")
  196. return f"偶发高额整数交易异常: {','.join(reasons)},需核查交易真实性"
  197. def recognize(self, csv_path: str = None, **kwargs) -> Dict[str, Any]:
  198. """
  199. 识别偶发高额整数交易异常
  200. Args:
  201. csv_path: CSV文件路径
  202. **kwargs: 其他参数
  203. Returns:
  204. Dict[str, Any]: 识别结果
  205. """
  206. try:
  207. # 使用父类的load_data方法加载标准化数据
  208. df = self.load_data(csv_path)
  209. print(f"🔍 {self.display_name}开始检查,共 {len(df)} 条记录")
  210. print(f" 检测规则: 整数倍({self.integer_multiple:,.0f}元) + 极端异常值({self.outlier_multiplier}倍) + 偶发高频 = 异常")
  211. # 检查必需字段
  212. required_fields = ['txId', 'datetime', 'txAmount', 'txDirection']
  213. missing_fields = [field for field in required_fields if field not in df.columns]
  214. if missing_fields:
  215. return {
  216. 'recognition_type': self.display_name,
  217. 'identified_count': 0,
  218. 'identified_anomalies': [],
  219. 'recognition_status': '失败',
  220. 'error': f'缺少必需字段: {missing_fields}'
  221. }
  222. # 确保datetime列已正确解析
  223. if not pd.api.types.is_datetime64_any_dtype(df['datetime']):
  224. df['datetime'] = pd.to_datetime(df['datetime'], errors='coerce')
  225. # 按时间排序,便于频率分析
  226. df = df.sort_values('datetime')
  227. # 计算整体平均交易金额(绝对值)
  228. avg_amount = df['txAmount'].abs().mean()
  229. print(f"📊 整体平均交易金额: ¥{avg_amount:,.2f}")
  230. print(f" 极端异常值阈值: ¥{avg_amount * self.outlier_multiplier:,.2f}")
  231. # ============ 第一步:筛选整数倍交易 ============
  232. integer_mask = df['txAmount'].apply(lambda x: self._is_integer_multiple(abs(x)))
  233. integer_transactions = df[integer_mask].copy()
  234. if len(integer_transactions) == 0:
  235. print(f"📊 未发现{self.integer_multiple:,.0f}元整数倍交易")
  236. return {
  237. 'recognition_type': self.display_name,
  238. 'identified_count': 0,
  239. 'identified_anomalies': [],
  240. 'recognition_status': '完成',
  241. 'statistics': {
  242. 'total_transactions': len(df),
  243. 'integer_transactions': 0,
  244. 'avg_amount': float(avg_amount),
  245. 'integer_multiple': self.integer_multiple
  246. }
  247. }
  248. print(f"📊 发现 {len(integer_transactions)} 笔{self.integer_multiple:,.0f}元整数倍交易")
  249. # ============ 第二步:筛选极端异常值 ============
  250. outlier_threshold = avg_amount * self.outlier_multiplier
  251. outlier_mask = integer_transactions['txAmount'].abs() > outlier_threshold
  252. outlier_transactions = integer_transactions[outlier_mask].copy()
  253. if len(outlier_transactions) == 0:
  254. print(f"📊 未发现极端异常值整数倍交易(金额≤¥{outlier_threshold:,.2f})")
  255. return {
  256. 'recognition_type': self.display_name,
  257. 'identified_count': 0,
  258. 'identified_anomalies': [],
  259. 'recognition_status': '完成',
  260. 'statistics': {
  261. 'total_transactions': len(df),
  262. 'integer_transactions': len(integer_transactions),
  263. 'outlier_transactions': 0,
  264. 'avg_amount': float(avg_amount),
  265. 'outlier_threshold': float(outlier_threshold)
  266. }
  267. }
  268. print(f"📊 发现 {len(outlier_transactions)} 笔极端异常值整数倍交易(金额>¥{outlier_threshold:,.2f})")
  269. # ============ 第三步:分析频率模式 ============
  270. frequency_info = self._analyze_frequency_pattern(outlier_transactions)
  271. print(f"📈 频率分析结果:")
  272. print(f" 总次数: {frequency_info['total_count']}")
  273. print(f" 是否偶发: {'是' if frequency_info['is_occasional'] else '否'} (标准差={frequency_info['gap_std']:.2f}天)")
  274. print(f" 是否高频: {'是' if frequency_info['is_high_frequency'] else '否'}")
  275. print(f" 涉及天数: {frequency_info['unique_dates']}天")
  276. # ============ 第四步:识别异常交易 ============
  277. identified_anomalies = []
  278. # 只有同时满足偶发且高频才标记为异常
  279. if frequency_info['is_occasional'] and frequency_info['is_high_frequency']:
  280. print(f"⚠️ 检测到偶发且高频的高额整数交易,开始标记异常...")
  281. for idx, row in outlier_transactions.iterrows():
  282. # 生成异常原因
  283. reason = self._generate_anomaly_reason(row, avg_amount, frequency_info)
  284. # 额外信息
  285. additional_info = {
  286. 'frequency_analysis': frequency_info,
  287. 'amount_analysis': {
  288. 'integer_multiple': self.integer_multiple,
  289. 'outlier_multiplier': self.outlier_multiplier,
  290. 'avg_amount': avg_amount,
  291. 'outlier_threshold': outlier_threshold,
  292. 'outlier_ratio': abs(row['txAmount']) / avg_amount if avg_amount > 0 else 0
  293. },
  294. 'integer_analysis': {
  295. 'multiple_count': row['txAmount'] / self.integer_multiple,
  296. 'is_integer_multiple': True
  297. }
  298. }
  299. # 生成异常记录
  300. anomaly = self.format_anomaly_record(
  301. row=row,
  302. reason=reason,
  303. severity=self.severity_level,
  304. check_type='occasional_high_integer_transaction',
  305. **additional_info
  306. )
  307. identified_anomalies.append(anomaly)
  308. print(f" ❌ 标记异常: ID={row['txId']}, ¥{row['txAmount']:,.2f}, {reason[:60]}...")
  309. else:
  310. print(f"✅ 未检测到偶发高频模式,不标记异常")
  311. print(f" 偶发性: {frequency_info['is_occasional']}")
  312. print(f" 高频性: {frequency_info['is_high_frequency']}")
  313. # ============ 第五步:结果统计 ============
  314. print(f"✅ {self.display_name}检查完成")
  315. print(f" 检查交易总数: {len(df)}")
  316. print(f" 整数倍交易数: {len(integer_transactions)}")
  317. print(f" 极端异常值数: {len(outlier_transactions)}")
  318. print(f" 异常交易数: {len(identified_anomalies)}")
  319. # 显示整数倍交易统计
  320. if len(integer_transactions) > 0:
  321. print("📋 整数倍交易统计:")
  322. total_integer_amount = integer_transactions['txAmount'].sum()
  323. avg_integer_amount = integer_transactions['txAmount'].mean()
  324. print(f" 总整数倍金额: ¥{total_integer_amount:,.2f}")
  325. print(f" 平均整数倍金额: ¥{avg_integer_amount:,.2f}")
  326. # 倍数分布
  327. integer_transactions['multiple'] = (integer_transactions['txAmount'] / self.integer_multiple).round()
  328. multiple_counts = integer_transactions['multiple'].value_counts().sort_index()
  329. print(f" 倍数分布: {dict(multiple_counts.head(10))}" +
  330. (", ..." if len(multiple_counts) > 10 else ""))
  331. # 按方向统计
  332. income_integer = integer_transactions[integer_transactions['txDirection'] == '收入']
  333. expense_integer = integer_transactions[integer_transactions['txDirection'] == '支出']
  334. print(f" 整数倍收入: {len(income_integer)}笔, ¥{income_integer['txAmount'].sum():,.2f}")
  335. print(f" 整数倍支出: {len(expense_integer)}笔, ¥{expense_integer['txAmount'].sum():,.2f}")
  336. return {
  337. 'recognition_type': self.display_name,
  338. 'identified_count': len(identified_anomalies),
  339. 'identified_anomalies': identified_anomalies,
  340. 'recognition_status': '完成',
  341. 'recognition_parameters': {
  342. 'integer_multiple': self.integer_multiple,
  343. 'outlier_multiplier': self.outlier_multiplier,
  344. 'frequency_window_days': self.frequency_window_days,
  345. 'min_occurrences_for_high_frequency': self.min_occurrences_for_high_frequency,
  346. 'gap_std_threshold': self.gap_std_threshold,
  347. 'severity_level': self.severity_level,
  348. 'avg_amount': float(avg_amount)
  349. },
  350. 'statistics': {
  351. 'total_transactions': len(df),
  352. 'integer_transactions': len(integer_transactions),
  353. 'outlier_transactions': len(outlier_transactions),
  354. 'anomalous_transactions': len(identified_anomalies),
  355. 'frequency_analysis': frequency_info,
  356. 'amount_statistics': {
  357. 'overall_avg': float(avg_amount),
  358. 'overall_max': float(df['txAmount'].max()),
  359. 'overall_min': float(df['txAmount'].min()),
  360. 'integer_avg': float(integer_transactions['txAmount'].mean()) if len(
  361. integer_transactions) > 0 else 0,
  362. 'integer_total': float(integer_transactions['txAmount'].sum()) if len(
  363. integer_transactions) > 0 else 0,
  364. 'outlier_avg': float(outlier_transactions['txAmount'].mean()) if len(
  365. outlier_transactions) > 0 else 0
  366. },
  367. 'direction_distribution': {
  368. 'integer_income': len(integer_transactions[integer_transactions['txDirection'] == '收入']),
  369. 'integer_expense': len(integer_transactions[integer_transactions['txDirection'] == '支出']),
  370. 'outlier_income': len(outlier_transactions[outlier_transactions['txDirection'] == '收入']),
  371. 'outlier_expense': len(outlier_transactions[outlier_transactions['txDirection'] == '支出'])
  372. }
  373. }
  374. }
  375. except FileNotFoundError as e:
  376. return {
  377. 'recognition_type': self.display_name,
  378. 'identified_count': 0,
  379. 'identified_anomalies': [],
  380. 'recognition_status': '失败',
  381. 'error': f'文件不存在: {str(e)}'
  382. }
  383. except Exception as e:
  384. import traceback
  385. traceback.print_exc()
  386. return {
  387. 'recognition_type': self.display_name,
  388. 'identified_count': 0,
  389. 'identified_anomalies': [],
  390. 'recognition_status': '失败',
  391. 'error': f'数据加载或处理失败: {str(e)}'
  392. }
  393. def get_summary(self) -> Dict[str, Any]:
  394. """获取识别器摘要"""
  395. summary = super().get_summary()
  396. summary.update({
  397. 'integer_multiple': self.integer_multiple,
  398. 'outlier_multiplier': self.outlier_multiplier,
  399. 'frequency_window_days': self.frequency_window_days,
  400. 'min_occurrences_for_high_frequency': self.min_occurrences_for_high_frequency,
  401. 'gap_std_threshold': self.gap_std_threshold,
  402. 'severity_level': self.severity_level,
  403. 'data_loaded': self._data is not None
  404. })
  405. return summary
  406. def get_config_summary(self) -> Dict[str, Any]:
  407. """获取配置摘要"""
  408. return {
  409. "整数倍基数": f"¥{self.integer_multiple:,.0f}元",
  410. "异常倍数阈值": f"{self.outlier_multiplier}倍",
  411. "频率分析窗口": f"{self.frequency_window_days}天",
  412. "高频最小次数": f"{self.min_occurrences_for_high_frequency}次",
  413. "偶发判断阈值": f"标准差>{self.gap_std_threshold}天",
  414. "检测逻辑": f"整数倍(¥{self.integer_multiple:,.0f}) + 极端异常值({self.outlier_multiplier}倍) + 偶发高频 = 异常",
  415. "严重程度": self.severity_level.upper(),
  416. "业务规则描述": "金额为整数倍、超过平均金额5倍且呈现偶发高频特征的交易需核查真实性"
  417. }