feat: risk_repository 风控四表读写(A3)

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]

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"] == "客户·赵**"