fix(risk): 合并预警的嵌套证据在研判入口摊平,RW-003 不再一律降级
docs/25 P2。核实后比报告描述的更麻烦一点:合并后的 evidence_snapshot 只留
product_id 与 merged_alerts,**各条规则原有的证据键被塞进了嵌套结构**
(risk_scan_service._merge_same_transaction_alerts:398-404)。而各研判函数读的是
**顶层键**(snapshot.get("ratio") 之类),于是合并过的预警一律读不到证据、降级成
"缺证据无法复核" —— 合并本来是为了少几条噪音,结果把这些预警的研判全废了。
修法:在**研判入口统一摊平**,而不是让每条规则各自去认嵌套结构。
- 新增 _flatten_merged_evidence(detail):顶层已有的键优先(来自 priority_score 最高的
主预警),再按顺序补入各子条目 evidence 里的键;merged_alerts 本身保留,
可追溯性不受影响;非合并结构原样返回。
- 列表级(_assess_list_rule)与详情级(_assess_detail_rule)两个入口都调用它,
所有规则(RW-003/007/012/015/018)一并受益。
新增 tests/unit/service/test_risk_judgement_merged_evidence.py(5 条):非合并结构原样返回、
嵌套键被抬到顶层、冲突时顶层优先、多子条目全部抬平,以及**报告症状的回归** ——
对比"没有 ratio"与"ratio 藏在 merged_alerts 里"两种输入,摊平后 RW-003 的研判结果
不再相同(原先两者都会降级成同一句话)。
ruff / mypy(136 文件) / 637 unit+contract 全绿。
This commit is contained in:
@@ -49,6 +49,7 @@ def assess_alert_detail(detail: dict[str, Any]) -> dict[str, Any]:
|
||||
|
||||
|
||||
def _assess_list_rule(rule: str, item: dict[str, Any]) -> dict[str, Any]:
|
||||
item = _flatten_merged_evidence(item)
|
||||
if rule == "RW-018":
|
||||
# 原先这里**完全不看 item**,无条件返回"可考虑放行",而且理由文本是硬编码的
|
||||
# "现有摘要显示交易来自有效定投工单"。于是列表里那些渠道并不匹配、或证据里
|
||||
@@ -101,7 +102,32 @@ def _assess_list_rule(rule: str, item: dict[str, Any]) -> dict[str, Any]:
|
||||
)
|
||||
|
||||
|
||||
def _flatten_merged_evidence(detail: dict[str, Any]) -> dict[str, Any]:
|
||||
"""把"合并预警"的嵌套证据摊回顶层,再交给各条规则的研判函数。
|
||||
|
||||
同一笔交易命中多条规则时,扫描器会把它们合并成一条
|
||||
(`risk_scan_service._merge_same_transaction_alerts`)。合并后的 `evidence_snapshot`
|
||||
只保留 `product_id` 和 `merged_alerts`,各条**原有的证据键被塞进
|
||||
`merged_alerts[].evidence`**;而各研判函数读的是**顶层键**(`snapshot.get("ratio")`
|
||||
之类),于是合并过的预警一律读不到证据、降级成"缺证据无法复核" ——
|
||||
合并本来是为了少几条噪音,结果把这些预警的研判全废掉了(docs/25 P2)。
|
||||
|
||||
在入口统一摊平:顶层已有的键优先(它来自 priority_score 最高的那条主预警),
|
||||
再按顺序补入各子条目的证据键。所有规则共用这一步,不必各自去认嵌套结构。
|
||||
"""
|
||||
snapshot = _mapping(detail.get("evidence_snapshot"))
|
||||
merged = snapshot.get("merged_alerts")
|
||||
if not isinstance(merged, list):
|
||||
return detail
|
||||
flat = dict(snapshot)
|
||||
for entry in merged:
|
||||
for key, value in _mapping(_mapping(entry).get("evidence")).items():
|
||||
flat.setdefault(key, value)
|
||||
return {**detail, "evidence_snapshot": flat}
|
||||
|
||||
|
||||
def _assess_detail_rule(rule: str, detail: dict[str, Any]) -> dict[str, Any]:
|
||||
detail = _flatten_merged_evidence(detail)
|
||||
if rule == "RW-003":
|
||||
return _assess_rw003(detail)
|
||||
if rule == "RW-007":
|
||||
|
||||
Reference in New Issue
Block a user