客服适当性改按第十二条匹配矩阵（更正我上一轮的复核结论）
你说得对：C ≥ R 是风控的口径，客服要走政策原文的矩阵。 上一轮我复核后说"两边一致、无需改动"是错的 —— 我只对比了政策文本与 SuitabilityService，漏了第三个东西：客服自己的知识库。POL-AST-012 就是第十二条矩阵 原文，而客服为 C1-C5 补的「能买什么产品」问答答案直接取自该矩阵（docs/24 第二节）。 于是同一个客服 Agent 对同一个问题给出相反答案： 问"C1 能买什么产品" → 知识检索答 R1、R2 可买（矩阵口径） 问"C1 能买这只 R2 吗" → 适当性出口答不能购买（严格 C ≥ R） 政策冲突的精确位置也不是"矩阵 vs 硬匹配"，而是第十四条**第 1 款与它自己的第 2、3 款** 矛盾：第 2 款说"低于一个等级以上"才拒绝（C1→R2 只低 1 级，够不上"以上"），第 3 款禁止 C1 买 R3+、C2 买 R4+（R2 不在禁止列表）。矩阵与第 2、3 款三方一致，孤立的是第 1 款。 改动： - suitability_service.py 新增 MATRIX_ALLOWED / MATRIX_NEEDS_DISCLOSURE，_decide 由 "C < R 即拒绝"改为按矩阵：C1→R2、C2→R3 直接可买；C3→R4、C4→R5 走第十五条豁免档 （reason_code=SUITABLE_WITH_DISCLOSURE，强制揭示 + 确认 + 录音）；低两级及以上仍拒绝。 - 客服话术分档：越级档不再说"在您的风险承受能力范围内" —— 那句只对 C ≥ R 成立， 用在 C1 买 R2 上会让客户以为自己的测评本来就覆盖这只产品。 - 风控侧不动，保留 C ≥ R：它要发现的是"越级成交且留痕不全"，这个差异是有意保留的。 测试：新增逐格对照政策原文的 25 格矩阵用例、豁免档用例、话术分档用例（+29）。 docs/25 第七节 #1 与 docs/24 第七节同步更正，包括写明我上一轮那个结论错在哪。 门禁：ruff 干净 / mypy 137 文件 / 693 unit+contract / 33 integration。
This commit is contained in:
@@ -63,6 +63,45 @@ def test_disclosure_and_recording_are_disclosed_when_required() -> None:
|
||||
assert "双录" in text
|
||||
|
||||
|
||||
def test_one_level_up_is_not_described_as_within_capacity() -> None:
|
||||
"""C1 买 R2 是矩阵允许的，但不能说成"在您的风险承受能力范围内"——它超出了等级。
|
||||
|
||||
这句错话会让客户以为自己的测评等级本来就覆盖 R2，下次买 R3 时就更难解释为什么不行。
|
||||
"""
|
||||
text = CustomerServiceAgent._suitability_text(
|
||||
"南方季季盈90天", 2, _decision(customer_risk_level=1)
|
||||
)
|
||||
|
||||
assert "在您的风险承受能力范围内" not in text
|
||||
assert "匹配矩阵允许购买的范围内" in text
|
||||
|
||||
|
||||
def test_disclosure_tier_explains_the_condition() -> None:
|
||||
"""C3 买 R4 属豁免档：要说清"高于您的等级"以及"签揭示书后可买"这个条件。"""
|
||||
text = CustomerServiceAgent._suitability_text(
|
||||
"南方稳健增利180天",
|
||||
4,
|
||||
_decision(
|
||||
customer_risk_level=3,
|
||||
reason_code="SUITABLE_WITH_DISCLOSURE",
|
||||
required_disclosure=True,
|
||||
requires_recording=True,
|
||||
),
|
||||
)
|
||||
|
||||
assert "高于您的风险承受能力等级" in text
|
||||
assert "签署产品风险揭示书后可以购买" in text
|
||||
assert "在您的风险承受能力范围内" not in text
|
||||
|
||||
|
||||
def test_same_level_still_uses_the_capacity_wording() -> None:
|
||||
text = CustomerServiceAgent._suitability_text(
|
||||
"南方季季盈90天", 2, _decision(customer_risk_level=2)
|
||||
)
|
||||
|
||||
assert "在您的风险承受能力范围内" in text
|
||||
|
||||
|
||||
def test_rejection_does_not_assert_a_reason() -> None:
|
||||
"""拒绝的原因可能是等级不匹配、测评过期或未测评，不能一律说成"超出承受能力"。"""
|
||||
text = CustomerServiceAgent._suitability_text(
|
||||
|
||||
@@ -101,14 +101,57 @@ def test_caller_cannot_declare_risk_facts(forged: dict[str, Any]) -> None:
|
||||
|
||||
|
||||
async def test_insufficient_authority_level_is_denied() -> None:
|
||||
"""低两个等级及以上仍必须拒绝（第十四条第 2、3 款，矩阵里的"❌ 禁止"）。"""
|
||||
decision = await service_with_row(authority_row(investor_type="C1")).evaluate(
|
||||
query(product_risk_level=2), context(), now=NOW
|
||||
query(product_risk_level=3), context(), now=NOW
|
||||
)
|
||||
assert decision.allowed is False
|
||||
assert decision.reason_code == "RISK_LEVEL_MISMATCH"
|
||||
assert decision.requires_recording is True
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("investor_type", "product_level", "allowed", "reason_code"),
|
||||
[
|
||||
# 逐格抄自 knowledge/policy/个人投资者适当性管理指南.md 第十二条矩阵。
|
||||
# 这张表的价值在于：任何一格被改动，都必须是有意为之并在此处说明理由。
|
||||
("C1", 1, True, "SUITABLE"), ("C1", 2, True, "SUITABLE"),
|
||||
("C1", 3, False, "RISK_LEVEL_MISMATCH"), ("C1", 4, False, "RISK_LEVEL_MISMATCH"),
|
||||
("C1", 5, False, "RISK_LEVEL_MISMATCH"),
|
||||
("C2", 1, True, "SUITABLE"), ("C2", 2, True, "SUITABLE"), ("C2", 3, True, "SUITABLE"),
|
||||
("C2", 4, False, "RISK_LEVEL_MISMATCH"), ("C2", 5, False, "RISK_LEVEL_MISMATCH"),
|
||||
("C3", 1, True, "SUITABLE"), ("C3", 2, True, "SUITABLE"), ("C3", 3, True, "SUITABLE"),
|
||||
("C3", 4, True, "SUITABLE_WITH_DISCLOSURE"), ("C3", 5, False, "RISK_LEVEL_MISMATCH"),
|
||||
("C4", 1, True, "SUITABLE"), ("C4", 2, True, "SUITABLE"), ("C4", 3, True, "SUITABLE"),
|
||||
("C4", 4, True, "SUITABLE"), ("C4", 5, True, "SUITABLE_WITH_DISCLOSURE"),
|
||||
("C5", 1, True, "SUITABLE"), ("C5", 2, True, "SUITABLE"), ("C5", 3, True, "SUITABLE"),
|
||||
("C5", 4, True, "SUITABLE"), ("C5", 5, True, "SUITABLE"),
|
||||
],
|
||||
)
|
||||
async def test_full_matrix_matches_policy_document(
|
||||
investor_type: str, product_level: int, allowed: bool, reason_code: str
|
||||
) -> None:
|
||||
"""客服回答必须与知识库里的矩阵一致 —— 这是同一个 Agent 的两条出口。"""
|
||||
decision = await service_with_row(authority_row(investor_type=investor_type)).evaluate(
|
||||
query(product_risk_level=product_level), context(), now=NOW
|
||||
)
|
||||
assert decision.allowed is allowed
|
||||
assert decision.reason_code == reason_code
|
||||
|
||||
|
||||
async def test_disclosure_tier_always_requires_disclosure_and_recording() -> None:
|
||||
"""C3→R4、C4→R5 是第十五条豁免档：可买，但必须揭示、确认、录音。"""
|
||||
for investor_type, product_level in (("C3", 4), ("C4", 5)):
|
||||
decision = await service_with_row(authority_row(investor_type=investor_type)).evaluate(
|
||||
query(product_risk_level=product_level), context(), now=NOW
|
||||
)
|
||||
assert decision.allowed is True
|
||||
assert decision.reason_code == "SUITABLE_WITH_DISCLOSURE"
|
||||
assert decision.required_disclosure is True
|
||||
assert decision.requires_confirmation is True
|
||||
assert decision.requires_recording is True
|
||||
|
||||
|
||||
async def test_expired_assessment_is_denied_even_for_eligible_level() -> None:
|
||||
row = authority_row(investor_type="C5", valid_until=NOW - timedelta(seconds=1))
|
||||
decision = await service_with_row(row).evaluate(query(product_risk_level=1), context(), now=NOW)
|
||||
|
||||
Reference in New Issue
Block a user