diff --git a/app/repository/core_ro.py b/app/repository/core_ro.py index b40aad4..bf95e4c 100644 --- a/app/repository/core_ro.py +++ b/app/repository/core_ro.py @@ -1,10 +1,22 @@ -"""Core 模拟库只读访问（jinrong_core · 无 HTTP API）。""" +"""Core 模拟库只读访问（jinrong_core · 无 HTTP API）。 + +AL-03 对齐 main 基准（2026-09-07）： +- 骨架为我方版本（utils/db.get_engine 统一引擎工厂——main 自带 create_engine 工厂不采纳， + 避免破坏统一引擎管理；get_trade_by_id / sum_trades_on_date / list_trades_range / + list_active_customers 等风控扩展方法保留）。 +- 自 main 吸收：get_customer_l0 扩列版（c.* + 风评新列 + risk_is_expired）、 + check_suitability + _suitability_result（R-02 矩阵判定，C×R 数据驱动）、 + list_products_for_customer（C-11）、list_trades、list_holdings 增 + min_subscribe_amount/term_days 列（limit=500 截断防护为我方 T-04 评审保留）。 +- 移植坑消解：main SQL 中 `(r.expires_at < CURDATE())` 为 MySQL 专属函数， + sqlite 测试库会炸——统一改为取回 expires_at 后 Python 端计算（_is_expired）。 +""" from __future__ import annotations from datetime import date, datetime, time, timedelta from decimal import Decimal -from typing import Any +from typing import Any, Literal from sqlalchemy import text from sqlalchemy.engine import Engine @@ -12,6 +24,29 @@ from sqlalchemy.engine import Engine from app.config.settings import settings from app.utils.db import get_engine +MatchResult = Literal[ + "allowed", "allowed_with_disclosure", "forbidden", "professional_exempt", "risk_expired" +] + + +def _as_date(value: Any) -> date | None: + """DB 取回的日期值统一转 date（MySQL DATE→date / sqlite TIMESTAMP→datetime / str 兜底）。""" + if value is None: + return None + if isinstance(value, datetime): + return value.date() + if isinstance(value, date): + return value + return date.fromisoformat(str(value)[:10]) + + +def _is_expired(value: Any, today: date | None = None) -> bool: + """FM-03 风评过期判定（Python 端，替代 main 的 CURDATE() SQL 表达式）。""" + d = _as_date(value) + if d is None: + return False + return d < (today or date.today()) + class CoreReadOnlyRepository: """仅 SELECT jinrong_core；禁止写操作。""" @@ -20,10 +55,14 @@ class CoreReadOnlyRepository: self._engine = engine or get_engine(settings.mysql_core_database) def get_customer_l0(self, customer_id: str) -> dict[str, Any] | None: + """L0 客户主档 + 正式风评（对齐用户信息数据示例 / 适当性指南；main 扩列版）。""" sql = text( """ - SELECT c.customer_id, c.display_name, c.age, c.occupation, c.open_date, - r.risk_code, r.evaluated_at AS risk_evaluated_at + SELECT c.*, + r.risk_code, r.questionnaire_score, r.max_loss_tolerance_pct, + r.investment_goal, r.investment_horizon, r.investor_category, + r.professional_approved_at, + r.evaluated_at AS risk_evaluated_at, r.expires_at AS risk_expires_at FROM core_customer c LEFT JOIN core_customer_risk r ON r.customer_id = c.customer_id WHERE c.customer_id = :cid AND c.is_active = 1 @@ -31,17 +70,220 @@ class CoreReadOnlyRepository: ) with self._engine.connect() as conn: row = conn.execute(sql, {"cid": customer_id}).mappings().first() - return dict(row) if row else None + if not row: + return None + data = dict(row) + # FM-03 过期判定 Python 端算（替代 main 的 CURDATE()，sqlite 兼容） + data["risk_is_expired"] = _is_expired(data.get("risk_expires_at")) + return data + + def check_suitability(self, customer_id: str, product_id: str) -> dict[str, Any]: + """R-02：基于 L0 C 等级 + 产品 R 等级 + 适当性矩阵判定（main 契约）。 + + 判定顺序：not_found → risk_expired(FM-03) → professional_exempt(JR-AST-PRO) + → 矩阵 forbidden(JR-AST-012) → 披露 → 年龄≥70 买 R3+ 网点确认(FM-01)。 + 返回字段与 risk_suitability_log 一一对应， + 见 docs/项目框架设计/表设计/07-risk_suitability_log说明.md。 + """ + sql = text( + """ + SELECT c.customer_id, c.age, c.is_hnw, + r.risk_code AS customer_risk_code, + r.investor_category, r.expires_at AS risk_expires_at, + p.product_id, p.product_name, p.min_risk_code AS product_risk_code, + p.min_subscribe_amount, p.term_days, p.requires_disclosure AS product_requires_disclosure, + sr.match_result AS matrix_match_result + FROM core_customer c + JOIN core_customer_risk r ON r.customer_id = c.customer_id + JOIN core_product p ON p.product_id = :pid + LEFT JOIN core_suitability_rule sr + ON sr.customer_risk_code = r.risk_code + AND sr.product_risk_code = p.min_risk_code + WHERE c.customer_id = :cid AND c.is_active = 1 + """ + ) + with self._engine.connect() as conn: + row = conn.execute(sql, {"cid": customer_id, "pid": product_id}).mappings().first() + if not row: + return self._suitability_result( + customer_id=customer_id, + product_id=product_id, + match_result="forbidden", + mismatch_type="not_found", + matched=False, + blocked=True, + reason="客户或产品不存在", + block_response_code="SUIT_NOT_FOUND", + ) + + data = dict(row) + base = { + "customer_id": data["customer_id"], + "product_id": data["product_id"], + "product_name": data.get("product_name"), + "customer_risk_code": data.get("customer_risk_code"), + "product_risk_code": data.get("product_risk_code"), + "investor_category": data.get("investor_category") or "ordinary", + "age": data.get("age"), + "is_hnw": data.get("is_hnw"), + "risk_expires_at": data.get("risk_expires_at"), + "risk_is_expired": _is_expired(data.get("risk_expires_at")), + "min_subscribe_amount": data.get("min_subscribe_amount"), + "term_days": data.get("term_days"), + } + + if base["risk_is_expired"]: + return self._suitability_result( + **base, + match_result="risk_expired", + mismatch_type="risk_expired", + matched=False, + blocked=True, + reason="风评已过期（FM-03），须重新测评", + block_response_code="SUIT_RISK_EXPIRED", + ) + + if base["investor_category"] == "professional": + return self._suitability_result( + **base, + match_result="professional_exempt", + mismatch_type="professional_exempt", + matched=True, + blocked=False, + requires_disclosure=False, + reason="专业投资者豁免适当性匹配", + block_response_code="SUIT_PROFESSIONAL_EXEMPT", + ) + + matrix = data.get("matrix_match_result") + if matrix == "forbidden" or matrix is None: + return self._suitability_result( + **base, + match_result="forbidden", + mismatch_type="risk_level", + matched=False, + blocked=True, + reason="客户风险等级与产品最低等级不匹配", + block_response_code="SUIT_RISK_MISMATCH", + ) + + requires_disclosure = matrix == "allowed_with_disclosure" or bool( + data.get("product_requires_disclosure") + ) + matched = True + blocked = False + match_result = "allowed_with_disclosure" if requires_disclosure else "allowed" + block_response_code = "SUIT_NEED_DISCLOSURE" if requires_disclosure else "SUIT_OK" + mismatch_type = "none" + reason: str | None = None + + age = base.get("age") or 0 + prod_r = base.get("product_risk_code") or "" + needs_branch_confirm = age >= 70 and prod_r >= "R3" + if needs_branch_confirm: + blocked = True + mismatch_type = "age_branch_confirm" + reason = "年龄70岁及以上购买R3及以上产品需网点当面确认（FM-01）" + block_response_code = "SUIT_AGE_CONFIRM" + + return self._suitability_result( + **base, + match_result=match_result, + mismatch_type=mismatch_type, + matched=matched, + blocked=blocked, + requires_disclosure=requires_disclosure, + needs_branch_confirm=needs_branch_confirm, + reason=reason, + block_response_code=block_response_code, + ) + + @staticmethod + def _suitability_result( + *, + customer_id: str | None = None, + product_id: str | None = None, + product_name: str | None = None, + customer_risk_code: str | None = None, + product_risk_code: str | None = None, + investor_category: str = "ordinary", + age: int | None = None, + is_hnw: bool | None = None, + risk_expires_at: Any = None, + risk_is_expired: bool = False, + min_subscribe_amount: Any = None, + term_days: Any = None, + match_result: str, + mismatch_type: str, + matched: bool, + blocked: bool, + requires_disclosure: bool = False, + needs_branch_confirm: bool = False, + reason: str | None = None, + block_response_code: str = "SUIT_OK", + ) -> dict[str, Any]: + return { + "customer_id": customer_id, + "product_id": product_id, + "product_name": product_name, + "customer_risk_code": customer_risk_code, + "product_risk_code": product_risk_code, + "investor_category": investor_category, + "age": age, + "is_hnw": is_hnw, + "risk_expires_at": risk_expires_at, + "risk_is_expired": risk_is_expired, + "min_subscribe_amount": min_subscribe_amount, + "term_days": term_days, + "match_result": match_result, + "mismatch_type": mismatch_type, + "matched": matched, + "blocked": blocked, + "requires_disclosure": requires_disclosure, + "needs_branch_confirm": needs_branch_confirm, + "reason": reason, + "block_response_code": block_response_code, + } + + def list_products_for_customer( + self, customer_id: str, limit: int = 50 + ) -> list[dict[str, Any]]: + """C-11：列出开放产品及与客户 C 等级的匹配结果（main 契约）。""" + sql = text( + """ + SELECT p.*, sr.match_result, + r.expires_at AS risk_expires_at, + r.risk_code AS customer_risk_code, + r.investor_category + FROM core_product p + CROSS JOIN core_customer_risk r + LEFT JOIN core_suitability_rule sr + ON sr.customer_risk_code = r.risk_code + AND sr.product_risk_code = p.min_risk_code + WHERE r.customer_id = :cid AND p.is_open = 1 + ORDER BY p.min_risk_code, p.product_id + LIMIT :lim + """ + ) + with self._engine.connect() as conn: + rows = conn.execute(sql, {"cid": customer_id, "lim": limit}).mappings().all() + result = [dict(r) for r in rows] + # FM-03 过期判定 Python 端算（替代 main 的 CURDATE()，sqlite 兼容） + for item in result: + item["risk_is_expired"] = _is_expired(item.pop("risk_expires_at", None)) + return result def list_holdings(self, customer_id: str, limit: int = 500) -> list[dict[str, Any]]: """持仓明细（市值降序）；limit 为 SQL 层保护上限（T-04 评审 P2）。 注意：调用方若要精确合计，须自行判断 len(rows) 是否触及 limit （core_tools.query_holdings 以 truncated 字段对外暴露）。 + main 扩列：min_subscribe_amount / term_days（C-11 起购与期限口径）。 """ sql = text( """ - SELECT h.*, p.product_name, p.min_risk_code, p.product_type + SELECT h.*, p.product_name, p.min_risk_code, p.product_type, + p.min_subscribe_amount, p.term_days FROM core_holding h JOIN core_product p ON p.product_id = h.product_id WHERE h.customer_id = :cid @@ -55,6 +297,29 @@ class CoreReadOnlyRepository: for r in conn.execute(sql, {"cid": customer_id, "lim": limit}).mappings() ] + def list_trades( + self, customer_id: str, since: date | None = None, limit: int = 50 + ) -> list[dict[str, Any]]: + """客户流水（时间降序 + 产品名 JOIN；main 新增查询能力）。""" + sql = text( + """ + SELECT t.*, p.product_name, p.min_risk_code + FROM core_trade t + JOIN core_product p ON p.product_id = t.product_id + WHERE t.customer_id = :cid + AND (:since IS NULL OR t.traded_at >= :since) + ORDER BY t.traded_at DESC + LIMIT :lim + """ + ) + with self._engine.connect() as conn: + return [ + dict(r) + for r in conn.execute( + sql, {"cid": customer_id, "since": since, "lim": limit} + ).mappings() + ] + def list_trades_range( self, customer_id: str, start: datetime, end: datetime, limit: int = 10000 ) -> list[dict[str, Any]]: