"""场景一 · 政企收入全链路穿透:拆单规避检测(R8)。 检测点: 1. 合同金额集中分布在审批阈值边缘(如阈值 80% 以上但不超阈值)。 2. 结合知识图谱穿透识别隐性实控人(多个客户经法人关联到同一实控人)。 满足上述模式则生成线索,附证据链与人话理由。 """ from __future__ import annotations from dataclasses import dataclass, field @dataclass class ContractRecord: """穿透分析输入:一份合同的关键信息。""" contract_id: str customer_key: str amount: float @dataclass class SplitFinding: """拆单检测结果。""" near_threshold: list[ContractRecord] = field(default_factory=list) ratio: float = 0.0 total_amount: float = 0.0 @property def hit(self) -> bool: return len(self.near_threshold) >= 3 def detect_threshold_edge( contracts: list[ContractRecord], approval_threshold: float, edge_ratio: float = 0.8, ) -> SplitFinding: """识别金额集中在审批阈值边缘 [edge_ratio*阈值, 阈值) 的合同。 这类"刚好低于阈值"的批量合同是典型的拆单规避特征。 """ if approval_threshold <= 0: raise ValueError("审批阈值必须为正数") lower = edge_ratio * approval_threshold near = [c for c in contracts if lower <= c.amount < approval_threshold] finding = SplitFinding( near_threshold=near, ratio=(len(near) / len(contracts)) if contracts else 0.0, total_amount=sum(c.amount for c in near), ) return finding def split_risk_score(finding: SplitFinding, shared_controller: bool) -> float: """综合评分:阈值边缘集中度 + 是否穿透到同一实控人。""" if not finding.hit: return 0.0 base = min(0.6, 0.1 * len(finding.near_threshold)) # 数量越多越可疑 base += 0.2 * finding.ratio if shared_controller: base += 0.3 # 同一实控人是强证据 return round(min(base, 1.0), 3) def build_rationale(finding: SplitFinding, threshold: float, shared_controller: bool) -> str: parts = [ f"检测到 {len(finding.near_threshold)} 份合同金额集中在审批阈值 " f"{threshold:.0f} 的边缘区间(占比 {finding.ratio:.0%}),", f"边缘合同金额合计约 {finding.total_amount:.0f}。", ] if shared_controller: parts.append("且经工商关联穿透,相关客户疑似同属一个隐性实控人,高度符合拆单规避特征。") else: parts.append("建议进一步穿透客户关联关系以确认是否同一实控人。") return "".join(parts)