| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492 |
- from pydantic import BaseModel, Field
- from typing import Dict, Any, Optional, Type
- import pandas as pd
- from datetime import timedelta
- from .enhanced_base_recognizer import EnhancedBaseRecognizer
- class OccasionalHighIntegerTransactionInput(BaseModel):
- """偶发高额整数交易识别工具输入"""
- csv_path: Optional[str] = Field(
- None,
- description="CSV文件路径(可选)。如果初始化时已提供csv_path,可以不用再次传入。"
- )
- class Config:
- arbitrary_types_allowed = True
- class OccasionalHighIntegerTransactionRecognizer(EnhancedBaseRecognizer):
- """
- 偶发高额整数交易异常识别器
- 异常规则定义:
- 银行流水核查中,若存在金额为10,000元整数倍的交易,且该类交易金额超过整体流水平均交易金额的5倍
- (构成极端异常值),同时此类极端异常交易呈现偶发且高频次发生的特征,可判定为金额维度存在
- 异常交易情形,需进一步核查交易真实性。
- """
- args_schema: Type[BaseModel] = OccasionalHighIntegerTransactionInput
- # 整数倍基数
- integer_multiple: float = Field(
- 10000.0,
- description="整数倍基数(元),检查是否为该金额的整数倍"
- )
- # 异常倍数阈值
- outlier_multiplier: float = Field(
- 5.0,
- description="异常倍数阈值,交易金额超过整体平均交易金额的多少倍视为极端异常值"
- )
- # 频率分析参数
- frequency_window_days: int = Field(
- 7,
- description="频率分析时间窗口(天),用于判断是否为高频发生"
- )
- min_occurrences_for_high_frequency: int = Field(
- 3,
- description="高频最小发生次数,在时间窗口内达到此次数视为高频"
- )
- # 偶发性判断参数
- gap_std_threshold: float = Field(
- 2.0,
- description="时间间隔标准差阈值,大于此值视为时间不规律(偶发)"
- )
- # 严重程度配置
- severity_level: str = Field(
- 'high',
- description="异常严重程度(high/medium/low)"
- )
- def __init__(self, csv_path: str = None, config: Dict[str, Any] = None, **kwargs):
- """
- 初始化偶发高额整数交易识别器
- Args:
- csv_path: CSV文件路径
- config: 配置参数
- **kwargs: 其他参数
- """
- # 调用父类的 __init__
- super().__init__(
- name="occasional_high_integer_recognizer",
- description="识别银行流水中偶发的高额整数倍交易异常,检测金额为整数倍、超过平均金额5倍且呈现偶发高频特征的交易。",
- display_name="偶发高额整数交易异常识别",
- csv_path=csv_path,
- config=config,
- **kwargs
- )
- # 从config获取配置,更新Field属性
- integer_config = self.get_config_value('occasional_integer_transaction', {})
- if integer_config:
- config_mapping = {
- 'integer_multiple': 'integer_multiple',
- 'outlier_multiplier': 'outlier_multiplier',
- 'frequency_window_days': 'frequency_window_days',
- 'min_occurrences_for_high_frequency': 'min_occurrences_for_high_frequency',
- 'gap_std_threshold': 'gap_std_threshold',
- 'severity_level': 'severity_level'
- }
- for config_key, attr_name in config_mapping.items():
- if config_key in integer_config:
- setattr(self, attr_name, integer_config[config_key])
- print(f"✅ {self.display_name} 初始化完成")
- print(f" 整数倍基数: ¥{self.integer_multiple:,.0f}")
- print(f" 异常倍数阈值: {self.outlier_multiplier}倍")
- print(f" 频率分析窗口: {self.frequency_window_days}天")
- print(f" 高频最小次数: {self.min_occurrences_for_high_frequency}次")
- print(f" 偶发判断阈值: 间隔标准差>{self.gap_std_threshold}")
- print(f" 严重程度: {self.severity_level}")
- def _is_integer_multiple(self, amount: float, tolerance: float = 0.01) -> bool:
- """
- 判断金额是否为整数倍
- Args:
- amount: 交易金额
- tolerance: 容差(元),解决浮点数精度问题
- Returns:
- bool: 是否为整数倍
- """
- if pd.isna(amount):
- return False
- # 计算余数
- remainder = abs(amount % self.integer_multiple)
- # 考虑浮点数精度,有两种情况:
- # 1. 余数接近于0(如10000 % 10000 = 0)
- # 2. 余数接近于整数倍基数(如10000 % 10000 = 0,但浮点误差可能为0.000001)
- return remainder < tolerance or abs(remainder - self.integer_multiple) < tolerance
- def _analyze_frequency_pattern(self, transactions: pd.DataFrame) -> Dict[str, Any]:
- """
- 分析交易频率模式
- Args:
- transactions: 交易数据
- Returns:
- Dict[str, Any]: 频率分析结果
- """
- if len(transactions) < 2:
- return {
- 'is_occasional': False,
- 'is_high_frequency': False,
- 'total_count': len(transactions),
- 'gap_std': 0.0,
- 'time_analysis': '数据不足,无法分析频率模式'
- }
- # 确保按时间排序
- sorted_transactions = transactions.sort_values('datetime')
- # 1. 计算时间间隔(天数)
- time_diffs = sorted_transactions['datetime'].diff().dt.total_seconds() / 86400.0
- time_diffs = time_diffs.dropna()
- if len(time_diffs) == 0:
- return {
- 'is_occasional': False,
- 'is_high_frequency': False,
- 'total_count': len(transactions),
- 'gap_std': 0.0,
- 'time_analysis': '时间间隔数据不足'
- }
- # 2. 判断是否偶发(时间间隔不规律)
- gap_std = time_diffs.std()
- is_occasional = gap_std > self.gap_std_threshold
- # 3. 判断是否高频(在一定时间内多次发生)
- # 按天统计发生次数
- date_counts = sorted_transactions['datetime'].dt.date.value_counts()
- # 检查是否有连续发生的情况
- dates = sorted(sorted_transactions['datetime'].dt.date.unique())
- date_diffs = [(dates[i + 1] - dates[i]).days for i in range(len(dates) - 1)]
- # 判断是否有在时间窗口内多次发生
- is_high_frequency = False
- if len(transactions) >= self.min_occurrences_for_high_frequency:
- # 检查是否有在frequency_window_days内达到min_occurrences_for_high_frequency次
- sliding_window_counts = []
- for i in range(len(dates)):
- window_start = dates[i]
- window_end = window_start + timedelta(days=self.frequency_window_days)
- count_in_window = sum(1 for d in dates if window_start <= d <= window_end)
- sliding_window_counts.append(count_in_window)
- max_in_window = max(sliding_window_counts) if sliding_window_counts else 0
- is_high_frequency = max_in_window >= self.min_occurrences_for_high_frequency
- return {
- 'is_occasional': is_occasional,
- 'is_high_frequency': is_high_frequency,
- 'total_count': len(transactions),
- 'gap_std': float(gap_std),
- 'gap_mean': float(time_diffs.mean()),
- 'date_counts': date_counts.to_dict(),
- 'unique_dates': len(date_counts),
- 'date_range': {
- 'start': dates[0].strftime('%Y-%m-%d') if dates else '',
- 'end': dates[-1].strftime('%Y-%m-%d') if dates else '',
- 'total_days': (dates[-1] - dates[0]).days + 1 if len(dates) > 1 else 1
- },
- 'time_analysis': f"时间间隔标准差: {gap_std:.2f}天,最大窗口内次数: {max_in_window if 'max_in_window' in locals() else 0}次"
- }
- def _generate_anomaly_reason(self, row: pd.Series, avg_amount: float,
- frequency_info: Dict[str, Any]) -> str:
- """
- 生成异常原因描述
- Args:
- row: 交易记录
- avg_amount: 整体平均交易金额
- frequency_info: 频率分析结果
- Returns:
- str: 异常原因描述
- """
- reasons = []
- # 整数倍特征
- multiple = row['txAmount'] / self.integer_multiple
- reasons.append(f"金额为¥{self.integer_multiple:,.0f}的{multiple:.0f}倍整数")
- # 极端异常值特征
- if avg_amount > 0:
- outlier_ratio = abs(row['txAmount']) / avg_amount
- reasons.append(f"金额超出整体平均{outlier_ratio:.1f}倍")
- # 频率特征
- if frequency_info['is_occasional']:
- reasons.append(f"交易时间不规律(间隔标准差{frequency_info['gap_std']:.1f}天)")
- if frequency_info['is_high_frequency']:
- reasons.append(f"高频发生({frequency_info['total_count']}次)")
- return f"偶发高额整数交易异常: {','.join(reasons)},需核查交易真实性"
- 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.integer_multiple:,.0f}元) + 极端异常值({self.outlier_multiplier}倍) + 偶发高频 = 异常")
- # 检查必需字段
- 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 = df.sort_values('datetime')
- # 计算整体平均交易金额(绝对值)
- avg_amount = df['txAmount'].abs().mean()
- print(f"📊 整体平均交易金额: ¥{avg_amount:,.2f}")
- print(f" 极端异常值阈值: ¥{avg_amount * self.outlier_multiplier:,.2f}")
- # ============ 第一步:筛选整数倍交易 ============
- integer_mask = df['txAmount'].apply(lambda x: self._is_integer_multiple(abs(x)))
- integer_transactions = df[integer_mask].copy()
- if len(integer_transactions) == 0:
- print(f"📊 未发现{self.integer_multiple:,.0f}元整数倍交易")
- return {
- 'recognition_type': self.display_name,
- 'identified_count': 0,
- 'identified_anomalies': [],
- 'recognition_status': '完成',
- 'statistics': {
- 'total_transactions': len(df),
- 'integer_transactions': 0,
- 'avg_amount': float(avg_amount),
- 'integer_multiple': self.integer_multiple
- }
- }
- print(f"📊 发现 {len(integer_transactions)} 笔{self.integer_multiple:,.0f}元整数倍交易")
- # ============ 第二步:筛选极端异常值 ============
- outlier_threshold = avg_amount * self.outlier_multiplier
- outlier_mask = integer_transactions['txAmount'].abs() > outlier_threshold
- outlier_transactions = integer_transactions[outlier_mask].copy()
- if len(outlier_transactions) == 0:
- print(f"📊 未发现极端异常值整数倍交易(金额≤¥{outlier_threshold:,.2f})")
- return {
- 'recognition_type': self.display_name,
- 'identified_count': 0,
- 'identified_anomalies': [],
- 'recognition_status': '完成',
- 'statistics': {
- 'total_transactions': len(df),
- 'integer_transactions': len(integer_transactions),
- 'outlier_transactions': 0,
- 'avg_amount': float(avg_amount),
- 'outlier_threshold': float(outlier_threshold)
- }
- }
- print(f"📊 发现 {len(outlier_transactions)} 笔极端异常值整数倍交易(金额>¥{outlier_threshold:,.2f})")
- # ============ 第三步:分析频率模式 ============
- frequency_info = self._analyze_frequency_pattern(outlier_transactions)
- print(f"📈 频率分析结果:")
- print(f" 总次数: {frequency_info['total_count']}")
- print(f" 是否偶发: {'是' if frequency_info['is_occasional'] else '否'} (标准差={frequency_info['gap_std']:.2f}天)")
- print(f" 是否高频: {'是' if frequency_info['is_high_frequency'] else '否'}")
- print(f" 涉及天数: {frequency_info['unique_dates']}天")
- # ============ 第四步:识别异常交易 ============
- identified_anomalies = []
- # 只有同时满足偶发且高频才标记为异常
- if frequency_info['is_occasional'] and frequency_info['is_high_frequency']:
- print(f"⚠️ 检测到偶发且高频的高额整数交易,开始标记异常...")
- for idx, row in outlier_transactions.iterrows():
- # 生成异常原因
- reason = self._generate_anomaly_reason(row, avg_amount, frequency_info)
- # 额外信息
- additional_info = {
- 'frequency_analysis': frequency_info,
- 'amount_analysis': {
- 'integer_multiple': self.integer_multiple,
- 'outlier_multiplier': self.outlier_multiplier,
- 'avg_amount': avg_amount,
- 'outlier_threshold': outlier_threshold,
- 'outlier_ratio': abs(row['txAmount']) / avg_amount if avg_amount > 0 else 0
- },
- 'integer_analysis': {
- 'multiple_count': row['txAmount'] / self.integer_multiple,
- 'is_integer_multiple': True
- }
- }
- # 生成异常记录
- anomaly = self.format_anomaly_record(
- row=row,
- reason=reason,
- severity=self.severity_level,
- check_type='occasional_high_integer_transaction',
- **additional_info
- )
- identified_anomalies.append(anomaly)
- print(f" ❌ 标记异常: ID={row['txId']}, ¥{row['txAmount']:,.2f}, {reason[:60]}...")
- else:
- print(f"✅ 未检测到偶发高频模式,不标记异常")
- print(f" 偶发性: {frequency_info['is_occasional']}")
- print(f" 高频性: {frequency_info['is_high_frequency']}")
- # ============ 第五步:结果统计 ============
- print(f"✅ {self.display_name}检查完成")
- print(f" 检查交易总数: {len(df)}")
- print(f" 整数倍交易数: {len(integer_transactions)}")
- print(f" 极端异常值数: {len(outlier_transactions)}")
- print(f" 异常交易数: {len(identified_anomalies)}")
- # 显示整数倍交易统计
- if len(integer_transactions) > 0:
- print("📋 整数倍交易统计:")
- total_integer_amount = integer_transactions['txAmount'].sum()
- avg_integer_amount = integer_transactions['txAmount'].mean()
- print(f" 总整数倍金额: ¥{total_integer_amount:,.2f}")
- print(f" 平均整数倍金额: ¥{avg_integer_amount:,.2f}")
- # 倍数分布
- integer_transactions['multiple'] = (integer_transactions['txAmount'] / self.integer_multiple).round()
- multiple_counts = integer_transactions['multiple'].value_counts().sort_index()
- print(f" 倍数分布: {dict(multiple_counts.head(10))}" +
- (", ..." if len(multiple_counts) > 10 else ""))
- # 按方向统计
- income_integer = integer_transactions[integer_transactions['txDirection'] == '收入']
- expense_integer = integer_transactions[integer_transactions['txDirection'] == '支出']
- print(f" 整数倍收入: {len(income_integer)}笔, ¥{income_integer['txAmount'].sum():,.2f}")
- print(f" 整数倍支出: {len(expense_integer)}笔, ¥{expense_integer['txAmount'].sum():,.2f}")
- return {
- 'recognition_type': self.display_name,
- 'identified_count': len(identified_anomalies),
- 'identified_anomalies': identified_anomalies,
- 'recognition_status': '完成',
- 'recognition_parameters': {
- 'integer_multiple': self.integer_multiple,
- 'outlier_multiplier': self.outlier_multiplier,
- 'frequency_window_days': self.frequency_window_days,
- 'min_occurrences_for_high_frequency': self.min_occurrences_for_high_frequency,
- 'gap_std_threshold': self.gap_std_threshold,
- 'severity_level': self.severity_level,
- 'avg_amount': float(avg_amount)
- },
- 'statistics': {
- 'total_transactions': len(df),
- 'integer_transactions': len(integer_transactions),
- 'outlier_transactions': len(outlier_transactions),
- 'anomalous_transactions': len(identified_anomalies),
- 'frequency_analysis': frequency_info,
- 'amount_statistics': {
- 'overall_avg': float(avg_amount),
- 'overall_max': float(df['txAmount'].max()),
- 'overall_min': float(df['txAmount'].min()),
- 'integer_avg': float(integer_transactions['txAmount'].mean()) if len(
- integer_transactions) > 0 else 0,
- 'integer_total': float(integer_transactions['txAmount'].sum()) if len(
- integer_transactions) > 0 else 0,
- 'outlier_avg': float(outlier_transactions['txAmount'].mean()) if len(
- outlier_transactions) > 0 else 0
- },
- 'direction_distribution': {
- 'integer_income': len(integer_transactions[integer_transactions['txDirection'] == '收入']),
- 'integer_expense': len(integer_transactions[integer_transactions['txDirection'] == '支出']),
- 'outlier_income': len(outlier_transactions[outlier_transactions['txDirection'] == '收入']),
- 'outlier_expense': len(outlier_transactions[outlier_transactions['txDirection'] == '支出'])
- }
- }
- }
- 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({
- 'integer_multiple': self.integer_multiple,
- 'outlier_multiplier': self.outlier_multiplier,
- 'frequency_window_days': self.frequency_window_days,
- 'min_occurrences_for_high_frequency': self.min_occurrences_for_high_frequency,
- 'gap_std_threshold': self.gap_std_threshold,
- 'severity_level': self.severity_level,
- 'data_loaded': self._data is not None
- })
- return summary
- def get_config_summary(self) -> Dict[str, Any]:
- """获取配置摘要"""
- return {
- "整数倍基数": f"¥{self.integer_multiple:,.0f}元",
- "异常倍数阈值": f"{self.outlier_multiplier}倍",
- "频率分析窗口": f"{self.frequency_window_days}天",
- "高频最小次数": f"{self.min_occurrences_for_high_frequency}次",
- "偶发判断阈值": f"标准差>{self.gap_std_threshold}天",
- "检测逻辑": f"整数倍(¥{self.integer_multiple:,.0f}) + 极端异常值({self.outlier_multiplier}倍) + 偶发高频 = 异常",
- "严重程度": self.severity_level.upper(),
- "业务规则描述": "金额为整数倍、超过平均金额5倍且呈现偶发高频特征的交易需核查真实性"
- }
|