| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473 |
- from pydantic import BaseModel, Field
- from typing import Dict, Any, Optional, Type, List, Set
- import pandas as pd
- from .enhanced_base_recognizer import EnhancedBaseRecognizer
- class NightTransactionInput(BaseModel):
- """夜间交易识别工具输入"""
- csv_path: Optional[str] = Field(
- None,
- description="CSV文件路径(可选)。如果初始化时已提供csv_path,可以不用再次传入。"
- )
- class Config:
- arbitrary_types_allowed = True
- class NightTransactionRecognizer(EnhancedBaseRecognizer):
- """
- 夜间交易(2-5点)异常识别器
- 基于图2规则定义:
- 1. 时间范围:特指凌晨2点至5点期间发生的交易
- 2. 时段性质:偏离日常资金往来常规时段,通常为非经营、非生活活动的常规休息时段
- 3. 异常判定:
- - 频繁出现资金收付记录
- - 交易金额较大
- - 缺乏合理交易背景
- 4. 合理性例外:特定行业经营需求、紧急资金周转等
- 核心逻辑:2-5点交易 + (高频或大额) + (无合理背景) = 异常夜间交易
- """
- args_schema: Type[BaseModel] = NightTransactionInput
- # 配置参数
- night_start_hour: int = Field(
- 2,
- description="夜间检测开始小时(0-23),默认2点"
- )
- night_end_hour: int = Field(
- 5,
- description="夜间检测结束小时(0-23),默认5点"
- )
- frequency_threshold_per_hour: int = Field(
- 3,
- description="高频交易阈值,每小时超过此笔数视为高频"
- )
- large_amount_threshold: float = Field(
- 50000.0,
- description="大额交易阈值(元),超过此金额的夜间交易视为大额异常"
- )
- # 行业特征关键词(24小时营业或夜间经营行业)
- night_industry_keywords: List[str] = Field(
- [
- "酒店", "宾馆", "KTV", "酒吧", "夜总会", "网吧", "便利店",
- "医院", "急救", "急诊", "消防", "公安", "保安", "物流",
- "运输", "出租车", "网约车", "外卖", "配送"
- ],
- description="夜间经营行业关键词,用于识别可能有合理背景的交易"
- )
- # 紧急情况关键词
- emergency_keywords: List[str] = Field(
- [
- "急救", "急诊", "抢救", "紧急", "urgent", "emergency",
- "抢险", "救援", "救灾", "应急", "加急"
- ],
- description="紧急情况关键词,用于识别可能有合理背景的交易"
- )
- def __init__(self, csv_path: str = None, config: Dict[str, Any] = None, **kwargs):
- """
- 初始化夜间交易识别器
- Args:
- csv_path: CSV文件路径
- config: 配置参数
- **kwargs: 其他参数
- """
- # 调用父类的 __init__
- super().__init__(
- name="night_transaction_recognizer",
- description="识别银行流水中的夜间交易异常(2-5点),检测高频、大额等异常特征。",
- display_name="夜间交易异常识别",
- csv_path=csv_path,
- config=config,
- **kwargs
- )
- # 从config获取配置,更新Field属性
- night_config = self.get_config_value('night_transaction', {})
- if night_config:
- config_mapping = {
- 'night_start_hour': 'night_start_hour',
- 'night_end_hour': 'night_end_hour',
- 'frequency_threshold_per_hour': 'frequency_threshold_per_hour',
- 'large_amount_threshold': 'large_amount_threshold',
- 'night_industry_keywords': 'night_industry_keywords',
- 'emergency_keywords': 'emergency_keywords'
- }
- for config_key, attr_name in config_mapping.items():
- if config_key in night_config:
- setattr(self, attr_name, night_config[config_key])
- # 验证时间配置
- self._validate_time_config()
- print(f"✅ {self.display_name} 初始化完成")
- print(f" 夜间时段: {self.night_start_hour:02d}:00 - {self.night_end_hour:02d}:00")
- print(f" 高频阈值: {self.frequency_threshold_per_hour}笔/小时")
- print(f" 大额阈值: ¥{self.large_amount_threshold:,.2f}")
- print(f" 夜间行业关键词: {len(self.night_industry_keywords)}个")
- print(f" 紧急情况关键词: {len(self.emergency_keywords)}个")
- def _validate_time_config(self):
- """验证时间配置合理性"""
- if not (0 <= self.night_start_hour <= 23):
- raise ValueError(f"夜间开始小时必须在0-23之间: {self.night_start_hour}")
- if not (0 <= self.night_end_hour <= 23):
- raise ValueError(f"夜间结束小时必须在0-23之间: {self.night_end_hour}")
- # 确保开始时间早于结束时间
- if self.night_start_hour >= self.night_end_hour:
- print(f"⚠️ 注意:夜间开始时间({self.night_start_hour}:00) >= 结束时间({self.night_end_hour}:00),"
- f"将按跨午夜处理")
- def _is_in_night_period(self, hour: int) -> bool:
- """
- 判断小时数是否在夜间时段内
- Args:
- hour: 小时数(0-23)
- Returns:
- bool: 是否在夜间时段
- """
- if self.night_start_hour < self.night_end_hour:
- # 正常情况:开始时间 < 结束时间
- return self.night_start_hour <= hour < self.night_end_hour
- else:
- # 跨午夜情况:开始时间 >= 结束时间
- return hour >= self.night_start_hour or hour < self.night_end_hour
- def _has_reasonable_background(self, row: pd.Series) -> bool:
- """
- 判断交易是否有合理背景
- 根据图2规则,合理的背景包括:
- 1. 夜间特定行业经营需求
- 2. 紧急资金周转
- Args:
- row: 交易记录
- Returns:
- bool: 是否有合理背景
- """
- # 检查交易摘要中的关键词
- summary = str(row.get('txSummary', '')).lower()
- counterparty = str(row.get('txCounterparty', '')).lower()
- # 合并检查文本
- check_text = f"{summary} {counterparty}"
- # 1. 检查是否为夜间行业
- for keyword in self.night_industry_keywords:
- if keyword.lower() in check_text:
- return True
- # 2. 检查是否为紧急情况
- for keyword in self.emergency_keywords:
- if keyword.lower() in check_text:
- return True
- # 3. 可以扩展其他合理背景检查逻辑
- return False
- def _detect_high_frequency_transactions(self,
- night_transactions: pd.DataFrame,
- date_col: str = 'date_only') -> Set[str]:
- """
- 检测高频夜间交易
- Args:
- night_transactions: 夜间交易数据
- date_col: 日期列名
- Returns:
- Set[str]: 高频交易ID集合
- """
- high_freq_ids = set()
- if len(night_transactions) == 0:
- return high_freq_ids
- # 按日期和小时分组统计
- if 'hour' in night_transactions.columns and date_col in night_transactions.columns:
- # 统计每小时交易笔数
- hourly_counts = night_transactions.groupby([date_col, 'hour']).size()
- for (trans_date, trans_hour), count in hourly_counts.items():
- if count > self.frequency_threshold_per_hour:
- # 获取该小时的所有交易
- mask = (night_transactions[date_col] == trans_date) & \
- (night_transactions['hour'] == trans_hour)
- hour_transactions = night_transactions[mask]
- # 收集交易ID
- for tx_id in hour_transactions['txId'].unique():
- high_freq_ids.add(str(tx_id))
- print(f" ⚠️ {trans_date} {trans_hour:02d}:00-{trans_hour + 1:02d}:00: "
- f"{count}笔交易,超过阈值{self.frequency_threshold_per_hour}笔")
- return high_freq_ids
- def _detect_large_amount_transactions(self, night_transactions: pd.DataFrame) -> Set[str]:
- """
- 检测大额夜间交易
- Args:
- night_transactions: 夜间交易数据
- Returns:
- Set[str]: 大额交易ID集合
- """
- large_amount_ids = set()
- if len(night_transactions) == 0:
- return large_amount_ids
- # 筛选大额交易
- large_amount_mask = night_transactions['txAmount'].abs() >= self.large_amount_threshold
- large_amount_tx = night_transactions[large_amount_mask]
- for _, row in large_amount_tx.iterrows():
- tx_id = str(row['txId'])
- large_amount_ids.add(tx_id)
- if len(large_amount_ids) > 0:
- print(f" ⚠️ 发现 {len(large_amount_ids)} 笔大额夜间交易(≥¥{self.large_amount_threshold:,.2f})")
- return large_amount_ids
- 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" 夜间时段: {self.night_start_hour:02d}:00 - {self.night_end_hour:02d}:00")
- print(f" 检测规则: 夜间交易 + (高频或大额) + (无合理背景) = 异常")
- # 检查必需字段
- 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')
- # 提取时间组件
- df['hour'] = df['datetime'].dt.hour
- df['minute'] = df['datetime'].dt.minute
- df['date_only'] = df['datetime'].dt.date
- # ============ 识别所有夜间交易 ============
- night_mask = df['hour'].apply(self._is_in_night_period)
- night_transactions = df[night_mask].copy()
- if len(night_transactions) == 0:
- print(f"✅ 未发现夜间交易({self.night_start_hour:02d}:00-{self.night_end_hour:02d}:00)")
- return {
- 'recognition_type': self.display_name,
- 'identified_count': 0,
- 'identified_anomalies': [],
- 'recognition_status': '完成',
- 'recognition_parameters': {
- 'night_period': f"{self.night_start_hour:02d}:00-{self.night_end_hour:02d}:00",
- 'frequency_threshold': f"{self.frequency_threshold_per_hour}笔/小时",
- 'large_amount_threshold': self.large_amount_threshold,
- 'total_checked': len(df),
- 'night_transactions_found': 0
- },
- 'statistics': {
- 'total_transactions': len(df),
- 'night_transaction_count': 0,
- 'night_transaction_ratio': 0.0
- }
- }
- print(f"📊 发现 {len(night_transactions)} 笔夜间交易")
- # ============ 合理性背景分析 ============
- reasonable_transactions = []
- for _, row in night_transactions.iterrows():
- if self._has_reasonable_background(row):
- reasonable_transactions.append(row['txId'])
- if reasonable_transactions:
- print(f"📊 其中 {len(reasonable_transactions)} 笔交易可能有合理背景(夜间行业/紧急情况)")
- # ============ 异常特征检测 ============
- identified_anomalies = []
- # 1. 检测高频交易
- high_freq_ids = self._detect_high_frequency_transactions(night_transactions)
- # 2. 检测大额交易
- large_amount_ids = self._detect_large_amount_transactions(night_transactions)
- # 3. 合并异常交易ID(排除有合理背景的)
- abnormal_ids = (high_freq_ids | large_amount_ids) - set(reasonable_transactions)
- # 4. 生成异常记录
- for tx_id in abnormal_ids:
- mask = night_transactions['txId'] == tx_id
- if mask.any():
- row = night_transactions[mask].iloc[0]
- # 判断异常类型
- is_high_freq = tx_id in high_freq_ids
- is_large_amount = tx_id in large_amount_ids
- # 生成异常原因
- reasons = []
- if is_high_freq:
- reasons.append("高频夜间交易")
- if is_large_amount:
- reasons.append(f"大额夜间交易(¥{row['txAmount']:,.2f}≥¥{self.large_amount_threshold:,.2f})")
- reason_str = ",".join(reasons)
- # 检查是否有合理背景但依然被标记为异常的原因
- additional_info = {
- 'hour': int(row['hour']),
- 'is_high_frequency': is_high_freq,
- 'is_large_amount': is_large_amount,
- 'amount': float(row['txAmount']),
- 'has_reasonable_background': str(row['txId']) in reasonable_transactions
- }
- anomaly = self.format_anomaly_record(
- row=row,
- reason=f"夜间{reason_str},缺乏合理交易背景",
- severity='high' if is_large_amount else 'medium',
- check_type='night_transaction_abnormal',
- **additional_info
- )
- identified_anomalies.append(anomaly)
- # ============ 结果统计和汇总 ============
- print(f"✅ {self.display_name}检查完成")
- print(f" 检查交易总数: {len(df)}")
- print(f" 夜间交易数: {len(night_transactions)}")
- print(f" 有合理背景: {len(reasonable_transactions)}")
- print(f" 高频异常: {len(high_freq_ids)}")
- print(f" 大额异常: {len(large_amount_ids)}")
- print(f" 最终异常数: {len(identified_anomalies)}")
- # 显示夜间交易时间分布
- if len(night_transactions) > 0:
- hour_distribution = night_transactions['hour'].value_counts().sort_index()
- print("📋 夜间交易时间分布:")
- for hour, count in hour_distribution.items():
- print(f" {hour:02d}:00-{hour + 1:02d}:00: {count}笔")
- # 显示前5笔异常交易详情
- if len(identified_anomalies) > 0:
- print("📋 异常夜间交易示例:")
- for i, anomaly in enumerate(identified_anomalies[:5], 1):
- time_str = f"{anomaly.get('txDate', '')} {anomaly.get('txTime', '')}"
- print(
- f" {i}. ID:{anomaly['txId']} | {time_str} | ¥{anomaly['txAmount']:,.2f} | {anomaly['recognition_reason']}")
- return {
- 'recognition_type': self.display_name,
- 'identified_count': len(identified_anomalies),
- 'identified_anomalies': identified_anomalies,
- 'recognition_status': '完成',
- 'recognition_parameters': {
- 'night_period': f"{self.night_start_hour:02d}:00-{self.night_end_hour:02d}:00",
- 'frequency_threshold_per_hour': self.frequency_threshold_per_hour,
- 'large_amount_threshold': self.large_amount_threshold,
- 'night_industry_keywords_count': len(self.night_industry_keywords),
- 'emergency_keywords_count': len(self.emergency_keywords),
- 'total_checked': len(df),
- 'night_transactions_found': len(night_transactions),
- 'reasonable_transactions': len(reasonable_transactions)
- },
- 'statistics': {
- 'total_transactions': len(df),
- 'night_transaction_count': len(night_transactions),
- 'night_transaction_ratio': len(night_transactions) / max(1, len(df)),
- 'reasonable_transaction_count': len(reasonable_transactions),
- 'high_frequency_count': len(high_freq_ids),
- 'large_amount_count': len(large_amount_ids),
- 'abnormal_night_transaction_count': len(identified_anomalies),
- 'hour_distribution': {
- str(hour): int(count)
- for hour, count in night_transactions['hour'].value_counts().items()
- } if len(night_transactions) > 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({
- 'night_start_hour': self.night_start_hour,
- 'night_end_hour': self.night_end_hour,
- 'frequency_threshold_per_hour': self.frequency_threshold_per_hour,
- 'large_amount_threshold': self.large_amount_threshold,
- 'night_industry_keywords_count': len(self.night_industry_keywords),
- 'emergency_keywords_count': len(self.emergency_keywords),
- 'data_loaded': self._data is not None
- })
- return summary
- def get_config_summary(self) -> Dict[str, Any]:
- """获取配置摘要"""
- return {
- "夜间时段": f"{self.night_start_hour:02d}:00 - {self.night_end_hour:02d}:00",
- "高频阈值": f"{self.frequency_threshold_per_hour}笔/小时",
- "大额阈值": f"¥{self.large_amount_threshold:,.2f}",
- "夜间行业关键词": f"{len(self.night_industry_keywords)}个",
- "紧急情况关键词": f"{len(self.emergency_keywords)}个",
- "检测逻辑": "夜间交易 + (高频或大额) + (无合理背景) = 异常"
- }
|