diff --git a/app/service/risk/profile_l3.py b/app/service/risk/profile_l3.py new file mode 100644 index 0000000..7fd508f --- /dev/null +++ b/app/service/risk/profile_l3.py @@ -0,0 +1,164 @@ +"""L3 监测画像写入(B3 · PRD FR-7 / R-05 最小写入)。 + +合并规则(防降级,PRD FR-7):monitor_tier 取最高档(normal < watch < high), +monitor_tags 追加合并不覆盖,risk_score 取 max,computed_at 每次写当前时间 +(列 NOT NULL 必须显式赋值),last_alert_id 联动最新预警。 + +并发:进程内锁按 customer_id 串行(多进程部署换 Redis SET NX,接口不变); +锁超时降级与跨进程竞态由 IntegrityError 兜底——重读合并后再 update,不丢更新。 +""" + +from __future__ import annotations + +import logging +from datetime import datetime +from typing import Any + +from sqlalchemy.exc import IntegrityError + +from app.repository.risk_repository import RiskRepository +from app.service.risk.alert_service import _run_locked # 同包复用聚合锁;B7 收敛至公共原语 + +logger = logging.getLogger(__name__) + +TIER_ORDER = ("normal", "watch", "high") +ALERT_TYPE_TIER = { + "aml": "high", + "pattern": "watch", + "large_amount": "watch", + "freq_trade": "watch", + "suitability": "normal", +} +AML_PENDING_TAG = "aml_hit_pending_review" + + +def tier_of(alert_type: str) -> str: + """alert_type → L3 档位映射(PRD FR-7 固定映射;未知类型拒绝写入)。""" + tier = ALERT_TYPE_TIER.get(alert_type) + if tier is None: + raise ValueError(f"unknown alert_type for L3 mapping: {alert_type}") + return tier + + +def highest_tier(a: str, b: str) -> str: + """取最高档(防降级核心原语)。""" + return a if TIER_ORDER.index(a) >= TIER_ORDER.index(b) else b + + +def merge_l3( + existing: dict[str, Any] | None, + *, + mapped_tier: str, + risk_score: int | None, + monitor_tags: list[str], + last_alert_id: str | None, + score_dimensions: dict[str, Any] | None = None, + computed_at: datetime | None = None, +) -> dict[str, Any]: + """纯函数:existing 行(repo.get_l3 输出)与新事件合并后的 L3 行。 + + existing 为 None 表示新客户首写;risk_score 双方均空时保持 NULL。 + """ + if existing is None: + tier, old_score, old_tags, old_dims = "normal", None, [], {} + else: + tier = existing.get("monitor_tier") or "normal" + old_score = existing.get("risk_score") + old_tags = existing.get("monitor_tags") or [] + old_dims = existing.get("score_dimensions") or {} + + scores = [int(s) for s in (old_score, risk_score) if s is not None] + return { + "monitor_tier": highest_tier(tier, mapped_tier), + "risk_score": max(scores) if scores else None, + "monitor_tags": sorted(set(old_tags) | set(monitor_tags)), + "score_dimensions": score_dimensions if score_dimensions is not None else old_dims, + "last_alert_id": last_alert_id, + "computed_at": computed_at or datetime.now(), + } + + +def upsert_profile_l3( + customer_id: str, + alert_type: str, + risk_score: int | None = None, + monitor_tags: list[str] | None = None, + last_alert_id: str | None = None, + score_dimensions: dict[str, Any] | None = None, + risk_repo: RiskRepository | None = None, + computed_at: datetime | None = None, +) -> dict[str, Any]: + """预警事件 → L3 upsert(B2 预警落库后调用;aml 自动追加待复核标签)。 + + 返回合并后的 L3 行(含 customer_id)。 + """ + repo = risk_repo or RiskRepository() + mapped_tier = tier_of(alert_type) + tags = list(monitor_tags or []) + if alert_type == "aml" and AML_PENDING_TAG not in tags: + tags.append(AML_PENDING_TAG) + + def _write(locked: bool) -> dict[str, Any]: + existing = repo.get_l3(customer_id) + merged = merge_l3( + existing, + mapped_tier=mapped_tier, + risk_score=risk_score, + monitor_tags=tags, + last_alert_id=last_alert_id, + score_dimensions=score_dimensions, + computed_at=computed_at, + ) + try: + if existing is None: + repo.insert_l3( + customer_id, + merged["monitor_tier"], + merged["risk_score"], + merged["score_dimensions"], + merged["monitor_tags"], + merged["last_alert_id"], + merged["computed_at"], + ) + else: + repo.update_l3( + customer_id, + merged["monitor_tier"], + merged["risk_score"], + merged["score_dimensions"], + merged["monitor_tags"], + merged["last_alert_id"], + merged["computed_at"], + ) + except IntegrityError: + # 首写竞态(锁超时降级/跨进程):另一线程已 insert,重读合并转更新 + if existing is not None: + raise + raced = repo.get_l3(customer_id) + merged = merge_l3( + raced, + mapped_tier=mapped_tier, + risk_score=risk_score, + monitor_tags=tags, + last_alert_id=last_alert_id, + score_dimensions=score_dimensions, + computed_at=computed_at, + ) + repo.update_l3( + customer_id, + merged["monitor_tier"], + merged["risk_score"], + merged["score_dimensions"], + merged["monitor_tags"], + merged["last_alert_id"], + merged["computed_at"], + ) + merged["customer_id"] = customer_id + return merged + + return _run_locked(f"l3:{customer_id}", _write) + + +def get_profile_l3(customer_id: str, risk_repo: RiskRepository | None = None) -> dict[str, Any] | None: + """L3 只读薄封装(对话线/引擎复用;Redis 缓存待 B7 lifespan 一并接入)。""" + return (risk_repo or RiskRepository()).get_l3(customer_id) diff --git a/tests/test_profile_l3.py b/tests/test_profile_l3.py new file mode 100644 index 0000000..1a9fca9 --- /dev/null +++ b/tests/test_profile_l3.py @@ -0,0 +1,247 @@ +"""profile_l3 单测(B3 · 最高档合并防降级 / AML 后大额不回落 / 并发首单)。 + +sqlite StaticPool 单连接共享内存库(同 test_alert_service 模式); +IntegrityError 兜底用 monkeypatch 模拟跨进程首写竞态。 +""" + +from datetime import datetime + +import pytest +from sqlalchemy import create_engine, text +from sqlalchemy.pool import StaticPool + +from app.repository.risk_repository import RiskRepository +from app.service.risk.profile_l3 import ( + AML_PENDING_TAG, + ALERT_TYPE_TIER, + get_profile_l3, + highest_tier, + merge_l3, + tier_of, + upsert_profile_l3, +) + + +@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 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 + ) + """ + ) + ) + repo = RiskRepository(engine=engine) + yield repo, engine + engine.dispose() + + +def _upsert(repo, cid, alert_type, score=None, alert_id=None, tags=None, dims=None): + return upsert_profile_l3( + cid, + alert_type, + risk_score=score, + monitor_tags=tags, + last_alert_id=alert_id, + score_dimensions=dims, + risk_repo=repo, + ) + + +def _row(engine, cid): + with engine.connect() as conn: + return conn.execute( + text( + "SELECT monitor_tier, risk_score, monitor_tags, last_alert_id, computed_at" + " FROM customer_profile_l3 WHERE customer_id = :cid" + ), + {"cid": cid}, + ).mappings().one() + + +# ---------- 映射与纯函数 ---------- + + +@pytest.mark.parametrize(("alert_type", "tier"), sorted(ALERT_TYPE_TIER.items())) +def test_alert_type_tier_mapping(alert_type, tier): + assert tier_of(alert_type) == tier + + +def test_unknown_alert_type_rejected(env): + repo, _ = env + with pytest.raises(ValueError): + _upsert(repo, "C1", "unknown_type", 70) + + +def test_highest_tier_order(): + assert highest_tier("normal", "watch") == "watch" + assert highest_tier("watch", "high") == "high" + assert highest_tier("high", "normal") == "high" + assert highest_tier("normal", "normal") == "normal" + + +def test_merge_new_customer(): + merged = merge_l3(None, mapped_tier="watch", risk_score=70, monitor_tags=["t1"], + last_alert_id="ALT-1") + assert merged["monitor_tier"] == "watch" + assert merged["risk_score"] == 70 + assert merged["monitor_tags"] == ["t1"] + assert merged["last_alert_id"] == "ALT-1" + assert merged["computed_at"] is not None + + +def test_merge_both_scores_none_stays_none(): + existing = {"monitor_tier": "watch", "risk_score": None, "monitor_tags": [], + "score_dimensions": {}} + merged = merge_l3(existing, mapped_tier="watch", risk_score=None, monitor_tags=[], + last_alert_id="ALT-2") + assert merged["risk_score"] is None + + +# ---------- upsert:首写与合并 ---------- + + +def test_first_event_inserts(env): + repo, engine = env + merged = _upsert(repo, "C1", "large_amount", 70, alert_id="ALT-1") + assert merged["monitor_tier"] == "watch" and merged["customer_id"] == "C1" + row = _row(engine, "C1") + assert row["monitor_tier"] == "watch" and row["risk_score"] == 70 + assert row["last_alert_id"] == "ALT-1" and row["computed_at"] is not None + assert get_profile_l3("C1", risk_repo=repo)["monitor_tier"] == "watch" + + +def test_normal_upgrades_to_watch(env): + repo, _ = env + _upsert(repo, "C1", "suitability", 90, alert_id="ALT-0") + merged = _upsert(repo, "C1", "pattern", 80, alert_id="ALT-1") + assert merged["monitor_tier"] == "watch" # normal → watch 升档 + + +def test_watch_does_not_degrade_to_normal(env): + """ Suitability 映射 normal:已 watch 的客户不被 suitability 事件拉低。""" + repo, engine = env + _upsert(repo, "C1", "pattern", 80, alert_id="ALT-1") + merged = _upsert(repo, "C1", "suitability", 90, alert_id="ALT-2") + assert merged["monitor_tier"] == "watch" + assert _row(engine, "C1")["monitor_tier"] == "watch" + + +def test_aml_marks_high_with_pending_review_tag(env): + repo, engine = env + merged = _upsert(repo, "C1", "aml", 95, alert_id="ALT-1") + assert merged["monitor_tier"] == "high" + assert AML_PENDING_TAG in merged["monitor_tags"] + assert AML_PENDING_TAG in _row(engine, "C1")["monitor_tags"] + + +def test_large_amount_after_aml_does_not_fall_back(env): + """B3 验收:AML 后大额 → tier 仍 high、tags 并集、score 取 max。""" + repo, engine = env + _upsert(repo, "C1", "aml", 95, alert_id="ALT-1") + merged = _upsert(repo, "C1", "large_amount", 70, alert_id="ALT-2", tags=["manual_review"]) + assert merged["monitor_tier"] == "high" + assert merged["risk_score"] == 95 # 取 max,不降 + assert set(merged["monitor_tags"]) == {AML_PENDING_TAG, "manual_review"} + assert merged["last_alert_id"] == "ALT-2" # 联动最新 + row = _row(engine, "C1") + assert row["monitor_tier"] == "high" and row["risk_score"] == 95 + + +def test_tags_accumulate_not_overwrite(env): + repo, _ = env + _upsert(repo, "C1", "pattern", 80, alert_id="ALT-1", tags=["freq"]) + merged = _upsert(repo, "C1", "pattern", 80, alert_id="ALT-2", tags=["freq"]) + assert merged["monitor_tags"] == ["freq"] # set 合并去重 + + +def test_score_dimensions_replaced_only_when_passed(env): + repo, _ = env + _upsert(repo, "C1", "aml", 95, alert_id="ALT-1", dims={"amount": 1}) + merged = _upsert(repo, "C1", "large_amount", 70, alert_id="ALT-2") + assert merged["score_dimensions"] == {"amount": 1} # 未传保留旧值 + merged = _upsert(repo, "C1", "large_amount", 70, alert_id="ALT-3", dims={"amount": 2}) + assert merged["score_dimensions"] == {"amount": 2} # 传入则替换 + + +# ---------- 并发与竞态 ---------- + + +def test_concurrent_first_upsert_single_row(env): + """并发冒烟:两线程同客户首单(aml + 大额)→ 1 行、high、tags 并集。""" + import json + from threading import Thread + + repo, engine = env + errors = [] + + def worker(alert_type, alert_id): + try: + _upsert(repo, "C1", alert_type, 70, alert_id=alert_id) + except Exception as exc: # pragma: no cover + errors.append(exc) + + threads = [ + Thread(target=worker, args=("aml", "ALT-A")), + Thread(target=worker, args=("large_amount", "ALT-B")), + ] + for t in threads: + t.start() + for t in threads: + t.join() + assert not errors, errors + with engine.connect() as conn: + count, payload = conn.execute( + text( + "SELECT COUNT(*), GROUP_CONCAT(monitor_tags) FROM customer_profile_l3" + " WHERE customer_id = 'C1'" + ) + ).fetchone() + assert count == 1 + tags = json.loads(payload) + assert AML_PENDING_TAG in tags + + +def test_integrity_error_falls_back_to_remerge(env, monkeypatch): + """跨进程竞态(评审 P2-7③):首读 None、insert 撞主键 → 重读合并转更新,不丢对方写入。""" + repo, engine = env + from sqlalchemy.exc import IntegrityError + + # 对方进程已写入 high(模拟 AML 先落库) + repo.insert_l3("C1", "high", 95, {}, [AML_PENDING_TAG], "ALT-AML", datetime.now()) + + calls = {"n": 0} + orig_get = type(repo).get_l3 + + def racing_get(self, cid): + calls["n"] += 1 + if calls["n"] == 1: + return None # 本进程读写在对方 insert 之前发生 + return orig_get(self, cid) + + monkeypatch.setattr(type(repo), "get_l3", racing_get) + try: + merged = _upsert(repo, "C1", "large_amount", 70, alert_id="ALT-2") + finally: + monkeypatch.undo() + + assert merged["monitor_tier"] == "high" # 重读后合并,不降级 + assert merged["risk_score"] == 95 + assert merged["last_alert_id"] == "ALT-2" + row = _row(engine, "C1") + assert row["monitor_tier"] == "high" and row["last_alert_id"] == "ALT-2"