From 74e35fed9660f7b66467b0ccce8c580392aa2d8c Mon Sep 17 00:00:00 2001 From: selfrelease Date: Wed, 17 Jun 2026 09:57:02 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E6=B7=BB=E5=8A=A0=E5=8F=AF=E9=85=8D?= =?UTF-8?q?=E7=BD=AE=E5=9C=BA=E6=99=AF=E6=B3=A8=E5=86=8C=E8=A1=A8=E4=B8=8E?= =?UTF-8?q?=E8=A7=84=E5=88=99=E5=BC=95=E6=93=8E?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 新增场景基类、配置加载、注册表与原语模块 - 添加 r10_refund_split 规则及场景 JSON Schema - 扩展 scan 引擎与 scenarios API - 新增场景注册表/配置/集成测试 - 更新前端 App、api、labels 支持新场景 --- backend/app/api/scenarios.py | 20 ++ backend/app/api/schemas.py | 9 + backend/app/engines/scan.py | 51 +++++ backend/app/main.py | 2 + backend/app/scenarios/__init__.py | 25 ++- backend/app/scenarios/base.py | 76 +++++++ backend/app/scenarios/churn_fraud.py | 50 +++++ backend/app/scenarios/config.py | 118 ++++++++++ backend/app/scenarios/primitives.py | 210 ++++++++++++++++++ backend/app/scenarios/registry.py | 36 +++ .../app/scenarios/rules/r10_refund_split.yaml | 23 ++ .../app/scenarios/rules/scenario.schema.json | 53 +++++ backend/app/scenarios/split_contract.py | 58 +++++ backend/requirements.txt | 1 + .../tests/integration/test_run_scenarios.py | 62 ++++++ backend/tests/test_config_scenarios.py | 100 +++++++++ backend/tests/test_scenario_registry.py | 74 ++++++ frontend/src/App.tsx | 16 +- frontend/src/api.ts | 8 + frontend/src/labels.ts | 8 + 20 files changed, 998 insertions(+), 2 deletions(-) create mode 100644 backend/app/api/scenarios.py create mode 100644 backend/app/scenarios/base.py create mode 100644 backend/app/scenarios/config.py create mode 100644 backend/app/scenarios/primitives.py create mode 100644 backend/app/scenarios/registry.py create mode 100644 backend/app/scenarios/rules/r10_refund_split.yaml create mode 100644 backend/app/scenarios/rules/scenario.schema.json create mode 100644 backend/tests/integration/test_run_scenarios.py create mode 100644 backend/tests/test_config_scenarios.py create mode 100644 backend/tests/test_scenario_registry.py diff --git a/backend/app/api/scenarios.py b/backend/app/api/scenarios.py new file mode 100644 index 0000000..e44c452 --- /dev/null +++ b/backend/app/api/scenarios.py @@ -0,0 +1,20 @@ +"""场景元数据 API：暴露已注册的审计场景，供前端动态渲染。 + +新增场景（实现 BaseScenario 并经 @register_scenario 注册）后，本端点自动反映， +无需改动前端代码。导入 registry 时会触发 app.scenarios 包的自动发现与注册。 +""" + +from __future__ import annotations + +from fastapi import APIRouter + +from app.api.schemas import ScenarioOut +from app.scenarios.registry import scenario_meta + +router = APIRouter(prefix="/scenarios", tags=["scenarios"]) + + +@router.get("", response_model=list[ScenarioOut]) +def list_scenarios() -> list[dict]: + """返回全部已注册场景的元数据（场景码、标题、展示名、风险域）。""" + return scenario_meta() diff --git a/backend/app/api/schemas.py b/backend/app/api/schemas.py index e869bd4..321c0b5 100644 --- a/backend/app/api/schemas.py +++ b/backend/app/api/schemas.py @@ -83,3 +83,12 @@ class DashboardSummary(BaseModel): by_confidence: dict[str, int] by_scenario: dict[str, int] total_amount_involved: float + + +class ScenarioOut(BaseModel): + """已注册审计场景的元数据，供前端动态渲染场景名（替代写死的 scenarioLabel）。""" + + code: str + title: str + label: str + risk_domain: str diff --git a/backend/app/engines/scan.py b/backend/app/engines/scan.py index c15a636..b23ef96 100644 --- a/backend/app/engines/scan.py +++ b/backend/app/engines/scan.py @@ -15,6 +15,7 @@ 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" @@ -98,3 +99,53 @@ def run_churn_scan( 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 diff --git a/backend/app/main.py b/backend/app/main.py index 14b7970..446276e 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -10,6 +10,7 @@ from app import __version__ from app.api.clues import router as clues_router from app.api.datahub import router as datahub_router from app.api.nlq import router as nlq_router +from app.api.scenarios import router as scenarios_router from app.config import get_settings @@ -30,6 +31,7 @@ app = FastAPI( app.include_router(datahub_router) app.include_router(clues_router) app.include_router(nlq_router) +app.include_router(scenarios_router) @app.get("/health") diff --git a/backend/app/scenarios/__init__.py b/backend/app/scenarios/__init__.py index f6df229..6098ced 100644 --- a/backend/app/scenarios/__init__.py +++ b/backend/app/scenarios/__init__.py @@ -1 +1,24 @@ -"""审计场景检测器：将业务数据中的异常模式转化为线索。""" +"""审计场景检测器：将业务数据中的异常模式转化为线索。 + +自动发现：导入本包下的全部场景模块以触发 @register_scenario 注册。 +新增场景时只需在 scenarios/ 下新增一个模块文件，无需改动任何现有代码即可被 +注册表（registry）与调度器（engines.scan.run_scenarios）发现。 +""" + +from __future__ import annotations + +import importlib +import pkgutil + +# 框架自身模块（非具体场景），不参与"代码场景"的自动导入 +_FRAMEWORK_MODULES = {"base", "registry", "primitives", "config"} + +# 1) 自动发现并导入"代码场景"模块（A 打底），触发 @register_scenario 注册 +for _module in pkgutil.iter_modules(__path__): + if _module.name not in _FRAMEWORK_MODULES: + importlib.import_module(f"{__name__}.{_module.name}") + +# 2) 加载"配置驱动场景"（B 快捷通道）：自动发现 rules/*.yaml 并注册，无需改代码 +from app.scenarios.config import load_yaml_scenarios # noqa: E402 + +load_yaml_scenarios() diff --git a/backend/app/scenarios/base.py b/backend/app/scenarios/base.py new file mode 100644 index 0000000..55c851e --- /dev/null +++ b/backend/app/scenarios/base.py @@ -0,0 +1,76 @@ +"""场景插件框架：统一的场景基类与产出物（A 打底）。 + +每个审计场景实现一个 BaseScenario 子类，自包含"检测→评分→产出线索草稿"， +并通过 @register_scenario 注册。扫描调度器（engines/scan.run_scenarios）统一遍历 +执行并落库，新增场景无需改动调度器、API 与前端。 +""" + +from __future__ import annotations + +import uuid +from abc import ABC, abstractmethod +from dataclasses import dataclass, field + +from sqlalchemy.orm import Session + + +@dataclass +class ClueDraft: + """场景产出的线索草稿：仅承载"内容"，落库由调度器统一完成。 + + 风险域、场景码等元数据来自场景类属性；模型/数据版本由调度器注入。 + """ + + score: float + rationale: str + evidence: dict + subjects: dict = field(default_factory=dict) + amount_involved: float | None = None + title: str | None = None # 覆盖场景默认标题；为 None 时用 BaseScenario.title + + +@dataclass +class ScanOutcome: + """单个场景一次扫描的结果：扫描覆盖数（证明全量性）+ 线索草稿列表。""" + + scanned_count: int + drafts: list[ClueDraft] = field(default_factory=list) + + +class BaseScenario(ABC): + """审计场景抽象基类。 + + 子类通过类属性声明元数据（场景码、标题、风险域、前端展示名、阈值）， + 并实现 scan() 完成检测与评分。命中阈值的草稿由调度器统一生成线索。 + """ + + # 场景码（如 "R8"），全局唯一，注册表按此索引 + code: str = "" + # 默认线索标题（草稿可覆盖） + title: str = "" + # 风险域（如 "收入"/"成本"） + risk_domain: str = "" + # 前端展示名（替代 labels.ts 中写死的 scenarioLabel） + label: str = "" + # 评分阈值：草稿 score >= 阈值才生成线索 + score_threshold: float = 0.5 + # 规则版本（配置驱动场景可用于追溯；代码场景一般为 None） + rule_version: str | None = None + + @abstractmethod + def scan( + self, + session: Session, + *, + data_version_id: uuid.UUID | None = None, + ) -> ScanOutcome: + """执行检测与评分，返回扫描覆盖数与命中草稿。 + + Args: + session: 数据库会话（如需自取数） + data_version_id: 当前数据版本 ID（用于线索可追溯） + + Returns: + ScanOutcome：scanned_count 体现全量覆盖，drafts 为候选线索 + """ + ... diff --git a/backend/app/scenarios/churn_fraud.py b/backend/app/scenarios/churn_fraud.py index 298cc8a..ee02f43 100644 --- a/backend/app/scenarios/churn_fraud.py +++ b/backend/app/scenarios/churn_fraud.py @@ -8,6 +8,9 @@ from __future__ import annotations from dataclasses import dataclass +from app.scenarios.base import BaseScenario, ClueDraft, ScanOutcome +from app.scenarios.registry import register_scenario + @dataclass class CohortPoint: @@ -83,3 +86,50 @@ def build_rationale(finding: ChurnFinding, mismatch: float) -> str: return ( f"未见明显断崖退订，但佣金与业务质量不匹配度为 {mismatch:.0%}，建议关注。" ) + + +@register_scenario +class ChurnFraudScenario(BaseScenario): + """场景二养卡骗补检测的插件封装：时序断崖 + 佣金质量不匹配 → 线索草稿。""" + + code = "R9" + title = "疑似养卡骗补（脉冲增长+规律退订）" + risk_domain = "成本" + label = "养卡骗补" + score_threshold = 0.5 # 与原 run_churn_scan 的 score>=0.5 一致 + + def __init__( + self, + retention_curve: list[CohortPoint], + commission_paid: float, + active_ratio: float, + zero_usage_ratio: float, + channel_key: str, + ) -> None: + self.retention_curve = retention_curve + self.commission_paid = commission_paid + self.active_ratio = active_ratio + self.zero_usage_ratio = zero_usage_ratio + self.channel_key = channel_key + + def scan(self, session, *, data_version_id=None) -> ScanOutcome: + finding = detect_pulse_decay(self.retention_curve) + mismatch = commission_quality_mismatch( + self.commission_paid, self.active_ratio, self.zero_usage_ratio + ) + score = churn_risk_score(finding, mismatch) + draft = ClueDraft( + score=score, + rationale=build_rationale(finding, mismatch), + evidence={ + "cliff_month": finding.cliff_month, + "max_drop": finding.max_drop, + "commission_paid": self.commission_paid, + "active_ratio": self.active_ratio, + "zero_usage_ratio": self.zero_usage_ratio, + "mismatch": mismatch, + }, + subjects={"channel": self.channel_key}, + amount_involved=self.commission_paid, + ) + return ScanOutcome(scanned_count=len(self.retention_curve), drafts=[draft]) diff --git a/backend/app/scenarios/config.py b/backend/app/scenarios/config.py new file mode 100644 index 0000000..2ba6b96 --- /dev/null +++ b/backend/app/scenarios/config.py @@ -0,0 +1,118 @@ +"""YAML 配置驱动场景（B 快捷通道）：用「配置 + 原语库」声明同构场景，免写 Python。 + +加载 scenarios/rules/*.yaml，经 pydantic 校验后为每条配置动态生成一个 BaseScenario +子类并注册。新增此类场景 = 新增一个 YAML 文件，无需改动任何 Python 代码。 +运行时与代码场景一致：实例化后注入数据行 rows，由 run_scenarios 统一调度落库。 +""" + +from __future__ import annotations + +import uuid +from pathlib import Path + +import yaml +from pydantic import BaseModel, Field, ValidationError + +from app.scenarios.base import BaseScenario, ClueDraft, ScanOutcome +from app.scenarios.primitives import get_primitive +from app.scenarios.registry import register_scenario + +_DEFAULT_RULES_DIR = Path(__file__).parent / "rules" + + +class ScenarioConfig(BaseModel): + """YAML 场景配置的 schema（pydantic 校验，字段缺失/类型错误即报错）。""" + + code: str = Field(min_length=1) + title: str = Field(min_length=1) + risk_domain: str = Field(min_length=1) + label: str = Field(min_length=1) + detector: str = Field(min_length=1) + params: dict = Field(default_factory=dict) + threshold: float = 0.5 + rationale_template: str = "{summary}" + rule_version: str | None = None + + +class ConfigScenario(BaseScenario): + """由 YAML 配置驱动的场景：运行时按 detector 调用原语，模板生成人话理由。 + + config 由动态子类（build_scenario_class）设置；数据行 rows 在实例化时注入。 + """ + + config: ScenarioConfig + + def __init__(self, rows: list[dict] | None = None) -> None: + self._rows = rows or [] + + def with_rows(self, rows: list[dict]) -> ConfigScenario: + """注入待扫描的数据行（链式返回自身）。""" + self._rows = rows + return self + + def scan(self, session, *, data_version_id: uuid.UUID | None = None) -> ScanOutcome: + fn = get_primitive(self.config.detector) + if fn is None: + raise ValueError(f"未知检测原语: {self.config.detector}") + result = fn(self._rows, self.config.params) + drafts: list[ClueDraft] = [] + if result.hit and result.score > 0: + try: + rationale = self.config.rationale_template.format( + summary=result.summary, **result.metrics + ) + except (KeyError, IndexError, ValueError): + rationale = result.summary + drafts.append( + ClueDraft( + score=result.score, + rationale=rationale, + evidence=result.evidence, + subjects=result.subjects, + ) + ) + return ScanOutcome(scanned_count=len(self._rows), drafts=drafts) + + +def build_scenario_class(config: ScenarioConfig) -> type[ConfigScenario]: + """按配置动态生成并注册一个 ConfigScenario 子类（类属性承载元数据）。""" + attrs = { + "code": config.code, + "title": config.title, + "risk_domain": config.risk_domain, + "label": config.label, + "score_threshold": config.threshold, + "rule_version": config.rule_version, + "config": config, + } + cls = type(f"ConfigScenario_{config.code}", (ConfigScenario,), attrs) + return register_scenario(cls) + + +def load_scenario_config(path: Path) -> ScenarioConfig: + """读取并校验单个 YAML 场景配置文件。""" + raw = yaml.safe_load(path.read_text(encoding="utf-8")) or {} + try: + config = ScenarioConfig(**raw) + except ValidationError as exc: + raise ValueError(f"场景配置 {path.name} 校验失败: {exc}") from exc + if get_primitive(config.detector) is None: + raise ValueError( + f"场景配置 {path.name} 引用了未注册的检测原语: {config.detector}" + ) + return config + + +def load_yaml_scenarios(rules_dir: Path | str | None = None) -> list[type[ConfigScenario]]: + """加载 rules 目录下全部 *.yaml 场景配置，动态生成并注册场景类。 + + 新增 YAML 文件即被发现，无需改动代码。返回新注册的场景类列表（便于测试）。 + """ + rules_dir = Path(rules_dir) if rules_dir else _DEFAULT_RULES_DIR + classes: list[type[ConfigScenario]] = [] + if not rules_dir.exists(): + return classes + for path in sorted(rules_dir.glob("*.yaml")): + config = load_scenario_config(path) + classes.append(build_scenario_class(config)) + return classes diff --git a/backend/app/scenarios/primitives.py b/backend/app/scenarios/primitives.py new file mode 100644 index 0000000..d098bfa --- /dev/null +++ b/backend/app/scenarios/primitives.py @@ -0,0 +1,210 @@ +"""检测原语库（B 快捷通道）：可被 YAML 配置场景复用的通用检测/评分模式。 + +每个原语接收数据行（list[dict]）与参数（dict），返回统一的 PrimitiveResult。 +新增"同构"场景时，可在 YAML 中通过 detector 引用原语、params 传参，无需编写 Python。 +若遇到原语无法表达的全新算法，则回到 A 方案实现一个 BaseScenario 子类（或新增原语）。 +""" + +from __future__ import annotations + +from collections.abc import Callable +from dataclasses import dataclass, field + +# 原语统一签名：(rows, params) -> PrimitiveResult +PrimitiveFn = Callable[[list[dict], dict], "PrimitiveResult"] + + +@dataclass +class PrimitiveResult: + """原语输出：是否命中、评分（0-1）、证据、主体、可读简述与可插值指标。""" + + hit: bool + score: float + evidence: dict = field(default_factory=dict) + subjects: dict = field(default_factory=dict) + summary: str = "" # 供 rationale_template 的 {summary} 占位 + metrics: dict = field(default_factory=dict) # 供 rationale_template 的其它占位 + + +# 全局原语注册表：name -> 原语函数 +PRIMITIVE_REGISTRY: dict[str, PrimitiveFn] = {} + + +def register_primitive(name: str) -> Callable[[PrimitiveFn], PrimitiveFn]: + """函数装饰器：把检测原语注册到全局表。""" + + def deco(fn: PrimitiveFn) -> PrimitiveFn: + PRIMITIVE_REGISTRY[name] = fn + return fn + + return deco + + +def get_primitive(name: str) -> PrimitiveFn | None: + """按名称查找已注册的检测原语。""" + return PRIMITIVE_REGISTRY.get(name) + + +def _to_float(value: object, default: float = 0.0) -> float: + try: + return float(value) # type: ignore[arg-type] + except (TypeError, ValueError): + return default + + +@register_primitive("threshold_edge") +def threshold_edge(rows: list[dict], params: dict) -> PrimitiveResult: + """金额集中在审批阈值边缘 [edge_ratio*threshold, threshold) 的拆单型模式。 + + params: + amount_field (str): 金额字段名 + threshold (float): 审批阈值（>0） + edge_ratio (float): 边缘下界比例，默认 0.8 + min_count (int): 触发命中的最小记录数，默认 3 + key_field (str, 可选): 主体字段名（用于汇总涉及主体） + """ + amount_field = params["amount_field"] + threshold = _to_float(params.get("threshold")) + edge_ratio = _to_float(params.get("edge_ratio", 0.8), 0.8) + min_count = int(params.get("min_count", 3)) + if threshold <= 0: + raise ValueError("threshold 必须为正数") + + lower = edge_ratio * threshold + near = [r for r in rows if lower <= _to_float(r.get(amount_field)) < threshold] + ratio = (len(near) / len(rows)) if rows else 0.0 + total = sum(_to_float(r.get(amount_field)) for r in near) + hit = len(near) >= min_count + + score = 0.0 + if hit: + score = min(0.6, 0.1 * len(near)) + 0.2 * ratio + score = round(min(score, 1.0), 3) + + subjects: dict = {} + key_field = params.get("key_field") + if key_field: + subjects = { + "keys": sorted( + {str(r.get(key_field)) for r in near if r.get(key_field) is not None} + ) + } + + return PrimitiveResult( + hit=hit, + score=score, + evidence={ + "near_count": len(near), + "edge_ratio": round(ratio, 3), + "near_amount": total, + "threshold": threshold, + }, + subjects=subjects, + summary=( + f"检测到 {len(near)} 条记录金额集中在审批阈值 {threshold:.0f} 的边缘区间" + f"（占比 {ratio:.0%}），合计约 {total:.0f}" + ), + metrics={ + "near_count": len(near), + "ratio": round(ratio, 3), + "near_amount": total, + "threshold": threshold, + }, + ) + + +@register_primitive("cliff_drop") +def cliff_drop(rows: list[dict], params: dict) -> PrimitiveResult: + """时序断崖：按 index 排序后，相邻点 value 骤降超过 cliff_drop 即命中。 + + params: + index_field (str): 序号/时间字段名 + value_field (str): 取值字段名（如留存率） + cliff_drop (float): 触发断崖的最小跌幅，默认 0.5 + """ + index_field = params["index_field"] + value_field = params["value_field"] + cliff = _to_float(params.get("cliff_drop", 0.5), 0.5) + + ordered = sorted(rows, key=lambda r: _to_float(r.get(index_field))) + max_drop = 0.0 + cliff_at: object = None + for prev, cur in zip(ordered, ordered[1:], strict=False): + drop = _to_float(prev.get(value_field)) - _to_float(cur.get(value_field)) + if drop > max_drop: + max_drop = drop + if drop >= cliff: + cliff_at = cur.get(index_field) + + hit = cliff_at is not None + max_drop = round(max_drop, 3) + score = round(min(0.4 + 0.5 * max_drop, 1.0), 3) if hit else round(0.3 * max_drop, 3) + + return PrimitiveResult( + hit=hit, + score=score, + evidence={"cliff_at": cliff_at, "max_drop": max_drop}, + summary=( + f"在 {cliff_at} 处出现断崖式骤降（最大跌幅 {max_drop:.0%}）" + if hit + else f"未见明显断崖（最大跌幅 {max_drop:.0%}）" + ), + metrics={"cliff_at": cliff_at, "max_drop": max_drop}, + ) + + +@register_primitive("ratio_threshold") +def ratio_threshold(rows: list[dict], params: dict) -> PrimitiveResult: + """聚合比率/阈值判定：对某字段做聚合并与阈值比较，命中则给定分。 + + params: + value_field (str): 取值字段名 + agg (str): 聚合方式 mean/sum/max/min/count，默认 mean + op (str): 比较运算 >=,>,<=,<，默认 >= + threshold (float): 比较阈值 + score (float): 命中评分，默认 0.8 + """ + value_field = params["value_field"] + agg = str(params.get("agg", "mean")) + op = str(params.get("op", ">=")) + threshold = _to_float(params.get("threshold")) + hit_score = _to_float(params.get("score", 0.8), 0.8) + + values = [_to_float(r.get(value_field)) for r in rows] + if not values and agg != "count": + return PrimitiveResult(hit=False, score=0.0, summary="无数据，未触发") + + if agg == "sum": + actual = sum(values) + elif agg == "max": + actual = max(values) + elif agg == "min": + actual = min(values) + elif agg == "count": + actual = float(len(values)) + else: # mean + actual = sum(values) / len(values) + + comparators = { + ">=": lambda a, b: a >= b, + ">": lambda a, b: a > b, + "<=": lambda a, b: a <= b, + "<": lambda a, b: a < b, + } + cmp = comparators.get(op) + if cmp is None: + raise ValueError(f"不支持的比较运算: {op}") + hit = cmp(actual, threshold) + score = round(hit_score, 3) if hit else 0.0 + + return PrimitiveResult( + hit=hit, + score=score, + evidence={"agg": agg, "actual": round(actual, 3), "op": op, "threshold": threshold}, + summary=( + f"{value_field} 的{agg}为 {actual:.3g}，{op} 阈值 {threshold:.3g}" + if hit + else f"{value_field} 的{agg}为 {actual:.3g}，未达阈值 {threshold:.3g}" + ), + metrics={"actual": round(actual, 3), "threshold": threshold}, + ) diff --git a/backend/app/scenarios/registry.py b/backend/app/scenarios/registry.py new file mode 100644 index 0000000..b7dcf1d --- /dev/null +++ b/backend/app/scenarios/registry.py @@ -0,0 +1,36 @@ +"""场景注册表：按场景码索引，供调度器与 API 统一发现已注册场景。 + +仿 app.ingest.registry 的设计：类装饰器自动注册，新增场景零改动核心代码。 +""" + +from __future__ import annotations + +from app.scenarios.base import BaseScenario + +# 全局注册表：scenario_code -> Scenario 类 +SCENARIO_REGISTRY: dict[str, type[BaseScenario]] = {} + + +def register_scenario(cls: type[BaseScenario]) -> type[BaseScenario]: + """类装饰器：将场景注册到全局表（按 code 索引）。""" + if cls.code: + SCENARIO_REGISTRY[cls.code] = cls + return cls + + +def get_scenario(code: str) -> type[BaseScenario] | None: + """按场景码查找已注册的场景类。""" + return SCENARIO_REGISTRY.get(code) + + +def scenario_meta() -> list[dict]: + """导出全部已注册场景的元数据，供 /scenarios API 与前端动态渲染。""" + return [ + { + "code": cls.code, + "title": cls.title, + "label": cls.label, + "risk_domain": cls.risk_domain, + } + for cls in SCENARIO_REGISTRY.values() + ] diff --git a/backend/app/scenarios/rules/r10_refund_split.yaml b/backend/app/scenarios/rules/r10_refund_split.yaml new file mode 100644 index 0000000..8299fa0 --- /dev/null +++ b/backend/app/scenarios/rules/r10_refund_split.yaml @@ -0,0 +1,23 @@ +# yaml-language-server: $schema=./scenario.schema.json +# 场景 R10（纯配置示例）· 大额退款拆分规避审批 +# 演示「B 快捷通道」：新增本场景仅需此 YAML 文件，无需改动任何 Python 代码。 +# 系统启动时由 app.scenarios.config.load_yaml_scenarios 自动发现、校验并注册。 + +code: R10 +title: 疑似大额退款拆分规避审批 +risk_domain: 收入 +label: 退款拆分 + +# 复用通用检测原语：金额集中在审批阈值边缘 +detector: threshold_edge +threshold: 0.0 # threshold_edge 命中（评分>0）即生成线索 +rule_version: rules@r10.v1 + +params: + amount_field: refund_amount # 数据行中的金额字段 + threshold: 500000 # 退款审批阈值（50 万） + edge_ratio: 0.8 # 边缘下界 = 0.8 * 阈值 + min_count: 3 # 至少 3 笔集中才判定命中 + key_field: customer # 主体字段，用于汇总涉及客户 + +rationale_template: "{summary}；疑似将整笔大额退款拆分至审批阈值以下以规避审批，建议穿透核查关联客户。" diff --git a/backend/app/scenarios/rules/scenario.schema.json b/backend/app/scenarios/rules/scenario.schema.json new file mode 100644 index 0000000..8d8f229 --- /dev/null +++ b/backend/app/scenarios/rules/scenario.schema.json @@ -0,0 +1,53 @@ +{ + "$schema": "http://json-schema.org/draft-07/schema#", + "title": "AIAudit 配置驱动场景", + "description": "配置驱动审计场景（B 快捷通道）的 schema，对应 app.scenarios.config.ScenarioConfig；权威校验以后端 pydantic 为准。", + "type": "object", + "required": ["code", "title", "risk_domain", "label", "detector"], + "additionalProperties": false, + "properties": { + "code": { + "type": "string", + "minLength": 1, + "description": "场景码，全局唯一，如 R10" + }, + "title": { + "type": "string", + "minLength": 1, + "description": "默认线索标题" + }, + "risk_domain": { + "type": "string", + "minLength": 1, + "description": "风险域，如 收入/成本" + }, + "label": { + "type": "string", + "minLength": 1, + "description": "前端展示名" + }, + "detector": { + "type": "string", + "minLength": 1, + "description": "检测原语名（见 app/scenarios/primitives.py）", + "examples": ["threshold_edge", "cliff_drop", "ratio_threshold"] + }, + "params": { + "type": "object", + "description": "传给检测原语的参数（键随原语而定）" + }, + "threshold": { + "type": "number", + "description": "评分阈值：草稿 score >= 阈值才生成线索", + "default": 0.5 + }, + "rationale_template": { + "type": "string", + "description": "线索理由模板，支持 {summary} 及原语 metrics 中的占位符" + }, + "rule_version": { + "type": ["string", "null"], + "description": "规则版本，用于线索可追溯" + } + } +} diff --git a/backend/app/scenarios/split_contract.py b/backend/app/scenarios/split_contract.py index 2be3a7a..7b4f5f9 100644 --- a/backend/app/scenarios/split_contract.py +++ b/backend/app/scenarios/split_contract.py @@ -10,6 +10,9 @@ from __future__ import annotations from dataclasses import dataclass, field +from app.scenarios.base import BaseScenario, ClueDraft, ScanOutcome +from app.scenarios.registry import register_scenario + @dataclass class ContractRecord: @@ -76,3 +79,58 @@ def build_rationale(finding: SplitFinding, threshold: float, shared_controller: else: parts.append("建议进一步穿透客户关联关系以确认是否同一实控人。") return "".join(parts) + + +@register_scenario +class SplitContractScenario(BaseScenario): + """场景一拆单检测的插件封装：检测→评分→产出线索草稿。 + + 数据通过构造注入（兼容现有 seed/测试的数据流）；后续可改为从 session 自取数， + 接口保持不变。 + """ + + code = "R8" + title = "疑似政企拆单规避审批" + risk_domain = "收入" + label = "政企拆单" + score_threshold = 0.0 # 命中即出线索（与原 run_split_contract_scan 的 score>0 一致） + + def __init__( + self, + contracts: list[ContractRecord], + approval_threshold: float, + shared_controller: bool = False, + ) -> None: + self.contracts = contracts + self.approval_threshold = approval_threshold + self.shared_controller = shared_controller + + def scan(self, session, *, data_version_id=None) -> ScanOutcome: + finding = detect_threshold_edge(self.contracts, self.approval_threshold) + score = split_risk_score(finding, self.shared_controller) + drafts: list[ClueDraft] = [] + if score > 0: + drafts.append( + ClueDraft( + score=score, + rationale=build_rationale( + finding, self.approval_threshold, self.shared_controller + ), + 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": self.approval_threshold, + "shared_controller": self.shared_controller, + }, + subjects={ + "customers": sorted( + {c.customer_key for c in finding.near_threshold} + ) + }, + amount_involved=finding.total_amount, + ) + ) + return ScanOutcome(scanned_count=len(self.contracts), drafts=drafts) diff --git a/backend/requirements.txt b/backend/requirements.txt index ce83ff0..f3fcf77 100644 --- a/backend/requirements.txt +++ b/backend/requirements.txt @@ -9,3 +9,4 @@ celery==5.4.0 redis==5.2.1 httpx==0.28.1 python-dotenv==1.0.1 +pyyaml==6.0.3 diff --git a/backend/tests/integration/test_run_scenarios.py b/backend/tests/integration/test_run_scenarios.py new file mode 100644 index 0000000..9d64ad7 --- /dev/null +++ b/backend/tests/integration/test_run_scenarios.py @@ -0,0 +1,62 @@ +"""通用场景调度器 run_scenarios 的落库集成测试（需 PostgreSQL）。 + +验证「场景插件 → 统一调度 → 阈值过滤 → 落库」链路，且与旧 run_xxx_scan 行为一致。 +""" + +from __future__ import annotations + +from app.clues.models import ClueStatus, ConfidenceTier +from app.engines import scan +from app.scenarios.churn_fraud import ChurnFraudScenario, CohortPoint +from app.scenarios.split_contract import ContractRecord, SplitContractScenario + + +def test_run_scenarios_creates_clues(session): + contracts = [ContractRecord(f"C{i}", f"CUST{i}", 850000) for i in range(8)] + curve = [CohortPoint(0, 1.0), CohortPoint(1, 0.95), CohortPoint(2, 0.1)] + scenarios = [ + SplitContractScenario( + contracts, approval_threshold=1_000_000, shared_controller=True + ), + ChurnFraudScenario( + retention_curve=curve, + commission_paid=300000, + active_ratio=0.05, + zero_usage_ratio=0.9, + channel_key="CH-001", + ), + ] + results = scan.run_scenarios(session, scenarios) + by_code = {r.scenario_code: r for r in results} + + assert by_code["R8"].scanned_count == 8 + assert len(by_code["R8"].clues) == 1 + r8_clue = by_code["R8"].clues[0] + assert r8_clue.confidence == ConfidenceTier.HIGH + assert r8_clue.status == ClueStatus.NEW + assert r8_clue.amount_involved > 0 + assert r8_clue.model_version == scan.MODEL_VERSION + + assert by_code["R9"].scanned_count == 3 + assert len(by_code["R9"].clues) == 1 + assert by_code["R9"].clues[0].scenario_code == "R9" + assert by_code["R9"].clues[0].subjects["channel"] == "CH-001" + + +def test_run_scenarios_filters_below_threshold(session): + # 干净数据：拆单不命中、养卡曲线平滑且佣金匹配 → 均不应生成线索 + contracts = [ContractRecord("C1", "A", 100000), ContractRecord("C2", "B", 3_000_000)] + curve = [CohortPoint(0, 1.0), CohortPoint(1, 0.98), CohortPoint(2, 0.97)] + scenarios = [ + SplitContractScenario(contracts, approval_threshold=1_000_000), + ChurnFraudScenario( + retention_curve=curve, + commission_paid=10000, + active_ratio=0.9, + zero_usage_ratio=0.05, + channel_key="CH-X", + ), + ] + results = scan.run_scenarios(session, scenarios) + for r in results: + assert r.clues == [] diff --git a/backend/tests/test_config_scenarios.py b/backend/tests/test_config_scenarios.py new file mode 100644 index 0000000..35b129b --- /dev/null +++ b/backend/tests/test_config_scenarios.py @@ -0,0 +1,100 @@ +"""配置驱动场景（B 快捷通道）测试：原语库、YAML 加载与校验、ConfigScenario 扫描。 + +纯逻辑，无需数据库。验证「新增 YAML → 自动注册 → 可扫描产出线索草稿」链路。 +""" + +from __future__ import annotations + +import pytest +from pydantic import ValidationError + +from app.scenarios.config import ( + ScenarioConfig, + load_yaml_scenarios, +) +from app.scenarios.primitives import cliff_drop, ratio_threshold, threshold_edge +from app.scenarios.registry import SCENARIO_REGISTRY, get_scenario + + +# ---------- 原语库 ---------- + +def test_threshold_edge_primitive_hits(): + rows = [{"refund_amount": 450000, "customer": f"C{i}"} for i in range(5)] + res = threshold_edge( + rows, + {"amount_field": "refund_amount", "threshold": 500000, "key_field": "customer"}, + ) + assert res.hit + assert res.score > 0 + assert res.evidence["near_count"] == 5 + assert len(res.subjects["keys"]) == 5 + + +def test_threshold_edge_primitive_no_hit_when_spread(): + rows = [{"refund_amount": 100000}, {"refund_amount": 900000}] + res = threshold_edge(rows, {"amount_field": "refund_amount", "threshold": 500000}) + assert not res.hit + assert res.score == 0.0 + + +def test_threshold_edge_requires_positive_threshold(): + with pytest.raises(ValueError): + threshold_edge([], {"amount_field": "x", "threshold": 0}) + + +def test_cliff_drop_primitive(): + rows = [{"m": 0, "v": 1.0}, {"m": 1, "v": 0.95}, {"m": 2, "v": 0.1}] + res = cliff_drop(rows, {"index_field": "m", "value_field": "v"}) + assert res.hit + assert res.evidence["cliff_at"] == 2 + + +def test_ratio_threshold_primitive(): + rows = [{"z": 0.9}, {"z": 0.95}] + res = ratio_threshold( + rows, {"value_field": "z", "agg": "mean", "op": ">=", "threshold": 0.8} + ) + assert res.hit + assert res.score == 0.8 + + +# ---------- YAML 配置校验 ---------- + +def test_scenario_config_validation_error(): + with pytest.raises(ValidationError): + ScenarioConfig(code="R99") # 缺 title/risk_domain/label/detector + + +def test_load_yaml_scenarios_registers_r10(): + load_yaml_scenarios() # 加载内置 rules/*.yaml（含 R10 示例） + assert "R10" in SCENARIO_REGISTRY + cls = get_scenario("R10") + assert cls is not None + assert cls.label == "退款拆分" + assert cls.risk_domain == "收入" + + +# ---------- ConfigScenario 扫描 ---------- + +def test_config_scenario_scan_produces_clue_draft(): + load_yaml_scenarios() + cls = get_scenario("R10") + rows = [{"refund_amount": 460000 + i * 5000, "customer": f"政企{i}"} for i in range(6)] + outcome = cls(rows=rows).scan(None) + assert outcome.scanned_count == 6 + assert len(outcome.drafts) == 1 + draft = outcome.drafts[0] + assert draft.score > 0 + assert "退款" in draft.rationale # rationale_template 生效 + assert "keys" in draft.subjects + + +def test_config_scenario_no_draft_when_clean(): + load_yaml_scenarios() + cls = get_scenario("R10") + rows = [ + {"refund_amount": 100000, "customer": "A"}, + {"refund_amount": 2_000_000, "customer": "B"}, + ] + outcome = cls(rows=rows).scan(None) + assert outcome.drafts == [] diff --git a/backend/tests/test_scenario_registry.py b/backend/tests/test_scenario_registry.py new file mode 100644 index 0000000..245e077 --- /dev/null +++ b/backend/tests/test_scenario_registry.py @@ -0,0 +1,74 @@ +"""场景插件框架测试：注册表、场景 scan() 产出、/scenarios API（纯逻辑，无需数据库）。""" + +from __future__ import annotations + +from fastapi.testclient import TestClient + +from app.main import app +from app.scenarios.base import ScanOutcome +from app.scenarios.churn_fraud import ChurnFraudScenario, CohortPoint +from app.scenarios.registry import SCENARIO_REGISTRY, get_scenario, scenario_meta +from app.scenarios.split_contract import ContractRecord, SplitContractScenario + + +def test_registry_contains_builtin_scenarios(): + # 导入 app.scenarios.* 已通过自动发现触发 @register_scenario 注册 R8/R9 + assert "R8" in SCENARIO_REGISTRY + assert "R9" in SCENARIO_REGISTRY + assert get_scenario("R8") is SplitContractScenario + assert get_scenario("R9") is ChurnFraudScenario + + +def test_scenario_meta_fields(): + meta = {m["code"]: m for m in scenario_meta()} + assert meta["R8"]["label"] == "政企拆单" + assert meta["R8"]["risk_domain"] == "收入" + assert meta["R9"]["label"] == "养卡骗补" + assert set(meta["R8"]) == {"code", "title", "label", "risk_domain"} + + +def test_split_scenario_scan_hits(): + contracts = [ContractRecord(f"C{i}", f"CUST{i}", 850000) for i in range(8)] + scenario = SplitContractScenario( + contracts, approval_threshold=1_000_000, shared_controller=True + ) + outcome = scenario.scan(None) + assert isinstance(outcome, ScanOutcome) + assert outcome.scanned_count == 8 + assert len(outcome.drafts) == 1 + draft = outcome.drafts[0] + assert draft.score >= scenario.score_threshold + assert draft.amount_involved > 0 + assert draft.evidence["shared_controller"] is True + + +def test_split_scenario_no_draft_when_clean(): + contracts = [ContractRecord("C1", "A", 100000), ContractRecord("C2", "B", 3_000_000)] + outcome = SplitContractScenario(contracts, approval_threshold=1_000_000).scan(None) + assert outcome.drafts == [] + assert outcome.scanned_count == 2 + + +def test_churn_scenario_scan_produces_draft(): + curve = [CohortPoint(0, 1.0), CohortPoint(1, 0.95), CohortPoint(2, 0.1)] + scenario = ChurnFraudScenario( + retention_curve=curve, + commission_paid=300000, + active_ratio=0.05, + zero_usage_ratio=0.9, + channel_key="CH-001", + ) + outcome = scenario.scan(None) + assert outcome.scanned_count == 3 + assert len(outcome.drafts) == 1 + draft = outcome.drafts[0] + assert draft.subjects["channel"] == "CH-001" + assert draft.score >= scenario.score_threshold + + +def test_scenarios_api_endpoint(): + client = TestClient(app) + resp = client.get("/scenarios") + assert resp.status_code == 200 + codes = {item["code"] for item in resp.json()} + assert {"R8", "R9"} <= codes diff --git a/frontend/src/App.tsx b/frontend/src/App.tsx index 3343552..81204fc 100644 --- a/frontend/src/App.tsx +++ b/frontend/src/App.tsx @@ -1,12 +1,26 @@ -import { useState } from "react"; +import { useEffect, useState } from "react"; import Dashboard from "./Dashboard"; import Clues from "./Clues"; import NLQ from "./NLQ"; +import { api } from "./api"; +import { setScenarioLabels } from "./labels"; type Tab = "dashboard" | "clues" | "nlq"; export default function App() { const [tab, setTab] = useState("dashboard"); + const [, setLabelsReady] = useState(false); + + useEffect(() => { + // 启动时从后端拉取场景元数据，动态填充场景展示名（失败则回退到内置兜底） + api + .listScenarios() + .then((items) => { + setScenarioLabels(items); + setLabelsReady(true); + }) + .catch(() => undefined); + }, []); return (
diff --git a/frontend/src/api.ts b/frontend/src/api.ts index e7f134c..f794efc 100644 --- a/frontend/src/api.ts +++ b/frontend/src/api.ts @@ -32,6 +32,13 @@ export interface NLQResponse { egress: boolean; } +export interface ScenarioMeta { + code: string; + title: string; + label: string; + risk_domain: string; +} + async function http(url: string, init?: RequestInit): Promise { const resp = await fetch(url, { headers: { "Content-Type": "application/json" }, @@ -65,4 +72,5 @@ export const api = { }), nlq: (question: string) => http("/nlq", { method: "POST", body: JSON.stringify({ question }) }), + listScenarios: () => http("/scenarios"), }; diff --git a/frontend/src/labels.ts b/frontend/src/labels.ts index 982eb05..d4c15d7 100644 --- a/frontend/src/labels.ts +++ b/frontend/src/labels.ts @@ -22,11 +22,19 @@ export const feedbackLabel: Record = { false_positive: "误报", }; +// 场景展示名：内置少量兜底，运行时由后端 /scenarios 填充（新增场景无需改前端） export const scenarioLabel: Record = { R8: "政企拆单", R9: "养卡骗补", }; +// 用后端返回的场景元数据就地填充展示名（组件仍读同一对象引用，无需改动） +export function setScenarioLabels(items: { code: string; label: string }[]): void { + for (const it of items) { + if (it.code && it.label) scenarioLabel[it.code] = it.label; + } +} + export const providerLabel: Record = { mock: "本地Mock", vllm: "本地vLLM",