from pydantic import BaseModel, Field from typing import Dict, Any, Optional, Type, List import pandas as pd from .enhanced_base_recognizer import EnhancedBaseRecognizer class InactiveAccountInput(BaseModel): """长期无交易账户识别工具输入""" csv_path: Optional[str] = Field( None, description="CSV文件路径(可选)。如果初始化时已提供csv_path,可以不用再次传入。" ) class Config: arbitrary_types_allowed = True class InactiveAccountRecognizer(EnhancedBaseRecognizer): """ 长期无交易账户识别器 异常规则定义: 若发现账户在指定周期内未产生任何流水明细,与正常经营或资金往来应具备的交易活跃度不符, 违背业务常理,可判定为流水存在异常,完整性存疑。 核心逻辑(根据您的要求调整): 1. 数据只属于一个账户,无需账户标识字段 2. 从数据最早日期开始计算连续无交易天数 3. 检查是否存在长时间无交易的"空白期" """ args_schema: Type[BaseModel] = InactiveAccountInput # 配置参数(简化版) inactive_period_days: int = Field( 30, description="无交易天数阈值(天),连续无交易超过此天数视为异常" ) # 严重程度配置 severity_level: str = Field( 'medium', 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="inactive_account_recognizer", description="识别在指定周期内无任何交易记录的异常账户,检查流水完整性。", display_name="长期无交易账户识别器", csv_path=csv_path, config=config, **kwargs ) # 从config获取配置,更新Field属性 inactive_config = self.get_config_value('inactive_account_check', {}) if inactive_config: config_mapping = { 'inactive_period_days': 'inactive_period_days', 'severity_level': 'severity_level' } for config_key, attr_name in config_mapping.items(): if config_key in inactive_config: setattr(self, attr_name, inactive_config[config_key]) print(f"✅ {self.display_name} 初始化完成") print(f" 无交易天数阈值: {self.inactive_period_days}天") print(f" 异常严重程度: {self.severity_level}") print(f" 账户假设: 整个文件视为单个账户") print(f" 基准日策略: 从最早交易日期开始检查") def _check_long_inactive_periods(self, df: pd.DataFrame) -> List[Dict[str, Any]]: """ 检查长时间无交易的空白期 Args: df: 交易数据(已按时间排序) Returns: List[Dict[str, Any]]: 发现的空白期列表 """ if len(df) < 2: return [] # 确保按时间排序 df = df.sort_values('datetime') inactive_periods = [] # 检查交易之间的时间间隔 for i in range(len(df) - 1): current_date = df.iloc[i]['datetime'] next_date = df.iloc[i + 1]['datetime'] # 计算天数差 days_diff = (next_date - current_date).days # 如果间隔超过阈值,记录为空白期 if days_diff > self.inactive_period_days: period_info = { 'start_date': current_date, 'end_date': next_date, 'inactive_days': days_diff, 'period_index': i, 'next_tx_id': df.iloc[i + 1]['txId'] } inactive_periods.append(period_info) return inactive_periods def _check_data_beginning_gap(self, df: pd.DataFrame, earliest_date: pd.Timestamp, data_coverage_days: int) -> Optional[Dict[str, Any]]: """ 检查数据开始前是否有空白期 Args: df: 交易数据 earliest_date: 数据中最早的交易日期 data_coverage_days: 数据覆盖的总天数 Returns: 空白期信息或None """ # 如果数据覆盖天数足够长(比如超过60天),但前面部分没有交易 # 这本身可能就是一个空白期的迹象 if data_coverage_days > self.inactive_period_days: # 找到第一条交易后的日期范围 second_date = df.sort_values('datetime').iloc[1]['datetime'] if len(df) > 1 else earliest_date first_gap = (second_date - earliest_date).days if len(df) > 1 else 0 if first_gap > self.inactive_period_days: return { 'type': 'beginning_gap', 'inactive_days': first_gap, 'start_date': earliest_date, 'end_date': second_date } return None 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.inactive_period_days}天以上的无交易空白期") # 检查必需字段 if 'datetime' not in df.columns: return { 'recognition_type': self.display_name, 'identified_count': 0, 'identified_anomalies': [], 'recognition_status': '失败', 'error': '缺少必需字段: datetime(时间信息)', 'recommendation': '请确保数据包含有效的日期时间信息' } # ============ 数据基本情况分析 ============ earliest_date = df['datetime'].min() latest_date = df['datetime'].max() data_coverage_days = (latest_date - earliest_date).days + 1 print(f"📊 数据基本情况:") print(f" 时间范围: {earliest_date.strftime('%Y-%m-%d')} 至 {latest_date.strftime('%Y-%m-%d')}") print(f" 数据覆盖天数: {data_coverage_days}天") print(f" 总交易笔数: {len(df)}") print(f" 日均交易笔数: {len(df) / data_coverage_days:.2f}笔/天") # ============ 检查数据是否足够 ============ if data_coverage_days < self.inactive_period_days: print(f"⚠️ 数据不足: 仅覆盖{data_coverage_days}天,小于阈值{self.inactive_period_days}天") print(f" 建议: 需要更长时间范围的数据才能准确判断") return { 'recognition_type': self.display_name, 'identified_count': 0, 'identified_anomalies': [], 'recognition_status': '完成', 'recognition_parameters': { 'inactive_period_days': self.inactive_period_days, 'severity_level': self.severity_level, 'data_coverage_days': data_coverage_days }, 'statistics': { 'total_transactions': len(df), 'data_coverage_days': data_coverage_days, 'date_range': { 'start': earliest_date.strftime('%Y-%m-%d'), 'end': latest_date.strftime('%Y-%m-%d') }, 'transaction_frequency': len(df) / data_coverage_days if data_coverage_days > 0 else 0 }, 'note': f'数据覆盖天数({data_coverage_days}天)不足阈值({self.inactive_period_days}天),无法准确判断' } # ============ 检查长时间空白期 ============ identified_anomalies = [] # 1. 检查交易之间的空白期 inactive_periods = self._check_long_inactive_periods(df) for period in inactive_periods: start_date = period['start_date'] end_date = period['end_date'] inactive_days = period['inactive_days'] next_tx_id = period['next_tx_id'] # 生成异常原因 reason = f"发现长时间无交易空白期:从{start_date.strftime('%Y-%m-%d')}到{end_date.strftime('%Y-%m-%d')},连续{inactive_days}天无任何交易,超过阈值{self.inactive_period_days}天" print(f" ❌ 发现空白期: {reason}") # 获取下一笔交易的详细信息 next_tx_data = df[df['txId'] == next_tx_id] if not next_tx_data.empty: next_tx = next_tx_data.iloc[0] # 使用真实交易数据创建异常记录 anomaly = { 'txId': str(next_tx_id), 'txDate': str(next_tx['txDate']), 'txTime': str(next_tx['txTime']), 'datetime': next_tx['datetime'] if 'datetime' in next_tx else end_date, 'txAmount': float(next_tx['txAmount']), 'txDirection': str(next_tx['txDirection']), 'recognition_reason': f"长期无交易异常:账户在{start_date.strftime('%Y-%m-%d')}至{end_date.strftime('%Y-%m-%d')}期间连续{inactive_days}天无任何交易,超过阈值{self.inactive_period_days}天。此笔交易({next_tx_id})为空白期后的首笔交易", 'severity': self.severity_level, 'status': '待核查', 'check_type': 'inactive_account_period', 'period_info': { 'start_date': start_date.strftime('%Y-%m-%d'), 'end_date': end_date.strftime('%Y-%m-%d'), 'inactive_days': inactive_days, 'threshold_days': self.inactive_period_days, 'next_tx_id': next_tx_id, 'next_tx_info': { 'txDate': str(next_tx['txDate']), 'txTime': str(next_tx['txTime']), 'txAmount': float(next_tx['txAmount']), 'txDirection': str(next_tx['txDirection']) } } } # 创建包含所有必要字段的Series anomaly_series = pd.Series({ 'txId': anomaly['txId'], 'txDate': anomaly['txDate'], 'txTime': anomaly['txTime'], 'txAmount': anomaly['txAmount'], 'txDirection': anomaly['txDirection'], 'txBalance': next_tx.get('txBalance', None), 'txSummary': next_tx.get('txSummary', ''), 'txCounterparty': next_tx.get('txCounterparty', ''), 'datetime': anomaly['datetime'] }) else: # 如果找不到下一笔交易,使用改进的默认格式 print(f"⚠️ 警告:未找到交易ID {next_tx_id} 的详细信息,使用默认格式") anomaly = { 'txId': str(next_tx_id), 'txDate': end_date.strftime('%Y-%m-%d'), 'txTime': '23:59:59', 'datetime': end_date, 'txAmount': 0.0, 'txDirection': '收入', # 默认设为收入 'recognition_reason': f"{reason}", 'severity': self.severity_level, 'status': '待核查', 'check_type': 'inactive_account_period', 'period_info': { 'start_date': start_date.strftime('%Y-%m-%d'), 'end_date': end_date.strftime('%Y-%m-%d'), 'inactive_days': inactive_days, 'threshold_days': self.inactive_period_days } } anomaly_series = pd.Series({ 'txId': anomaly['txId'], 'txDate': anomaly['txDate'], 'txTime': anomaly['txTime'], 'txAmount': anomaly['txAmount'], 'txDirection': anomaly['txDirection'], 'datetime': anomaly['datetime'] }) # 格式化异常记录 formatted_anomaly = self.format_anomaly_record( row=anomaly_series, reason=anomaly['recognition_reason'], severity=anomaly['severity'], check_type=anomaly['check_type'], **anomaly['period_info'] ) identified_anomalies.append(formatted_anomaly) # 2. 检查整体交易活跃度(如果整个数据期交易都很少) avg_transactions_per_day = len(df) / data_coverage_days if avg_transactions_per_day < 0.1: # 平均每天不足0.1笔交易 print(f"⚠️ 交易活跃度极低: 平均{avg_transactions_per_day:.2f}笔/天") # 生成低活跃度异常 low_activity_reason = f"账户整体交易活跃度极低:{data_coverage_days}天内仅{len(df)}笔交易,平均{avg_transactions_per_day:.2f}笔/天,不符合正常资金往来特征" # 使用最后一笔交易作为异常记录的基础 last_tx = df.iloc[-1] if len(df) > 0 else None if last_tx is not None: anomaly_series = pd.Series({ 'txId': 'LOW_ACTIVITY_OVERALL', 'txDate': last_tx['txDate'], 'txTime': last_tx['txTime'], 'txAmount': float(last_tx['txAmount']), 'txDirection': str(last_tx['txDirection']), 'txBalance': last_tx.get('txBalance', None), 'txSummary': last_tx.get('txSummary', ''), 'txCounterparty': last_tx.get('txCounterparty', ''), 'datetime': last_tx.get('datetime', latest_date) }) else: anomaly_series = pd.Series({ 'txId': 'LOW_ACTIVITY_OVERALL', 'txDate': latest_date.strftime('%Y-%m-%d'), 'txTime': '23:59:59', 'txAmount': 0.0, 'txDirection': '收入', 'datetime': latest_date }) formatted_anomaly = self.format_anomaly_record( row=anomaly_series, reason=low_activity_reason, severity='low' if self.severity_level == 'medium' else self.severity_level, # 降一级严重度 check_type='low_activity_overall', activity_metrics={ 'data_coverage_days': data_coverage_days, 'total_transactions': len(df), 'avg_transactions_per_day': avg_transactions_per_day } ) identified_anomalies.append(formatted_anomaly) # ============ 结果统计 ============ print(f"✅ {self.display_name}检查完成") print(f" 检查结果:") print(f" 空白期数量: {len(inactive_periods)}") print(f" 异常记录数: {len(identified_anomalies)}") if len(inactive_periods) == 0: print(f" ✅ 未发现超过{self.inactive_period_days}天的空白期") return { 'recognition_type': self.display_name, 'identified_count': len(identified_anomalies), 'identified_anomalies': identified_anomalies, 'recognition_status': '完成', 'recognition_parameters': { 'inactive_period_days': self.inactive_period_days, 'severity_level': self.severity_level, 'check_strategy': '从最早交易日期开始检查空白期' }, 'statistics': { 'total_transactions': len(df), 'data_coverage_days': data_coverage_days, 'date_range': { 'start': earliest_date.strftime('%Y-%m-%d'), 'end': latest_date.strftime('%Y-%m-%d') }, 'transaction_frequency': avg_transactions_per_day, 'inactive_periods_count': len(inactive_periods), 'inactive_periods_details': [ { 'start_date': p['start_date'].strftime('%Y-%m-%d'), 'end_date': p['end_date'].strftime('%Y-%m-%d'), 'inactive_days': p['inactive_days'], 'next_tx_id': p['next_tx_id'] } for p in inactive_periods ] } } 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({ 'inactive_period_days': self.inactive_period_days, 'severity_level': self.severity_level, 'data_loaded': self._data is not None, 'check_strategy': '从最早交易日期开始检查连续无交易空白期' }) return summary def get_config_summary(self) -> Dict[str, Any]: """获取配置摘要""" return { "无交易天数阈值": f"{self.inactive_period_days}天", "异常严重程度": self.severity_level.upper(), "检测逻辑": f"从最早交易日期开始,检查连续{self.inactive_period_days}天以上的无交易空白期", "账户假设": "整个文件视为单个账户", "基准日策略": "以数据中最早交易日期为起点", "业务规则描述": "连续长时间无任何交易,与正常经营或资金往来的交易活跃度不符" }