feat: 预警聚合/去重/审计/PubSub 推送 alert_service(B2)
This commit is contained in:
@@ -219,6 +219,25 @@ class RiskRepository:
|
||||
row["payload"] = json.loads(row["payload"]) if isinstance(row["payload"], str) else row["payload"]
|
||||
return row
|
||||
|
||||
# ---------- audit_log（风控判定审计 · 只 INSERT，PRD §7.3）----------
|
||||
|
||||
def insert_audit_log(self, entry: dict[str, Any]) -> None:
|
||||
sql = text(
|
||||
"""
|
||||
INSERT INTO audit_log
|
||||
(trace_id, event_type, agent_type, actor_id, customer_id, rule_id,
|
||||
input_summary, decision, risk_score, handler_id, handler_result, handler_comment)
|
||||
VALUES (:trace_id, :event_type, :agent_type, :actor_id, :customer_id, :rule_id,
|
||||
:input_summary, :decision, :risk_score, :handler_id, :handler_result, :handler_comment)
|
||||
"""
|
||||
)
|
||||
params = dict(entry)
|
||||
params["input_summary"] = json.dumps(
|
||||
entry.get("input_summary") or {}, ensure_ascii=False, default=str
|
||||
)
|
||||
with self._engine.begin() as conn:
|
||||
conn.execute(sql, params)
|
||||
|
||||
# ---------- risk_suitability_log ----------
|
||||
|
||||
def insert_suitability_log(self, log: dict[str, Any]) -> None:
|
||||
|
||||
@@ -0,0 +1,286 @@
|
||||
"""预警单聚合与通知（B2 · 架构 §5.2 / PRD FR-4）。
|
||||
|
||||
职责：单事件命中规则的**聚合决策与编排**（merge 原语 append_alert_event 在 repo，A3 评审口径）；
|
||||
进程内锁防并发首单双单（多进程部署时换 Redis SET NX，接口不变）；审计落库；Pub/Sub 通知广播。
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import logging
|
||||
import threading
|
||||
from datetime import date, datetime, time
|
||||
from typing import Any, Callable
|
||||
from uuid import uuid4
|
||||
|
||||
from app.config.settings import settings
|
||||
from app.repository.risk_repository import EVENT_ALERT_TYPES, RiskRepository
|
||||
from app.service.risk.rules import RuleHit
|
||||
from app.utils.trace import current_trace, new_trace
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
LOCK_TIMEOUT_SECONDS = 2.0
|
||||
TIER_SCORE = {"large_amount": 70, "freq_trade": 50, "pattern": 80, "suitability": 90, "aml": 95}
|
||||
|
||||
|
||||
class RedisAlertPublisher:
|
||||
"""同步 Redis Pub/Sub 发布器（事件线与同步调用链一致；B7 lifespan 管理连接）。"""
|
||||
|
||||
def __init__(self, url: str | None = None) -> None:
|
||||
self._url = url or settings.redis_url
|
||||
self._client = None
|
||||
|
||||
def publish(self, channel: str, payload: dict[str, Any]) -> None:
|
||||
if self._client is None:
|
||||
import redis
|
||||
|
||||
self._client = redis.Redis.from_url(self._url, decode_responses=True)
|
||||
self._client.publish(channel, json.dumps(payload, ensure_ascii=False))
|
||||
|
||||
|
||||
_publisher: RedisAlertPublisher | None = None
|
||||
|
||||
|
||||
def set_publisher(publisher: RedisAlertPublisher | None) -> None:
|
||||
"""测试/集成注入点（B7 lifespan 初始化，测试注入 fake）。"""
|
||||
global _publisher
|
||||
_publisher = publisher
|
||||
|
||||
|
||||
def _publish(payload: dict[str, Any]) -> None:
|
||||
try:
|
||||
(_publisher or RedisAlertPublisher()).publish("risk:pub:alert", payload)
|
||||
except Exception: # 通知失败不阻塞预警落库（DB 为权威）
|
||||
logger.exception("publish risk:pub:alert failed")
|
||||
|
||||
|
||||
# ---------- 进程内聚合锁（架构 §5.2；拿不到锁降级独立出单，宁多勿漏） ----------
|
||||
|
||||
_locks: dict[str, threading.Lock] = {}
|
||||
_locks_guard = threading.Lock()
|
||||
|
||||
|
||||
def _lock_for(key: str) -> threading.Lock:
|
||||
with _locks_guard:
|
||||
lock = _locks.get(key)
|
||||
if lock is None:
|
||||
lock = threading.Lock()
|
||||
_locks[key] = lock
|
||||
return lock
|
||||
|
||||
|
||||
def _run_locked(key: str, fn: Callable[[bool], Any]) -> Any:
|
||||
"""锁内执行 fn(locked=True)；获取超时降级 fn(locked=False)。"""
|
||||
lock = _lock_for(key)
|
||||
if not lock.acquire(timeout=LOCK_TIMEOUT_SECONDS):
|
||||
logger.warning("agg lock timeout, degrade to standalone alert: %s", key)
|
||||
return fn(locked=False)
|
||||
try:
|
||||
return fn(locked=True)
|
||||
finally:
|
||||
lock.release()
|
||||
|
||||
|
||||
def _day_start(day: date | None = None) -> datetime:
|
||||
return datetime.combine(day or date.today(), time.min)
|
||||
|
||||
|
||||
def _new_alert_id() -> str:
|
||||
return f"ALT-{datetime.now():%Y%m%d}-{uuid4().hex[:8].upper()}"
|
||||
|
||||
|
||||
def _audit(
|
||||
risk_repo: RiskRepository,
|
||||
*,
|
||||
customer_id: str | None,
|
||||
rule_id: str | None,
|
||||
decision: str,
|
||||
risk_score: int | None,
|
||||
input_summary: dict[str, Any],
|
||||
event_type: str = "risk_judgement",
|
||||
actor_id: str = "SYSTEM",
|
||||
) -> None:
|
||||
risk_repo.insert_audit_log(
|
||||
{
|
||||
"trace_id": current_trace() or new_trace(),
|
||||
"event_type": event_type,
|
||||
"agent_type": "risk",
|
||||
"actor_id": actor_id,
|
||||
"customer_id": customer_id,
|
||||
"rule_id": rule_id,
|
||||
"input_summary": input_summary,
|
||||
"decision": decision,
|
||||
"risk_score": risk_score,
|
||||
"handler_id": None,
|
||||
"handler_result": None,
|
||||
"handler_comment": None,
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
def _publish_alert(alert: dict[str, Any]) -> None:
|
||||
_publish(
|
||||
{
|
||||
"alert_id": alert["alert_id"],
|
||||
"alert_type": alert["alert_type"],
|
||||
"customer_id_mask": alert["customer_id"][:6] + "**",
|
||||
"risk_score": alert["risk_score"],
|
||||
"trace_id": alert["trace_id"],
|
||||
"notify_role": ["risk_officer"] + (["compliance"] if alert["alert_type"] == "aml" else []),
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
def record_trade_alerts(
|
||||
trade: dict[str, Any],
|
||||
hits: list[RuleHit],
|
||||
risk_repo: RiskRepository | None = None,
|
||||
customer_context: dict[str, Any] | None = None,
|
||||
) -> dict[str, Any] | None:
|
||||
"""交易事件预警入口（引擎 B4 调用）。
|
||||
|
||||
hits 为空 → 审计 pass；否则事件类命中聚合为一张单（PRD FR-4：同客户同日仅一张事件类
|
||||
pending 单，alert_type 随最高分规则动态更新）。
|
||||
"""
|
||||
risk_repo = risk_repo or RiskRepository()
|
||||
if not hits:
|
||||
_audit(risk_repo, customer_id=trade["customer_id"], rule_id=None, decision="pass",
|
||||
risk_score=None, input_summary={"trade_id": trade["trade_id"]})
|
||||
return None
|
||||
|
||||
best = max(hits, key=lambda h: h.risk_score)
|
||||
triggered_rules = sorted({h.rule_id for h in hits})
|
||||
risk_score = max(h.risk_score for h in hits)
|
||||
event = {
|
||||
"trade_id": trade["trade_id"],
|
||||
"product_id": trade["product_id"],
|
||||
"trade_type": trade["trade_type"],
|
||||
"amount": str(trade["amount"]),
|
||||
"traded_at": str(trade["traded_at"]),
|
||||
"matched_rules": [h.rule_id for h in hits],
|
||||
"details": {h.rule_id: h.detail for h in hits},
|
||||
}
|
||||
|
||||
def _agg(locked: bool) -> dict[str, Any]:
|
||||
if locked:
|
||||
pending = risk_repo.find_pending_event_alert(trade["customer_id"], _day_start())
|
||||
if pending:
|
||||
risk_repo.append_alert_event(
|
||||
pending["alert_id"], event, triggered_rules, risk_score, best.alert_type
|
||||
)
|
||||
updated = risk_repo.get_alert(pending["alert_id"])
|
||||
_audit(risk_repo, customer_id=trade["customer_id"], rule_id=",".join(triggered_rules),
|
||||
decision="alert_appended", risk_score=updated["risk_score"],
|
||||
input_summary=event)
|
||||
_publish_alert(updated)
|
||||
return updated
|
||||
alert = {
|
||||
"alert_id": _new_alert_id(),
|
||||
"trace_id": current_trace() or new_trace(),
|
||||
"customer_id": trade["customer_id"],
|
||||
"trade_id": trade["trade_id"],
|
||||
"alert_type": best.alert_type,
|
||||
"triggered_rules": triggered_rules,
|
||||
"risk_score": risk_score,
|
||||
"status": "pending_review",
|
||||
"payload": {
|
||||
"product_id": trade["product_id"],
|
||||
"events": [event],
|
||||
"customer_context": customer_context or {},
|
||||
},
|
||||
}
|
||||
risk_repo.insert_alert(alert)
|
||||
_audit(risk_repo, customer_id=trade["customer_id"], rule_id=",".join(triggered_rules),
|
||||
decision="alert_created", risk_score=risk_score, input_summary=event)
|
||||
_publish_alert(alert)
|
||||
return alert
|
||||
|
||||
return _run_locked(f"agg:event:{trade['customer_id']}:{date.today()}", _agg)
|
||||
|
||||
|
||||
def record_suitability_alert(
|
||||
trade_request: dict[str, Any],
|
||||
rule_id: str,
|
||||
block_reason: str,
|
||||
risk_repo: RiskRepository | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""R-02 阻断预警（网关 FR-1 调用）：同客户+产品+日仅一张 pending 单。"""
|
||||
risk_repo = risk_repo or RiskRepository()
|
||||
event = {
|
||||
"trade_id": trade_request["trade_id"],
|
||||
"product_id": trade_request["product_id"],
|
||||
"trade_type": trade_request["trade_type"],
|
||||
"amount": str(trade_request["amount"]),
|
||||
"traded_at": str(trade_request["traded_at"]),
|
||||
"matched_rules": [rule_id],
|
||||
"details": {rule_id: block_reason},
|
||||
}
|
||||
best_type, score = "suitability", TIER_SCORE["suitability"]
|
||||
|
||||
def _agg(locked: bool) -> dict[str, Any]:
|
||||
if locked:
|
||||
pending = risk_repo.find_pending_suitability_alert(
|
||||
trade_request["customer_id"], trade_request["product_id"], _day_start()
|
||||
)
|
||||
if pending:
|
||||
risk_repo.append_alert_event(
|
||||
pending["alert_id"], event, [rule_id], score, best_type
|
||||
)
|
||||
updated = risk_repo.get_alert(pending["alert_id"])
|
||||
_audit(risk_repo, customer_id=trade_request["customer_id"], rule_id=rule_id,
|
||||
decision="suitability_alert_appended", risk_score=score,
|
||||
input_summary=event, event_type="suitability_block")
|
||||
_publish_alert(updated)
|
||||
return updated
|
||||
alert = {
|
||||
"alert_id": _new_alert_id(),
|
||||
"trace_id": current_trace() or new_trace(),
|
||||
"customer_id": trade_request["customer_id"],
|
||||
"trade_id": trade_request["trade_id"],
|
||||
"alert_type": best_type,
|
||||
"triggered_rules": [rule_id],
|
||||
"risk_score": score,
|
||||
"status": "pending_review",
|
||||
"payload": {
|
||||
"product_id": trade_request["product_id"],
|
||||
"events": [event],
|
||||
"customer_context": {},
|
||||
},
|
||||
}
|
||||
risk_repo.insert_alert(alert)
|
||||
_audit(risk_repo, customer_id=trade_request["customer_id"], rule_id=rule_id,
|
||||
decision="suitability_alert_created", risk_score=score,
|
||||
input_summary=event, event_type="suitability_block")
|
||||
_publish_alert(alert)
|
||||
return alert
|
||||
|
||||
return _run_locked(
|
||||
f"agg:suitability:{trade_request['customer_id']}:{trade_request['product_id']}:{date.today()}",
|
||||
_agg,
|
||||
)
|
||||
|
||||
|
||||
def record_aml_alert(
|
||||
customer_id: str,
|
||||
aml_detail: dict[str, Any],
|
||||
risk_repo: RiskRepository | None = None,
|
||||
) -> dict[str, Any]:
|
||||
"""R-03 AML 命中：独立出单不聚合（PRD FR-4），单事件单张。"""
|
||||
risk_repo = risk_repo or RiskRepository()
|
||||
alert = {
|
||||
"alert_id": _new_alert_id(),
|
||||
"trace_id": current_trace() or new_trace(),
|
||||
"customer_id": customer_id,
|
||||
"trade_id": aml_detail.get("trade_id"),
|
||||
"alert_type": "aml",
|
||||
"triggered_rules": ["AML-001"],
|
||||
"risk_score": TIER_SCORE["aml"],
|
||||
"status": "pending_review",
|
||||
"payload": {"product_id": aml_detail.get("product_id"), "events": [aml_detail], "customer_context": {}},
|
||||
}
|
||||
risk_repo.insert_alert(alert)
|
||||
_audit(risk_repo, customer_id=customer_id, rule_id="AML-001", decision="aml_alert_created",
|
||||
risk_score=TIER_SCORE["aml"], input_summary=aml_detail, event_type="aml_hit")
|
||||
_publish_alert(alert)
|
||||
return alert
|
||||
@@ -0,0 +1,195 @@
|
||||
"""alert_service 单测（B2 · 聚合/去重/审计/推送 + 并发冒烟）。
|
||||
|
||||
sqlite StaticPool 单连接共享内存库，跨线程可用；publisher 注入 fake。
|
||||
"""
|
||||
|
||||
import json
|
||||
from datetime import datetime
|
||||
from decimal import Decimal
|
||||
from threading import Thread
|
||||
|
||||
import pytest
|
||||
from sqlalchemy import create_engine, text
|
||||
from sqlalchemy.pool import StaticPool
|
||||
|
||||
from app.repository.risk_repository import RiskRepository
|
||||
from app.service.risk import alert_service
|
||||
from app.service.risk.alert_service import (
|
||||
record_aml_alert,
|
||||
record_suitability_alert,
|
||||
record_trade_alerts,
|
||||
set_publisher,
|
||||
)
|
||||
from app.service.risk.rules import RULE_ALERT_TYPES, RuleHit
|
||||
|
||||
|
||||
class FakePublisher:
|
||||
def __init__(self):
|
||||
self.messages = []
|
||||
|
||||
def publish(self, channel, payload):
|
||||
self.messages.append((channel, json.loads(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_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
|
||||
)
|
||||
"""
|
||||
)
|
||||
)
|
||||
repo = RiskRepository(engine=engine)
|
||||
pub = FakePublisher()
|
||||
set_publisher(pub)
|
||||
yield repo, pub, engine
|
||||
set_publisher(None)
|
||||
engine.dispose()
|
||||
|
||||
|
||||
def _trade(trade_id, amount="600000", customer="C1"):
|
||||
return {
|
||||
"trade_id": trade_id,
|
||||
"customer_id": customer,
|
||||
"product_id": "P1",
|
||||
"trade_type": "subscribe",
|
||||
"amount": Decimal(amount),
|
||||
"traded_at": datetime(2026, 9, 6, 14, 0, 0),
|
||||
}
|
||||
|
||||
|
||||
def _hit(rule_id, score):
|
||||
return RuleHit(rule_id, RULE_ALERT_TYPES[rule_id], score, "detail")
|
||||
|
||||
|
||||
def _counts(engine, table, where="1=1", params=None):
|
||||
with engine.connect() as conn:
|
||||
return conn.execute(text(f"SELECT COUNT(*) FROM {table} WHERE {where}"), params or {}).scalar_one()
|
||||
|
||||
|
||||
def test_no_hits_records_pass_audit(env):
|
||||
repo, _, engine = env
|
||||
result = record_trade_alerts(_trade("T1"), [], risk_repo=repo)
|
||||
assert result is None
|
||||
assert _counts(engine, "risk_alert") == 0
|
||||
assert _counts(engine, "audit_log", "decision = 'pass'") == 1
|
||||
|
||||
|
||||
def test_first_large_trade_creates_agg_alert(env):
|
||||
repo, pub, engine = env
|
||||
hits = [_hit("RISK-001", 70), _hit("RISK-002", 70)]
|
||||
alert = record_trade_alerts(_trade("T1"), hits, risk_repo=repo)
|
||||
assert alert["alert_type"] == "large_amount"
|
||||
assert alert["triggered_rules"] == ["RISK-001", "RISK-002"]
|
||||
assert alert["risk_score"] == 70
|
||||
assert alert["status"] == "pending_review"
|
||||
assert len(alert["payload"]["events"]) == 1
|
||||
assert alert["trace_id"].startswith("trc-")
|
||||
assert _counts(engine, "risk_alert") == 1
|
||||
assert _counts(engine, "audit_log", "decision = 'alert_created'") == 1
|
||||
(channel, payload), = pub.messages
|
||||
assert channel == "risk:pub:alert"
|
||||
assert payload["alert_id"] == alert["alert_id"]
|
||||
assert payload["customer_id_mask"].endswith("**")
|
||||
|
||||
|
||||
def test_same_day_second_trade_appends_not_duplicates(env):
|
||||
repo, pub, engine = env
|
||||
record_trade_alerts(_trade("T1"), [_hit("RISK-001", 70), _hit("RISK-002", 70)], risk_repo=repo)
|
||||
alert = record_trade_alerts(_trade("T2"), [_hit("RISK-001", 70)], risk_repo=repo)
|
||||
assert _counts(engine, "risk_alert") == 1 # 同日仅一张事件类单
|
||||
got = repo.get_alert(alert["alert_id"])
|
||||
assert len(got["payload"]["events"]) == 2
|
||||
assert got["triggered_rules"] == ["RISK-001", "RISK-002"]
|
||||
assert _counts(engine, "audit_log", "decision = 'alert_appended'") == 1
|
||||
assert len(pub.messages) == 2 # 每次事件都广播，消费端按 alert_id 聚合
|
||||
|
||||
|
||||
def test_alert_type_follows_highest_score_rule(env):
|
||||
"""类型随最高分规则动态更新（PRD FR-4；评审 P1-3 口径）。"""
|
||||
repo, _, _ = env
|
||||
first = record_trade_alerts(_trade("T1"), [_hit("RISK-001", 70)], risk_repo=repo)
|
||||
assert first["alert_type"] == "large_amount"
|
||||
updated = record_trade_alerts(_trade("T2", "800000"), [_hit("RISK-005", 80)], risk_repo=repo)
|
||||
assert updated["alert_id"] == first["alert_id"]
|
||||
got = repo.get_alert(first["alert_id"])
|
||||
assert got["alert_type"] == "pattern" and got["risk_score"] == 80
|
||||
assert set(got["triggered_rules"]) == {"RISK-001", "RISK-005"}
|
||||
|
||||
|
||||
def test_suitability_dedup_per_product(env):
|
||||
repo, _, engine = env
|
||||
req1 = _trade("T1", "100000")
|
||||
req2 = _trade("T2", "100000")
|
||||
a1 = record_suitability_alert(req1, "SUIT-001", "C1 不可购 R4", risk_repo=repo)
|
||||
a2 = record_suitability_alert(req2, "SUIT-001", "C1 不可购 R4", risk_repo=repo)
|
||||
assert a1["alert_id"] == a2["alert_id"] # 同客户+产品+日去重
|
||||
got = repo.get_alert(a1["alert_id"])
|
||||
assert got["alert_type"] == "suitability" and len(got["payload"]["events"]) == 2
|
||||
# 不同产品独立出单
|
||||
req3 = _trade("T3", "100000")
|
||||
req3["product_id"] = "P2"
|
||||
a3 = record_suitability_alert(req3, "SUIT-001", "x", risk_repo=repo)
|
||||
assert a3["alert_id"] != a1["alert_id"]
|
||||
assert _counts(engine, "risk_alert") == 2
|
||||
|
||||
|
||||
def test_aml_standalone_alert_with_compliance_notify(env):
|
||||
repo, pub, _ = env
|
||||
alert = record_aml_alert("C1", {"list_type": "sanction", "match": 0.93}, risk_repo=repo)
|
||||
assert alert["alert_type"] == "aml" and alert["risk_score"] == 95
|
||||
(_, payload), = pub.messages
|
||||
assert "compliance" in payload["notify_role"]
|
||||
|
||||
|
||||
def test_concurrent_first_alerts_single_row(env):
|
||||
"""并发冒烟（B2 验收）：两线程同客户同日首单 → 预警单 1 张、events 含两笔。"""
|
||||
repo, _, engine = env
|
||||
errors = []
|
||||
|
||||
def worker(trade_id):
|
||||
try:
|
||||
record_trade_alerts(
|
||||
_trade(trade_id), [_hit("RISK-001", 70), _hit("RISK-002", 70)], risk_repo=repo
|
||||
)
|
||||
except Exception as exc: # pragma: no cover
|
||||
errors.append(exc)
|
||||
|
||||
threads = [Thread(target=worker, args=(f"T{i}",)) for i in (1, 2)]
|
||||
for t in threads:
|
||||
t.start()
|
||||
for t in threads:
|
||||
t.join()
|
||||
assert not errors, errors
|
||||
assert _counts(engine, "risk_alert") == 1
|
||||
with engine.connect() as conn:
|
||||
events = conn.execute(text("SELECT payload FROM risk_alert")).scalar_one()
|
||||
assert len(json.loads(events)["events"]) == 2
|
||||
Reference in New Issue
Block a user