210 lines
8.7 KiB
Python
210 lines
8.7 KiB
Python
"""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 text
|
|
|
|
from _ddl import create_sqlite_engine
|
|
|
|
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 = []
|
|
self.deletes = []
|
|
|
|
def publish(self, channel, payload):
|
|
self.messages.append((channel, payload))
|
|
|
|
def delete(self, *keys):
|
|
self.deletes.append(keys)
|
|
|
|
|
|
@pytest.fixture()
|
|
def env():
|
|
engine = create_sqlite_engine() # DDL 单一事实源（B4 评审 P3-12）
|
|
with engine.begin() as conn:
|
|
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"]
|
|
# FR-4 payload 完整性（B4 评审 P1-1）：客户上下文 = L0 摘要 + 近 30 天统计
|
|
ev_alert = next(
|
|
a for a in (repo.get_alert(aid) for aid in result["alert_ids"])
|
|
if a["alert_type"] == "large_amount"
|
|
)
|
|
ctx = ev_alert["payload"]["customer_context"]
|
|
assert ctx["l0"]["display_name"] == "李四" and ctx["l0"]["age"] == 35
|
|
assert ctx["trades_30d"]["count"] >= 1
|
|
|
|
|
|
def test_second_trade_same_day_appends_to_same_alert(env):
|
|
"""B4 评审 P2-3：同日第二笔 → 追加同一事件单，triggered_rules/risk_score/L3 联动。"""
|
|
core, repo, _, engine = env
|
|
with engine.begin() as conn:
|
|
_seed_trade(conn, "T1", "600000")
|
|
_seed_trade(conn, "T2", "200000", at=datetime(2026, 9, 6, 14, 1, 0))
|
|
r1 = process_trade_event(_trade("T1", "600000"), core_ro=core, risk_repo=repo)
|
|
r2 = process_trade_event(
|
|
_trade("T2", "200000", at=datetime(2026, 9, 6, 14, 1, 0)), core_ro=core, risk_repo=repo
|
|
)
|
|
assert r2["alert_ids"] == [r1["alert_ids"][0]] # 同日仅一张事件类单
|
|
alert = repo.get_alert(r1["alert_ids"][0])
|
|
assert len(alert["payload"]["events"]) == 2
|
|
assert alert["triggered_rules"] == ["RISK-001", "RISK-002"] # T2 仅命中累计，追加合并
|
|
assert alert["risk_score"] == 70 # 取 max
|
|
assert _counts(engine, "audit_log", "decision='alert_appended'") == 1
|
|
assert repo.get_l3("C1")["last_alert_id"] == alert["alert_id"] # L3 联动最新
|
|
|
|
|
|
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")
|