| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354 |
- from pydantic import BaseModel, Field
- from typing import Dict, Any, Optional, Type
- import pandas as pd
- from .enhanced_base_recognizer import EnhancedBaseRecognizer
- class HighFrequencyInput(BaseModel):
- """高频交易识别工具输入"""
- csv_path: Optional[str] = Field(
- None,
- description="CSV文件路径(可选)。如果初始化时已提供csv_path,可以不用再次传入。"
- )
- class Config:
- arbitrary_types_allowed = True
- class HighFrequencyTransactionRecognizer(EnhancedBaseRecognizer):
- """
- 高频交易异常识别器
- 严格遵循业务规则定义:
- 银行流水高频交易(按小时维度)的定义为:
- 以"日期 + 小时"为统计单位,通过汇总该时间间隔内的交易笔数与交易金额,
- 识别出单小时交易笔数超过 10 笔的特定时段组合,
- 此类在短时间内(1 小时)集中发生、交易频次密集的资金收付行为,
- 即为银行流水高频交易,其核心特征是交易笔数在单位小时内达到预设阈值(10 笔),
- 体现出资金往来的集中性与活跃度异常。
- """
- args_schema: Type[BaseModel] = HighFrequencyInput
- # 配置参数 - 严格按照业务规则
- frequency_threshold: int = Field(
- 10,
- description="高频交易阈值,每小时超过此笔数视为高频交易(业务规则要求10笔)"
- )
- def __init__(self, csv_path: str = None, config: Dict[str, Any] = None, **kwargs):
- """
- 初始化高频交易识别器
- Args:
- csv_path: CSV文件路径
- config: 配置参数
- **kwargs: 其他参数
- """
- # 调用父类的 __init__
- super().__init__(
- name="high_frequency_recognizer",
- description="识别银行流水中的高频交易异常,严格按照业务规则:单小时交易笔数超过10笔即为高频异常。",
- display_name="高频交易异常识别",
- csv_path=csv_path,
- config=config,
- **kwargs
- )
- # 从config获取配置
- high_freq_config = self.get_config_value('high_frequency', {})
- if high_freq_config and 'frequency_threshold' in high_freq_config:
- self.frequency_threshold = high_freq_config['frequency_threshold']
- print(f"✅ {self.display_name} 初始化完成")
- print(f" 严格遵循业务规则:单小时交易笔数 > {self.frequency_threshold}笔 = 高频异常")
- def _calculate_hourly_statistics(self, df: pd.DataFrame) -> pd.DataFrame:
- """
- 计算每小时的交易统计数据
- Args:
- df: 标准化后的交易数据
- Returns:
- pd.DataFrame: 每小时统计结果
- """
- # 提取日期和小时
- df['date_hour'] = df['datetime'].dt.strftime('%Y-%m-%d %H')
- df['date'] = df['datetime'].dt.date
- df['hour'] = df['datetime'].dt.hour
- # 按照业务规则:以"日期 + 小时"为统计单位
- hour_stats = []
- # 按日期+小时分组
- for (date_val, hour_val), group in df.groupby(['date', 'hour']):
- transaction_count = len(group)
- amount_sum = group['txAmount'].sum()
- # 统计交易方向分布
- direction_counts = group['txDirection'].value_counts().to_dict()
- hour_stats.append({
- 'date': date_val,
- 'hour': hour_val,
- 'transaction_count': transaction_count,
- 'amount_sum': amount_sum,
- 'amount_avg': amount_sum / transaction_count if transaction_count > 0 else 0,
- 'in_count': direction_counts.get('收入', 0),
- 'out_count': direction_counts.get('支出', 0)
- })
- return pd.DataFrame(hour_stats)
- def _identify_high_frequency_periods(self, hour_stats: pd.DataFrame) -> pd.DataFrame:
- """
- 识别高频交易时段
- Args:
- hour_stats: 每小时统计数据
- Returns:
- pd.DataFrame: 标记了高频时段的结果
- """
- # 严格按照业务规则:单小时交易笔数超过10笔
- hour_stats['is_high_frequency'] = hour_stats['transaction_count'] > self.frequency_threshold
- return hour_stats
- def _get_transactions_in_period(self, df: pd.DataFrame, date_val: pd.Timestamp.date, hour_val: int) -> pd.DataFrame:
- """
- 获取指定时段内的所有交易
- Args:
- df: 原始交易数据
- date_val: 日期
- hour_val: 小时
- Returns:
- pd.DataFrame: 指定时段内的交易
- """
- return df[
- (df['datetime'].dt.date == date_val) &
- (df['datetime'].dt.hour == hour_val)
- ].copy()
- 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" 业务规则: 以'日期 + 小时'为统计单位")
- print(f" 高频阈值: 单小时交易笔数 > {self.frequency_threshold}笔")
- # 检查必需字段
- 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')
- # 检查是否有无效的时间数据
- invalid_times = df['datetime'].isna().sum()
- if invalid_times > 0:
- print(f"⚠️ 警告: 有 {invalid_times} 条记录的时间解析失败,将跳过这些记录")
- df = df[df['datetime'].notna()]
- if len(df) == 0:
- return {
- 'recognition_type': self.display_name,
- 'identified_count': 0,
- 'identified_anomalies': [],
- 'recognition_status': '完成',
- 'statistics': {'total_valid_transactions': 0}
- }
- # ============ 按照业务规则:以"日期 + 小时"为统计单位 ============
- hour_stats = self._calculate_hourly_statistics(df)
- if len(hour_stats) == 0:
- return {
- 'recognition_type': self.display_name,
- 'identified_count': 0,
- 'identified_anomalies': [],
- 'recognition_status': '完成',
- 'statistics': {
- 'total_transactions': len(df),
- 'hour_periods': 0,
- 'high_frequency_periods': 0
- }
- }
- print(f"📊 按'日期+小时'分组统计完成,共 {len(hour_stats)} 个时段")
- # ============ 识别高频时段 ============
- hour_stats = self._identify_high_frequency_periods(hour_stats)
- # 高频时段统计
- high_freq_periods = hour_stats[hour_stats['is_high_frequency']]
- print(f"📊 发现 {len(high_freq_periods)} 个高频时段(> {self.frequency_threshold}笔/小时)")
- # ============ 生成异常记录 ============
- identified_anomalies = []
- # 为每个高频时段创建异常记录
- for _, period_row in high_freq_periods.iterrows():
- date_val = period_row['date']
- hour_val = period_row['hour']
- # 获取该时段内的所有交易
- period_transactions = self._get_transactions_in_period(df, date_val, hour_val)
- # 为每笔交易创建异常记录
- for _, tx_row in period_transactions.iterrows():
- # 生成异常原因 - 严格按照业务规则描述
- reason = f"属于高频交易时段:{date_val} {hour_val:02d}:00-{hour_val + 1:02d}:00,该时段共{period_row['transaction_count']}笔交易,超过阈值{self.frequency_threshold}笔,体现资金往来的集中性与活跃度异常"
- # 额外信息
- additional_info = {
- 'period_info': {
- 'date': date_val.strftime('%Y-%m-%d'),
- 'hour': int(hour_val),
- 'start_time': f"{hour_val:02d}:00",
- 'end_time': f"{hour_val + 1:02d}:00",
- 'transaction_count': int(period_row['transaction_count']),
- 'amount_sum': float(period_row['amount_sum']),
- 'amount_avg': float(period_row['amount_avg']),
- 'in_count': int(period_row['in_count']),
- 'out_count': int(period_row['out_count'])
- },
- 'business_rule': {
- 'statistic_unit': "日期 + 小时",
- 'threshold': self.frequency_threshold,
- 'description': "单小时交易笔数超过阈值,体现资金往来的集中性与活跃度异常"
- }
- }
- anomaly = self.format_anomaly_record(
- row=tx_row,
- reason=reason,
- severity='high', # 高频交易通常视为高风险
- check_type='high_frequency_transaction',
- **additional_info
- )
- identified_anomalies.append(anomaly)
- # ============ 结果统计 ============
- print(f"✅ {self.display_name}检查完成")
- print(f" 检查交易总数: {len(df)}")
- print(f" 统计时段总数: {len(hour_stats)}")
- print(f" 高频时段数: {len(high_freq_periods)}")
- print(f" 异常交易笔数: {len(identified_anomalies)}")
- # 显示高频时段详情
- if len(high_freq_periods) > 0:
- print("📋 高频时段详情(按交易笔数排序):")
- sorted_periods = high_freq_periods.sort_values('transaction_count', ascending=False)
- for i, (_, row) in enumerate(sorted_periods.iterrows(), 1):
- time_range = f"{row['hour']:02d}:00-{row['hour'] + 1:02d}:00"
- print(f" {i}. {row['date']} {time_range}: {row['transaction_count']}笔交易")
- print(f" 收入: {row['in_count']}笔,支出: {row['out_count']}笔")
- print(f" 总金额: ¥{row['amount_sum']:,.2f},笔均: ¥{row['amount_avg']:,.2f}")
- # 显示整体统计
- if len(hour_stats) > 0:
- max_transactions = hour_stats['transaction_count'].max()
- avg_transactions = hour_stats['transaction_count'].mean()
- print(f"📊 整体统计: 最高{max_transactions}笔/小时,平均{avg_transactions:.1f}笔/小时")
- return {
- 'recognition_type': self.display_name,
- 'identified_count': len(identified_anomalies),
- 'identified_anomalies': identified_anomalies,
- 'recognition_status': '完成',
- 'recognition_parameters': {
- 'frequency_threshold': self.frequency_threshold,
- 'statistic_unit': "日期 + 小时",
- 'business_rule': "单小时交易笔数超过阈值即视为高频交易异常"
- },
- 'statistics': {
- 'total_transactions': len(df),
- 'total_periods': len(hour_stats),
- 'high_frequency_periods': len(high_freq_periods),
- 'period_statistics': {
- 'max_transactions_per_hour': int(hour_stats['transaction_count'].max()),
- 'min_transactions_per_hour': int(hour_stats['transaction_count'].min()),
- 'avg_transactions_per_hour': float(hour_stats['transaction_count'].mean()),
- 'max_amount_per_hour': float(hour_stats['amount_sum'].max()),
- 'min_amount_per_hour': float(hour_stats['amount_sum'].min()),
- 'avg_amount_per_hour': float(hour_stats['amount_sum'].mean())
- },
- 'high_frequency_details': [
- {
- 'date': row['date'].strftime('%Y-%m-%d'),
- 'hour': int(row['hour']),
- 'transaction_count': int(row['transaction_count']),
- 'amount_sum': float(row['amount_sum']),
- 'in_count': int(row['in_count']),
- 'out_count': int(row['out_count'])
- }
- for _, row in high_freq_periods.iterrows()
- ] if len(high_freq_periods) > 0 else []
- }
- }
- 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({
- 'frequency_threshold': self.frequency_threshold,
- 'business_rule': '单小时交易笔数超过阈值即视为高频交易异常',
- 'data_loaded': self._data is not None
- })
- return summary
- def get_config_summary(self) -> Dict[str, Any]:
- """获取配置摘要"""
- return {
- "高频交易阈值": f"{self.frequency_threshold}笔/小时",
- "统计单位": "日期 + 小时",
- "检测逻辑": "单小时交易笔数 > 阈值 = 高频交易异常",
- "业务规则描述": "识别短时间内集中发生、交易频次密集的资金收付行为"
- }
|