high_frequency_transaction_recognizer.py 15 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354
  1. from pydantic import BaseModel, Field
  2. from typing import Dict, Any, Optional, Type
  3. import pandas as pd
  4. from .enhanced_base_recognizer import EnhancedBaseRecognizer
  5. class HighFrequencyInput(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 HighFrequencyTransactionRecognizer(EnhancedBaseRecognizer):
  14. """
  15. 高频交易异常识别器
  16. 严格遵循业务规则定义:
  17. 银行流水高频交易(按小时维度)的定义为:
  18. 以"日期 + 小时"为统计单位,通过汇总该时间间隔内的交易笔数与交易金额,
  19. 识别出单小时交易笔数超过 10 笔的特定时段组合,
  20. 此类在短时间内(1 小时)集中发生、交易频次密集的资金收付行为,
  21. 即为银行流水高频交易,其核心特征是交易笔数在单位小时内达到预设阈值(10 笔),
  22. 体现出资金往来的集中性与活跃度异常。
  23. """
  24. args_schema: Type[BaseModel] = HighFrequencyInput
  25. # 配置参数 - 严格按照业务规则
  26. frequency_threshold: int = Field(
  27. 10,
  28. description="高频交易阈值,每小时超过此笔数视为高频交易(业务规则要求10笔)"
  29. )
  30. def __init__(self, csv_path: str = None, config: Dict[str, Any] = None, **kwargs):
  31. """
  32. 初始化高频交易识别器
  33. Args:
  34. csv_path: CSV文件路径
  35. config: 配置参数
  36. **kwargs: 其他参数
  37. """
  38. # 调用父类的 __init__
  39. super().__init__(
  40. name="high_frequency_recognizer",
  41. description="识别银行流水中的高频交易异常,严格按照业务规则:单小时交易笔数超过10笔即为高频异常。",
  42. display_name="高频交易异常识别",
  43. csv_path=csv_path,
  44. config=config,
  45. **kwargs
  46. )
  47. # 从config获取配置
  48. high_freq_config = self.get_config_value('high_frequency', {})
  49. if high_freq_config and 'frequency_threshold' in high_freq_config:
  50. self.frequency_threshold = high_freq_config['frequency_threshold']
  51. print(f"✅ {self.display_name} 初始化完成")
  52. print(f" 严格遵循业务规则:单小时交易笔数 > {self.frequency_threshold}笔 = 高频异常")
  53. def _calculate_hourly_statistics(self, df: pd.DataFrame) -> pd.DataFrame:
  54. """
  55. 计算每小时的交易统计数据
  56. Args:
  57. df: 标准化后的交易数据
  58. Returns:
  59. pd.DataFrame: 每小时统计结果
  60. """
  61. # 提取日期和小时
  62. df['date_hour'] = df['datetime'].dt.strftime('%Y-%m-%d %H')
  63. df['date'] = df['datetime'].dt.date
  64. df['hour'] = df['datetime'].dt.hour
  65. # 按照业务规则:以"日期 + 小时"为统计单位
  66. hour_stats = []
  67. # 按日期+小时分组
  68. for (date_val, hour_val), group in df.groupby(['date', 'hour']):
  69. transaction_count = len(group)
  70. amount_sum = group['txAmount'].sum()
  71. # 统计交易方向分布
  72. direction_counts = group['txDirection'].value_counts().to_dict()
  73. hour_stats.append({
  74. 'date': date_val,
  75. 'hour': hour_val,
  76. 'transaction_count': transaction_count,
  77. 'amount_sum': amount_sum,
  78. 'amount_avg': amount_sum / transaction_count if transaction_count > 0 else 0,
  79. 'in_count': direction_counts.get('收入', 0),
  80. 'out_count': direction_counts.get('支出', 0)
  81. })
  82. return pd.DataFrame(hour_stats)
  83. def _identify_high_frequency_periods(self, hour_stats: pd.DataFrame) -> pd.DataFrame:
  84. """
  85. 识别高频交易时段
  86. Args:
  87. hour_stats: 每小时统计数据
  88. Returns:
  89. pd.DataFrame: 标记了高频时段的结果
  90. """
  91. # 严格按照业务规则:单小时交易笔数超过10笔
  92. hour_stats['is_high_frequency'] = hour_stats['transaction_count'] > self.frequency_threshold
  93. return hour_stats
  94. def _get_transactions_in_period(self, df: pd.DataFrame, date_val: pd.Timestamp.date, hour_val: int) -> pd.DataFrame:
  95. """
  96. 获取指定时段内的所有交易
  97. Args:
  98. df: 原始交易数据
  99. date_val: 日期
  100. hour_val: 小时
  101. Returns:
  102. pd.DataFrame: 指定时段内的交易
  103. """
  104. return df[
  105. (df['datetime'].dt.date == date_val) &
  106. (df['datetime'].dt.hour == hour_val)
  107. ].copy()
  108. def recognize(self, csv_path: str = None, **kwargs) -> Dict[str, Any]:
  109. """
  110. 识别高频交易异常 - 严格按照业务规则
  111. Args:
  112. csv_path: CSV文件路径
  113. **kwargs: 其他参数
  114. Returns:
  115. Dict[str, Any]: 识别结果
  116. """
  117. try:
  118. # 使用父类的load_data方法加载标准化数据
  119. df = self.load_data(csv_path)
  120. print(f"🔍 {self.display_name}开始检查,共 {len(df)} 条记录")
  121. print(f" 业务规则: 以'日期 + 小时'为统计单位")
  122. print(f" 高频阈值: 单小时交易笔数 > {self.frequency_threshold}笔")
  123. # 检查必需字段
  124. required_fields = ['txId', 'datetime', 'txAmount', 'txDirection']
  125. missing_fields = [field for field in required_fields if field not in df.columns]
  126. if missing_fields:
  127. return {
  128. 'recognition_type': self.display_name,
  129. 'identified_count': 0,
  130. 'identified_anomalies': [],
  131. 'recognition_status': '失败',
  132. 'error': f'缺少必需字段: {missing_fields}'
  133. }
  134. # 确保datetime列已正确解析
  135. if not pd.api.types.is_datetime64_any_dtype(df['datetime']):
  136. df['datetime'] = pd.to_datetime(df['datetime'], errors='coerce')
  137. # 检查是否有无效的时间数据
  138. invalid_times = df['datetime'].isna().sum()
  139. if invalid_times > 0:
  140. print(f"⚠️ 警告: 有 {invalid_times} 条记录的时间解析失败,将跳过这些记录")
  141. df = df[df['datetime'].notna()]
  142. if len(df) == 0:
  143. return {
  144. 'recognition_type': self.display_name,
  145. 'identified_count': 0,
  146. 'identified_anomalies': [],
  147. 'recognition_status': '完成',
  148. 'statistics': {'total_valid_transactions': 0}
  149. }
  150. # ============ 按照业务规则:以"日期 + 小时"为统计单位 ============
  151. hour_stats = self._calculate_hourly_statistics(df)
  152. if len(hour_stats) == 0:
  153. return {
  154. 'recognition_type': self.display_name,
  155. 'identified_count': 0,
  156. 'identified_anomalies': [],
  157. 'recognition_status': '完成',
  158. 'statistics': {
  159. 'total_transactions': len(df),
  160. 'hour_periods': 0,
  161. 'high_frequency_periods': 0
  162. }
  163. }
  164. print(f"📊 按'日期+小时'分组统计完成,共 {len(hour_stats)} 个时段")
  165. # ============ 识别高频时段 ============
  166. hour_stats = self._identify_high_frequency_periods(hour_stats)
  167. # 高频时段统计
  168. high_freq_periods = hour_stats[hour_stats['is_high_frequency']]
  169. print(f"📊 发现 {len(high_freq_periods)} 个高频时段(> {self.frequency_threshold}笔/小时)")
  170. # ============ 生成异常记录 ============
  171. identified_anomalies = []
  172. # 为每个高频时段创建异常记录
  173. for _, period_row in high_freq_periods.iterrows():
  174. date_val = period_row['date']
  175. hour_val = period_row['hour']
  176. # 获取该时段内的所有交易
  177. period_transactions = self._get_transactions_in_period(df, date_val, hour_val)
  178. # 为每笔交易创建异常记录
  179. for _, tx_row in period_transactions.iterrows():
  180. # 生成异常原因 - 严格按照业务规则描述
  181. reason = f"属于高频交易时段:{date_val} {hour_val:02d}:00-{hour_val + 1:02d}:00,该时段共{period_row['transaction_count']}笔交易,超过阈值{self.frequency_threshold}笔,体现资金往来的集中性与活跃度异常"
  182. # 额外信息
  183. additional_info = {
  184. 'period_info': {
  185. 'date': date_val.strftime('%Y-%m-%d'),
  186. 'hour': int(hour_val),
  187. 'start_time': f"{hour_val:02d}:00",
  188. 'end_time': f"{hour_val + 1:02d}:00",
  189. 'transaction_count': int(period_row['transaction_count']),
  190. 'amount_sum': float(period_row['amount_sum']),
  191. 'amount_avg': float(period_row['amount_avg']),
  192. 'in_count': int(period_row['in_count']),
  193. 'out_count': int(period_row['out_count'])
  194. },
  195. 'business_rule': {
  196. 'statistic_unit': "日期 + 小时",
  197. 'threshold': self.frequency_threshold,
  198. 'description': "单小时交易笔数超过阈值,体现资金往来的集中性与活跃度异常"
  199. }
  200. }
  201. anomaly = self.format_anomaly_record(
  202. row=tx_row,
  203. reason=reason,
  204. severity='high', # 高频交易通常视为高风险
  205. check_type='high_frequency_transaction',
  206. **additional_info
  207. )
  208. identified_anomalies.append(anomaly)
  209. # ============ 结果统计 ============
  210. print(f"✅ {self.display_name}检查完成")
  211. print(f" 检查交易总数: {len(df)}")
  212. print(f" 统计时段总数: {len(hour_stats)}")
  213. print(f" 高频时段数: {len(high_freq_periods)}")
  214. print(f" 异常交易笔数: {len(identified_anomalies)}")
  215. # 显示高频时段详情
  216. if len(high_freq_periods) > 0:
  217. print("📋 高频时段详情(按交易笔数排序):")
  218. sorted_periods = high_freq_periods.sort_values('transaction_count', ascending=False)
  219. for i, (_, row) in enumerate(sorted_periods.iterrows(), 1):
  220. time_range = f"{row['hour']:02d}:00-{row['hour'] + 1:02d}:00"
  221. print(f" {i}. {row['date']} {time_range}: {row['transaction_count']}笔交易")
  222. print(f" 收入: {row['in_count']}笔,支出: {row['out_count']}笔")
  223. print(f" 总金额: ¥{row['amount_sum']:,.2f},笔均: ¥{row['amount_avg']:,.2f}")
  224. # 显示整体统计
  225. if len(hour_stats) > 0:
  226. max_transactions = hour_stats['transaction_count'].max()
  227. avg_transactions = hour_stats['transaction_count'].mean()
  228. print(f"📊 整体统计: 最高{max_transactions}笔/小时,平均{avg_transactions:.1f}笔/小时")
  229. return {
  230. 'recognition_type': self.display_name,
  231. 'identified_count': len(identified_anomalies),
  232. 'identified_anomalies': identified_anomalies,
  233. 'recognition_status': '完成',
  234. 'recognition_parameters': {
  235. 'frequency_threshold': self.frequency_threshold,
  236. 'statistic_unit': "日期 + 小时",
  237. 'business_rule': "单小时交易笔数超过阈值即视为高频交易异常"
  238. },
  239. 'statistics': {
  240. 'total_transactions': len(df),
  241. 'total_periods': len(hour_stats),
  242. 'high_frequency_periods': len(high_freq_periods),
  243. 'period_statistics': {
  244. 'max_transactions_per_hour': int(hour_stats['transaction_count'].max()),
  245. 'min_transactions_per_hour': int(hour_stats['transaction_count'].min()),
  246. 'avg_transactions_per_hour': float(hour_stats['transaction_count'].mean()),
  247. 'max_amount_per_hour': float(hour_stats['amount_sum'].max()),
  248. 'min_amount_per_hour': float(hour_stats['amount_sum'].min()),
  249. 'avg_amount_per_hour': float(hour_stats['amount_sum'].mean())
  250. },
  251. 'high_frequency_details': [
  252. {
  253. 'date': row['date'].strftime('%Y-%m-%d'),
  254. 'hour': int(row['hour']),
  255. 'transaction_count': int(row['transaction_count']),
  256. 'amount_sum': float(row['amount_sum']),
  257. 'in_count': int(row['in_count']),
  258. 'out_count': int(row['out_count'])
  259. }
  260. for _, row in high_freq_periods.iterrows()
  261. ] if len(high_freq_periods) > 0 else []
  262. }
  263. }
  264. except FileNotFoundError as e:
  265. return {
  266. 'recognition_type': self.display_name,
  267. 'identified_count': 0,
  268. 'identified_anomalies': [],
  269. 'recognition_status': '失败',
  270. 'error': f'文件不存在: {str(e)}'
  271. }
  272. except Exception as e:
  273. import traceback
  274. traceback.print_exc()
  275. return {
  276. 'recognition_type': self.display_name,
  277. 'identified_count': 0,
  278. 'identified_anomalies': [],
  279. 'recognition_status': '失败',
  280. 'error': f'数据加载或处理失败: {str(e)}'
  281. }
  282. def get_summary(self) -> Dict[str, Any]:
  283. """获取识别器摘要"""
  284. summary = super().get_summary()
  285. summary.update({
  286. 'frequency_threshold': self.frequency_threshold,
  287. 'business_rule': '单小时交易笔数超过阈值即视为高频交易异常',
  288. 'data_loaded': self._data is not None
  289. })
  290. return summary
  291. def get_config_summary(self) -> Dict[str, Any]:
  292. """获取配置摘要"""
  293. return {
  294. "高频交易阈值": f"{self.frequency_threshold}笔/小时",
  295. "统计单位": "日期 + 小时",
  296. "检测逻辑": "单小时交易笔数 > 阈值 = 高频交易异常",
  297. "业务规则描述": "识别短时间内集中发生、交易频次密集的资金收付行为"
  298. }