158 lines
5.9 KiB
Python
158 lines
5.9 KiB
Python
"""适当性校验服务（R-02 · 全系统唯一阻断点 · SUIT-001~008）。
|
||
|
||
规则权威：docs/PRD/附-风控规则表.md §1。返回语义（PRD FR-2）：
|
||
- is_matched = 纯等级矩阵结果（含 SUIT-006 封顶后判定）
|
||
- blocked = 最终是否阻断（= NOT is_matched 或 SUIT-008 测评过期）
|
||
- reasons[] 列明各规则判定；落库 customer_risk_level 记**原测评等级**
|
||
- 数据归属校验（G-01）在 FastAPI 依赖层完成，本服务不做
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
from dataclasses import dataclass, field
|
||
from datetime import date, datetime
|
||
from decimal import Decimal
|
||
from typing import Any
|
||
|
||
from app.repository.core_ro import CoreReadOnlyRepository
|
||
from app.repository.risk_repository import RiskRepository
|
||
from app.utils.trace import current_trace
|
||
|
||
TIER_ORDER = {"C1": 1, "C2": 2, "C3": 3, "C4": 4, "C5": 5}
|
||
|
||
|
||
@dataclass(frozen=True)
|
||
class SuitabilityResult:
|
||
customer_level: str # 原测评等级（落库用）
|
||
effective_level: str # SUIT-006 封顶后判定等级
|
||
product_level: str
|
||
is_matched: bool
|
||
blocked: bool
|
||
reasons: list[str] = field(default_factory=list)
|
||
block_reason: str = "" # 阻断主因（人类可读，不含文案模板）
|
||
|
||
|
||
def grade_number(grade: str) -> int:
|
||
"""C1~C5 / R1~R5 → 1~5；非法值抛错（数据污染即 Fail-fast）。"""
|
||
n = TIER_ORDER.get(grade)
|
||
if n is None:
|
||
n = TIER_ORDER.get(grade.replace("R", "C"))
|
||
if n is None:
|
||
raise ValueError(f"invalid risk grade: {grade!r}")
|
||
return n
|
||
|
||
|
||
def cap_by_age(level: str, age: int | None) -> tuple[str, str | None]:
|
||
"""SUIT-006：年龄 ≥70 按最高 C3 封顶；age IS NULL 跳过并提示人工复核。"""
|
||
if age is None:
|
||
return level, "年龄缺失，建议人工复核"
|
||
if age >= 70 and grade_number(level) > grade_number("C3"):
|
||
return "C3", f"客户测评 {level} 因年龄≥70 按 C3 处理（SUIT-006）"
|
||
return level, None
|
||
|
||
|
||
def is_assessment_valid(evaluated_at: date | datetime | str | None, today: date, valid_days: int) -> bool:
|
||
"""SUIT-008：测评有效期默认 365 天（.env 可配）。兼容 str 输入（驱动差异防御）。"""
|
||
if evaluated_at is None:
|
||
return False
|
||
if isinstance(evaluated_at, str):
|
||
evaluated_at = date.fromisoformat(evaluated_at[:10])
|
||
if isinstance(evaluated_at, datetime):
|
||
evaluated_at = evaluated_at.date()
|
||
return (today - evaluated_at).days < valid_days
|
||
|
||
|
||
def match_by_matrix(customer_level: str, product_level: str) -> bool:
|
||
"""SUIT-001~005：客户等级序号 ≥ 产品等级序号方可购买。"""
|
||
return grade_number(customer_level) >= grade_number(product_level)
|
||
|
||
|
||
def check_core(
|
||
customer: dict[str, Any],
|
||
product: dict[str, Any],
|
||
valid_days: int = 365,
|
||
today: date | None = None,
|
||
) -> SuitabilityResult:
|
||
"""纯函数核心（无 IO）：customer=core_customer+risk 合并行，product=core_product 行。
|
||
|
||
键约定：customer.risk_code（C1~C5）、customer.age（可 None）、
|
||
customer.risk_evaluated_at（date/datetime）；product.min_risk_code（R1~R5）。
|
||
"""
|
||
today = today or date.today()
|
||
raw_level: str = customer["risk_code"]
|
||
product_level: str = product["min_risk_code"]
|
||
reasons: list[str] = []
|
||
|
||
effective_level, cap_reason = cap_by_age(raw_level, customer.get("age"))
|
||
if cap_reason:
|
||
reasons.append(cap_reason)
|
||
|
||
is_matched = match_by_matrix(effective_level, product_level)
|
||
if is_matched:
|
||
reasons.append(f"等级矩阵通过：{effective_level} 可购 {product_level}（SUIT-001~005）")
|
||
else:
|
||
reasons.append(f"等级矩阵不通过：{effective_level} 不可购 {product_level}（SUIT-001~005）")
|
||
|
||
expired = not is_assessment_valid(customer.get("risk_evaluated_at"), today, valid_days)
|
||
if expired:
|
||
reasons.append(f"风险测评已过期（有效期 {valid_days} 天），请重新测评（SUIT-008）")
|
||
|
||
blocked = (not is_matched) or expired
|
||
if not is_matched:
|
||
block_reason = f"您的风险等级为{raw_level}（判定按{effective_level}），该产品为{product_level}，风险不匹配"
|
||
elif expired:
|
||
block_reason = "您的风险测评已过期，无法购买新产品，请重新测评"
|
||
else:
|
||
block_reason = ""
|
||
|
||
return SuitabilityResult(
|
||
customer_level=raw_level,
|
||
effective_level=effective_level,
|
||
product_level=product_level,
|
||
is_matched=is_matched,
|
||
blocked=blocked,
|
||
reasons=reasons,
|
||
block_reason=block_reason,
|
||
)
|
||
|
||
|
||
def suitability_check(
|
||
customer_id: str,
|
||
product_id: str,
|
||
core_ro: CoreReadOnlyRepository | None = None,
|
||
risk_repo: RiskRepository | None = None,
|
||
valid_days: int = 365,
|
||
today: date | None = None,
|
||
) -> SuitabilityResult:
|
||
"""服务入口：查 L0 事实 → 纯函数判定 → 落 risk_suitability_log（每次校验必落）。
|
||
|
||
阻断时的预警单由调用方（交易网关 FR-1）生成，本函数只落校验日志。
|
||
"""
|
||
core_ro = core_ro or CoreReadOnlyRepository()
|
||
risk_repo = risk_repo or RiskRepository()
|
||
|
||
customer = core_ro.get_customer_l0(customer_id)
|
||
if customer is None:
|
||
raise LookupError(f"customer not found: {customer_id}")
|
||
product = core_ro.get_product(product_id)
|
||
if product is None:
|
||
raise LookupError(f"product not found: {product_id}")
|
||
|
||
result = check_core(customer, product, valid_days=valid_days, today=today)
|
||
|
||
risk_repo.insert_suitability_log(
|
||
{
|
||
"trace_id": current_trace(),
|
||
"customer_id": customer_id,
|
||
"product_id": product_id,
|
||
"customer_risk_level": result.customer_level,
|
||
"product_risk_level": result.product_level,
|
||
"is_matched": int(result.is_matched),
|
||
"is_blocked": int(result.blocked),
|
||
"block_reason": result.block_reason or None,
|
||
"request_ref": None,
|
||
"profile_l1_version": None,
|
||
}
|
||
)
|
||
return result
|