freedakgmail
80 lines
2.4 KiB
Python
80 lines
2.4 KiB
Python
"""场景检测器单元测试（纯逻辑，无需数据库）。"""
|
|
|
|
from app.scenarios.churn_fraud import (
|
|
CohortPoint,
|
|
churn_risk_score,
|
|
commission_quality_mismatch,
|
|
detect_pulse_decay,
|
|
)
|
|
from app.scenarios.split_contract import (
|
|
ContractRecord,
|
|
detect_threshold_edge,
|
|
split_risk_score,
|
|
)
|
|
|
|
# ---------- 场景一：政企拆单 (R8) ----------
|
|
|
|
def test_threshold_edge_detects_split():
|
|
# 阈值 100 万，8 份合同集中在 79万-99万
|
|
contracts = [ContractRecord(f"C{i}", f"CUST{i}", 810000 + i * 20000) for i in range(8)]
|
|
finding = detect_threshold_edge(contracts, approval_threshold=1_000_000)
|
|
assert finding.hit
|
|
assert len(finding.near_threshold) == 8
|
|
|
|
|
|
def test_threshold_edge_no_split_when_amounts_spread():
|
|
contracts = [
|
|
ContractRecord("C1", "A", 100000),
|
|
ContractRecord("C2", "B", 2_000_000),
|
|
]
|
|
finding = detect_threshold_edge(contracts, approval_threshold=1_000_000)
|
|
assert not finding.hit
|
|
|
|
|
|
def test_split_score_higher_with_shared_controller():
|
|
contracts = [ContractRecord(f"C{i}", f"CUST{i}", 850000) for i in range(8)]
|
|
finding = detect_threshold_edge(contracts, 1_000_000)
|
|
s_no = split_risk_score(finding, shared_controller=False)
|
|
s_yes = split_risk_score(finding, shared_controller=True)
|
|
assert s_yes > s_no
|
|
assert s_yes <= 1.0
|
|
|
|
|
|
def test_threshold_must_be_positive():
|
|
import pytest
|
|
|
|
with pytest.raises(ValueError):
|
|
detect_threshold_edge([], approval_threshold=0)
|
|
|
|
|
|
# ---------- 场景二：养卡骗补 (R9) ----------
|
|
|
|
def test_pulse_decay_detects_cliff():
|
|
curve = [
|
|
CohortPoint(0, 1.0),
|
|
CohortPoint(1, 0.95),
|
|
CohortPoint(2, 0.92),
|
|
CohortPoint(3, 0.10), # 第3个月断崖
|
|
]
|
|
finding = detect_pulse_decay(curve)
|
|
assert finding.pulse_then_decay
|
|
assert finding.cliff_month == 3
|
|
|
|
|
|
def test_no_cliff_for_smooth_curve():
|
|
curve = [CohortPoint(i, 1.0 - 0.05 * i) for i in range(5)]
|
|
finding = detect_pulse_decay(curve)
|
|
assert not finding.pulse_then_decay
|
|
|
|
|
|
def test_commission_mismatch_high_for_zero_usage():
|
|
m = commission_quality_mismatch(commission_paid=100000, active_ratio=0.05, zero_usage_ratio=0.9)
|
|
assert m > 0.7
|
|
|
|
|
|
def test_churn_score_combines_signals():
|
|
curve = [CohortPoint(0, 1.0), CohortPoint(1, 0.2)]
|
|
finding = detect_pulse_decay(curve)
|
|
score = churn_risk_score(finding, mismatch=0.8)
|
|
assert 0.0 < score <= 1.0
|