Files
group_fqcd_jr/app/service/suitability_service.py
张胜宇 35effab70f feat(W20): 新增 E2c-my 出口 + 展示层净化 + P2 自助流程豁免(金标零回归)
用户口径:既智能又安全,不靠"答不上就转人工";按本人画像测评列出可购买的产品,
但不引导购买。

出口由五个扩为六个:
- 新增 E2c-my:按客户本人权威测评等级给出可购买范围 + 在售清单
  (只列代码/名称/类别/风险等级,按代码升序;出口后置引导词护栏,命中即降级为仅范围)
- 点名档位(「我可以买 R3 的产品吗」)先给一句直接裁决再列范围
  —— 修复前只给清单,"不"字始终没说出来
- 新工具 query_eligible_products(suitability:read)+ 发布 release 220

真实业务缺陷:
- P2_SELF_SERVICE_PATTERNS:问「怎么修改绑定的银行卡」不再建单转人工(FAQ 里就有答案);
  「帮我把绑定银行卡换一下」仍走 P2
- 展示层净化 render_plain / drop_yield_claims / prettify_title 接在 E3/E4/E5b/澄清四处
  (E4 模型会照抄证据包的 markdown 标记)
- 空结果护栏:语料里 15 个纯收益切片被净化清空时回退 E5b,不给空气泡

测试:定向 334 passed;全量回归 1969 passed / 3 skipped / 0 failed(基线 1946);
46 条金标 M-1 46/46=100%、M-4 100%、M-6 5/46=10.9%、M-7~M-10 全 0 —— 与
score_w11b 逐项一致(零回归)。

落档:D1.6 §4.49 / D2.1 v6.36 / D1.1 §30
2026-09-20 18:55:22 +08:00

474 lines
20 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""公共投顾适当性校验服务。
适当性是治理边界,不属于任何一个业务 Agent。该服务只读客户/产品风险信息,
返回不可变决定;拒绝决定必须留下审计记录,且不会修改交易或产品数据。
合规要点(B2 修复):
1. **客户风险等级只能来自服务端权威来源**:取 ``fin_risk_assessment`` 中该客户最新一条
测评(``assessed_at`` 最新),调用方传入的 ``customer_risk_level`` 一律不接受
(DTO ``extra="forbid"`` 直接拒绝伪造字段)。
2. **测评有效期以权威记录的 ``valid_until`` 为准**,调用方不能自报过期时间;测评缺失或
过期一律失败关闭,不做静默降级。
3. **专业投资者身份来自 ``sys_user``**(``is_professional_investor`` 且
``professional_investor_status='已认定'``),而非调用方参数;已认定的专业投资者可豁免
C/R 等级匹配,但仍强制风险揭示、确认与录音,且不能绕过测评有效期与审计。
"""
import re
from collections.abc import Callable, Mapping
from datetime import UTC, datetime
from typing import Any, Literal
from pydantic import BaseModel, ConfigDict, Field
from sqlalchemy import text
from app.core.contracts import RequestContext
from app.core.errors import ForbiddenAgentError
from app.infrastructure.db import SessionFactory
from app.model.audit import InteractionAudit
PROFESSIONAL_INVESTOR_CERTIFIED = "已认定"
RISK_LEVEL_SOURCE = "fin_risk_assessment"
PROFESSIONAL_INVESTOR_SOURCE = "sys_user"
CUSTOMER_SCOPE_EXEMPT_ROLES = frozenset({"admin", "super_admin"})
# 《个人投资者适当性管理指南》第十二条匹配矩阵(客户等级 → 可购买的产品等级)。
#
# 矩阵与第十四条的**第 2、3 款**一致:只禁止"低两个等级及以上"的越级,低一个等级要看
# 档位(C1→R2、C2→R3 直接可买;C3→R4、C4→R5 需签风险揭示书)。真正冲突的是第十四条
# **第 1 款**"必须大于或等于"—— 它与同一条第 2、3 款自相矛盾。
#
# 2026-09-11 业务裁定:**客服回答按矩阵,风控扫描保留 C ≥ R**。
# 理由是两者的职责不同:客服要给客户一个与知识库(`POL-AST-012`)一致的"能不能买",
# 矩阵才是客户看得见的口径;风控要发现的是"越级成交且留痕不全",用更严的 C ≥ R 去
# 事后核查。改之前两边是**同一个 Agent 自相矛盾**:问"C1 能买什么产品"答"R1、R2 可买"
# (矩阵),问"C1 能买这只 R2 吗"却答"不能购买"(C ≥ R)。
MATRIX_ALLOWED: dict[int, frozenset[int]] = {
1: frozenset({1, 2}),
2: frozenset({1, 2, 3}),
3: frozenset({1, 2, 3, 4}),
4: frozenset({1, 2, 3, 4, 5}),
5: frozenset({1, 2, 3, 4, 5}),
}
# 矩阵里标"⚠️ 需签署风险揭示书"的档位,即第十五条豁免档。
MATRIX_NEEDS_DISCLOSURE: dict[int, frozenset[int]] = {
3: frozenset({4}),
4: frozenset({5}),
}
_INVESTOR_TYPE_PATTERN = re.compile(r"^C([1-5])$")
AuthorityReason = Literal[
"AUTHORITY_OK",
"CUSTOMER_NOT_FOUND",
"ASSESSMENT_MISSING",
"ASSESSMENT_EXPIRED",
"RISK_LEVEL_INVALID",
]
# 只读查询:客户专业投资者身份(sys_user)+ 最新一条风险测评(fin_risk_assessment)。
# 不读取 answers 问卷原文,避免把敏感测评明细带入服务层。
_AUTHORITY_SQL = text(
"""
SELECT u.is_professional_investor,
u.professional_investor_status,
a.investor_type,
a.assessed_at,
a.valid_until
FROM sys_user u
LEFT JOIN fin_risk_assessment a ON a.id = (
SELECT x.id FROM fin_risk_assessment x
WHERE x.customer_id = u.id
ORDER BY x.assessed_at DESC, x.id DESC
LIMIT 1
)
WHERE u.id = :customer_id
"""
)
class RiskAuthorityProfile(BaseModel):
"""服务端权威风险画像(只读汇总,不含测评问卷原文)。"""
model_config = ConfigDict(extra="forbid", frozen=True)
customer_id: str
customer_risk_level: int | None = None
professional_investor: bool = False
assessed_at: datetime | None = None
valid_until: datetime | None = None
authority_reason: AuthorityReason = "AUTHORITY_OK"
class SuitabilityToolInput(BaseModel):
"""ToolExecutor 使用的严格输入模型,避免业务 Agent 自行拼接规则。
调用方只能声明“给谁、买什么等级的产品、是否需要揭示/确认”,
风险等级与测评有效期一律由服务端权威来源解析。
"""
model_config = ConfigDict(extra="forbid", frozen=True)
customer_id: str = Field(min_length=1, max_length=20, pattern=r"^[0-9]+$")
product_risk_level: int = Field(ge=1, le=5)
product_requires_disclosure: bool = True
requires_confirmation: bool = False
class SuitabilityDecision(BaseModel):
model_config = ConfigDict(extra="forbid", frozen=True)
allowed: bool
reason_code: str
required_disclosure: bool
requires_confirmation: bool
requires_recording: bool
customer_risk_level: int | None = None
risk_level_source: str = RISK_LEVEL_SOURCE
professional_investor: bool = False
assessment_valid_until: datetime | None = None
def _as_utc(value: object) -> datetime | None:
"""库内 DATETIME 为 UTC naive,统一规范化为带时区,便于安全比较。"""
if not isinstance(value, datetime):
return None
return value if value.tzinfo is not None else value.replace(tzinfo=UTC)
def _risk_level_from_investor_type(investor_type: object) -> int | None:
if not isinstance(investor_type, str):
return None
matched = _INVESTOR_TYPE_PATTERN.match(investor_type.strip().upper())
return int(matched.group(1)) if matched is not None else None
class SuitabilityService:
"""执行 C1-C5/R1-R5 的公共、只读适当性规则。"""
def __init__(self, *, session_factory: Callable[[], Any] | None = None) -> None:
self._session_factory: Callable[[], Any] = session_factory or SessionFactory
async def evaluate(
self, request: SuitabilityToolInput, context: RequestContext, *, now: datetime | None = None
) -> SuitabilityDecision:
current = now or datetime.now(UTC)
self._assert_customer_scope(request.customer_id, context)
profile = await self._load_authority_profile(request.customer_id)
return self._decide(request, profile, current)
async def authority_for_customer(self, customer_id: int) -> RiskAuthorityProfile:
"""Expose the read-only risk authority for other advisory services."""
return await self._load_authority_profile(str(customer_id))
async def _load_authority_profile(self, customer_id: str) -> RiskAuthorityProfile:
async with self._session_factory() as session:
result = await session.execute(_AUTHORITY_SQL, {"customer_id": int(customer_id)})
row: Mapping[str, Any] | None = result.mappings().first()
if row is None:
return RiskAuthorityProfile(
customer_id=customer_id, authority_reason="CUSTOMER_NOT_FOUND"
)
level = _risk_level_from_investor_type(row["investor_type"])
valid_until = _as_utc(row["valid_until"])
if row["investor_type"] is None:
reason: AuthorityReason = "ASSESSMENT_MISSING"
elif level is None:
reason = "RISK_LEVEL_INVALID"
elif valid_until is None:
# 有效期缺失视为不可用,绝不按“长期有效”放行。
reason = "ASSESSMENT_EXPIRED"
else:
reason = "AUTHORITY_OK"
return RiskAuthorityProfile(
customer_id=customer_id,
customer_risk_level=level,
professional_investor=(
bool(row["is_professional_investor"])
and str(row["professional_investor_status"]) == PROFESSIONAL_INVESTOR_CERTIFIED
),
assessed_at=_as_utc(row["assessed_at"]),
valid_until=valid_until,
authority_reason=reason,
)
def _decide(
self, request: SuitabilityToolInput, profile: RiskAuthorityProfile, current: datetime
) -> SuitabilityDecision:
if profile.authority_reason == "CUSTOMER_NOT_FOUND":
return self._denied("CUSTOMER_NOT_FOUND", request, profile)
if profile.authority_reason == "ASSESSMENT_MISSING":
return self._denied("ASSESSMENT_MISSING", request, profile)
if profile.authority_reason == "RISK_LEVEL_INVALID":
return self._denied("RISK_LEVEL_INVALID", request, profile)
if profile.valid_until is None or profile.valid_until <= current:
# 测评过期即拒绝:专业投资者也不能绕过有效期。
return self._denied("ASSESSMENT_EXPIRED", request, profile)
if profile.customer_risk_level is None:
return self._denied("ASSESSMENT_MISSING", request, profile)
if profile.professional_investor:
# 已认定专业投资者可豁免等级匹配,但必须揭示、确认并录音留痕。
return SuitabilityDecision(
allowed=True,
reason_code="SUITABLE_PROFESSIONAL_INVESTOR",
required_disclosure=True,
requires_confirmation=True,
requires_recording=True,
customer_risk_level=profile.customer_risk_level,
professional_investor=True,
assessment_valid_until=profile.valid_until,
)
# 按第十二条匹配矩阵裁决(见 MATRIX_ALLOWED 的说明):不再用"C < R 即拒绝",
# 那样会把矩阵允许的 C1→R2、C2→R3 以及豁免档 C3→R4、C4→R5 一起拒掉。
level = profile.customer_risk_level
product_level = request.product_risk_level
if product_level not in MATRIX_ALLOWED.get(level, frozenset()):
return self._denied("RISK_LEVEL_MISMATCH", request, profile)
needs_disclosure = product_level in MATRIX_NEEDS_DISCLOSURE.get(level, frozenset())
required_disclosure = request.product_requires_disclosure or needs_disclosure
return SuitabilityDecision(
allowed=True,
reason_code="SUITABLE_WITH_DISCLOSURE" if needs_disclosure else "SUITABLE",
required_disclosure=required_disclosure,
requires_confirmation=request.requires_confirmation or required_disclosure,
requires_recording=required_disclosure or request.requires_confirmation,
customer_risk_level=level,
professional_investor=False,
assessment_valid_until=profile.valid_until,
)
async def evaluate_and_audit(
self, request: SuitabilityToolInput, context: RequestContext, *, now: datetime | None = None
) -> SuitabilityDecision:
decision = await self.evaluate(request, context, now=now)
# 适当性决定是受监管业务决策,拒绝和通过都留痕;只记录权威来源摘要,
# 不保存测评问卷原文,也不保存调用方自报的任何等级。
async with self._session_factory() as session, session.begin():
actor_id = int(context.user_id) if context.user_id.isdecimal() else None
session.add(InteractionAudit(
actor_type="agent",
actor_id=actor_id,
portal=context.portal,
action_type="suitability.checked",
detail={
"trace_id": context.trace_id,
"status": "allowed" if decision.allowed else "denied",
"reason_code": decision.reason_code,
"customer_id": request.customer_id,
"customer_risk_level": decision.customer_risk_level,
"risk_level_source": decision.risk_level_source,
"professional_investor": decision.professional_investor,
"professional_investor_source": PROFESSIONAL_INVESTOR_SOURCE,
"assessment_valid_until": (
decision.assessment_valid_until.isoformat()
if decision.assessment_valid_until is not None
else None
),
"product_risk_level": request.product_risk_level,
},
created_at=datetime.now(UTC).replace(tzinfo=None),
))
return decision
@staticmethod
def _assert_customer_scope(customer_id: str, context: RequestContext) -> None:
"""公共鉴权:除管理员外不得查询他人风险测评。"""
if set(context.roles).intersection(CUSTOMER_SCOPE_EXEMPT_ROLES):
return
if customer_id == context.user_id or customer_id in context.customer_ids:
return
raise ForbiddenAgentError("不能查询该客户的风险测评")
@staticmethod
def _denied(
reason_code: str, request: SuitabilityToolInput, profile: RiskAuthorityProfile
) -> SuitabilityDecision:
return SuitabilityDecision(
allowed=False,
reason_code=reason_code,
required_disclosure=request.product_requires_disclosure,
requires_confirmation=True,
requires_recording=True,
customer_risk_level=profile.customer_risk_level,
professional_investor=profile.professional_investor,
assessment_valid_until=profile.valid_until,
)
# ---------------------------------------------------------------------------
# `E2c-my`:按客户**权威**等级给出可购买产品清单(只读)
# ---------------------------------------------------------------------------
#: 产品上架状态字面。与 `trade_service` / 门户产品页取的是同一口径(`fin_product.status`)。
PRODUCT_STATUS_LISTED = "上市"
class EligibleProductQuery(BaseModel):
"""工具入参:只能声明「查谁的」—— 等级、清单与排序一律由服务端决定。"""
model_config = ConfigDict(extra="forbid", frozen=True)
customer_id: str = Field(min_length=1, max_length=20, pattern=r"^[0-9]+$")
class EligibleProduct(BaseModel):
"""清单里的一只产品。
**字段刻意只有四个**:代码 / 名称 / 类别 / 风险等级。净值、收益率、涨跌幅、费率
一律不进这个模型 —— 一旦进了,这份清单就从「按等级筛选的公开产品表」变成
「推介材料」,而客服出口只能陈述、不能引导。
"""
model_config = ConfigDict(extra="forbid", frozen=True)
product_code: str
product_name: str
product_category: str
risk_level: str
class EligibleProductView(BaseModel):
model_config = ConfigDict(extra="forbid", frozen=True)
customer_id: str
customer_risk_level: int | None = None
risk_level_source: str = RISK_LEVEL_SOURCE
authority_reason: AuthorityReason = "AUTHORITY_OK"
assessment_valid_until: datetime | None = None
allowed_levels: tuple[str, ...] = ()
disclosure_levels: tuple[str, ...] = ()
products: tuple[EligibleProduct, ...] = ()
excluded_count: int = 0
_ELIGIBLE_SQL = text(
"""
SELECT product_code, product_name, product_category, risk_level
FROM fin_product
WHERE status = :status
ORDER BY product_code
"""
)
class EligibleProductsService:
"""只读:按客户**权威**等级给出「可购买产品」清单。
四条硬约束(缺任何一条都失败关闭,返回空清单而不是猜一个范围):
1. 客户等级只认 ``fin_risk_assessment``(与 ``check_suitability`` **同一个**权威口径),
**不采信调用方**,也不采信客户在问句里自报的等级;
2. 清单只取 ``fin_product`` 中 ``status='上市'`` 的产品,字段只有代码 / 名称 / 类别 /
风险等级 —— **不含净值、收益率、涨跌幅、费率**;
3. 顺序固定**按产品代码升序**,不按收益 / 规模 / 热度排 —— 排序就是隐性推荐;
4. 匹配范围取 ``MATRIX_ALLOWED``(第十二条矩阵),需签风险揭示书的档位单独标出
(``MATRIX_NEEDS_DISCLOSURE``),把「能买」与「需签字后才能买」讲清楚。
"""
def __init__(self, *, session_factory: Callable[[], Any] | None = None) -> None:
self._session_factory: Callable[[], Any] = session_factory or SessionFactory
async def load(
self, request: EligibleProductQuery, context: RequestContext
) -> EligibleProductView:
SuitabilityService._assert_customer_scope(request.customer_id, context)
authority = await self._authority(request.customer_id)
level = authority.customer_risk_level
if authority.authority_reason != "AUTHORITY_OK" or level is None:
# 失败关闭:等级拿不到就**不列清单**,也不退化成「全都列一遍」。
await self._audit(request, context, authority, ())
return EligibleProductView(
customer_id=request.customer_id,
authority_reason=authority.authority_reason,
assessment_valid_until=authority.valid_until,
)
allowed = MATRIX_ALLOWED.get(level, frozenset())
disclosure = MATRIX_NEEDS_DISCLOSURE.get(level, frozenset())
listed = await self._listed_products()
picked = tuple(item for item in listed if _level_number(item.risk_level) in allowed)
excluded = len(listed) - len(picked)
await self._audit(request, context, authority, picked)
return EligibleProductView(
customer_id=request.customer_id,
customer_risk_level=level,
authority_reason=authority.authority_reason,
assessment_valid_until=authority.valid_until,
allowed_levels=tuple(f"R{n}" for n in sorted(allowed)),
disclosure_levels=tuple(f"R{n}" for n in sorted(disclosure)),
products=picked,
excluded_count=excluded,
)
async def _authority(self, customer_id: str) -> RiskAuthorityProfile:
return await SuitabilityService(
session_factory=self._session_factory
).authority_for_customer(int(customer_id))
async def _listed_products(self) -> tuple[EligibleProduct, ...]:
async with self._session_factory() as session:
result = await session.execute(
_ELIGIBLE_SQL, {"status": PRODUCT_STATUS_LISTED}
)
rows = result.mappings().all()
return tuple(
EligibleProduct(
product_code=str(row["product_code"]),
product_name=str(row["product_name"]),
product_category=str(row["product_category"]),
risk_level=str(row["risk_level"]),
)
for row in rows
)
async def _audit(
self,
request: EligibleProductQuery,
context: RequestContext,
authority: RiskAuthorityProfile,
picked: tuple[EligibleProduct, ...],
) -> None:
"""留痕:谁在什么时候按哪个等级取走了一份可买清单(只记摘要,不记产品明细)。"""
async with self._session_factory() as session, session.begin():
actor_id = int(context.user_id) if context.user_id.isdecimal() else None
session.add(InteractionAudit(
actor_type="agent",
actor_id=actor_id,
portal=context.portal,
action_type="suitability.eligible_products",
detail={
"trace_id": context.trace_id,
"customer_id": request.customer_id,
"customer_risk_level": authority.customer_risk_level,
"risk_level_source": RISK_LEVEL_SOURCE,
"authority_reason": authority.authority_reason,
"listed_count": len(picked),
},
created_at=datetime.now(UTC).replace(tzinfo=None),
))
def _level_number(label: str) -> int:
"""``"R3"`` → ``3``;无法解析时返回 ``0``(任何 ``allowed`` 集合都不含 ``0`` ⇒ 不入选)。"""
matched = re.fullmatch(r"R([1-5])", str(label).strip())
return int(matched.group(1)) if matched else 0
async def query_eligible_products_tool(
arguments: EligibleProductQuery, context: RequestContext
) -> dict[str, Any]:
"""ToolExecutor 入口:返回 JSON 可序列化的可买清单视图。"""
view = await EligibleProductsService().load(arguments, context)
return view.model_dump(mode="json")
async def suitability_tool_handler(
arguments: SuitabilityToolInput, context: RequestContext
) -> dict[str, Any]:
decision = await SuitabilityService().evaluate_and_audit(arguments, context)
return decision.model_dump(mode="json")