selfrelease
74e35fed96
- 新增场景基类、配置加载、注册表与原语模块 - 添加 r10_refund_split 规则及场景 JSON Schema - 扩展 scan 引擎与 scenarios API - 新增场景注册表/配置/集成测试 - 更新前端 App、api、labels 支持新场景
63 lines
2.3 KiB
Python
63 lines
2.3 KiB
Python
"""通用场景调度器 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 == []
|