From e8e5f7e98166d2cdbc7a61b374a1746a2de8de14 Mon Sep 17 00:00:00 2001 From: YUAN Date: Sun, 6 Sep 2026 17:19:54 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20AML=20=E5=90=8D=E5=8D=95=E5=8C=B9?= =?UTF-8?q?=E9=85=8D/=E5=85=A8=E9=87=8F=E6=89=AB=E6=8F=8F=20+=20=E4=BA=8B?= =?UTF-8?q?=E4=BB=B6=E5=BC=95=E6=93=8E=E7=BC=96=E6=8E=92=20+=20R-05=20scor?= =?UTF-8?q?ing=20=E5=8D=A0=E4=BD=8D(B4)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- app/repository/core_ro.py | 8 + app/service/risk/aml_service.py | 103 ++++++++++++ app/service/risk/engine.py | 100 ++++++++++++ app/service/risk/scoring.py | 18 +++ tests/test_aml_service.py | 228 ++++++++++++++++++++++++++ tests/test_risk_engine.py | 278 ++++++++++++++++++++++++++++++++ 6 files changed, 735 insertions(+) create mode 100644 app/service/risk/aml_service.py create mode 100644 app/service/risk/engine.py create mode 100644 app/service/risk/scoring.py create mode 100644 tests/test_aml_service.py create mode 100644 tests/test_risk_engine.py diff --git a/app/repository/core_ro.py b/app/repository/core_ro.py index e058150..f01f2e8 100644 --- a/app/repository/core_ro.py +++ b/app/repository/core_ro.py @@ -119,6 +119,14 @@ class CoreReadOnlyRepository: row = conn.execute(sql, {"pid": product_id}).mappings().first() return dict(row) if row else None + def list_active_customers(self) -> list[dict[str, Any]]: + """全量在册客户（id + display_name；AML scan_all 全量扫描用，仅 SELECT）。""" + sql = text( + "SELECT customer_id, display_name FROM core_customer WHERE is_active = 1" + ) + with self._engine.connect() as conn: + return [dict(r) for r in conn.execute(sql).mappings()] + def list_customers_by_advisor(self, advisor_id: str) -> list[str]: sql = text( """ diff --git a/app/service/risk/aml_service.py b/app/service/risk/aml_service.py new file mode 100644 index 0000000..6c64edc --- /dev/null +++ b/app/service/risk/aml_service.py @@ -0,0 +1,103 @@ +"""AML 名单匹配（B4 · PRD FR-5 / AML-001~002）。 + +一期降级口径：仅 display_name 归一化（去空白 + 大小写折叠）+ difflib 相似度； +证件/银行卡匹配待 Core 提供证件数据后启用（表 id_no/bank_card_no 已预留）。 +阈值：名单行 match_threshold 优先（表默认 0.85），缺省回落 settings.risk_aml_default_threshold。 +命中动作（独立 aml 预警单 + L3 high 标记）由调用方编排：engine（交易触发）/ +scan_all（手动全量扫描）；不冻结、不自动上报（附表 §2 行为边界）。 +""" + +from __future__ import annotations + +import difflib +import logging +from typing import Any + +from app.config.settings import settings +from app.repository.core_ro import CoreReadOnlyRepository +from app.repository.risk_repository import RiskRepository +from app.service.risk.alert_service import record_aml_alert +from app.service.risk.profile_l3 import upsert_profile_l3 + +logger = logging.getLogger(__name__) + + +def normalize_name(name: str) -> str: + """去全部空白 + 大小写折叠（PRD FR-5 一期归一化口径）。""" + return "".join(name.split()).casefold() + + +def similarity(a: str, b: str) -> float: + """归一化后的相似度（difflib SequenceMatcher ratio ∈ [0, 1]）。""" + return difflib.SequenceMatcher(None, a, b).ratio() + + +def match_name( + customer_name: str, entries: list[dict[str, Any]] +) -> list[dict[str, Any]]: + """客户姓名 ↔ 名单条目匹配，返回命中明细（进预警 payload / 审计）。 + + entries 为 repo.list_active_aml_entries() 输出；ratio ≥ 行阈值即命中。 + """ + norm = normalize_name(customer_name) + hits: list[dict[str, Any]] = [] + for e in entries: + threshold = float(e.get("match_threshold") or settings.risk_aml_default_threshold) + ratio = similarity(norm, normalize_name(e["full_name"])) + if ratio >= threshold: + hits.append( + { + "list_id": e["list_id"], + "list_type": e["list_type"], + "matched_name": e["full_name"], + "similarity": round(ratio, 4), + "threshold": threshold, + "list_version": e["list_version"], + "source": e.get("source"), + } + ) + return hits + + +def match_customer( + customer_id: str, + core_ro: CoreReadOnlyRepository | None = None, + risk_repo: RiskRepository | None = None, +) -> list[dict[str, Any]]: + """按客户匹配活跃名单（客户不存在返回空，不抛错）。""" + core = core_ro or CoreReadOnlyRepository() + repo = risk_repo or RiskRepository() + l0 = core.get_customer_l0(customer_id) + if not l0: + return [] + return match_name(l0["display_name"], repo.list_active_aml_entries()) + + +def scan_all( + core_ro: CoreReadOnlyRepository | None = None, + risk_repo: RiskRepository | None = None, +) -> dict[str, Any]: + """手动全量扫描（PRD FR-5 触发时机 2，B6 接 POST /api/risk/aml/scan）。 + + 每个命中客户：一张 aml 独立预警单（matches 全量进 payload）+ L3 high 标记。 + """ + core = core_ro or CoreReadOnlyRepository() + repo = risk_repo or RiskRepository() + entries = repo.list_active_aml_entries() + customers = core.list_active_customers() + alerts: list[str] = [] + hit_customers = 0 + for c in customers: + hits = match_name(c["display_name"], entries) + if not hits: + continue + hit_customers += 1 + alert = record_aml_alert( + c["customer_id"], {"trigger": "scan", "matches": hits}, risk_repo=repo + ) + upsert_profile_l3( + 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} diff --git a/app/service/risk/engine.py b/app/service/risk/engine.py new file mode 100644 index 0000000..9afc749 --- /dev/null +++ b/app/service/risk/engine.py @@ -0,0 +1,100 @@ +"""风控事件引擎（B4 · 架构 §3.1 ④ / PRD FR-3）。 + +编排（网关 B5 在 core_trade 落库后**同步调用**，不用消息队列）： +当日流水上下文（core_ro）→ RISK-001~005 纯函数 → AML 姓名匹配 → 预警落库 +（alert_service：聚合/去重/审计/推送）→ L3 upsert（profile_l3）。 +审计 pass（未命中）与命中审计均由 alert_service 完成，本层不重复落审计。 + +RISK-004 窗口以 trade["traded_at"] 为事件时点（非墙钟 now）：rebuild_alerts +幂等重放可复现窗口判定，演示脚本不受执行时刻影响。 + +客户事件钩子 on_customer_created/on_customer_updated 为 FR-5 预留（本期 no-op， +模拟环境无开户流程）。 +""" + +from __future__ import annotations + +from datetime import datetime +from typing import Any + +from app.repository.core_ro import CoreReadOnlyRepository +from app.repository.risk_repository import RiskRepository +from app.service.risk.alert_service import record_aml_alert, record_trade_alerts +from app.service.risk.aml_service import match_customer +from app.service.risk.profile_l3 import upsert_profile_l3 +from app.service.risk.rules import RiskThresholds, run_rules + + +def _as_datetime(value: Any) -> datetime: + if isinstance(value, datetime): + return value + if isinstance(value, str): + return datetime.fromisoformat(value) + raise TypeError(f"traded_at must be datetime/str, got {type(value)!r}") + + +def _normalize_trades(trades: list[dict[str, Any]]) -> list[dict[str, Any]]: + """驱动差异防御：sqlite text 查询返回 str 时间，统一转 datetime（MySQL 驱动本就返回 datetime）。""" + for t in trades: + if isinstance(t.get("traded_at"), str): + t["traded_at"] = datetime.fromisoformat(t["traded_at"]) + return trades + + +def process_trade_event( + trade: dict[str, Any], + core_ro: CoreReadOnlyRepository | None = None, + risk_repo: RiskRepository | None = None, + thresholds: RiskThresholds | None = None, +) -> dict[str, Any]: + """处理一笔已落库交易（PRD FR-1 ②③b 之后）。 + + 返回 {"triggered_rules": [...], "alert_ids": [...], "aml_hit": bool}， + 网关据此拼装响应（FR-1 ⑤：blocked=false + trade_id + 触发规则列表）。 + """ + core = core_ro or CoreReadOnlyRepository() + repo = risk_repo or RiskRepository() + th = thresholds or RiskThresholds.from_settings() + + event_at = _as_datetime(trade["traded_at"]) + day_start = event_at.replace(hour=0, minute=0, second=0, microsecond=0) + trades = _normalize_trades(core.list_trades(trade["customer_id"], since=day_start, limit=1000)) + + result: dict[str, Any] = {"triggered_rules": [], "alert_ids": [], "aml_hit": False} + + hits = run_rules(trades, th, now=event_at) + # 无条件走预警编排：空 hits 由 alert_service 落 pass 审计（架构 §3.1 ④ 未命中分支） + alert = record_trade_alerts(trade, hits, risk_repo=repo) + if hits: + result["triggered_rules"] = sorted({h.rule_id for h in hits}) + if alert: + result["alert_ids"].append(alert["alert_id"]) + best = max(hits, key=lambda h: h.risk_score) + upsert_profile_l3( + trade["customer_id"], + best.alert_type, + last_alert_id=alert["alert_id"] if alert else None, + risk_repo=repo, + ) + + aml_hits = match_customer(trade["customer_id"], core_ro=core, risk_repo=repo) + if aml_hits: + result["aml_hit"] = True + alert = record_aml_alert( + trade["customer_id"], + {"trigger": "trade", "trade_id": trade.get("trade_id"), "matches": aml_hits}, + risk_repo=repo, + ) + result["alert_ids"].append(alert["alert_id"]) + upsert_profile_l3( + trade["customer_id"], "aml", last_alert_id=alert["alert_id"], risk_repo=repo + ) + + return result + + +def on_customer_created(customer_id: str) -> None: + """AML 开户触发预留（本期 no-op；模拟环境无开户流程，PRD FR-5）。""" + +def on_customer_updated(customer_id: str) -> None: + """客户信息变更触发预留（本期 no-op；PRD FR-5）。""" diff --git a/app/service/risk/scoring.py b/app/service/risk/scoring.py new file mode 100644 index 0000000..0e00906 --- /dev/null +++ b/app/service/risk/scoring.py @@ -0,0 +1,18 @@ +"""R-05 动态评分预留接口（FR-7 · 本期静态映射，评分模型后置）。""" + +from __future__ import annotations + +from typing import Any + +from app.repository.risk_repository import RiskRepository + + +def recompute_customer_score( + customer_id: str, risk_repo: RiskRepository | None = None +) -> dict[str, Any]: + """重算客户 L3 risk_score（R-05 评分模型接入点，签名冻结）。 + + 本期抛 NotImplementedError：score 口径一期保持 NULL（profile_l3 docstring + 留痕）；接入时同步将 merge_l3 的 risk_score 改为保留 existing 值。 + """ + raise NotImplementedError("R-05 dynamic scoring not implemented (static mapping phase)") diff --git a/tests/test_aml_service.py b/tests/test_aml_service.py new file mode 100644 index 0000000..396f623 --- /dev/null +++ b/tests/test_aml_service.py @@ -0,0 +1,228 @@ +"""aml_service 单测（B4 · 归一化/相似度/行阈值优先/scan_all 编排）。 + +sqlite StaticPool 内存库；名单阈值边界用可控 threshold 值驱动（1.0 严格、 +0.5 宽松），避免依赖 difflib 具体分值。 +""" + +from datetime import datetime + +import pytest +from sqlalchemy import create_engine, text +from sqlalchemy.pool import StaticPool + +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.aml_service import match_customer, match_name, normalize_name, scan_all, similarity +from app.service.risk.profile_l3 import AML_PENDING_TAG + + +class FakePublisher: + def __init__(self): + self.messages = [] + + def publish(self, channel, payload): + assert isinstance(payload, dict) + self.messages.append((channel, payload)) + + +@pytest.fixture() +def env(): + engine = create_engine( + "sqlite://", + poolclass=StaticPool, + connect_args={"check_same_thread": False}, + ) + with engine.begin() as conn: + conn.execute( + text( + """ + 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 + ) + """ + ) + ) + conn.execute( + text( + """ + 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 + ) + """ + ) + ) + conn.execute( + text( + """ + CREATE TABLE core_customer_risk ( + customer_id VARCHAR(64), risk_code VARCHAR(8), evaluated_at TIMESTAMP + ) + """ + ) + ) + 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 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 + ) + """ + ) + ) + conn.execute( + text( + """ + 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 + ) + """ + ) + ) + # 名单：默认阈值 0.85 两行、严格 1.0 一行、行阈值覆盖（0.5/0.99 同名对）一停用行 + conn.execute( + text( + """ + INSERT INTO risk_aml_list (list_id, list_type, full_name, match_threshold, source, list_version, is_active) VALUES + ('SAN-1', 'sanction', '张某某', 0.85, 'mock', 'v1', 1), + ('PEP-1', 'pep', '李四', 0.85, 'mock', 'v1', 1), + ('EXACT-1', 'sanction', '孙七', 1.0, 'mock', 'v1', 1), + ('WIDE-1', 'terror', '张三', 0.5, 'mock', 'v1', 1), + ('STRICT-1', 'sanction', '张三', 0.99, 'mock', 'v1', 1), + ('OFF-1', 'pep', '王五', 0.85, 'mock', 'v1', 0) + """ + ) + ) + conn.execute( + text( + """ + INSERT INTO core_customer (customer_id, display_name, age, is_active) VALUES + ('C1', '张某某', 40, 1), ('C2', '李四', 35, 1), + ('C3', '赵六六', 28, 1), ('C4', '王五', 50, 1) + """ + ) + ) + core = CoreReadOnlyRepository(engine=engine) + repo = RiskRepository(engine=engine) + pub = FakePublisher() + alert_service.set_publisher(pub) + yield core, repo, pub + alert_service.set_publisher(None) + engine.dispose() + + +def test_normalize_name_strips_space_and_case(): + assert normalize_name(" Ab C ") == "abc" + assert normalize_name("ABC") == normalize_name("abc") + + +def test_similarity_identical_is_one(): + assert similarity(normalize_name("张某某"), normalize_name("张某某")) == 1.0 + + +def test_exact_name_hit_with_default_threshold(env): + _, repo, _ = env + hits = match_name("张某某", repo.list_active_aml_entries()) + assert [h["list_id"] for h in hits] == ["SAN-1"] + assert hits[0]["similarity"] == 1.0 + assert hits[0]["list_type"] == "sanction" + assert hits[0]["list_version"] == "v1" + + +def test_threshold_boundary_is_inclusive(env): + """ratio ≥ threshold 命中：1.0 严格名单同名命中、近似名不命中。""" + _, repo, _ = env + entries = repo.list_active_aml_entries() + assert [h["list_id"] for h in match_name("孙七", entries)] == ["EXACT-1"] + assert match_name("孙七七", entries) == [] # ratio < 1.0 不命中 + + +def test_row_threshold_overrides_default(env): + """行阈值优先：0.5 宽松行命中、0.99 严格行不命中（同名对）。""" + _, repo, _ = env + hits = match_name("张三三", repo.list_active_aml_entries()) + assert [h["list_id"] for h in hits] == ["WIDE-1"] + + +def test_inactive_entry_excluded(env): + _, repo, _ = env + assert match_name("王五", repo.list_active_aml_entries()) == [] # OFF-1 停用 + + +def test_match_customer_missing_returns_empty(env): + core, repo, _ = env + assert match_customer("C999", core_ro=core, risk_repo=repo) == [] + + +def test_match_customer_by_id(env): + core, repo, _ = env + hits = match_customer("C2", core_ro=core, risk_repo=repo) + assert [h["list_id"] for h in hits] == ["PEP-1"] + + +def test_scan_all_creates_alert_and_l3(env): + core, repo, pub = env + summary = scan_all(core_ro=core, risk_repo=repo) + assert summary["scanned"] == 4 + assert summary["hit_customers"] == 2 # C1 张某某、C2 李四（C3 赵六六 不命中任一） + assert len(summary["alerts"]) == 2 + with core._engine.connect() as conn: + types = [ + r[0] + for r in conn.execute(text("SELECT alert_type FROM risk_alert")).fetchall() + ] + tiers = dict( + conn.execute( + text("SELECT customer_id, monitor_tier FROM customer_profile_l3") + ).fetchall() + ) + assert types == ["aml", "aml"] + assert tiers == {"C1": "high", "C2": "high"} + payload = repo.get_alert(summary["alerts"][0])["payload"] + assert payload["events"][0]["trigger"] == "scan" + assert payload["events"][0]["matches"][0]["list_id"] in ("SAN-1", "PEP-1") + tags = repo.get_l3("C1")["monitor_tags"] + assert AML_PENDING_TAG in tags + assert len(pub.messages) == 2 # 每个命中客户一次紧急推送 + + +def test_scan_all_no_hit_creates_nothing(env): + core, repo, pub = 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": []} + with core._engine.connect() as conn: + assert conn.execute(text("SELECT COUNT(*) FROM risk_alert")).scalar_one() == 0 + assert pub.messages == [] diff --git a/tests/test_risk_engine.py b/tests/test_risk_engine.py new file mode 100644 index 0000000..0182cec --- /dev/null +++ b/tests/test_risk_engine.py @@ -0,0 +1,278 @@ +"""risk engine 集成冒烟（B4 · 规则编排 + AML 组合 + L3/审计/推送贯通）。 + +引擎输入为已落库交易；当日流水经 sqlite core_trade 提供给规则层。 +RISK-004 窗口以 trade["traded_at"] 为事件时点（幂等重放口径）。 +""" + +from datetime import datetime +from decimal import Decimal + +import pytest +from sqlalchemy import create_engine, text +from sqlalchemy.pool import StaticPool + +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.engine import on_customer_created, on_customer_updated, process_trade_event +from app.service.risk.profile_l3 import AML_PENDING_TAG +from app.service.risk.scoring import recompute_customer_score + + +class FakePublisher: + def __init__(self): + self.messages = [] + + def publish(self, channel, payload): + self.messages.append((channel, payload)) + + +@pytest.fixture() +def env(): + engine = create_engine( + "sqlite://", + poolclass=StaticPool, + connect_args={"check_same_thread": False}, + ) + with engine.begin() as conn: + conn.execute( + text( + """ + 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 + ) + """ + ) + ) + conn.execute( + text( + """ + CREATE TABLE core_customer_risk ( + customer_id VARCHAR(64), risk_code VARCHAR(8), evaluated_at TIMESTAMP + ) + """ + ) + ) + conn.execute( + text( + """ + CREATE TABLE core_product ( + product_id VARCHAR(64) PRIMARY KEY, product_name VARCHAR(128), + min_risk_code VARCHAR(8), product_type VARCHAR(32) + ) + """ + ) + ) + conn.execute( + text( + """ + 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 + ) + """ + ) + ) + conn.execute( + text( + """ + 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 + ) + """ + ) + ) + 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 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 + ) + """ + ) + ) + conn.execute( + text( + """ + 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 + ) + """ + ) + ) + conn.execute( + text( + """ + INSERT INTO core_customer (customer_id, display_name, age, is_active) VALUES + ('C1', '张某某', 40, 1), ('C2', '李四', 35, 1) + """ + ) + ) + conn.execute( + text( + "INSERT INTO core_product (product_id, product_name, min_risk_code, product_type)" + " VALUES ('P1', '测试混合基金', 'R3', 'mixed')" + ) + ) + conn.execute( + text( + "INSERT INTO risk_aml_list (list_id, list_type, full_name, match_threshold, source, list_version, is_active)" + " VALUES ('PEP-1', 'pep', '李四', 0.85, 'mock', 'v1', 1)" + ) + ) + core = CoreReadOnlyRepository(engine=engine) + repo = RiskRepository(engine=engine) + pub = FakePublisher() + alert_service.set_publisher(pub) + yield core, repo, pub, engine + alert_service.set_publisher(None) + engine.dispose() + + +def _trade(trade_id, amount="600000", customer="C1", ttype="subscribe", + at=datetime(2026, 9, 6, 14, 0, 0)): + return { + "trade_id": trade_id, + "customer_id": customer, + "product_id": "P1", + "trade_type": ttype, + "amount": Decimal(amount), + "trade_status": "confirmed", + "traded_at": at, + } + + +def _seed_trade(conn, trade_id, amount, customer="C1", ttype="subscribe", + at=datetime(2026, 9, 6, 14, 0, 0)): + conn.execute( + text( + "INSERT INTO core_trade (trade_id, customer_id, product_id, trade_type, amount," + " trade_status, traded_at) VALUES (:tid, :cid, 'P1', :tt, :amt, 'confirmed', :at)" + ), + {"tid": trade_id, "cid": customer, "tt": ttype, "amt": amount, "at": at}, + ) + + +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_large_amount_triggers_alert_l3_and_push(env): + core, repo, pub, engine = env + with engine.begin() as conn: + _seed_trade(conn, "T1", "600000") + result = process_trade_event(_trade("T1", "600000"), core_ro=core, risk_repo=repo) + # 单笔 60 万同时命中 RISK-001（单笔）与 RISK-002（当日累计含本笔），聚合一张单 + assert result["triggered_rules"] == ["RISK-001", "RISK-002"] + assert len(result["alert_ids"]) == 1 and result["aml_hit"] is False + alert = repo.get_alert(result["alert_ids"][0]) + assert alert["alert_type"] == "large_amount" and alert["risk_score"] == 70 + assert _counts(engine, "customer_profile_l3", "customer_id='C1' AND monitor_tier='watch'") == 1 + assert _counts(engine, "audit_log", "decision='alert_created'") == 1 + (channel, payload), = pub.messages + assert channel == "risk:pub:alert" and payload["risk_score"] == 70 + + +def test_small_trade_passes_without_alert(env): + core, repo, pub, engine = env + with engine.begin() as conn: + _seed_trade(conn, "T1", "1000") + result = process_trade_event(_trade("T1", "1000"), core_ro=core, risk_repo=repo) + assert result == {"triggered_rules": [], "alert_ids": [], "aml_hit": False} + assert _counts(engine, "risk_alert") == 0 + assert _counts(engine, "audit_log", "decision='pass'") == 1 + assert _counts(engine, "customer_profile_l3") == 0 + assert pub.messages == [] + + +def test_aml_and_event_rule_both_fire(env): + """命中名单客户发起大额 → 事件单 + aml 独立单、L3 high、紧急推送含 compliance。""" + core, repo, pub, engine = env + with engine.begin() as conn: + _seed_trade(conn, "T1", "600000", customer="C2") + result = process_trade_event(_trade("T1", "600000", customer="C2"), + core_ro=core, risk_repo=repo) + assert result["aml_hit"] is True + assert result["triggered_rules"] == ["RISK-001", "RISK-002"] # 含本笔累计 + assert len(result["alert_ids"]) == 2 + types = {a["alert_type"] for a in (repo.get_alert(aid) for aid in result["alert_ids"])} + assert types == {"large_amount", "aml"} + l3 = repo.get_l3("C2") + assert l3["monitor_tier"] == "high" and AML_PENDING_TAG in l3["monitor_tags"] + assert l3["risk_score"] is None # score 一期不写（P3-4 口径） + assert len(pub.messages) == 2 + (_, aml_payload), = [m for m in pub.messages if m[1]["alert_type"] == "aml"] + assert "compliance" in aml_payload["notify_role"] + + +def test_probe_window_uses_trade_time_not_wall_clock(env): + """RISK-004 窗口以 traded_at 为事件时点：流水时间集中即可命中，与执行时刻无关。""" + core, repo, _, engine = env + with engine.begin() as conn: + _seed_trade(conn, "T1", "450000", at=datetime(2026, 9, 6, 13, 59, 30)) + _seed_trade(conn, "T2", "450000", at=datetime(2026, 9, 6, 13, 59, 50)) + _seed_trade(conn, "T3", "450000", at=datetime(2026, 9, 6, 14, 0, 10)) + # 引擎契约：网关先落库再调用，本笔 T3 已在流水内（FR-1 ②③b→④） + result = process_trade_event( + _trade("T3", "450000", at=datetime(2026, 9, 6, 14, 0, 10)), + core_ro=core, risk_repo=repo, + ) + assert "RISK-004" in result["triggered_rules"] + alert = repo.get_alert(result["alert_ids"][0]) + assert alert["alert_type"] == "pattern" and alert["risk_score"] == 80 + + +def test_convert_type_does_not_trigger_rules(env): + """convert 不落库也不会进事件线（引擎防御过滤，PRD FR-1）。""" + core, repo, pub, engine = env + with engine.begin() as conn: + _seed_trade(conn, "T0", "600000", ttype="convert") + result = process_trade_event(_trade("T1", "600000", ttype="convert"), + core_ro=core, risk_repo=repo) + assert result["triggered_rules"] == [] and result["aml_hit"] is False + assert _counts(engine, "risk_alert") == 0 + assert _counts(engine, "audit_log", "decision='pass'") == 1 + assert pub.messages == [] + + +def test_customer_event_hooks_noop(): + assert on_customer_created("C1") is None + assert on_customer_updated("C1") is None + + +def test_scoring_placeholder_raises(): + with pytest.raises(NotImplementedError): + recompute_customer_score("C1")