| 123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450 |
- 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}天以上的无交易空白期",
- "账户假设": "整个文件视为单个账户",
- "基准日策略": "以数据中最早交易日期为起点",
- "业务规则描述": "连续长时间无任何交易,与正常经营或资金往来的交易活跃度不符"
- }
|