45 lines
2.1 KiB
Python
45 lines
2.1 KiB
Python
"""适当性校验（工作台本地桩实现，契约与投顾Agent 共享包保持一致）。
|
||
|
||
common_const §7 决议：公共适当性校验函数沉淀为独立共享包 `suitability`，工作台与
|
||
Agent 复用同一实现、禁止各自复制。在 Agent 侧共享包发布前，本模块提供**契约一致的
|
||
本地桩**：`check_suitability(customer_risk, product_risk) -> {ok, reason}`；
|
||
共享包就绪后仅替换 import，不改变调用方。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
# 风险等级 → 序号（C_n / R_n 的 n）。兼容两套口径：标准 C1-C5/R1-R5 与历史中文等级，
|
||
# 避免问卷侧尚未完成 C1-C5 归一化时发送终审误判。
|
||
# 假设：中文等级与 C 级一一对应（保守=C1/稳健=C2/平衡=C3/进取=C4/激进=C5）。
|
||
_RISK_RANK = {
|
||
"C1": 1, "C2": 2, "C3": 3, "C4": 4, "C5": 5,
|
||
"R1": 1, "R2": 2, "R3": 3, "R4": 4, "R5": 5,
|
||
"保守": 1, "稳健": 2, "平衡": 3, "进取": 4, "激进": 5,
|
||
}
|
||
|
||
|
||
def _rank(level: str | None) -> int | None:
|
||
"""风险等级统一转序号；未知/空返回 None（表示无法判定）。"""
|
||
if not level:
|
||
return None
|
||
return _RISK_RANK.get(str(level).strip())
|
||
|
||
|
||
def check_suitability(customer_risk: str | None, product_risk: str | None) -> dict:
|
||
"""适当性硬规则：产品风险 R 不得高于客户风险 C（R_n ≤ C_n）。
|
||
|
||
返回 {"ok": bool, "reason": str}。无法判定（缺等级/未知编码）视为不通过并说明原因，
|
||
宁可拦截不放行（合规红线，发送环节 fail-closed）。
|
||
"""
|
||
customer_rank = _rank(customer_risk)
|
||
product_rank = _rank(product_risk)
|
||
if customer_rank is None:
|
||
return {"ok": False, "reason": "客户无有效风险等级，无法进行适当性校验"}
|
||
if product_rank is None:
|
||
return {"ok": False, "reason": f"产品风险等级无法识别: {product_risk}"}
|
||
if product_rank > customer_rank:
|
||
return {
|
||
"ok": False,
|
||
"reason": f"产品风险等级 {product_risk} 高于客户风险等级 {customer_risk}",
|
||
}
|
||
return {"ok": True, "reason": ""}
|