From 2a95b65ec8e8e338c30848f62ebc32b81b04a61e Mon Sep 17 00:00:00 2001 From: YUAN Date: Sun, 6 Sep 2026 18:05:44 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E6=A8=A1=E6=8B=9F=E4=BA=A4=E6=98=93?= =?UTF-8?q?=E7=BD=91=E5=85=B3(suitability=20=E9=98=BB=E6=96=AD=E2=86=92?= =?UTF-8?q?=E8=90=BD=E5=BA=93=E2=86=92=E5=BC=95=E6=93=8E,=20convert=20400,?= =?UTF-8?q?=20B5)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- app/api/simulate.py | 35 +++++ app/gateway/__init__.py | 10 ++ app/gateway/gateway_repository.py | 66 +++++++++ app/gateway/trade_gateway.py | 159 ++++++++++++++++++++ tests/test_trade_gateway.py | 235 ++++++++++++++++++++++++++++++ 5 files changed, 505 insertions(+) create mode 100644 app/api/simulate.py create mode 100644 app/gateway/__init__.py create mode 100644 app/gateway/gateway_repository.py create mode 100644 app/gateway/trade_gateway.py create mode 100644 tests/test_trade_gateway.py diff --git a/app/api/simulate.py b/app/api/simulate.py new file mode 100644 index 0000000..a42d078 --- /dev/null +++ b/app/api/simulate.py @@ -0,0 +1,35 @@ +"""模拟交易网关路由（PRD FR-1 · 薄路由，不含业务）。 + +鉴权：依赖 T-01 JWT（risk_demo 或客户本人 customer_id == subject）； +`get_auth_context` B6 落地后在此挂 `Depends`，本路由签名不变（开发计划 B5/B6 边界）。 +trace：B7 中间件贯通；B7 前由 service 层 ensure_trace 兜底。 +""" + +from __future__ import annotations + +from decimal import Decimal + +from fastapi import APIRouter, HTTPException +from pydantic import BaseModel, Field + +from app.gateway.trade_gateway import UnsupportedTradeType, submit_trade + +router = APIRouter(prefix="/api/simulate", tags=["simulate"]) + + +class TradeRequest(BaseModel): + customer_id: str = Field(..., min_length=1) + product_id: str = Field(..., min_length=1) + trade_type: str = Field(..., description="subscribe | redeem；convert 显式拒绝") + amount: Decimal = Field(..., gt=0, description="交易金额（元），必须为正数") + + +@router.post("/trade") +def submit_trade_api(req: TradeRequest) -> dict: + """模拟交易（FR-1）：适当性阻断或放行+引擎判定，返回 blocked + trade_id。""" + try: + return submit_trade(req.model_dump()) + except UnsupportedTradeType as exc: + raise HTTPException(status_code=400, detail=str(exc)) from exc + except LookupError as exc: + raise HTTPException(status_code=404, detail=str(exc)) from exc diff --git a/app/gateway/__init__.py b/app/gateway/__init__.py new file mode 100644 index 0000000..fea7541 --- /dev/null +++ b/app/gateway/__init__.py @@ -0,0 +1,10 @@ +"""模拟交易网关（PRD FR-1 · D5；外部 Core 交易系统替身）。 + +模块例外条款（FRAMEWORK §3）：本模块的 gateway_repository 是唯一允许 +INSERT jinrong_core.core_trade 的代码入口；生产环境由真实交易系统回调替代， +整个 app/gateway/ 退役。 + +链路（架构 §3.1）：参数校验（convert→400）→ suitability_check（FR-2， +不匹配→阻断，不落 trade）→ 匹配→INSERT core_trade→同步调规则引擎（FR-3） +→ 返回 blocked + trade_id + 触发规则。阻断/放行全量审计（agent_type='platform'）。 +""" diff --git a/app/gateway/gateway_repository.py b/app/gateway/gateway_repository.py new file mode 100644 index 0000000..fb63d2d --- /dev/null +++ b/app/gateway/gateway_repository.py @@ -0,0 +1,66 @@ +"""模拟交易网关写侧（PRD FR-1；FRAMEWORK §3 例外条款）。 + +仅本类可 INSERT jinrong_core.core_trade；core_ro 仍只读。生产环境由真实 +交易系统回调替代，本类随 app/gateway/ 退役。 +""" + +from __future__ import annotations + +from datetime import datetime +from decimal import Decimal +from typing import Any + +from sqlalchemy import create_engine, text +from sqlalchemy.engine import Engine + +from app.config.settings import settings + + +class GatewayRepository: + """core_trade 唯一写入口；仅 INSERT，不改不删（模拟网关语义）。""" + + def __init__(self, engine: Engine | None = None) -> None: + self._engine = engine or self._default_engine() + + @staticmethod + def _default_engine() -> Engine: + pwd = settings.mysql_password + auth = f"{settings.mysql_user}:{pwd}" if pwd else settings.mysql_user + url = ( + f"mysql+pymysql://{auth}@{settings.mysql_host}:{settings.mysql_port}" + f"/{settings.mysql_core_database}?charset=utf8mb4" + ) + return create_engine(url, pool_pre_ping=True) + + def insert_trade( + self, + trade_id: str, + customer_id: str, + product_id: str, + trade_type: str, + amount: Decimal, + traded_at: datetime, + trade_status: str = "confirmed", + ) -> None: + sql = text( + """ + INSERT INTO core_trade + (trade_id, customer_id, product_id, trade_type, amount, + trade_status, traded_at) + VALUES (:tid, :cid, :pid, :ttype, :amount, :status, :at) + """ + ) + with self._engine.begin() as conn: + conn.execute( + sql, + { + "tid": trade_id, + "cid": customer_id, + "pid": product_id, + "ttype": trade_type, + # str 无损传递：MySQL DECIMAL 隐式转换；sqlite text SQL 不支持 Decimal 绑定 + "amount": str(amount), + "status": trade_status, + "at": traded_at, + }, + ) diff --git a/app/gateway/trade_gateway.py b/app/gateway/trade_gateway.py new file mode 100644 index 0000000..002007a --- /dev/null +++ b/app/gateway/trade_gateway.py @@ -0,0 +1,159 @@ +"""交易网关服务（PRD FR-1 · 架构 §3.1）。 + +submit_trade 为唯一入口：参数校验 → suitability_check（FR-2，落校验日志； +不匹配→ suitability 预警单 + 阻断响应，交易不落 core_trade）→ 匹配 → +INSERT core_trade → 同步调规则引擎（FR-3）→ 返回 blocked + trade_id + +触发规则。阻断/放行全量审计（agent_type='platform'，FR-1 §6）。 + +鉴权归路由层（T-01/B6 的 get_auth_context：risk_demo 或客户本人）； +trace 由调用方中间件贯通，本层 ensure_trace 兜底（脚本/测试直调场景）。 +""" + +from __future__ import annotations + +from datetime import datetime +from decimal import Decimal +from typing import Any +from uuid import uuid4 + +from app.gateway.gateway_repository import GatewayRepository +from app.repository.core_ro import CoreReadOnlyRepository +from app.repository.risk_repository import RiskRepository +from app.service.risk.engine import process_trade_event +from app.service.risk.rules import RiskThresholds +from app.service.risk.alert_service import record_suitability_alert +from app.service.suitability import SuitabilityResult, suitability_check +from app.utils.trace import ensure_trace + +SUPPORTED_TRADE_TYPES = ("subscribe", "redeem") +CONVERT_MESSAGE = "转换交易暂不支持，请分别发起申购/赎回" +ADVICE = "请联系持证投资顾问" +RECORDED_NOTICE = "本次请求已记录" + + +class UnsupportedTradeType(ValueError): + """trade_type 非法（convert 显式 400；未知类型兜底拒绝，PRD FR-1）。""" + + +def _new_trade_id(now: datetime) -> str: + return f"TRD-{now:%Y%m%d}-{uuid4().hex[:8].upper()}" + + +def _audit( + repo: RiskRepository, + *, + decision: str, + trade_id: str, + req: dict[str, Any], + rule_id: str | None = None, + detail: dict[str, Any] | None = None, +) -> None: + from app.utils.trace import current_trace, new_trace + + repo.insert_audit_log( + { + "trace_id": current_trace() or new_trace(), + "event_type": "trade_request", + "agent_type": "platform", + "actor_id": "SYSTEM", # B6 接入 AuthContext 后透传 actor_id + "customer_id": req.get("customer_id"), + "rule_id": rule_id, + "input_summary": { + "trade_id": trade_id, + "product_id": req.get("product_id"), + "trade_type": req.get("trade_type"), + "amount": str(req.get("amount")), + **(detail or {}), + }, + "decision": decision, + "risk_score": None, + "handler_id": None, + "handler_result": None, + "handler_comment": None, + } + ) + + +def submit_trade( + req: dict[str, Any], + core_ro: CoreReadOnlyRepository | None = None, + risk_repo: RiskRepository | None = None, + gateway_repo: GatewayRepository | None = None, + thresholds: RiskThresholds | None = None, + now: datetime | None = None, +) -> dict[str, Any]: + """处理一笔模拟交易请求（PRD FR-1 流程 ①~⑤）。 + + req：{customer_id, product_id, trade_type, amount}；amount 转 Decimal。 + 返回 FR-1 ⑤ 响应体：阻断 {blocked, trade_id, block_reason, reasons, advice, + notice}；放行 {blocked, trade_id, triggered_rules, alert_ids, aml_hit}。 + """ + core = core_ro or CoreReadOnlyRepository() + repo = risk_repo or RiskRepository() + writer = gateway_repo or GatewayRepository() + th = thresholds or RiskThresholds.from_settings() + ensure_trace() + + trade_type = str(req.get("trade_type", "")) + if trade_type == "convert": + raise UnsupportedTradeType(CONVERT_MESSAGE) + if trade_type not in SUPPORTED_TRADE_TYPES: + raise UnsupportedTradeType(f"不支持的交易类型: {trade_type}（仅 subscribe/redeem）") + + now = now or datetime.now() + trade_id = _new_trade_id(now) + amount = Decimal(str(req["amount"])) + traded_at = now + + result: SuitabilityResult = suitability_check( + req["customer_id"], req["product_id"], core_ro=core, risk_repo=repo, request_ref=trade_id + ) + if result.blocked: + record_suitability_alert( + { + "trade_id": trade_id, + "customer_id": req["customer_id"], + "product_id": req["product_id"], + "trade_type": trade_type, + "amount": amount, + "traded_at": traded_at, + }, + rule_id=result.rule_id, + block_reason=result.block_reason, + risk_repo=repo, + ) + _audit( + repo, + decision="suitability_blocked", + trade_id=trade_id, + req=req, + rule_id=result.rule_id, + ) + return { + "blocked": True, + "trade_id": trade_id, + "block_reason": result.block_reason, + "reasons": list(result.reasons), + "advice": ADVICE, + "notice": RECORDED_NOTICE, + } + + writer.insert_trade( + trade_id, req["customer_id"], req["product_id"], trade_type, amount, traded_at + ) + engine_result = process_trade_event( + { + "trade_id": trade_id, + "customer_id": req["customer_id"], + "product_id": req["product_id"], + "trade_type": trade_type, + "amount": amount, + "trade_status": "confirmed", + "traded_at": traded_at, + }, + core_ro=core, + risk_repo=repo, + thresholds=th, + ) + _audit(repo, decision="trade_accepted", trade_id=trade_id, req=req) + return {"blocked": False, "trade_id": trade_id, **engine_result} diff --git a/tests/test_trade_gateway.py b/tests/test_trade_gateway.py new file mode 100644 index 0000000..9841164 --- /dev/null +++ b/tests/test_trade_gateway.py @@ -0,0 +1,235 @@ +"""trade_gateway 集成测试（B5 · FR-1：convert 400 / 阻断不落 trade / 放行贯通引擎）。 + +服务层直测三路径 + TestClient 验 HTTP 语义（sqlite 全套表，驱动差异由引擎层 +_normalize_trades 兜底）。API 层经 monkeypatch 注入 sqlite 仓储。 +""" + +from datetime import datetime, timedelta +from decimal import Decimal + +import pytest +from fastapi import FastAPI +from fastapi.testclient import TestClient +from sqlalchemy import create_engine, text +from sqlalchemy.pool import StaticPool + +from app.api.simulate import router as simulate_router +from app.gateway import trade_gateway as tg +from app.gateway.gateway_repository import GatewayRepository +from app.gateway.trade_gateway import UnsupportedTradeType, submit_trade +from app.repository.core_ro import CoreReadOnlyRepository +from app.repository.risk_repository import RiskRepository +from app.service.risk import alert_service +from app.service.risk.profile_l3 import AML_PENDING_TAG + + +class FakePublisher: + def __init__(self): + self.messages = [] + + def publish(self, channel, payload): + self.messages.append((channel, payload)) + + +DDL = [ + """CREATE TABLE core_customer ( + customer_id VARCHAR(64) PRIMARY KEY, display_name VARCHAR(128), age INTEGER, + occupation VARCHAR(64), open_date DATE, is_active TINYINT DEFAULT 1)""", + """CREATE TABLE core_customer_risk ( + customer_id VARCHAR(64), risk_code VARCHAR(8), evaluated_at TIMESTAMP)""", + """CREATE TABLE core_product ( + product_id VARCHAR(64) PRIMARY KEY, product_name VARCHAR(128), + min_risk_code VARCHAR(8), product_type VARCHAR(32))""", + """CREATE TABLE core_trade ( + trade_id VARCHAR(64) PRIMARY KEY, customer_id VARCHAR(64), product_id VARCHAR(64), + trade_type VARCHAR(16), amount DECIMAL, trade_status VARCHAR(16), traded_at TIMESTAMP)""", + """CREATE TABLE risk_aml_list ( + id INTEGER PRIMARY KEY AUTOINCREMENT, list_id VARCHAR(64), list_type VARCHAR(16), + full_name VARCHAR(128), match_threshold REAL, source VARCHAR(64), + list_version VARCHAR(16), effective_date DATE, is_active TINYINT DEFAULT 1, + created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP)""", + """CREATE TABLE risk_alert ( + alert_id VARCHAR(64) PRIMARY KEY, trace_id VARCHAR(64), customer_id VARCHAR(64), + trade_id VARCHAR(64), alert_type VARCHAR(16), triggered_rules TEXT, + risk_score INTEGER, status VARCHAR(24) DEFAULT 'pending_review', payload TEXT, + handler_id VARCHAR(64), handler_result VARCHAR(64), handler_comment VARCHAR(512), + created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, handled_at TIMESTAMP)""", + """CREATE TABLE audit_log ( + id INTEGER PRIMARY KEY, trace_id VARCHAR(64), event_type VARCHAR(64), + agent_type VARCHAR(16), actor_id VARCHAR(64), customer_id VARCHAR(64), + rule_id VARCHAR(64), input_summary TEXT, decision VARCHAR(64), risk_score INTEGER, + handler_id VARCHAR(64), handler_result VARCHAR(64), handler_comment VARCHAR(512), + created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP)""", + """CREATE TABLE risk_suitability_log ( + id INTEGER PRIMARY KEY AUTOINCREMENT, trace_id VARCHAR(64), customer_id VARCHAR(64), + product_id VARCHAR(64), customer_risk_level VARCHAR(8), product_risk_level VARCHAR(8), + is_matched TINYINT, is_blocked TINYINT, block_reason VARCHAR(512), + request_ref VARCHAR(64), profile_l1_version VARCHAR(32), + created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP)""", + """CREATE TABLE customer_profile_l3 ( + customer_id VARCHAR(64) PRIMARY KEY, monitor_tier VARCHAR(16) NOT NULL, + risk_score INTEGER, score_dimensions TEXT, monitor_tags TEXT, + last_alert_id VARCHAR(64), computed_at TIMESTAMP NOT NULL, + updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP)""", +] + + +@pytest.fixture() +def env(): + engine = create_engine( + "sqlite://", poolclass=StaticPool, connect_args={"check_same_thread": False} + ) + with engine.begin() as conn: + for ddl in DDL: + conn.execute(text(ddl)) + conn.execute( + text( + "INSERT INTO core_customer (customer_id, display_name, age, is_active) VALUES" + " ('CUST-1001', '客户·王**', 28, 1), ('CUST-3001', '客户·孙**', 45, 1)" + ) + ) + conn.execute( + text( + "INSERT INTO core_customer_risk (customer_id, risk_code, evaluated_at) VALUES" + " ('CUST-1001', 'C1', :t), ('CUST-3001', 'C3', :t)" + ), + {"t": datetime.now() - timedelta(days=30)}, + ) + conn.execute( + text( + "INSERT INTO core_product (product_id, product_name, min_risk_code, product_type) VALUES" + " ('PROD-161725', '科技成长主题', 'R4', 'mixed')," + " ('PROD-510300', '沪深300指数', 'R3', 'index')" + ) + ) + core = CoreReadOnlyRepository(engine=engine) + repo = RiskRepository(engine=engine) + writer = GatewayRepository(engine=engine) + pub = FakePublisher() + alert_service.set_publisher(pub) + yield core, repo, writer, pub, engine + alert_service.set_publisher(None) + engine.dispose() + + +def _req(customer="CUST-1001", product="PROD-161725", ttype="subscribe", amount="100000"): + return { + "customer_id": customer, + "product_id": product, + "trade_type": ttype, + "amount": amount, + } + + +def _counts(engine, table, where="1=1"): + with engine.connect() as conn: + return conn.execute(text(f"SELECT COUNT(*) FROM {table} WHERE {where}")).scalar_one() + + +# ---------- 服务层 ---------- + + +def test_convert_rejected(env): + core, repo, writer, _, _ = env + with pytest.raises(UnsupportedTradeType, match="转换交易暂不支持"): + submit_trade(_req(ttype="convert"), core_ro=core, risk_repo=repo, gateway_repo=writer) + with pytest.raises(UnsupportedTradeType, match="不支持的交易类型"): + submit_trade(_req(ttype="purchase"), core_ro=core, risk_repo=repo, gateway_repo=writer) + assert _counts(writer._engine, "core_trade") == 0 + assert _counts(writer._engine, "audit_log", "decision='invalid_type'") == 0 + + +def test_blocked_does_not_touch_core_trade(env): + """A-1：C1 客户买 R4 → blocked=true(SUIT-001)、不落 core_trade、日志/预警/审计齐全。""" + core, repo, writer, pub, engine = env + resp = submit_trade(_req(), core_ro=core, risk_repo=repo, gateway_repo=writer, + now=datetime(2026, 9, 6, 14, 0, 0)) + assert resp["blocked"] is True + assert "SUIT-001" in resp["block_reason"] + assert resp["advice"] == "请联系持证投资顾问" and resp["notice"] == "本次请求已记录" + assert _counts(engine, "core_trade") == 0 # 阻断不落交易 + assert _counts(engine, "risk_suitability_log", "is_blocked=1 AND request_ref='" + resp["trade_id"] + "'") == 1 + assert _counts(engine, "risk_alert", "alert_type='suitability'") == 1 + assert _counts(engine, "audit_log", "agent_type='platform' AND decision='suitability_blocked'") == 1 + assert _counts(engine, "audit_log", "agent_type='platform' AND decision='trade_accepted'") == 0 + assert len(pub.messages) == 1 # suitability 预警推送 + + +def test_accepted_trades_and_engine_fires(env): + """A-3：C3 客户 60 万买 R3 → 落库 confirmed + 事件预警单(RISK-001/002) + 审计放行。""" + core, repo, writer, pub, engine = env + resp = submit_trade( + _req(customer="CUST-3001", product="PROD-510300", amount="600000"), + core_ro=core, risk_repo=repo, gateway_repo=writer, now=datetime(2026, 9, 6, 14, 0, 0), + ) + assert resp["blocked"] is False + assert resp["trade_id"].startswith("TRD-20260906-") + assert resp["triggered_rules"] == ["RISK-001", "RISK-002"] # 含本笔累计 + assert len(resp["alert_ids"]) == 1 + with engine.connect() as conn: + row = conn.execute( + text("SELECT trade_status, amount FROM core_trade WHERE trade_id=:t"), + {"t": resp["trade_id"]}, + ).mappings().one() + assert row["trade_status"] == "confirmed" and Decimal(str(row["amount"])) == Decimal("600000") + alert = repo.get_alert(resp["alert_ids"][0]) + assert alert["alert_type"] == "large_amount" and alert["risk_score"] == 70 + assert _counts(engine, "risk_suitability_log", "is_blocked=0") == 1 + assert _counts(engine, "audit_log", "agent_type='platform' AND decision='trade_accepted'") == 1 + assert _counts(engine, "customer_profile_l3", "monitor_tier='watch'") == 1 + assert len(pub.messages) == 1 + + +def test_missing_customer_returns_lookup_error(env): + core, repo, writer, _, _ = env + with pytest.raises(LookupError): + submit_trade(_req(customer="CUST-9999"), core_ro=core, risk_repo=repo, gateway_repo=writer) + + +# ---------- API 层（TestClient；仓储注入 sqlite） ---------- + + +@pytest.fixture() +def client(env, monkeypatch): + core, repo, writer, pub, engine = env + monkeypatch.setattr(tg, "CoreReadOnlyRepository", lambda: core) + monkeypatch.setattr(tg, "RiskRepository", lambda: repo) + monkeypatch.setattr(tg, "GatewayRepository", lambda: writer) + app = FastAPI() + app.include_router(simulate_router) + with TestClient(app) as c: + yield c + + +def test_api_convert_returns_400(client): + r = client.post("/api/simulate/trade", json=_req(ttype="convert")) + assert r.status_code == 400 + assert "转换交易暂不支持" in r.json()["detail"] + + +def test_api_blocked_returns_200_with_blocked_true(client): + r = client.post("/api/simulate/trade", json=_req()) # C1 买 R4 + assert r.status_code == 200 + body = r.json() + assert body["blocked"] is True and "SUIT-001" in body["block_reason"] + + +def test_api_accepted_returns_200_with_trade_id(client): + r = client.post( + "/api/simulate/trade", + json=_req(customer="CUST-3001", product="PROD-510300", amount="600000"), + ) + assert r.status_code == 200 + body = r.json() + assert body["blocked"] is False and body["trade_id"].startswith("TRD-") + assert body["triggered_rules"] == ["RISK-001", "RISK-002"] + + +def test_api_non_positive_amount_returns_422(client): + r = client.post("/api/simulate/trade", json=_req(amount="0")) + assert r.status_code == 422 + + +def test_api_unknown_customer_returns_404(client): + r = client.post("/api/simulate/trade", json=_req(customer="CUST-9999")) + assert r.status_code == 404