inactive_account_recognizer.py 20 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450
  1. from pydantic import BaseModel, Field
  2. from typing import Dict, Any, Optional, Type, List
  3. import pandas as pd
  4. from .enhanced_base_recognizer import EnhancedBaseRecognizer
  5. class InactiveAccountInput(BaseModel):
  6. """长期无交易账户识别工具输入"""
  7. csv_path: Optional[str] = Field(
  8. None,
  9. description="CSV文件路径(可选)。如果初始化时已提供csv_path,可以不用再次传入。"
  10. )
  11. class Config:
  12. arbitrary_types_allowed = True
  13. class InactiveAccountRecognizer(EnhancedBaseRecognizer):
  14. """
  15. 长期无交易账户识别器
  16. 异常规则定义:
  17. 若发现账户在指定周期内未产生任何流水明细,与正常经营或资金往来应具备的交易活跃度不符,
  18. 违背业务常理,可判定为流水存在异常,完整性存疑。
  19. 核心逻辑(根据您的要求调整):
  20. 1. 数据只属于一个账户,无需账户标识字段
  21. 2. 从数据最早日期开始计算连续无交易天数
  22. 3. 检查是否存在长时间无交易的"空白期"
  23. """
  24. args_schema: Type[BaseModel] = InactiveAccountInput
  25. # 配置参数(简化版)
  26. inactive_period_days: int = Field(
  27. 30,
  28. description="无交易天数阈值(天),连续无交易超过此天数视为异常"
  29. )
  30. # 严重程度配置
  31. severity_level: str = Field(
  32. 'medium',
  33. description="异常严重程度(high/medium/low)"
  34. )
  35. def __init__(self, csv_path: str = None, config: Dict[str, Any] = None, **kwargs):
  36. """
  37. 初始化长期无交易账户识别器
  38. Args:
  39. csv_path: CSV文件路径
  40. config: 配置参数
  41. **kwargs: 其他参数
  42. """
  43. # 调用父类的 __init__
  44. super().__init__(
  45. name="inactive_account_recognizer",
  46. description="识别在指定周期内无任何交易记录的异常账户,检查流水完整性。",
  47. display_name="长期无交易账户识别器",
  48. csv_path=csv_path,
  49. config=config,
  50. **kwargs
  51. )
  52. # 从config获取配置,更新Field属性
  53. inactive_config = self.get_config_value('inactive_account_check', {})
  54. if inactive_config:
  55. config_mapping = {
  56. 'inactive_period_days': 'inactive_period_days',
  57. 'severity_level': 'severity_level'
  58. }
  59. for config_key, attr_name in config_mapping.items():
  60. if config_key in inactive_config:
  61. setattr(self, attr_name, inactive_config[config_key])
  62. print(f"✅ {self.display_name} 初始化完成")
  63. print(f" 无交易天数阈值: {self.inactive_period_days}天")
  64. print(f" 异常严重程度: {self.severity_level}")
  65. print(f" 账户假设: 整个文件视为单个账户")
  66. print(f" 基准日策略: 从最早交易日期开始检查")
  67. def _check_long_inactive_periods(self, df: pd.DataFrame) -> List[Dict[str, Any]]:
  68. """
  69. 检查长时间无交易的空白期
  70. Args:
  71. df: 交易数据(已按时间排序)
  72. Returns:
  73. List[Dict[str, Any]]: 发现的空白期列表
  74. """
  75. if len(df) < 2:
  76. return []
  77. # 确保按时间排序
  78. df = df.sort_values('datetime')
  79. inactive_periods = []
  80. # 检查交易之间的时间间隔
  81. for i in range(len(df) - 1):
  82. current_date = df.iloc[i]['datetime']
  83. next_date = df.iloc[i + 1]['datetime']
  84. # 计算天数差
  85. days_diff = (next_date - current_date).days
  86. # 如果间隔超过阈值,记录为空白期
  87. if days_diff > self.inactive_period_days:
  88. period_info = {
  89. 'start_date': current_date,
  90. 'end_date': next_date,
  91. 'inactive_days': days_diff,
  92. 'period_index': i,
  93. 'next_tx_id': df.iloc[i + 1]['txId']
  94. }
  95. inactive_periods.append(period_info)
  96. return inactive_periods
  97. def _check_data_beginning_gap(self, df: pd.DataFrame, earliest_date: pd.Timestamp,
  98. data_coverage_days: int) -> Optional[Dict[str, Any]]:
  99. """
  100. 检查数据开始前是否有空白期
  101. Args:
  102. df: 交易数据
  103. earliest_date: 数据中最早的交易日期
  104. data_coverage_days: 数据覆盖的总天数
  105. Returns:
  106. 空白期信息或None
  107. """
  108. # 如果数据覆盖天数足够长(比如超过60天),但前面部分没有交易
  109. # 这本身可能就是一个空白期的迹象
  110. if data_coverage_days > self.inactive_period_days:
  111. # 找到第一条交易后的日期范围
  112. second_date = df.sort_values('datetime').iloc[1]['datetime'] if len(df) > 1 else earliest_date
  113. first_gap = (second_date - earliest_date).days if len(df) > 1 else 0
  114. if first_gap > self.inactive_period_days:
  115. return {
  116. 'type': 'beginning_gap',
  117. 'inactive_days': first_gap,
  118. 'start_date': earliest_date,
  119. 'end_date': second_date
  120. }
  121. return None
  122. def recognize(self, csv_path: str = None, **kwargs) -> Dict[str, Any]:
  123. """
  124. 识别长期无交易账户异常
  125. Args:
  126. csv_path: CSV文件路径
  127. **kwargs: 其他参数
  128. Returns:
  129. Dict[str, Any]: 识别结果
  130. """
  131. try:
  132. # 使用父类的load_data方法加载标准化数据
  133. df = self.load_data(csv_path)
  134. print(f"🔍 {self.display_name}开始检查,共 {len(df)} 条记录")
  135. print(f" 检查规则: 从最早交易日期开始,检查是否存在连续{self.inactive_period_days}天以上的无交易空白期")
  136. # 检查必需字段
  137. if 'datetime' not in df.columns:
  138. return {
  139. 'recognition_type': self.display_name,
  140. 'identified_count': 0,
  141. 'identified_anomalies': [],
  142. 'recognition_status': '失败',
  143. 'error': '缺少必需字段: datetime(时间信息)',
  144. 'recommendation': '请确保数据包含有效的日期时间信息'
  145. }
  146. # ============ 数据基本情况分析 ============
  147. earliest_date = df['datetime'].min()
  148. latest_date = df['datetime'].max()
  149. data_coverage_days = (latest_date - earliest_date).days + 1
  150. print(f"📊 数据基本情况:")
  151. print(f" 时间范围: {earliest_date.strftime('%Y-%m-%d')} 至 {latest_date.strftime('%Y-%m-%d')}")
  152. print(f" 数据覆盖天数: {data_coverage_days}天")
  153. print(f" 总交易笔数: {len(df)}")
  154. print(f" 日均交易笔数: {len(df) / data_coverage_days:.2f}笔/天")
  155. # ============ 检查数据是否足够 ============
  156. if data_coverage_days < self.inactive_period_days:
  157. print(f"⚠️ 数据不足: 仅覆盖{data_coverage_days}天,小于阈值{self.inactive_period_days}天")
  158. print(f" 建议: 需要更长时间范围的数据才能准确判断")
  159. return {
  160. 'recognition_type': self.display_name,
  161. 'identified_count': 0,
  162. 'identified_anomalies': [],
  163. 'recognition_status': '完成',
  164. 'recognition_parameters': {
  165. 'inactive_period_days': self.inactive_period_days,
  166. 'severity_level': self.severity_level,
  167. 'data_coverage_days': data_coverage_days
  168. },
  169. 'statistics': {
  170. 'total_transactions': len(df),
  171. 'data_coverage_days': data_coverage_days,
  172. 'date_range': {
  173. 'start': earliest_date.strftime('%Y-%m-%d'),
  174. 'end': latest_date.strftime('%Y-%m-%d')
  175. },
  176. 'transaction_frequency': len(df) / data_coverage_days if data_coverage_days > 0 else 0
  177. },
  178. 'note': f'数据覆盖天数({data_coverage_days}天)不足阈值({self.inactive_period_days}天),无法准确判断'
  179. }
  180. # ============ 检查长时间空白期 ============
  181. identified_anomalies = []
  182. # 1. 检查交易之间的空白期
  183. inactive_periods = self._check_long_inactive_periods(df)
  184. for period in inactive_periods:
  185. start_date = period['start_date']
  186. end_date = period['end_date']
  187. inactive_days = period['inactive_days']
  188. next_tx_id = period['next_tx_id']
  189. # 生成异常原因
  190. reason = f"发现长时间无交易空白期:从{start_date.strftime('%Y-%m-%d')}到{end_date.strftime('%Y-%m-%d')},连续{inactive_days}天无任何交易,超过阈值{self.inactive_period_days}天"
  191. print(f" ❌ 发现空白期: {reason}")
  192. # 获取下一笔交易的详细信息
  193. next_tx_data = df[df['txId'] == next_tx_id]
  194. if not next_tx_data.empty:
  195. next_tx = next_tx_data.iloc[0]
  196. # 使用真实交易数据创建异常记录
  197. anomaly = {
  198. 'txId': str(next_tx_id),
  199. 'txDate': str(next_tx['txDate']),
  200. 'txTime': str(next_tx['txTime']),
  201. 'datetime': next_tx['datetime'] if 'datetime' in next_tx else end_date,
  202. 'txAmount': float(next_tx['txAmount']),
  203. 'txDirection': str(next_tx['txDirection']),
  204. 'recognition_reason': f"长期无交易异常:账户在{start_date.strftime('%Y-%m-%d')}至{end_date.strftime('%Y-%m-%d')}期间连续{inactive_days}天无任何交易,超过阈值{self.inactive_period_days}天。此笔交易({next_tx_id})为空白期后的首笔交易",
  205. 'severity': self.severity_level,
  206. 'status': '待核查',
  207. 'check_type': 'inactive_account_period',
  208. 'period_info': {
  209. 'start_date': start_date.strftime('%Y-%m-%d'),
  210. 'end_date': end_date.strftime('%Y-%m-%d'),
  211. 'inactive_days': inactive_days,
  212. 'threshold_days': self.inactive_period_days,
  213. 'next_tx_id': next_tx_id,
  214. 'next_tx_info': {
  215. 'txDate': str(next_tx['txDate']),
  216. 'txTime': str(next_tx['txTime']),
  217. 'txAmount': float(next_tx['txAmount']),
  218. 'txDirection': str(next_tx['txDirection'])
  219. }
  220. }
  221. }
  222. # 创建包含所有必要字段的Series
  223. anomaly_series = pd.Series({
  224. 'txId': anomaly['txId'],
  225. 'txDate': anomaly['txDate'],
  226. 'txTime': anomaly['txTime'],
  227. 'txAmount': anomaly['txAmount'],
  228. 'txDirection': anomaly['txDirection'],
  229. 'txBalance': next_tx.get('txBalance', None),
  230. 'txSummary': next_tx.get('txSummary', ''),
  231. 'txCounterparty': next_tx.get('txCounterparty', ''),
  232. 'datetime': anomaly['datetime']
  233. })
  234. else:
  235. # 如果找不到下一笔交易,使用改进的默认格式
  236. print(f"⚠️ 警告:未找到交易ID {next_tx_id} 的详细信息,使用默认格式")
  237. anomaly = {
  238. 'txId': str(next_tx_id),
  239. 'txDate': end_date.strftime('%Y-%m-%d'),
  240. 'txTime': '23:59:59',
  241. 'datetime': end_date,
  242. 'txAmount': 0.0,
  243. 'txDirection': '收入', # 默认设为收入
  244. 'recognition_reason': f"{reason}",
  245. 'severity': self.severity_level,
  246. 'status': '待核查',
  247. 'check_type': 'inactive_account_period',
  248. 'period_info': {
  249. 'start_date': start_date.strftime('%Y-%m-%d'),
  250. 'end_date': end_date.strftime('%Y-%m-%d'),
  251. 'inactive_days': inactive_days,
  252. 'threshold_days': self.inactive_period_days
  253. }
  254. }
  255. anomaly_series = pd.Series({
  256. 'txId': anomaly['txId'],
  257. 'txDate': anomaly['txDate'],
  258. 'txTime': anomaly['txTime'],
  259. 'txAmount': anomaly['txAmount'],
  260. 'txDirection': anomaly['txDirection'],
  261. 'datetime': anomaly['datetime']
  262. })
  263. # 格式化异常记录
  264. formatted_anomaly = self.format_anomaly_record(
  265. row=anomaly_series,
  266. reason=anomaly['recognition_reason'],
  267. severity=anomaly['severity'],
  268. check_type=anomaly['check_type'],
  269. **anomaly['period_info']
  270. )
  271. identified_anomalies.append(formatted_anomaly)
  272. # 2. 检查整体交易活跃度(如果整个数据期交易都很少)
  273. avg_transactions_per_day = len(df) / data_coverage_days
  274. if avg_transactions_per_day < 0.1: # 平均每天不足0.1笔交易
  275. print(f"⚠️ 交易活跃度极低: 平均{avg_transactions_per_day:.2f}笔/天")
  276. # 生成低活跃度异常
  277. low_activity_reason = f"账户整体交易活跃度极低:{data_coverage_days}天内仅{len(df)}笔交易,平均{avg_transactions_per_day:.2f}笔/天,不符合正常资金往来特征"
  278. # 使用最后一笔交易作为异常记录的基础
  279. last_tx = df.iloc[-1] if len(df) > 0 else None
  280. if last_tx is not None:
  281. anomaly_series = pd.Series({
  282. 'txId': 'LOW_ACTIVITY_OVERALL',
  283. 'txDate': last_tx['txDate'],
  284. 'txTime': last_tx['txTime'],
  285. 'txAmount': float(last_tx['txAmount']),
  286. 'txDirection': str(last_tx['txDirection']),
  287. 'txBalance': last_tx.get('txBalance', None),
  288. 'txSummary': last_tx.get('txSummary', ''),
  289. 'txCounterparty': last_tx.get('txCounterparty', ''),
  290. 'datetime': last_tx.get('datetime', latest_date)
  291. })
  292. else:
  293. anomaly_series = pd.Series({
  294. 'txId': 'LOW_ACTIVITY_OVERALL',
  295. 'txDate': latest_date.strftime('%Y-%m-%d'),
  296. 'txTime': '23:59:59',
  297. 'txAmount': 0.0,
  298. 'txDirection': '收入',
  299. 'datetime': latest_date
  300. })
  301. formatted_anomaly = self.format_anomaly_record(
  302. row=anomaly_series,
  303. reason=low_activity_reason,
  304. severity='low' if self.severity_level == 'medium' else self.severity_level, # 降一级严重度
  305. check_type='low_activity_overall',
  306. activity_metrics={
  307. 'data_coverage_days': data_coverage_days,
  308. 'total_transactions': len(df),
  309. 'avg_transactions_per_day': avg_transactions_per_day
  310. }
  311. )
  312. identified_anomalies.append(formatted_anomaly)
  313. # ============ 结果统计 ============
  314. print(f"✅ {self.display_name}检查完成")
  315. print(f" 检查结果:")
  316. print(f" 空白期数量: {len(inactive_periods)}")
  317. print(f" 异常记录数: {len(identified_anomalies)}")
  318. if len(inactive_periods) == 0:
  319. print(f" ✅ 未发现超过{self.inactive_period_days}天的空白期")
  320. return {
  321. 'recognition_type': self.display_name,
  322. 'identified_count': len(identified_anomalies),
  323. 'identified_anomalies': identified_anomalies,
  324. 'recognition_status': '完成',
  325. 'recognition_parameters': {
  326. 'inactive_period_days': self.inactive_period_days,
  327. 'severity_level': self.severity_level,
  328. 'check_strategy': '从最早交易日期开始检查空白期'
  329. },
  330. 'statistics': {
  331. 'total_transactions': len(df),
  332. 'data_coverage_days': data_coverage_days,
  333. 'date_range': {
  334. 'start': earliest_date.strftime('%Y-%m-%d'),
  335. 'end': latest_date.strftime('%Y-%m-%d')
  336. },
  337. 'transaction_frequency': avg_transactions_per_day,
  338. 'inactive_periods_count': len(inactive_periods),
  339. 'inactive_periods_details': [
  340. {
  341. 'start_date': p['start_date'].strftime('%Y-%m-%d'),
  342. 'end_date': p['end_date'].strftime('%Y-%m-%d'),
  343. 'inactive_days': p['inactive_days'],
  344. 'next_tx_id': p['next_tx_id']
  345. }
  346. for p in inactive_periods
  347. ]
  348. }
  349. }
  350. except FileNotFoundError as e:
  351. return {
  352. 'recognition_type': self.display_name,
  353. 'identified_count': 0,
  354. 'identified_anomalies': [],
  355. 'recognition_status': '失败',
  356. 'error': f'文件不存在: {str(e)}'
  357. }
  358. except Exception as e:
  359. import traceback
  360. traceback.print_exc()
  361. return {
  362. 'recognition_type': self.display_name,
  363. 'identified_count': 0,
  364. 'identified_anomalies': [],
  365. 'recognition_status': '失败',
  366. 'error': f'数据加载或处理失败: {str(e)}'
  367. }
  368. def get_summary(self) -> Dict[str, Any]:
  369. """获取识别器摘要"""
  370. summary = super().get_summary()
  371. summary.update({
  372. 'inactive_period_days': self.inactive_period_days,
  373. 'severity_level': self.severity_level,
  374. 'data_loaded': self._data is not None,
  375. 'check_strategy': '从最早交易日期开始检查连续无交易空白期'
  376. })
  377. return summary
  378. def get_config_summary(self) -> Dict[str, Any]:
  379. """获取配置摘要"""
  380. return {
  381. "无交易天数阈值": f"{self.inactive_period_days}天",
  382. "异常严重程度": self.severity_level.upper(),
  383. "检测逻辑": f"从最早交易日期开始,检查连续{self.inactive_period_days}天以上的无交易空白期",
  384. "账户假设": "整个文件视为单个账户",
  385. "基准日策略": "以数据中最早交易日期为起点",
  386. "业务规则描述": "连续长时间无任何交易,与正常经营或资金往来的交易活跃度不符"
  387. }