diff --git a/app/api/risk.py b/app/api/risk.py index 54e3ae6..7499a8f 100644 --- a/app/api/risk.py +++ b/app/api/risk.py @@ -9,8 +9,8 @@ handler_result 枚举由请求模型 Literal 校验（repo 不校验，开发计 直调 suitability/check 每次落 audit（request_ref='api:suitability_check'， 评审 P2-3/P3-9）；aml/scan 幂等防护挂账 B9b 前（评审 P3-6）。 挂载：main.py include（B7）；错误体统一 ApiError → 手册 §10 结构 -（utils/response.register_error_handlers，挂账④；预警类响应固定 disclaimer -挂账 B9b 核查单①）。 +（utils/response.register_error_handlers，挂账④）。 +预警类 API 响应体固定 disclaimer（PRD §6/规则表 §5 · B9b 核查单①）。 """ from __future__ import annotations @@ -33,6 +33,9 @@ from app.utils.trace import current_trace, new_trace router = APIRouter(prefix="/api/risk", tags=["risk"]) +# 预警系统级声明（规则表 §5，区别于 G-08 客户免责声明；PRD §6 固定出现在预警类响应体） +ALERT_DISCLAIMER = "本预警由系统自动生成，最终判定需经风控专员人工审核。" + def _repo() -> RiskRepository: """仓储入口（测试 monkeypatch 点）。""" @@ -76,7 +79,8 @@ def list_alerts_api( page=page, page_size=page_size, ) - return {"items": rows, "total": total, "page": page, "page_size": page_size} + return {"items": rows, "total": total, "page": page, "page_size": page_size, + "disclaimer": ALERT_DISCLAIMER} @router.post("/alerts/{alert_id}/handle") @@ -85,13 +89,14 @@ def handle_alert_api(alert_id: str, req: HandleRequest, auth: AuthContext = Depe if not auth.has_role("risk_officer"): deny(auth, "AUTH_403_ROLE", _repo(), message="risk_officer only") try: - return handle_alert( + handled = handle_alert( alert_id, req.handler_result, auth.actor_id, req.handler_comment, risk_repo=_repo() ) except LookupError as exc: # NotFoundError 子类（B6 评审 P3-5 收敛） raise ApiError(404, "NOT_FOUND", str(exc)) from exc except StateConflict as exc: raise ApiError(409, "STATE_CONFLICT", str(exc)) from exc + return {**handled, "disclaimer": ALERT_DISCLAIMER} @router.post("/suitability/check") @@ -157,4 +162,4 @@ def aml_scan_api(auth: AuthContext = Depends(get_auth_context)) -> dict: "handler_comment": None, } ) - return summary + return {**summary, "disclaimer": ALERT_DISCLAIMER} diff --git a/app/repository/risk_repository.py b/app/repository/risk_repository.py index 39710a4..33f9636 100644 --- a/app/repository/risk_repository.py +++ b/app/repository/risk_repository.py @@ -72,6 +72,25 @@ class RiskRepository: row = conn.execute(sql, params).mappings().first() return self._parse_alert(dict(row)) if row else None + def find_recent_aml_alert(self, customer_id: str, day_start: datetime) -> dict | None: + """该客户当日最近一张 aml 单（scan 幂等检查 · B9b 核查单②/B6 评审 P3-6）。 + + 不限 status——已处置单也算「已出」，同扫描日重扫不重复出单（模拟每日 + 批量口径：跨日再扫出新单）。交易触发的当日 aml 单同样命中（当日命中 + 已留痕，scan 无需重复出单）。 + """ + sql = text( + """ + SELECT * FROM risk_alert + WHERE customer_id = :cid AND alert_type = 'aml' AND created_at >= :day_start + ORDER BY created_at DESC + LIMIT 1 + """ + ) + with self._engine.connect() as conn: + row = conn.execute(sql, {"cid": customer_id, "day_start": day_start}).mappings().first() + return self._parse_alert(dict(row)) if row else None + def insert_alert(self, alert: dict[str, Any]) -> None: sql = text( """ diff --git a/app/service/risk/aml_service.py b/app/service/risk/aml_service.py index 6daa159..92d95ba 100644 --- a/app/service/risk/aml_service.py +++ b/app/service/risk/aml_service.py @@ -16,6 +16,7 @@ from __future__ import annotations import difflib import logging +from datetime import date, datetime, time from typing import Any from app.config.settings import settings @@ -87,19 +88,28 @@ def scan_all( """手动全量扫描（PRD FR-5 触发时机 2，B6 接 POST /api/risk/aml/scan）。 每个命中客户：一张 aml 独立预警单（matches 全量进 payload）+ L3 high 标记。 + 幂等防护（B9b 核查单②/B6 评审 P3-6）：同客户当日已有 aml 单（含交易触发 + 与已处置单）不再重复出单，计入 skipped_existing——模拟每日批量口径，重复 + 点击不刷单；读后写竞态在手动单发操作下可接受（无并发出单场景）。 """ core = core_ro or CoreReadOnlyRepository() repo = risk_repo or RiskRepository() ensure_trace() # 手动扫描入口兜底归因（B6 API 场景保留中间件 trace，评审 P3-8） entries = repo.list_active_aml_entries() customers = core.list_active_customers() + day_start = datetime.combine(date.today(), time.min) alerts: list[str] = [] + skipped_existing: list[str] = [] hit_customers = 0 for c in customers: hits = match_name(c["display_name"], entries) if not hits: continue hit_customers += 1 + existing = repo.find_recent_aml_alert(c["customer_id"], day_start) + if existing: + skipped_existing.append(existing["alert_id"]) + continue alert = record_aml_alert( c["customer_id"], {"trigger": "scan", "matches": hits}, risk_repo=repo ) @@ -107,5 +117,13 @@ def scan_all( c["customer_id"], "aml", last_alert_id=alert["alert_id"], risk_repo=repo ) alerts.append(alert["alert_id"]) - logger.info("AML scan_all: scanned=%d hit_customers=%d", len(customers), hit_customers) - return {"scanned": len(customers), "hit_customers": hit_customers, "alerts": alerts} + logger.info( + "AML scan_all: scanned=%d hit_customers=%d new_alerts=%d skipped_existing=%d", + len(customers), hit_customers, len(alerts), len(skipped_existing), + ) + return { + "scanned": len(customers), + "hit_customers": hit_customers, + "alerts": alerts, + "skipped_existing": skipped_existing, + } diff --git a/docs/memory/FLOW.md b/docs/memory/FLOW.md index 3afb50f..6c00cd2 100644 --- a/docs/memory/FLOW.md +++ b/docs/memory/FLOW.md @@ -48,9 +48,9 @@ RBAC 联调账号：scripts/dev/rbac-seed-reference.md Client → Gateway(JWT/RBAC) → api/chat → agent_service(LangGraph) → Tools → 存储 → 响应 + audit_log ``` -当前：**风控事件线已集成 + 测试基建完成**（交易网关 → 规则引擎 → 预警/AML/L3 → 4 API，`main.py` 路由挂载 + trace 中间件 + lifespan 完成，B1~B8；B8 = conftest 演示数据校验 + 集成测试 A-1~A-5/A-7/A-9 + `tests/_ddl.py` DDL 单一事实源；B7/B8 均经独立 AI 评审闭环；B9a = subscribe_alerts/rebuild_alerts 演示与补偿脚本，204 测试绿）；剩 B9b 演示走查；对话链路待 T-01/T-07。 +当前：**风控事件线已集成 + 测试基建完成**（交易网关 → 规则引擎 → 预警/AML/L3 → 4 API，`main.py` 路由挂载 + trace 中间件 + lifespan 完成，B1~B8；B8 = conftest 演示数据校验 + 集成测试 A-1~A-5/A-7/A-9 + `tests/_ddl.py` DDL 单一事实源；B7/B8 均经独立 AI 评审闭环；B9a = subscribe_alerts/rebuild_alerts 演示与补偿脚本；B9b = 演示走查 + B6/B7/B8 核查单①~⑥收口（disclaimer/aml scan 幂等/集成测试前置断言），208 测试绿，演示 SOP 见 docs/项目框架设计/演示SOP-风控模块.md，M2 tag risk-m2）；阶段 B 完结；对话链路待 T-01/T-07。 -**本机已就位状态（2026-09-06 · 已完成上述 ①~⑤，无需重做）：** `.env` 已配置（学习项目，`MYSQL_PASSWORD=123456`；NEO4J/DEEPSEEK 留空暂不影响风控）；`jinrong_core` + `jinrong_agent` 已灌库（28 客户 / AML 名单 8 条 / 演示测评已刷新 / 归属同步 28 行）；`python -m pytest` 204 绿（集成测试真连本机 MySQL；未灌库的机器自动 skip 集成模块，单测不受影响）；本机 Redis 服务在跑（`redis://127.0.0.1:6379/0`，B9a 订阅脚本已验证）。 +**本机已就位状态（2026-09-06 · 已完成上述 ①~⑤，无需重做）：** `.env` 已配置（学习项目，`MYSQL_PASSWORD=123456`；NEO4J/DEEPSEEK 留空暂不影响风控）；`jinrong_core` + `jinrong_agent` 已灌库（28 客户 / AML 名单 8 条 / 演示测评已刷新 / 归属同步 28 行）；`python -m pytest` 208 绿（集成测试真连本机 MySQL；未灌库的机器自动 skip 集成模块，单测不受影响）；本机 Redis 服务在跑（`redis://127.0.0.1:6379/0`，B9a 订阅脚本已验证）。 **本机已知坑：** `mysql.exe` 不在 PATH（位于 `C:\Program Files\MySQL\MySQL Server 8.0\bin`）；`reset.ps1` 的交互式 `-p` 在自动化执行时会卡死——脚本化重灌用 `MYSQL_PWD` 环境变量传密码（交互执行不受影响，不把密码写进仓库脚本）；`redis` 包 requirements 有但初始 `pip install -r` 漏装（B9a 于 2026-09-07 补装，重装环境时留意）。 diff --git a/docs/memory/MEMORY.md b/docs/memory/MEMORY.md index 919557b..3e6acfd 100644 --- a/docs/memory/MEMORY.md +++ b/docs/memory/MEMORY.md @@ -9,7 +9,7 @@ **项目是什么：** 金融四 Agent（客户财富 / 代理人 / 数据分析 / 风控）共用数据层与合规底座；**不**互调 LLM，跨 Agent 走 L1/L2/L3 画像与预警表。 -**当前进度：** 需求与表设计已定 · **风控模块已落地 B1~B8+B9a**（规则/预警聚合/L3/AML/引擎/交易网关/鉴权+4 API/适当性校验/main 集成+挂账①~⑦/B8 conftest+集成测试/**B9a 订阅+重放补偿脚本**，B7/B8/B9a 均经独立 AI 评审闭环，**204 测试绿**）· 剩 **B9b 演示走查** · **JWT(T-01) / 审计中间件(T-02) / LangGraph 对话线(T-07) 未做**（chat/knowledge/admin 仍空壳）。**开发在分支 `feature/risk`（未合入 main）。** +**当前进度：** 需求与表设计已定 · **风控模块 B1~B9b 全部完成（M2 已达成，tag risk-m2）**（规则/预警聚合/L3/AML/引擎/交易网关/鉴权+4 API/适当性校验/main 集成/B8 测试基建/B9a 订阅+重放脚本/B9b 演示走查+核查单①~⑥收口，B7/B8/B9a 均经独立 AI 评审闭环，**208 测试绿**）· 阶段 B 完结剩 M2 文字复核 → **JWT(T-01) / 审计中间件(T-02) / LangGraph 对话线(T-07) 未做**（chat/knowledge/admin 仍空壳）。**开发在分支 `feature/risk`（未合入 main）。** **仓库地图：** @@ -51,7 +51,7 @@ > **本机（2026-09-06）①~⑤已执行、`.env` 已配置，勿重做**；本机状态与已知坑（mysql.exe 路径 / reset.ps1 交互式 -p）见 `FLOW.md` §0 尾注。 -**下一步开发（见 TODO）：** 风控 B9b 演示走查（reset → demo SQL → Swagger 过 A-1~A-5/A-7~A-9 + B6/B7/B8 挂账核查单①~⑥）；随后 Wave 0 T-01 JWT / T-02 审计中间件 / T-06 / T-07 LangGraph。 +**下一步开发（见 TODO）：** 风控 M2 文字收尾复核 → Wave 0 T-01 JWT / T-02 审计中间件 / T-06 / T-07 LangGraph。演示走查按 `docs/项目框架设计/演示SOP-风控模块.md`。 **禁止（改代码前必记）：** Core 正式 C1~C5 不可被画像覆盖 · 审计表只 INSERT · 代理人草稿不外发 · 仅 R-02 可阻断交易 · 四 Agent 不互调 LLM。 @@ -166,6 +166,6 @@ RBAC 联调账号：scripts/dev/rbac-seed-reference.md 2. 改动属于 api / service / tool / repository 哪一层？ 3. 是否需 customer_id 归属与 JWT RBAC？ 4. Core 是模拟库只读还是 agent 库读写？ -5. 如何验证？（`python -m pytest` 全量（当前 204 绿）· uvicorn 启动 + /health · SQL / sync 脚本 · 对照 REQUIREMENTS 验收列） +5. 如何验证？（`python -m pytest` 全量（当前 208 绿）· uvicorn 启动 + /health · SQL / sync 脚本 · 对照 REQUIREMENTS 验收列） 大任务：FRAMEWORK/FLOW 与实现状态不符时先更新 memory 再编码（用户确认跳过除外）。 diff --git a/docs/memory/TODO.md b/docs/memory/TODO.md index e393b37..c39eb83 100644 --- a/docs/memory/TODO.md +++ b/docs/memory/TODO.md @@ -5,12 +5,10 @@ ## 进行中 -- [ ] 风控 **B9b 演示链路走查**（reset → demo SQL → Swagger 过 A-1~A-5/A-7~A-9 + 核查单①~⑥）——走查前先处理核查单⑥（B8 复审 P2-1 演示库耦合前置断言） +- [ ] 风控 M2 收尾：MEMORY/TODO 状态复核（开发计划 M4 前置）——B1~B9b 已全部完成（M2 tag 已打），复核边界文字对齐后转 Wave 0 ## 待办（推荐顺序） -- [ ] 风控 M2 收尾：MEMORY/TODO 状态复核（开发计划 M4 前置） - - [ ] T-01 Auth SDK / JWT 中间件（对照 `02-JWT-RBAC鉴权手册.md`） - [ ] T-02 audit_log 中间件 + trace_id 贯通 - [ ] T-05 本机执行 `reset.ps1` + `01-mysql-共用底座.sql` + sync 脚本验收 @@ -22,7 +20,7 @@ ### 风控模块（PRD v1.0 已冻结 · `docs/PRD/PRD-风控监测Agent.md`，事件驱动线不依赖 T-07 可先行） -- [ ] T-30 风控事件线：`app/gateway/` 交易网关 + `service/risk/` 规则引擎（RISK-001~005）+ 预警单聚合 + L3 最小写入 + `risk:pub:alert` 推送 + AML（含 `risk_aml_list` 种子）+ 演示数据脚本验收 A-1~A-5/A-9 —— **进度（2026-09-06）：B1 规则纯函数 / B2 预警服务 / B3 L3 写入（risk_score 一期不写）/ B4 AML+引擎编排 / B5 交易网关 / B6 鉴权+4 API+处置编排 / B7 main 集成（路由挂载 + trace 中间件 + lifespan 启动期 debug 校验；挂账①锁公共化 ②L3 缓存 DEL ③死代码 ④统一错误体 ⑤启动校验 ⑥引擎工厂 ⑦处置原子事务均落地；⑧ input_guard_log 双写仍随 T-02）均完成并经独立 AI 评审闭环（B7 复审有条件通过→P1-1 dispose 已修闭环；183 测试绿；P2/P3 已登记开发计划 B7/B8/B9b/T-02）；**B8 已完成（2026-09-06）：conftest（演示数据校验/幂等代跑 prepare_risk_demo/TRD-TEST- teardown + L3 快照还原）+ 集成测试 11 例（A-1~A-5/A-7/A-9 + trace 一致性 + 审计 JSON）+ sqlite DDL 单一事实源 `_ddl.py`（10 文件收敛，localtime 时区收敛 B5 P3-4）+ L3 DEL 行为断言（P2-1）+ locks 文案（P3-2）；B8 复审有条件通过→P1-1 conftest skip 路径炸收集已修闭环（故障注入验证），P2-1 演示库耦合前置断言挂 B9b 核查单⑥**；**B9a 已完成（2026-09-07）：`scripts/demo/subscribe_alerts.py`（risk:pub:alert 订阅演示：连接自检/--duration/心跳）+ `rebuild_alerts.py`（按 trade_id 幂等重放：`find_alerts_by_trade` 以 payload LIKE 查已入单（含 aml/已处置单）防重复出单与重复 append；engine_error 中断笔 skip 时 warning 提示人工核对；core_ro 新增只读 `get_trade_by_id`）+ tests/test_demo_scripts.py 8 例，204 绿；真库手工验证通过（rebuilt→skipped→missing exit1；订阅端到端收到假消息+真交易推送且 trace 贯通；中间态警示路径复验），验证现场已清理；独立 AI 评审有条件通过→P2-1 中间态警示/P2-2 非首笔幂等用例已闭环**；剩 B9b 演示走查** +- [ ] T-30 风控事件线：`app/gateway/` 交易网关 + `service/risk/` 规则引擎（RISK-001~005）+ 预警单聚合 + L3 最小写入 + `risk:pub:alert` 推送 + AML（含 `risk_aml_list` 种子）+ 演示数据脚本验收 A-1~A-5/A-9 —— **进度（2026-09-06）：B1 规则纯函数 / B2 预警服务 / B3 L3 写入（risk_score 一期不写）/ B4 AML+引擎编排 / B5 交易网关 / B6 鉴权+4 API+处置编排 / B7 main 集成（路由挂载 + trace 中间件 + lifespan 启动期 debug 校验；挂账①锁公共化 ②L3 缓存 DEL ③死代码 ④统一错误体 ⑤启动校验 ⑥引擎工厂 ⑦处置原子事务均落地；⑧ input_guard_log 双写仍随 T-02）均完成并经独立 AI 评审闭环（B7 复审有条件通过→P1-1 dispose 已修闭环；183 测试绿；P2/P3 已登记开发计划 B7/B8/B9b/T-02）；**B8 已完成（2026-09-06）：conftest（演示数据校验/幂等代跑 prepare_risk_demo/TRD-TEST- teardown + L3 快照还原）+ 集成测试 11 例（A-1~A-5/A-7/A-9 + trace 一致性 + 审计 JSON）+ sqlite DDL 单一事实源 `_ddl.py`（10 文件收敛，localtime 时区收敛 B5 P3-4）+ L3 DEL 行为断言（P2-1）+ locks 文案（P3-2）；B8 复审有条件通过→P1-1 conftest skip 路径炸收集已修闭环（故障注入验证），P2-1 演示库耦合前置断言挂 B9b 核查单⑥**；**B9a 已完成（2026-09-07）：`scripts/demo/subscribe_alerts.py`（risk:pub:alert 订阅演示：连接自检/--duration/心跳）+ `rebuild_alerts.py`（按 trade_id 幂等重放：`find_alerts_by_trade` 以 payload LIKE 查已入单（含 aml/已处置单）防重复出单与重复 append；engine_error 中断笔 skip 时 warning 提示人工核对；core_ro 新增只读 `get_trade_by_id`）+ tests/test_demo_scripts.py 8 例；真库手工验证通过（rebuilt→skipped→missing exit1；订阅端到端收到假消息+真交易推送且 trace 贯通；中间态警示路径复验），验证现场已清理；独立 AI 评审有条件通过→P2-1 中间态警示/P2-2 非首笔幂等用例已闭环**；**B9b 已完成（2026-09-07）：前置代码三件（①预警类 API 固定 disclaimer ②scan_all 幂等防护 `find_recent_aml_alert`+`skipped_existing` ③核查单⑥集成测试前置断言 A-1/A-3/A-4 清洁检查+A-5 按 trade_id 定位），208 绿；演示 SOP `docs/项目框架设计/演示SOP-风控模块.md`；真服务走查（uvicorn+真库）A-1~A-5/A-7/A-9+核查单①②④⑤ 全 PASS（11/11，A-6 归 M3），核查单⑥防护现场复现验证；走查演示数据按 SOP 保留、pytest 集成前需重灌；M2 达成打 tag risk-m2（M1 未打 tag 留痕）——核查单①~⑥ 全部收口，风控阶段 B 完结** - [ ] T-31 `service/suitability.py` 公共校验（SUIT-001~008）+ `POST /api/risk/suitability/check` + 单测验收 A-8 —— **代码已完成（A1~A4 · 77 测试绿；B6 起该 API 带鉴权+直调审计）；MySQL 手工 SQL 对照挂账至 B9b 执行（阶段 A 评审 P2-8）** - [ ] T-32 预警台账与人工处置 API（`GET /alerts`、`POST /handle`，risk_officer/compliance 权限）+ 对话线（依赖 T-01/T-03/T-07）验收 A-6/A-7 —— **API 部分已由 B6 覆盖并复审通过（A-7/A-9 用例绿）；对话线仍依赖 T-01/T-03/T-07** diff --git a/docs/项目框架设计/开发计划-风控模块.md b/docs/项目框架设计/开发计划-风控模块.md index 4475d89..5a1df69 100644 --- a/docs/项目框架设计/开发计划-风控模块.md +++ b/docs/项目框架设计/开发计划-风控模块.md @@ -39,6 +39,8 @@ **B8 复审（独立 AI 评审 · 2026-09-06）：有条件通过 → 已闭环。** 开发计划 B8 行 a/b/c + DDL 单一事实源/localtime + TRD-TEST- 注入 + A-1~A-5/A-7/A-9/trace/审计 JSON 全部达成，196 绿实跑，B7 复审 P2-1/P3-2 落地。**P1-1 `conftest.py` `except pytest.SkipRequested` 引用不存在属性，演示数据缺失时 skip 路径反噬整个 pytest 收集（未灌库机器单测全跑不了）——已修（Skipped 继承 BaseException 直传，故障注入验证：错密码收集期 skip 不 error）**；P3 顺手项已落：teardown `l3_snapshot` 哨兵（防 setup 失败掩盖原始异常）、A-1 补 trace 环①响应头与环②suitability_log.trace_id。挂账：**P2-1 集成测试对演示库当日状态隐式耦合（A-4 依赖 CUST-9527 当日无历史交易、A-5 aml 单查询未按本测试 trade_id 定位、A-1 同日已有 pending 单会并入旧单）→ B9b 核查单⑥，演示与 pytest 同日交叉前必须处理**；P3-3 `_state` 模块级顺序耦合（单跑 A-7 等 KeyError，文件头已声明保序，可接受留痕）；P3-4 阈值冒烟用例在 .env 覆盖阈值时会红（注释已声明前提）；P3-5 teardown 时间窗隐含 MySQL 时钟==本机时钟（docstring 已声明单机约定）。 | B9b | 演示链路走查：`reset.ps1` → `prepare_risk_demo.sql` → agent 库建表 → `seed-aml-list.sql` → Swagger 逐条过 **A-1~A-5、A-7~A-9（A-6 归 M3）**。**B6 挂账核查单：① 预警类 API 响应体含固定 disclaimer「本预警由系统自动生成，最终判定需经风控专员人工审核」（PRD §6/规则表 §5，B6 复审 P3-7）② aml/scan 幂等防护（重复扫描同命中客户重复出单，演示点击即复现，B6 评审 P3-6）③ B6 时代码以 TestClient 独立挂 router 等价验证，本次补一次真 Swagger 手测（B6 复审遗漏⑤）④ analyst 台账只读权限扩展待 Wave 3 分析 Agent 接入时定权限矩阵（手册 §5.3 有 risk:alert:read，现 fail-closed 拒绝，B6 复审观察③）⑤ 生产/演示机 `app_env=development` 误配检查（B7 复审 P3-3：误配时 debug 头可达且启动校验放行，列入演示 SOP）⑥ B8 复审 P2-1：集成测试对演示库当日状态的隐式耦合前置断言（A-4 前断言 CUST-9527 当日事件计数==0、A-5 aml 单按本测试 trade_id 定位、A-1 同日 pending 单并入防护），演示与 pytest 同日交叉执行前必须处理** | 演示 SOP | 按 PRD §8 验收表逐条打勾 | B8、B9a | +**B9b 完成（2026-09-07）：** 走查前置代码三件：①预警类 API（GET alerts/POST handle/POST aml/scan）响应体加固定 `disclaimer`（核查单①）②`scan_all` 幂等防护——新增 `find_recent_aml_alert`，同客户当日已有 aml 单（含交易触发/已处置）不再出单，summary 增 `skipped_existing`（核查单②）③集成测试前置断言（核查单⑥）：A-1/A-3/A-4 清洁检查（交叉时 fail+重灌提示）、A-5 aml 单按本笔 trade_id 定位。208 测试绿（+4：scan 幂等×2/disclaimer/交易触发后 scan skip）。演示 SOP 落地 `docs/项目框架设计/演示SOP-风控模块.md`（含脚本化重灌 MYSQL_PWD 方式）。真服务走查（uvicorn 8123 + 真库）：**A-1~A-5/A-7/A-9 + 核查单①②④⑤ 全 PASS（11/11）**，核查单⑥防护价值现场复现（走查残留数据致 A-1 定位失效 → 重灌后过，正是 P2-1 预警的场景）。核查单③=本次真服务手测（区别于 B6 TestClient 等价验证）；⑤本机 APP_ENV=development 确认。走查演示数据按 SOP 保留，pytest 集成测试前需重灌。**里程碑 M2 达成（阶段 B 全链路 + A-1~A-5/A-7~A-9 验收），commit 打 tag risk-m2（M1 时点未打 tag，留痕）**。 + ## 阶段 C · 对话线（依赖 Wave 0 的 T-01 JWT / T-03 输入防护 / T-07 LangGraph） | # | 任务 | 产出 | 验证方式 | 依赖 | diff --git a/docs/项目框架设计/演示SOP-风控模块.md b/docs/项目框架设计/演示SOP-风控模块.md new file mode 100644 index 0000000..120f1f8 --- /dev/null +++ b/docs/项目框架设计/演示SOP-风控模块.md @@ -0,0 +1,89 @@ +# 演示 SOP · 风控监测 Agent（B9b） + +> 版本：v1.0 · 2026-09-07 · 配套 PRD v1.0 §8 验收表 + 开发计划 B9b 核查单 +> 适用：演示机/本机（`app_env=development`，debug 鉴权头可用；生产部署禁用，见核查单⑤） +> 首次走查记录：2026-09-07，A-1~A-5/A-7~A-9 + 核查单①~⑥ 全部通过（11/11 PASS） + +--- + +## 1. 前置条件 + +- 本机 bootstrap ①~⑤ 已执行（`.env`、依赖、双库、种子）——见 `docs/memory/FLOW.md` §0； + 已就位机器勿重跑，只按本 SOP §2 重灌即可。 +- `redis` 包已安装（requirements 内；B9a 曾发现环境漏装，重装环境时留意）。 +- 本机 MySQL 可达、Redis 服务在跑（订阅演示需要；不可达时推送降级仅日志，走查其余项不受影响）。 + +## 2. 演示链路重灌（每次演示/走查前执行） + +**目的**：清空上一轮演示/走查残留（trade/预警/审计/L3），使验收断言从干净状态出发。 + +**脚本化方式（推荐，自动化执行不卡交互）**——Git Bash： + +```bash +export MYSQL_PWD=<密码> +MYSQL="C:/Program Files/MySQL/MySQL Server 8.0/bin/mysql.exe" # mysql.exe 不在 PATH（FLOW §0 坑） +"$MYSQL" -h 127.0.0.1 -u root -e "DROP DATABASE IF EXISTS jinrong_core; DROP DATABASE IF EXISTS jinrong_agent;" +for f in scripts/core/00-create-database.sql scripts/core/01-ddl.sql scripts/core/02-seed-base.sql \ + scripts/core/03-seed-customers.sql scripts/core/04-seed-holdings.sql scripts/core/05-seed-trades.sql \ + scripts/core/06-seed-nav.sql "docs/项目框架设计/表设计/01-mysql-共用底座.sql" \ + "docs/项目框架设计/表设计/02-mysql-agent专用.sql" scripts/agent/seed-aml-list.sql \ + scripts/demo/prepare_risk_demo.sql; do + "$MYSQL" -h 127.0.0.1 -u root --default-character-set=utf8mb4 < "$f" +done +python scripts/sync/sync_advisor_rel.py # 归属同步（28 行） +``` + +**交互方式**：`.\scripts\core\reset.ps1`（PowerShell，交互输密码；`-SkipNeo4j` 可跳过 Neo4j 同步）， +再手动执行上表后 4 个 SQL（agent 库建表 ×2 / seed-aml-list / prepare_risk_demo）。 + +**重灌完成标志**：`seed-aml-list.sql` 输出 8 条名单；`prepare_risk_demo.sql` 输出 7 行测评 +（6 行 90 天、CUST-1004 保持过期）；`sync_advisor_rel` 输出 upserted 28 rows。 + +## 3. 启动与入口 + +```bash +uvicorn app.main:app --reload # Swagger: http://127.0.0.1:8000/docs +python scripts/demo/subscribe_alerts.py # 另开终端：订阅预警推送（Ctrl+C 退出） +``` + +debug 鉴权头（dev 过渡，T-01 后换 JWT）：`X-Debug-Role` + `X-Debug-Actor`。 +走查角色：`risk_officer`(STAFF-90001) / `compliance`(STAFF-40001) / `customer`(CUST-1001/CUST-1002) +/ `advisor`(STAFF-10087，名下无 CUST-3001) / `risk_demo`(STAFF-DEMO，simulate 白名单)。 + +## 4. 验收走查（PRD §8 · A-6 归 M3 对话线） + +Swagger 逐条执行；预期响应与库内核验点如下（`trade_id` 以实际响应为准）。 + +| # | 操作（Swagger） | 预期 | +| --- | --- | --- | +| A-1 | `POST /api/simulate/trade`，customer 头 CUST-1001，body：CUST-1001 / PROD-161725 / subscribe / 10000 | `blocked=true`，`block_reason` 含 SUIT-001；库核验：`core_trade` 无该 trade_id；`risk_suitability_log.is_blocked=1`；`risk_alert`（trade_id 定位）suitability/pending；audit `suitability_blocked`（agent_type=platform）；订阅端收到推送且 trace 与响应头一致 | +| A-2 | 同上，risk_demo 头，CUST-4001 / PROD-161725 / subscribe / 20000 | `blocked=true`，reasons 含 SUIT-006（≥70 按 C3）与 SUIT-003，**无 SUIT-008** | +| A-3 | 同上，CUST-3001 / PROD-510300 / subscribe / 500000 | `blocked=false`，`triggered_rules=["RISK-001","RISK-002"]`，预警单 score=70/pending；`core_trade` 落库；audit `trade_accepted`；**订阅端实时打印推送** | +| A-4 | 同上，CUST-9527 / PROD-510300 / subscribe / 1000，连发 4 笔 | 第 3 笔触发 RISK-003；第 4 笔 `alert_ids` 与第 3 笔相同（并入同单），该客户 freq 类单数=1，payload.events 长度=2 | +| A-5 | 同上，CUST-1002 / PROD-005828 / subscribe / 100 | `aml_hit=true`；aml 独立单 score=95（按本笔 trade_id 定位）；L3 `monitor_tier=high`；compliance 头 `GET /api/risk/alerts` 仅见 aml 单且含该单；`core_customer.is_active` 仍=1（不冻户） | +| A-7 | `POST /api/risk/alerts/{A-3的alert_id}/handle`，officer 头，`confirmed_suspicious` → 再处置一次 → 换 compliance 头再试 | 200（状态 confirmed_suspicious，handler_id=STAFF-90001）→ 409 STATE_CONFLICT → 403；audit 留痕（alert_handle/alert_handled） | +| A-9 | `POST /api/risk/suitability/check`：customer CUST-1002 查 CUST-3001；advisor STAFF-10087 查 CUST-3001 | 403 `AUTH_403_NOT_OWNER` / `AUTH_403_NOT_ASSIGNED`；audit `event_type=authz, decision=forbidden` 各 1 条 | + +**B9a 脚本演示（运维位）**： + +```bash +python scripts/demo/rebuild_alerts.py # 补偿重放：rebuilt → 再跑 skipped → 不存在 exit 1 +``` + +## 5. 核查单（B6/B7/B8 挂账收口） + +| # | 挂账来源 | 检查项 | 结论（2026-09-07 走查） | +| --- | --- | --- | --- | +| ① | B6 复审 P3-7 | 预警类 API（GET alerts / POST handle / POST aml/scan）响应体含固定 `disclaimer`「本预警由系统自动生成，最终判定需经风控专员人工审核。」 | ✅ 已实现并真服务验证 | +| ② | B6 评审 P3-6 | aml/scan 幂等：同客户当日已有 aml 单（含交易触发/已处置）不重复出单，响应 `skipped_existing` 指向既有单 | ✅ 已实现并真服务验证 | +| ③ | B6 复审遗漏⑤ | 真服务手测（区别于 TestClient 等价验证）：uvicorn 真库走查 A-1~A-9 全量 | ✅ 本 SOP §4 即该手测 | +| ④ | B6 复审观察③ | analyst 角色台账 fail-closed：`GET /api/risk/alerts` → 403（risk:alert:read 权限扩展待 Wave 3 分析 Agent 时定矩阵） | ✅ 403 保持 | +| ⑤ | B7 复审 P3-3 | 生产/演示机误配检查：本机 `.env` `APP_ENV=development`（演示机允许 debug 头）；`app_env != development` 时启动期拒绝 debug 依赖（单测覆盖），部署非 dev 机时必须核对 | ✅ 本机=development 已确认 | +| ⑥ | B8 复审 P2-1 | 集成测试对演示库当日状态的隐式耦合 → 已加前置断言（A-1/A-3/A-4 清洁检查 fail+提示、A-5 按 trade_id 定位），代码随 B9b 落地 | ✅ 已落地；交叉数据在场时 pytest 显式 fail 提示重灌 | + +## 6. 走查后状态与注意事项 + +- 走查产生的交易/预警/审计/L3 **保留在演示库**（演示痕迹即演示资产）；再次演示前回到 §2 重灌。 +- **重灌后才能跑 pytest 集成测试**：核查单⑥前置断言会对当日演示数据显式 fail（防交叉失真）。 +- B9a subscribe 窗口未开时推送不丢失告警——Redis Pub/Sub 无持久化，演示时先开订阅再发交易。 +- A-6（对话线验收）归 M3（T-01/T-03/T-07 就绪后补）。 diff --git a/tests/test_aml_service.py b/tests/test_aml_service.py index eb400b7..f2a1423 100644 --- a/tests/test_aml_service.py +++ b/tests/test_aml_service.py @@ -171,7 +171,50 @@ def test_scan_all_no_hit_creates_nothing(env): with core._engine.begin() as conn: conn.execute(text("UPDATE risk_aml_list SET is_active = 0")) summary = scan_all(core_ro=core, risk_repo=repo) - assert summary == {"scanned": 4, "hit_customers": 0, "alerts": []} + assert summary == {"scanned": 4, "hit_customers": 0, "alerts": [], "skipped_existing": []} with core._engine.connect() as conn: assert conn.execute(text("SELECT COUNT(*) FROM risk_alert")).scalar_one() == 0 assert pub.messages == [] + + +def test_scan_all_idempotent_same_day(env): + """B9b 核查单②（B6 评审 P3-6）：同日重扫不重复出单，重复点击防护。""" + core, repo, pub = env + first = scan_all(core_ro=core, risk_repo=repo) + assert len(first["alerts"]) == 2 and first["skipped_existing"] == [] + second = scan_all(core_ro=core, risk_repo=repo) + assert second["alerts"] == [] + assert sorted(second["skipped_existing"]) == sorted(first["alerts"]) + assert len(pub.messages) == 2 # 重扫无新推送（出单时的推送各一次） + + +def test_scan_skips_customer_already_alerted_by_trade(env): + """当日交易触发已出 aml 单的客户，scan 不再重复出单（同日命中留痕一次）。""" + from decimal import Decimal + + from app.service.risk.engine import process_trade_event + + core, repo, pub = env + with core._engine.begin() as conn: + conn.execute( + text( + "INSERT INTO core_trade (trade_id, customer_id, product_id, trade_type," + " amount, trade_status, traded_at)" + " VALUES ('T-SCAN-1', 'C2', 'P1', 'subscribe', 1000, 'confirmed', :at)" + ), + {"at": datetime(2026, 9, 7, 10, 0, 0)}, + ) + result = process_trade_event( + { + "trade_id": "T-SCAN-1", "customer_id": "C2", "product_id": "P1", + "trade_type": "subscribe", "amount": Decimal("1000"), + "trade_status": "confirmed", "traded_at": datetime(2026, 9, 7, 10, 0, 0), + }, + core_ro=core, risk_repo=repo, + ) + assert result["aml_hit"] is True # 交易触发已出 C2 的 aml 单 + summary = scan_all(core_ro=core, risk_repo=repo) + assert len(summary["alerts"]) == 1 # 仅 C1（无当日 aml 单） + assert summary["skipped_existing"] == [ + aid for aid in result["alert_ids"] if repo.get_alert(aid)["alert_type"] == "aml" + ] diff --git a/tests/test_integration_risk.py b/tests/test_integration_risk.py index 3b5e39c..bab3e21 100644 --- a/tests/test_integration_risk.py +++ b/tests/test_integration_risk.py @@ -91,12 +91,71 @@ def _audit_by_trade(env, trade_id, event_type=None): return _one(env, sql, et=event_type or "trade_request", tid=trade_id) +# ---------- 核查单⑥前置断言（B8 复审 P2-1：演示与 pytest 同日交叉防护） ---------- +# 各用例对演示库当日状态有隐式假设；交叉执行时不静默产出错误结论，而是 fail 并提示处理。 + + +def _fail_if_dirty(cond_row, message): + assert cond_row is None, f"演示/测试同日交叉防护（核查单⑥）：{message}" + + +def _assert_a1_clean(env): + """A-1 假定当日无 CUST-1001×PROD-161725 pending suitability 单： + 同日交叉时本笔会并入旧单（新 trade_id 不在 trade_id 列），断言必炸。""" + _fail_if_dirty( + _one( + env, + "SELECT alert_id FROM risk_alert WHERE customer_id = 'CUST-1001'" + " AND alert_type = 'suitability' AND status = 'pending_review'" + " AND payload LIKE '%PROD-161725%' AND created_at >= CURDATE()", + ), + "CUST-1001 今日已有 PROD-161725 pending suitability 单；" + "先按演示 SOP 重灌或清理当日演示单后再跑集成测试", + ) + + +def _assert_a3_clean(env): + """A-3 假定 CUST-3001 当日无 confirmed 交易（RISK-003 频次口径）且无当日 + pending 事件单（聚合锚点：有 pending 单则本笔 append 进旧单，规则断言漂移）。""" + _fail_if_dirty( + _core_one( + env, + "SELECT trade_id FROM core_trade WHERE customer_id = 'CUST-3001'" + " AND trade_status = 'confirmed' AND traded_at >= CURDATE() LIMIT 1", + ), + "CUST-3001 今日已有 confirmed 交易（影响 RISK-002 累计/RISK-003 频次口径）", + ) + _fail_if_dirty( + _one( + env, + "SELECT alert_id FROM risk_alert WHERE customer_id = 'CUST-3001'" + " AND alert_type IN ('large_amount','freq_trade','pattern')" + " AND status = 'pending_review' AND created_at >= CURDATE()", + ), + "CUST-3001 今日已有 pending 事件类预警单（本笔会并入旧单而非新建）", + ) + + +def _assert_a4_clean(env): + """A-4 假定 CUST-9527 当日无 confirmed 交易：同日交叉时首笔即可能触发规则， + 「第 3 笔才触发 / events==2」断言失真。""" + _fail_if_dirty( + _core_one( + env, + "SELECT trade_id FROM core_trade WHERE customer_id = 'CUST-9527'" + " AND trade_status = 'confirmed' AND traded_at >= CURDATE() LIMIT 1", + ), + "CUST-9527 今日已有 confirmed 交易（A-4 频次计数失真）", + ) + + # ---------- A-1：C1 申购 R4 → SUIT-001 阻断（含 trace 一致性） ---------- def test_a1_suit001_blocked_with_trace_consistency(iclient, risk_demo_env): c, fake = iclient env = risk_demo_env + _assert_a1_clean(env) trace = f"trc-integ-{uuid4().hex[:12]}" r = c.post( "/api/simulate/trade", @@ -178,6 +237,7 @@ def test_a2_age70_cap_suit006_no_suit008(iclient, risk_demo_env): def test_a3_large_amount_alert_and_publish(iclient, risk_demo_env): c, fake = iclient env = risk_demo_env + _assert_a3_clean(env) r = c.post( "/api/simulate/trade", json=_trade("CUST-3001", "PROD-510300", "subscribe", 500000), @@ -213,6 +273,7 @@ def test_a3_large_amount_alert_and_publish(iclient, risk_demo_env): def test_a4_freq_trade_merged_into_same_alert(iclient, risk_demo_env): c, _ = iclient env = risk_demo_env + _assert_a4_clean(env) alert_id = None for _ in range(3): r = c.post( @@ -263,9 +324,12 @@ def test_a5_aml_hit_independent_alert(iclient, risk_demo_env): assert body["blocked"] is False and body["aml_hit"] is True trade_id = body["trade_id"] + # aml 单按本测试 trade_id 定位（核查单⑥：同日旧 aml 单不得干扰断言） alert = _one( env, - "SELECT * FROM risk_alert WHERE customer_id = 'CUST-1002' AND alert_type = 'aml'", + "SELECT * FROM risk_alert WHERE customer_id = 'CUST-1002' AND alert_type = 'aml'" + " AND trade_id = :t", + t=trade_id, ) assert alert is not None assert alert["risk_score"] == 95 and alert["status"] == "pending_review" diff --git a/tests/test_risk_api.py b/tests/test_risk_api.py index 71d5efe..057c43c 100644 --- a/tests/test_risk_api.py +++ b/tests/test_risk_api.py @@ -300,8 +300,10 @@ def test_aml_scan_officer_only_with_audit(client, env): r = client.post("/api/risk/aml/scan", headers=OFFICER) assert r.status_code == 200 body = r.json() - assert body == {"scanned": 2, "hit_customers": 1, "alerts": body["alerts"]} # CUST-3001 命中 + assert body["scanned"] == 2 and body["hit_customers"] == 1 # CUST-3001 命中 assert len(body["alerts"]) == 1 + assert body["skipped_existing"] == [] # 首扫无既有单 + assert body["disclaimer"] == risk_api.ALERT_DISCLAIMER # B9b 核查单① l3 = repo.get_l3("CUST-3001") assert l3["monitor_tier"] == "high" # scan 命中标记 L3 # L3 写侧缓存 DEL 钩子（B7 挂账②）：命中客户 upsert 后失效读缓存 @@ -312,6 +314,31 @@ def test_aml_scan_officer_only_with_audit(client, env): assert client.post("/api/risk/aml/scan", headers=headers).status_code == 403 +def test_aml_scan_idempotent_same_day(client, env): + """B9b 核查单②（B6 评审 P3-6）：同日重扫不重复出单，重复点击防护。""" + repo, engine = env + first = client.post("/api/risk/aml/scan", headers=OFFICER).json() + second = client.post("/api/risk/aml/scan", headers=OFFICER).json() + assert second["alerts"] == [] + assert second["skipped_existing"] == first["alerts"] + # aml 单总数不变（env 预置 ALT-A1 一张 + scan 新出一张；重扫零新增） + assert _counts(engine, "risk_alert", "alert_type='aml'") == 2 + + +def test_alert_disclaimer_on_alert_apis(client): + """B9b 核查单①（B6 复审 P3-7）：预警类 API 响应体固定 disclaimer（规则表 §5）。""" + r = client.get("/api/risk/alerts", headers=OFFICER) + body = r.json() + assert r.status_code == 200 and body["disclaimer"] == risk_api.ALERT_DISCLAIMER + pending = next(i for i in body["items"] if i["status"] == "pending_review") + r = client.post( + f"/api/risk/alerts/{pending['alert_id']}/handle", + json={"handler_result": "confirmed_normal"}, + headers=OFFICER, + ) + assert r.status_code == 200 and r.json()["disclaimer"] == risk_api.ALERT_DISCLAIMER + + # ---------- 非 dev 环境拒绝 debug 鉴权 ----------