feat: AML 名单匹配/全量扫描 + 事件引擎编排 + R-05 scoring 占位(B4)

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(
             """

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}

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）。"""

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)")

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 == []

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")