"""全量穿透扫描编排(P1.5)。 把场景检测器的结果转化为线索,记录扫描覆盖范围(证明全量性)与数据版本(可追溯)。 当前为同步执行;后续可包装为 Celery 异步任务(接口保持不变)。 """ from __future__ import annotations import uuid from dataclasses import dataclass from sqlalchemy.orm import Session from app.clues import service as clue_svc from app.clues.models import Clue from app.scenarios import churn_fraud as cf from app.scenarios import split_contract as sc from app.scenarios.base import BaseScenario MODEL_VERSION = "mock-llm@0.1" @dataclass class ScanResult: scenario_code: str scanned_count: int clue: Clue | None def run_split_contract_scan( session: Session, contracts: list[sc.ContractRecord], approval_threshold: float, shared_controller: bool = False, data_version_id: uuid.UUID | None = None, ) -> ScanResult: """场景一拆单扫描:检测→评分→(命中则)生成线索。""" finding = sc.detect_threshold_edge(contracts, approval_threshold) score = sc.split_risk_score(finding, shared_controller) clue = None if score > 0: rationale = sc.build_rationale(finding, approval_threshold, shared_controller) clue = clue_svc.create_clue( session, title="疑似政企拆单规避审批", risk_domain="收入", scenario_code="R8", score=score, rationale=rationale, evidence={ "near_threshold_contracts": [c.contract_id for c in finding.near_threshold], "edge_ratio": finding.ratio, "near_threshold_amount": finding.total_amount, "approval_threshold": approval_threshold, "shared_controller": shared_controller, }, subjects={"customers": sorted({c.customer_key for c in finding.near_threshold})}, amount_involved=finding.total_amount, model_version=MODEL_VERSION, data_version_id=data_version_id, ) return ScanResult("R8", len(contracts), clue) def run_churn_scan( session: Session, retention_curve: list[cf.CohortPoint], commission_paid: float, active_ratio: float, zero_usage_ratio: float, channel_key: str, data_version_id: uuid.UUID | None = None, ) -> ScanResult: """场景二养卡骗补扫描:时序断崖 + 佣金质量不匹配→线索。""" finding = cf.detect_pulse_decay(retention_curve) mismatch = cf.commission_quality_mismatch(commission_paid, active_ratio, zero_usage_ratio) score = cf.churn_risk_score(finding, mismatch) clue = None if score >= 0.5: rationale = cf.build_rationale(finding, mismatch) clue = clue_svc.create_clue( session, title="疑似养卡骗补(脉冲增长+规律退订)", risk_domain="成本", scenario_code="R9", score=score, rationale=rationale, evidence={ "cliff_month": finding.cliff_month, "max_drop": finding.max_drop, "commission_paid": commission_paid, "active_ratio": active_ratio, "zero_usage_ratio": zero_usage_ratio, "mismatch": mismatch, }, subjects={"channel": channel_key}, amount_involved=commission_paid, model_version=MODEL_VERSION, data_version_id=data_version_id, ) return ScanResult("R9", len(retention_curve), clue) @dataclass class ScenarioRunResult: """通用调度下单个场景的落库结果:扫描覆盖数 + 生成的线索列表。""" scenario_code: str scanned_count: int clues: list[Clue] def run_scenarios( session: Session, scenarios: list[BaseScenario], *, data_version_id: uuid.UUID | None = None, actor: str = "system", ) -> list[ScenarioRunResult]: """通用场景调度器:遍历已实例化的场景,检测→按阈值过滤→统一落库。 新增场景无需改动本函数:实现 BaseScenario 子类并经 @register_scenario 注册即可。 每个场景自报扫描覆盖数(证明全量性),超过阈值的草稿由本函数统一调 create_clue 落库。 """ results: list[ScenarioRunResult] = [] for scenario in scenarios: outcome = scenario.scan(session, data_version_id=data_version_id) created: list[Clue] = [] for draft in outcome.drafts: if draft.score >= scenario.score_threshold: created.append( clue_svc.create_clue( session, title=draft.title or scenario.title, risk_domain=scenario.risk_domain, scenario_code=scenario.code, score=draft.score, rationale=draft.rationale, evidence=draft.evidence, subjects=draft.subjects, amount_involved=draft.amount_involved, model_version=MODEL_VERSION, rule_version=scenario.rule_version, data_version_id=data_version_id, actor=actor, ) ) results.append( ScenarioRunResult(scenario.code, outcome.scanned_count, created) ) return results