XingHuo/app/service/suitability.py

"""适当性校验服务（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 typing import Any

from app.config.settings import settings
from app.repository.core_ro import CoreReadOnlyRepository
from app.repository.risk_repository import RiskRepository
from app.utils.exceptions import NotFoundError
from app.utils.trace import current_trace, new_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 = ""  # 阻断主因（人类可读，含精确规则编号）
    rule_id: str = ""  # 判定主因编号：客户维度 SUIT-00n / SUIT-008 / SUIT-PASS


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/str）；product.min_risk_code（R1~R5）。

    rule_id 口径（对齐 PRD §8 A-1/A-2 断言）：等级不匹配时取**客户判定等级**对应的
    SUIT-00n（C1 仅可购 R1→SUIT-001）；仅过期阻断时 SUIT-008；通过时 SUIT-PASS。
    """
    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)
    level_rule = f"SUIT-00{grade_number(effective_level)}"
    if is_matched:
        reasons.append(f"等级矩阵通过：{effective_level} 可购 {product_level}（{level_rule}）")
    else:
        reasons.append(
            f"等级矩阵不通过：{effective_level} 仅限其等级内产品，不可购 {product_level}（{level_rule}）"
        )

    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:
        rule_id = level_rule
        block_reason = (
            f"您的风险等级为{raw_level}（判定按{effective_level}），"
            f"该产品为{product_level}，风险不匹配（{rule_id}）"
        )
    elif expired:
        rule_id = "SUIT-008"
        block_reason = "您的风险测评已过期，无法购买新产品，请重新测评（SUIT-008）"
    else:
        rule_id = "SUIT-PASS"
        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,
        rule_id=rule_id,
    )


def suitability_check(
    customer_id: str,
    product_id: str,
    core_ro: CoreReadOnlyRepository | None = None,
    risk_repo: RiskRepository | None = None,
    valid_days: int | None = None,
    today: date | None = None,
    request_ref: str | None = None,
) -> SuitabilityResult:
    """服务入口：查 L0 事实 → 纯函数判定 → 落 risk_suitability_log（每次校验必落）。

    valid_days 缺省取 settings.risk_assessment_valid_days（SUIT-008 .env 可配）；
    request_ref 由网关传 trade_id（B5）。

    阻断时的预警单由调用方（交易网关 FR-1）生成，本函数只落校验日志。
    """
    core_ro = core_ro or CoreReadOnlyRepository()
    risk_repo = risk_repo or RiskRepository()
    valid_days = valid_days if valid_days is not None else settings.risk_assessment_valid_days

    customer = core_ro.get_customer_l0(customer_id)
    if customer is None:
        raise NotFoundError(f"customer not found: {customer_id}")
    product = core_ro.get_product(product_id)
    if product is None:
        raise NotFoundError(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() or new_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": request_ref,
            "profile_l1_version": None,
        }
    )
    return result