From 4eb8b462cd116620c1d87ca91d01a2eada4aa865 Mon Sep 17 00:00:00 2001 From: YUAN Date: Sun, 6 Sep 2026 15:15:51 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20risk=5Frepository=20=E9=A3=8E=E6=8E=A7?= =?UTF-8?q?=E5=9B=9B=E8=A1=A8=E8=AF=BB=E5=86=99(A3)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- app/repository/risk_repository.py | 325 ++++++++++++++++++++++++++++++ tests/test_risk_repository.py | 237 ++++++++++++++++++++++ 2 files changed, 562 insertions(+) create mode 100644 app/repository/risk_repository.py create mode 100644 tests/test_risk_repository.py diff --git a/app/repository/risk_repository.py b/app/repository/risk_repository.py new file mode 100644 index 0000000..b30457e --- /dev/null +++ b/app/repository/risk_repository.py @@ -0,0 +1,325 @@ +"""jinrong_agent 风控四表读写（risk_alert / risk_suitability_log / customer_profile_l3 / risk_aml_list）。 + +仅服务风控模块（PRD v1.0 §5.1 权限总表）；不含业务判定逻辑（聚合合并/状态机入参 +由 service 层决定后传入）。JSON 列以 dict 交互，内部序列化。 +""" + +from __future__ import annotations + +import json +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 + +EVENT_ALERT_TYPES = ("large_amount", "freq_trade", "pattern") + + +class RiskRepository: + """风控产出表读写；不修改表结构（PRD 冻结约束）。""" + + 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_database}?charset=utf8mb4" + ) + return create_engine(url, pool_pre_ping=True) + + # ---------- risk_alert ---------- + + def find_pending_event_alert(self, customer_id: str, day_start: datetime) -> dict | None: + """当日该客户的事件类 pending 单（聚合锚点；PRD FR-4 同日一张）。""" + return self._find_pending(customer_id, day_start, types=EVENT_ALERT_TYPES) + + def find_pending_suitability_alert( + self, customer_id: str, product_id: str, day_start: datetime + ) -> dict | None: + row = self._find_pending(customer_id, day_start, types=("suitability",)) + if row and row["payload"].get("product_id") == product_id: + return row + return None + + def _find_pending( + self, customer_id: str, day_start: datetime, types: tuple[str, ...] + ) -> dict | None: + placeholders = ", ".join(f":t{i}" for i in range(len(types))) + params: dict[str, Any] = {"cid": customer_id, "day_start": day_start} + for i, t in enumerate(types): + params[f"t{i}"] = t + sql = text( + f""" + SELECT * FROM risk_alert + WHERE customer_id = :cid + AND status = 'pending_review' + AND alert_type IN ({placeholders}) + AND created_at >= :day_start + ORDER BY created_at DESC + LIMIT 1 + """ + ) + with self._engine.connect() as conn: + row = conn.execute(sql, params).mappings().first() + return self._parse_alert(dict(row)) if row else None + + def insert_alert(self, alert: dict[str, Any]) -> None: + sql = text( + """ + INSERT INTO risk_alert + (alert_id, trace_id, customer_id, trade_id, alert_type, triggered_rules, + risk_score, status, payload) + VALUES (:alert_id, :trace_id, :customer_id, :trade_id, :alert_type, :triggered_rules, + :risk_score, :status, :payload) + """ + ) + with self._engine.begin() as conn: + conn.execute(sql, self._dump_alert(alert)) + + def append_alert_event( + self, + alert_id: str, + event: dict[str, Any], + triggered_rules: list[str], + risk_score: int, + alert_type: str, + ) -> None: + """读改写：追加 payload.events、合并 triggered_rules、risk_score 取 max、alert_type 更新。""" + with self._engine.begin() as conn: + row = conn.execute( + text("SELECT payload, triggered_rules, risk_score FROM risk_alert WHERE alert_id = :aid"), + {"aid": alert_id}, + ).mappings().first() + if row is None: + raise ValueError(f"alert not found: {alert_id}") + payload = json.loads(row["payload"]) if isinstance(row["payload"], str) else row["payload"] + events = payload.setdefault("events", []) + events.append(event) + old_rules = ( + json.loads(row["triggered_rules"]) + if isinstance(row["triggered_rules"], str) + else row["triggered_rules"] + ) + merged_rules = sorted(set(old_rules) | set(triggered_rules)) + old_score = int(row["risk_score"] or 0) + new_score = max(old_score, int(risk_score)) + conn.execute( + text( + """ + UPDATE risk_alert + SET payload = :payload, triggered_rules = :rules, + risk_score = :score, alert_type = :atype + WHERE alert_id = :aid + """ + ), + { + "payload": json.dumps(payload, ensure_ascii=False), + "rules": json.dumps(merged_rules), + "score": new_score, + "atype": alert_type, + "aid": alert_id, + }, + ) + + def get_alert(self, alert_id: str) -> dict | None: + with self._engine.connect() as conn: + row = conn.execute( + text("SELECT * FROM risk_alert WHERE alert_id = :aid"), {"aid": alert_id} + ).mappings().first() + return self._parse_alert(dict(row)) if row else None + + def list_alerts( + self, + status: str | None = None, + alert_type: str | None = None, + customer_id: str | None = None, + start: datetime | None = None, + end: datetime | None = None, + page: int = 1, + page_size: int = 20, + ) -> tuple[list[dict], int]: + """预警台账分页查询（PRD FR-4）；过滤参数全部可选。""" + where = ["1=1"] + params: dict[str, Any] = {} + if status: + where.append("status = :status") + params["status"] = status + if alert_type: + where.append("alert_type = :atype") + params["atype"] = alert_type + if customer_id: + where.append("customer_id = :cid") + params["cid"] = customer_id + if start: + where.append("created_at >= :start") + params["start"] = start + if end: + where.append("created_at < :end") + params["end"] = end + where_sql = " AND ".join(where) + with self._engine.connect() as conn: + total = conn.execute( + text(f"SELECT COUNT(*) FROM risk_alert WHERE {where_sql}"), params + ).scalar_one() + rows = conn.execute( + text( + f""" + SELECT * FROM risk_alert WHERE {where_sql} + ORDER BY created_at DESC + LIMIT :lim OFFSET :off + """ + ), + {**params, "lim": page_size, "off": (page - 1) * page_size}, + ).mappings() + return [self._parse_alert(dict(r)) for r in rows], int(total) + + def update_alert_status( + self, alert_id: str, handler_result: str, handler_id: str, handler_comment: str | None + ) -> bool: + """状态机：仅 pending_review 可处置（PRD FR-4）；返回是否发生变更。""" + sql = text( + """ + UPDATE risk_alert + SET status = :result, handler_id = :hid, handler_result = :result, + handler_comment = :comment, handled_at = :handled_at + WHERE alert_id = :aid AND status = 'pending_review' + """ + ) + with self._engine.begin() as conn: + res = conn.execute( + sql, + { + "result": handler_result, + "hid": handler_id, + "comment": handler_comment, + "handled_at": datetime.now(), + "aid": alert_id, + }, + ) + return res.rowcount == 1 + + @staticmethod + def _dump_alert(alert: dict[str, Any]) -> dict[str, Any]: + out = dict(alert) + out["triggered_rules"] = json.dumps(alert["triggered_rules"]) + out["payload"] = json.dumps(alert["payload"], ensure_ascii=False, default=str) + return out + + @staticmethod + def _parse_alert(row: dict[str, Any]) -> dict[str, Any]: + row["triggered_rules"] = json.loads(row["triggered_rules"]) + row["payload"] = json.loads(row["payload"]) if isinstance(row["payload"], str) else row["payload"] + return row + + # ---------- risk_suitability_log ---------- + + def insert_suitability_log(self, log: dict[str, Any]) -> None: + sql = text( + """ + INSERT INTO risk_suitability_log + (trace_id, customer_id, product_id, customer_risk_level, product_risk_level, + is_matched, is_blocked, block_reason, request_ref, profile_l1_version) + VALUES (:trace_id, :customer_id, :product_id, :customer_risk_level, :product_risk_level, + :is_matched, :is_blocked, :block_reason, :request_ref, :profile_l1_version) + """ + ) + with self._engine.begin() as conn: + conn.execute(sql, log) + + # ---------- customer_profile_l3 ---------- + + def get_l3(self, customer_id: str) -> dict | None: + with self._engine.connect() as conn: + row = conn.execute( + text("SELECT * FROM customer_profile_l3 WHERE customer_id = :cid"), {"cid": customer_id} + ).mappings().first() + if not row: + return None + data = dict(row) + for key in ("score_dimensions", "monitor_tags"): + if isinstance(data.get(key), str): + data[key] = json.loads(data[key]) + return data + + def insert_l3( + self, + customer_id: str, + monitor_tier: str, + risk_score: int | None, + score_dimensions: dict | list | None, + monitor_tags: list[str], + last_alert_id: str | None, + computed_at: datetime, + ) -> None: + """新行插入；行已存在时由 service 层读后改走 update_l3（合并逻辑在 service）。""" + sql = text( + """ + INSERT INTO customer_profile_l3 + (customer_id, monitor_tier, risk_score, score_dimensions, monitor_tags, + last_alert_id, computed_at) + VALUES (:cid, :tier, :score, :dims, :tags, :last_alert, :computed_at) + """ + ) + with self._engine.begin() as conn: + conn.execute( + sql, + { + "cid": customer_id, + "tier": monitor_tier, + "score": risk_score, + "dims": json.dumps(score_dimensions or {}, ensure_ascii=False), + "tags": json.dumps(monitor_tags, ensure_ascii=False), + "last_alert": last_alert_id, + "computed_at": computed_at, + }, + ) + + def update_l3( + self, + customer_id: str, + monitor_tier: str, + risk_score: int | None, + score_dimensions: dict | list | None, + monitor_tags: list[str], + last_alert_id: str | None, + computed_at: datetime, + ) -> None: + """整行更新（service 层完成最高档/tags 合并后调用；computed_at NOT NULL 必传）。""" + sql = text( + """ + UPDATE customer_profile_l3 + SET monitor_tier = :tier, risk_score = :score, score_dimensions = :dims, + monitor_tags = :tags, last_alert_id = :last_alert, computed_at = :computed_at + WHERE customer_id = :cid + """ + ) + with self._engine.begin() as conn: + conn.execute( + sql, + { + "cid": customer_id, + "tier": monitor_tier, + "score": risk_score, + "dims": json.dumps(score_dimensions or {}, ensure_ascii=False), + "tags": json.dumps(monitor_tags, ensure_ascii=False), + "last_alert": last_alert_id, + "computed_at": computed_at, + }, + ) + + # ---------- risk_aml_list ---------- + + def list_active_aml_entries(self) -> list[dict]: + with self._engine.connect() as conn: + rows = conn.execute( + text("SELECT * FROM risk_aml_list WHERE is_active = 1") + ).mappings() + return [dict(r) for r in rows] diff --git a/tests/test_risk_repository.py b/tests/test_risk_repository.py new file mode 100644 index 0000000..61a0f3c --- /dev/null +++ b/tests/test_risk_repository.py @@ -0,0 +1,237 @@ +"""risk_repository 单测（A3 · sqlite 内存库；JSON/ENUM 列用 TEXT 兼容）。 + +upsert_l3 的 ON DUPLICATE KEY UPDATE 更新分支为 MySQL 方言，sqlite 仅测插入路径， +更新分支由 B9b 演示走查做手工 SQL 对照。 +""" + +import json +from datetime import datetime +from decimal import Decimal + +import pytest +from sqlalchemy import create_engine, text + +from app.repository.risk_repository import RiskRepository + + +@pytest.fixture() +def repo(): + engine = create_engine("sqlite:///:memory:") + with engine.begin() as conn: + conn.execute( + text( + """ + 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 + ) + """ + ) + ) + conn.execute( + text( + """ + CREATE TABLE customer_profile_l3 ( + customer_id VARCHAR(64) PRIMARY KEY, + monitor_tier VARCHAR(8) DEFAULT 'normal', + risk_score INTEGER, + score_dimensions TEXT, + monitor_tags TEXT, + last_alert_id VARCHAR(64), + computed_at TIMESTAMP, + updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP + ) + """ + ) + ) + conn.execute( + text( + """ + CREATE TABLE risk_aml_list ( + id INTEGER PRIMARY KEY, + list_id VARCHAR(64), + list_type VARCHAR(8), + full_name VARCHAR(128), + match_threshold NUMERIC(3,2) DEFAULT 0.85, + is_active INTEGER DEFAULT 1 + ) + """ + ) + ) + + def make_alert(alert_id, alert_type="large_amount", cid="C1", rules=None, score=70): + return { + "alert_id": alert_id, + "trace_id": "trc-test", + "customer_id": cid, + "trade_id": "TRD-1", + "alert_type": alert_type, + "triggered_rules": rules or ["RISK-001"], + "risk_score": score, + "status": "pending_review", + "payload": {"product_id": "P1", "events": []}, + } + + yield RiskRepository(engine=engine), engine, make_alert + engine.dispose() + + +def test_insert_and_get_alert(repo): + r, _, make_alert = repo + r.insert_alert(make_alert("ALT-1")) + got = r.get_alert("ALT-1") + assert got["alert_type"] == "large_amount" + assert got["triggered_rules"] == ["RISK-001"] + assert got["payload"]["product_id"] == "P1" + assert got["status"] == "pending_review" + + +def test_find_pending_event_alert(repo): + r, _, make_alert = repo + r.insert_alert(make_alert("ALT-E", alert_type="large_amount")) + found = r.find_pending_event_alert("C1", datetime(2026, 1, 1)) + assert found is not None and found["alert_id"] == "ALT-E" + # suitability 类型不在事件类查询范围 + r.insert_alert(make_alert("ALT-S", alert_type="suitability")) + assert r.find_pending_event_alert("C1", datetime(2026, 1, 1))["alert_id"] == "ALT-E" + + +def test_find_pending_suitability_alert_matches_product(repo): + r, _, make_alert = repo + r.insert_alert(make_alert("ALT-S", alert_type="suitability")) + assert r.find_pending_suitability_alert("C1", "P1", datetime(2026, 1, 1)) is not None + assert r.find_pending_suitability_alert("C1", "P2", datetime(2026, 1, 1)) is None + + +def test_append_alert_event_merges(repo): + r, _, make_alert = repo + r.insert_alert(make_alert("ALT-1", rules=["RISK-001"], score=70)) + r.append_alert_event( + "ALT-1", + event={"trade_id": "TRD-2", "amount": "600000"}, + triggered_rules=["RISK-002", "RISK-001"], + risk_score=50, + alert_type="large_amount", + ) + got = r.get_alert("ALT-1") + assert got["triggered_rules"] == ["RISK-001", "RISK-002"] # 合并未重复 + assert got["risk_score"] == 70 # 取 max，不被 50 降 + assert len(got["payload"]["events"]) == 1 + assert got["payload"]["events"][0]["trade_id"] == "TRD-2" + # 再追加一笔，score 升到 90 + r.append_alert_event("ALT-1", {"trade_id": "TRD-3"}, ["RISK-005"], 80, "pattern") + got = r.get_alert("ALT-1") + assert got["risk_score"] == 80 + assert got["alert_type"] == "pattern" # 类型随最高分规则更新 + assert len(got["payload"]["events"]) == 2 + + +def test_append_missing_alert_raises(repo): + r, _, _ = repo + with pytest.raises(ValueError): + r.append_alert_event("NOPE", {}, ["RISK-001"], 70, "large_amount") + + +def test_update_alert_status_state_machine(repo): + r, _, make_alert = repo + r.insert_alert(make_alert("ALT-1")) + assert r.update_alert_status("ALT-1", "confirmed_suspicious", "STAFF-30001", "确认可疑") is True + got = r.get_alert("ALT-1") + assert got["status"] == "confirmed_suspicious" + assert got["handler_id"] == "STAFF-30001" + assert got["handled_at"] is not None + # 已处置不可再改（状态机只允许 pending_review → 三态） + assert r.update_alert_status("ALT-1", "confirmed_normal", "STAFF-30002", None) is False + + +def test_list_alerts_filters_and_pages(repo): + r, _, make_alert = repo + for i in range(1, 4): + r.insert_alert(make_alert(f"ALT-{i}", alert_type="large_amount" if i < 3 else "aml")) + items, total = r.list_alerts(alert_type="large_amount", page=1, page_size=20) + assert total == 2 and len(items) == 2 + items, total = r.list_alerts(alert_type="aml") + assert total == 1 and items[0]["alert_id"] == "ALT-3" + items, total = r.list_alerts(customer_id="C1") + assert total == 3 + + +def test_insert_suitability_log(repo): + r, engine, _ = repo + with engine.begin() as conn: + conn.execute( + text( + "CREATE TABLE risk_suitability_log (" + "id INTEGER PRIMARY KEY, trace_id VARCHAR(64), customer_id VARCHAR(64)," + " product_id VARCHAR(64), customer_risk_level VARCHAR(2)," + " product_risk_level VARCHAR(2), is_matched INTEGER, is_blocked INTEGER," + " block_reason VARCHAR(512), request_ref VARCHAR(64), profile_l1_version INTEGER," + " created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP)" + ) + ) + log = { + "trace_id": "trc-x", + "customer_id": "C1", + "product_id": "P1", + "customer_risk_level": "C1", + "product_risk_level": "R4", + "is_matched": 0, + "is_blocked": 1, + "block_reason": "SUIT-001", + "request_ref": None, + "profile_l1_version": None, + } + r.insert_suitability_log(log) + row = engine.connect().execute( + text("SELECT customer_risk_level, is_blocked FROM risk_suitability_log") + ).first() + assert row[0] == "C1" and row[1] == 1 + + +def test_l3_insert_and_get(repo): + r, _, _ = repo + r.insert_l3( + "C1", "high", 95, {"aml": True}, ["aml_hit_pending_review"], "ALT-AML-1", datetime(2026, 9, 6) + ) + got = r.get_l3("C1") + assert got["monitor_tier"] == "high" + assert got["monitor_tags"] == ["aml_hit_pending_review"] + assert got["score_dimensions"] == {"aml": True} + + +def test_l3_update_after_insert(repo): + r, _, _ = repo + r.insert_l3("C1", "watch", 80, {}, ["pattern"], "ALT-1", datetime(2026, 9, 6)) + r.update_l3("C1", "high", 95, {"aml": True}, ["pattern", "aml_hit_pending_review"], "ALT-2", + datetime(2026, 9, 6, 12)) + got = r.get_l3("C1") + assert got["monitor_tier"] == "high" + assert got["monitor_tags"] == ["pattern", "aml_hit_pending_review"] + assert got["last_alert_id"] == "ALT-2" + + +def test_list_active_aml_entries(repo): + r, engine, _ = repo + with engine.begin() as conn: + conn.execute( + text( + "INSERT INTO risk_aml_list (list_id, list_type, full_name, is_active) VALUES" + "('AML-1', 'sanction', '客户·赵**', 1)," + "('AML-2', 'pep', '客户·已停**', 0)" + ) + ) + entries = r.list_active_aml_entries() + assert len(entries) == 1 + assert entries[0]["full_name"] == "客户·赵**"