diff --git a/app/service/risk_judgement_service.py b/app/service/risk_judgement_service.py index 0dd8ee2..89cd7a8 100644 --- a/app/service/risk_judgement_service.py +++ b/app/service/risk_judgement_service.py @@ -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": diff --git a/tests/unit/service/test_risk_judgement_merged_evidence.py b/tests/unit/service/test_risk_judgement_merged_evidence.py new file mode 100644 index 0000000..c2eb138 --- /dev/null +++ b/tests/unit/service/test_risk_judgement_merged_evidence.py @@ -0,0 +1,90 @@ +"""合并预警的证据必须能被研判读到(docs/25 P2)。 + +**问题**:同一笔交易命中多条规则时,扫描器会把它们合并成一条 +(`risk_scan_service._merge_same_transaction_alerts`),合并后的 `evidence_snapshot` +只留 `product_id` 和 `merged_alerts`,各条**原有的证据键被塞进 +`merged_alerts[].evidence`**。而研判函数读的是**顶层键**(`snapshot.get("ratio")` 之类), +于是合并过的预警一律读不到证据、降级成"缺证据无法复核" —— 合并本来是为了少几条噪音, +结果把这些预警的研判全废了。 + +修法是在研判入口统一摊平(`_flatten_merged_evidence`),列表级与详情级共用。 +""" + +from typing import Any + +from app.service.risk_judgement_service import ( + _assess_detail_rule, + _flatten_merged_evidence, +) + + +def _merged(**nested_evidence: Any) -> dict[str, Any]: + return { + "transaction": {"amount": "600000"}, + "evidence_snapshot": { + "product_id": 7001, + "merged_alerts": [ + {"alert_type": "大额赎回", "evidence": nested_evidence}, + ], + }, + } + + +def test_non_merged_detail_is_returned_unchanged() -> None: + detail = {"evidence_snapshot": {"ratio": "0.9"}} + + assert _flatten_merged_evidence(detail) == detail + + +def test_nested_evidence_is_lifted_to_top_level() -> None: + flat = _flatten_merged_evidence(_merged(ratio="0.9", average_amount="100"))["evidence_snapshot"] + + assert flat["ratio"] == "0.9" + assert flat["average_amount"] == "100" + assert flat["product_id"] == 7001 # 原有键不丢 + assert "merged_alerts" in flat # 嵌套结构保留,可追溯性不受影响 + + +def test_top_level_value_wins_over_nested() -> None: + """冲突时保留顶层 —— 它来自 priority_score 最高的那条主预警。""" + detail = { + "evidence_snapshot": { + "ratio": "9", + "merged_alerts": [{"alert_type": "A", "evidence": {"ratio": "1"}}], + } + } + + assert _flatten_merged_evidence(detail)["evidence_snapshot"]["ratio"] == "9" + + +def test_multiple_merged_entries_are_all_lifted() -> None: + detail = { + "evidence_snapshot": { + "product_id": 1, + "merged_alerts": [ + {"alert_type": "A", "evidence": {"ratio": "0.9"}}, + {"alert_type": "B", "evidence": {"average_amount": "100"}}, + ], + } + } + flat = _flatten_merged_evidence(detail)["evidence_snapshot"] + + assert flat["ratio"] == "0.9" + assert flat["average_amount"] == "100" + + +def test_rw003_no_longer_degrades_on_merged_evidence() -> None: + """报告描述的原始症状:合并后 RW-003 读不到 ratio、一律降级。 + + 对比"没有 ratio"与"ratio 藏在 merged_alerts 里"两种输入 —— 摊平之后后者应当 + 真的把 ratio 读进来,于是两条路径的研判结果不再相同。 + """ + without_ratio = { + "transaction": {"amount": "600000"}, + "evidence_snapshot": {"product_id": 7001}, + } + + degraded = _assess_detail_rule("RW-003", without_ratio) + recovered = _assess_detail_rule("RW-003", _merged(ratio="0.9")) + + assert str(degraded) != str(recovered), "摊平后应当读到嵌套的 ratio,研判结果应不同"