Files
group_xinghuo_jinrong/docs/项目框架设计/实现方案-风控追加需求v1.1-C4C6.md
T

34 KiB
Raw Blame History

实现方案 · 风控追加需求 v1.1(FR-8 / FR-9 / FR-10 → C4~C6)

依据:PRD v1.1 §4A(已冻结并入)· 附-风控规则表 v1.1 · 开发计划 v1.2 C4C6 任务行 + 挂账 #1#9 分支 risk-control-agent(旧称 feature/risk 已过时)· 基线:T-21 完成后(425 用例 collect 已验证,全量执行绿待跑) 本文为编码依据,函数签名/文件路径可直接照抄;与 PRD 冲突时以 PRD 为准。

评审修订记录(v1.1 · 2026-09-07):经独立 AI 评审 FAIL → 修订本文。P0-1 C4 聚合锚点与 C6 行为链单碰撞(§2.1/§2.5 锚点排除 agent_behavior);P1-1 升级 payload 写入统一读改写(§3.2);P1-2 回归机制改 autouse monkeypatch settings + from_settings 补读字段(§2.3/§6.2);P1-3 追加时 alert_type 按合并后规则重算(§2.5);P1-4 L2 推送通知链累积(§3.3);P2-1~P2-6 顺手收口(§2.2/§3.5/§4.2/§4.3/§6.2)。


0. 三个需求一句话 + 落库前提核实结论

需求 规则 一句话 触发方式 优先级
FR-8 RISK-006 客户持仓 R4+R5 市值占比 ≥ 阈值(默认 80%)→ 并入当日事件类预警单 交易落库后同步(引擎内追加) P1
FR-9 RISK-007 pending 单超时(4h/24h,AML 1h/4h)→ payload 写升级标记 + 推送升级,不改 status 定时扫描 15 分钟 P0
FR-10 RISK-008 代理人行为链(A 诱导调仓 / B 越权试探 / C 越权查询)→ 代理人维度独立预警单 定时扫描 30 分钟 P1

DDL 核实结论(决定方案形态,均已确认):

项 结论 对方案的影响
risk_alert.alert_type ENUM 五值,无 concentration/agent_behavior 一律落 pattern + payload.alert_subtype 区分
risk_alert.status ENUM 四值,无 escalated 升级信息全由 payload.escalation_level/escalated_at 承载
risk_alert.payload JSON NOT NULL 直接写 dict(repo _dump_alert 已序列化)
audit_log.event_type VARCHAR(64)(非 ENUM) risk_concentration/alert_escalation/agent_behavior_detected 三个新事件类型可直接写入,无需改表
core_product.min_risk_code CHAR(2),取值 R1~R5 RISK-006 判定 min_risk_code in ("R4","R5")(sqlite 测试 DDL 为 VARCHAR(8),同口径兼容)
agent_session/Agent 准入 AGENT_ACCESS_MATRIX["risk"].roles 现为 ("risk_officer","service_risk") risk_manager 需增补(见 §3.3,注意对话线例外)

1. 模块与文件划分总表

1.1 新增文件(6 个)

文件 职责 所属任务
app/service/risk/escalation_service.py RISK-007 扫描 + 升级决策 + 幂等 payload 写入 + 降噪推送(纯 service,可被脚本/测试直调) C5
app/service/risk/agent_behavior_service.py RISK-008 滑动窗口聚合 audit_log + 行为链判定 + 出单(纯 service) C6
scripts/cron/escalation_scan.py 定时任务壳:sys.path 引导 → new_trace() → 调 escalation_service → 打印摘要(15min 周期挂系统 cron) C5
scripts/cron/agent_behavior_scan.py 同上(30min 周期) C6
tests/test_escalation_service.py C5 单测(含幂等/边界/降噪) C5
tests/test_agent_behavior_service.py C6 单测(A/B/C 边界 + 本人排除 + 去重) C6

1.2 修改文件(10 个)

文件 改动 所属任务
app/config/settings.py 追加 11 个配置项(§5) C4~C6
.env.example 同步登记(含注释) C4~C6
app/service/risk/rules.py RuleHit 加 alert_subtype 字段;新增 rule_concentration 纯函数;RiskThresholds 加 concentration_threshold C4
app/service/risk/engine.py process_trade_event 追加持仓规则分支(RISK-001~005 之后、AML 之前) C4
app/service/risk/alert_service.py record_trade_alerts 维护 payload.alert_subtype;_publish_alert 支持覆盖 notify_role/附带 subtype C4(C5/C6 复用)
app/repository/risk_repository.py 新增 append_alert_subtypes / update_alert_escalation / list_pending_alerts_all / find_agent_behavior_alert / list_audit_events C5/C6
app/repository/core_ro.py 新增 concentration_profile(customer_id, limit=500)(R4+R5 聚合一次 SQL 搞定,避免 Python 端 N+1) C4
app/api/deps.py AGENT_ACCESS_MATRIX["risk"]["roles"] 增补 risk_manager C5
app/api/risk.py list_alerts_api 加 risk_manager 全量只读分支;chat 相关不动 C5
app/api/chat.py risk 分支准入后加显式守卫:roles 含 risk_manager → deny(对话线维持仅 risk_officer) C5
app/service/risk/chat_tools.py customer_context 扩展 concentration_ratio;新增 query_overdue_alerts / query_agent_behavior 两个 Tool + 注册表条目 C4/C5/C6
app/service/tool_service.py _INTENT_KEYWORDS["risk"] 追加两条意图词组;summarize 加三个 Tool 摘要分支 C4/C5/C6
app/gateway/trade_gateway.py submit_trade 加可选参数 actor_id,透传到 _audit(C6 条件 A 数据源前提) C6
app/api/simulate.py 调 submit_trade 时传 actor_id=auth.actor_id C6
scripts/core/02-seed-base.sql 追加 STAFF-31001/31002(staff_type='risk_officer',roles='["risk_manager"]') C5 前置
tests/conftest.py 新增 backdated_alert / backdated_audit_event fixture(免等待测试基建) C5/C6

chat_tools.py / tool_service.py 三个任务都碰,按 C4→C5→C6 顺序合并提交,避免冲突。


2. C4 · RISK-006 集中度预警(FR-8)

2.1 数据流转

交易落库 → engine.process_trade_event
  ├─ run_rules(RISK-001~005)                (现有,不动)
  ├─ [新增] core.concentration_profile(customer_id)
  │    → rule_concentration(profile, th) → RuleHit(alert_subtype="concentration")
  │    命中 → 并入 hits,走既有 record_trade_alerts(聚合锚点 = find_pending_event_alert
  │           **排除 payload.alert_subtype 含 'agent_behavior' 的单**——客户维度事件单与
  │           代理人维度行为链单是两条出单线,聚合并存、互不影响,PRD 4A.0 #3;
  │           评审 P0-1 收口,见 §2.5 锚点过滤)
  │    + L3:upsert_profile_l3(customer_id, "pattern", monitor_tags=["high_risk_concentration"])
  │    + 审计:event_type='risk_concentration'(金额 DESENS-005 截断)
  └─ match_customer(AML)                    (现有,不动)

关键点:RISK-006 不新建出单路径——并入 record_trade_alerts 既有聚合,risk_score=60 经 max() 语义不覆盖 RISK-001/002 的 70;alert_type 随最高分规则动态更新的现有逻辑不变。

2.2 core_ro 新增:concentration_profile(挂账 #1 收口)

def concentration_profile(self, customer_id: str, limit: int = 500) -> dict[str, Any]:
    """持仓集中度画像(RISK-006 专用,一次 SQL 聚合,避免 N+1)。

    返回 {r45_value: Decimal, total_value: Decimal, holdings_truncated: bool, rows: [...]}。
    rows 仅保留 market_value 降序前 limit 条明细(进审计 input_summary 用摘要,不进 payload 全量)。
    触及 limit → holdings_truncated=True(调用方按保守口径视同达标,PRD FR-8 截断防护)。
    """
    # SQL:SELECT h.market_value, p.min_risk_code FROM core_holding h
    #      JOIN core_product p ... WHERE h.customer_id=:cid
    #      ORDER BY h.market_value DESC LIMIT :lim+1   ← 多取 1 行判定截断
    # Python 端按 min_risk_code in ("R4","R5") 分组求和(Decimal)

采用「SQL 取明细 + Python 聚合」而非 SUM(CASE):明细行还要进审计摘要,且 sqlite/MySQL DDL 差异下 Python 判 min_risk_code 最稳。limit+1 探测截断,避免二次 COUNT 查询。 注:PRD FR-8 字面为「调用 core_ro.list_holdings」;本方法为挂账 #1 的收口(聚合封装,内部同源查询,市值口径不变),偏离字面已在方法 docstring 注明。

2.3 rules.py 改动

@dataclass(frozen=True)
class RuleHit:
    rule_id: str
    alert_type: str
    risk_score: int
    detail: str
    alert_subtype: str | None = None   # 新增:RISK-006→"concentration",RISK-008→"agent_behavior",RISK-001~005 恒 None

RULE_SCORES["RISK-006"] = 60
RULE_ALERT_TYPES["RISK-006"] = "pattern"

@dataclass(frozen=True)
class RiskThresholds:
    ...
    concentration_threshold: float = 0.80          # settings.risk_concentration_threshold

# from_settings() 必须同步补读:concentration_threshold=s.risk_concentration_threshold
# (评审 P1-2:引擎/网关测试全部走 from_settings 默认路径,漏读字段会导致
#   autouse 回归 fixture 的 monkeypatch 失效)

def rule_concentration(profile: dict[str, Any], th: RiskThresholds) -> RuleHit | None:
    """RISK-006 集中度:R4+R5 市值占比 ≥ 阈值(纯函数)。

    total_value == 0(空仓)不触发;holdings_truncated=True 视同达标(保守告警)。
    detail 含 r45_value/total_value/占比(Decimal→str)。
    """

不动 run_rules 签名——RISK-001~005 吃 trades、RISK-006 吃持仓画像,输入域不同;引擎分别调用后合并 hits,比改 run_rules 兼容面小(现有 13 处 run_rules 调用/断言零改动)。

2.4 engine.py 接入点

# process_trade_event 内,hits = run_rules(...) 之后:
profile = core.concentration_profile(trade["customer_id"])          # 截断防护在 rule 内判定
conc = rule_concentration(profile, th)
if conc:
    hits = [*hits, conc]        # 并入后走 record_trade_alerts 既有聚合/审计/推送
    upsert_profile_l3(trade["customer_id"], "pattern",
                      monitor_tags=["high_risk_concentration"],
                      last_alert_id=... )                            # 聚合出单后补(同现有 best 分支)
    _audit_concentration(repo, trade, profile, conc)                 # event_type='risk_concentration'

注意顺序:RISK-006 分支放在 run_rules 之后、record_trade_alerts 之前,使 record_trade_alerts 一次调用覆盖合并;L3 upsert 与专属审计放在出单之后(拿 alert_id)。命中 RISK-006 但 RISK-001~005 未命中时,record_trade_alerts 同样正常建单(hits 非空即走聚合分支),无需为「仅集中度命中」写独立分支。

2.5 alert_service.py 改动

① 聚合锚点排除代理人维度单(评审 P0-1 收口,必须做)

RiskRepository.find_pending_event_alert 改为「查候选 + Python 过滤」:

def find_pending_event_alert(self, customer_id: str, day_start: datetime) -> dict | None:
    """当日该客户的事件类 pending 单(聚合锚点)——**排除代理人维度行为链单**。

    LIMIT 1 会取到最新 pattern 单;若它恰是 agent_behavior 单而更早还有客户维度
    事件单,会误判无锚点导致当日出第二张客户维度单。故候选取 LIMIT 50 再过滤:
    SELECT * FROM risk_alert
    WHERE customer_id=:cid AND status='pending_review'
      AND alert_type IN (事件类) AND created_at>=:day_start
    ORDER BY created_at DESC LIMIT 50
    → Python 过滤:'agent_behavior' not in (payload.get("alert_subtype") or []) → 取最新
    """

依据 PRD §4A.0 #3:FR-4「同客户同自然日仅一张」只对客户维度事件单成立,RISK-008 行为链单(同 pattern 但 payload.actor_type='agent')是独立出单线,两者不得互相并入。C6 的 find_agent_behavior_alert 已按 subtype 过滤,天然安全(单向污染只在 C4 侧)。

② alert_subtype 集合维护

subtypes = sorted({h.alert_subtype for h in hits if h.alert_subtype})
# 新建单:仅当 subtypes 非空才写 payload["alert_subtype"] = subtypes(空集不注入该键,评审 P2-3)
# 追加单(锚点命中):risk_repo.append_alert_event(..., extra_subtypes=subtypes)

③ 追加时 alert_type 按合并后规则重算(评审 P1-3,既有缺陷顺带修正)

现状缺陷:append_alert_event 用「本批 best.alert_type」更新——既有 large_amount(70) 单,本批仅命中 RISK-006(60) 时会被错误翻转为 pattern。修法(alert_service 追加分支):

merged_rules = sorted(set(pending["triggered_rules"]) | {h.rule_id for h in hits})
best_rule = max((r for r in merged_rules if r in RULE_SCORES), key=lambda r: RULE_SCORES[r])
risk_repo.append_alert_event(..., risk_score=max(old, 本批max), alert_type=RULE_ALERT_TYPES[best_rule])

(RULE_SCORES/RULE_ALERT_TYPES 从 rules.py 导入——alert_service 已依赖 rules.RuleHit,无新增分层问题。编码后跑全量回归确认无现有断言依赖旧翻转行为。)

  • RiskRepository.append_alert_event 加可选参数 extra_subtypes: list[str] | None = None:合并进 payload.setdefault("alert_subtype", [])(sorted set 并集),老单无该字段时首次追加自动创建。向后兼容:不传时行为与现在完全一致,现有用例不受影响。
  • _publish_alert 加可选参数 notify_role: list[str] | None = None, extra: dict | None = None:缺省保持 ["risk_officer"](+aml compliance) 现状;extra 合并进推送体(C5 放 escalation_level,C4/C6 放 alert_subtype)。02-redis-keys.md 同步增补字段说明(PRD §13 联动清单)。

2.6 对话线:customer_context 扩展

chat_tools.customer_context 返回体追加:

"profile": {"concentration_ratio": 0.9, "r45_value": "...", "total_value": "...",
            "holdings_truncated": False}   # 复用 core_ro.concentration_profile,脱敏后百分比

调用同一聚合函数,不重复实现口径。tool_service.summarize 的 customer_context 分支加一句「高风险持仓占比 90%(仅供参考)」。意图词无需新增(customer_context 词组已有)。


3. C5 · RISK-007 时效升级(FR-9)

3.1 鉴权与角色联动(C5 开工前置,挂账 #6)

  1. 种子:scripts/core/02-seed-base.sql 追加:
INSERT INTO core_staff VALUES ('STAFF-31001','风控经理甲','risk_officer','["risk_manager"]',1,...);
INSERT INTO core_staff VALUES ('STAFF-31002','风控经理乙','risk_officer','["risk_manager"]',1,...);
  1. app/api/deps.py:AGENT_ACCESS_MATRIX["risk"]["roles"] → ("risk_officer", "risk_manager", "service_risk")。理由:HTTP 台账请求经 get_auth_context 的交叉校验,不加则 manager 连 GET /api/risk/alerts 都过不去(401 通道 AUTH_403_AGENT_MISMATCH)。
  2. app/api/chat.py(对话线例外):risk 分支在 assert_agent_access 之后加守卫:
if "risk_manager" in auth.roles:
    deny(auth, "AUTH_403_ROLE", repo, message="对话线仅限 risk_officer,请走 HTTP 台账")

这是对 PRD 4A.1「AGENT_ACCESS_MATRIX 增补行」与「对话线不放行 risk_manager」两条的联合落地:矩阵放行解决 HTTP 通道,chat 层显式拒绝保住 FR-6 冻结口径。Tool 层 assert_tool_access 不动(manager 根本进不了对话线,天然 fail-closed)。

  1. app/api/risk.py::list_alerts_api:
if auth.has_role("risk_officer"):
    ...  # 现有全量
elif auth.has_role("risk_manager"):
    ...  # 全量只读(同 risk_officer 查询路径,无处置入口)
elif auth.has_role("compliance"):
    ...  # 现有 aml 强制过滤
else:
    deny(...)

handle_alert_api 白名单不动(manager 无 risk_officer 角色 → 自然 403,PRD「零代码改动」)。JWT 手册 §5.3 增补为文档动作,随 C5 提交。

3.2 仓储新增(risk_repository.py)

def list_pending_alerts_all(self, page_size: int = 1000) -> list[dict]:
    """全部 pending_review 单(C5 扫描输入;演示规模一次取回,Python 端判级)。
    复用 _parse_alert。"""

def update_alert_escalation(self, alert_id: str, escalation_level: int,
                            escalated_at: datetime, trace_id: str) -> bool:
    """payload 升级标记独占写入(RISK-007;status/handler_* 列一律不碰)。

    实现:**统一读改写**(评审 P1-1 定案)——SELECT payload → Python 合并
    escalation_level/escalated_at/escalation_trace_id → UPDATE payload。
    与现有 append_alert_event 同模式(跨 MySQL/sqlite 已验证可行),不引入
    JSON_SET 方言分支;并发窗口由「单发定时脚本 + 幂等闸门(仅升不降)」兜底。
    WHERE alert_id=:aid,返回 rowcount==1。"""

人工处置 API 走 handle_alert_with_audit(只写 status/handler_*/handled_at 列),与 update_alert_escalation 写集不相交,互不覆盖——对应 A-11「payload 与人工处置字段互不覆盖」断言。

3.3 escalation_service.py(核心职责)

"""预警处置时效升级(C5 · PRD FR-9 / 规则表 v1.1 RISK-007 补充约束)。"""

def scan_and_escalate(
    now: datetime | None = None,
    risk_repo: RiskRepository | None = None,
    thresholds: EscalationThresholds | None = None,
) -> dict[str, Any]:
    """单次扫描入口(定时脚本/测试/手动演示共用)。

    EscalationThresholds(frozen dataclass, from_settings()):
      l1_hours=4, l2_hours=24, aml_l1_hours=1, aml_l2_hours=4

    流程:
    1. rows = repo.list_pending_alerts_all()
    2. 逐单计算 computed_level:
       overdue_h = (now - created_at).total_seconds()/3600
       aml 单用 aml_l1/l2,普通单用 l1/l2;level = 2 if overdue>=l2 else 1 if overdue>=l1 else 0
    3. 幂等闸门:computed_level > (payload.escalation_level or 0) 才动作;
       同级/降级一律跳过(每级别至多推送一次,A-11 幂等断言锚点)
    4. 动作顺序(LEVEL_1 与 LEVEL_2 同机制,PRD 拍板「先持久化再推送」):
       a. repo.update_alert_escalation(alert_id, level, now, trace_id)
       b. 审计 event_type='alert_escalation'(input_summary 含 alert_id/已达级别/
          超时时长/升级原因;trace 贯通)
    5. 降噪合并:按 (customer_id, level) 分组,同组多单合并为一次推送
       (alert_ids 列表进推送体;审计仍逐单落,不合并)
    6. 推送:redis_gateway.publish("risk:pub:alert", {..., "escalation_level": lvl,
       "notify_role": 通知链**累积**(评审 P1-4 修正):
       L1 → ["risk_officer", "risk_manager"]
       L2 → ["risk_officer", "risk_manager", "compliance"]   ← manager 不因升到 L2 而移出
       → 复用 alert_service._publish_alert 的 extra/notify_role 扩展
    7. 任务级审计一条:decision='scan_completed',input_summary={scanned, escalated, merged, skipped}
    返回 {"scanned": n, "escalated": [...], "merged_notices": m, "skipped": k}
    """

边界细则(编码时执行):

  • created_at 为 naive datetime(repo 直返),now 缺省 datetime.now(),同口径相减。
  • AML 判定按 alert_type == "aml";suitability 单同样纳入扫描(属 pending_review,超时同样积压——PRD 未排除,从「时效监控」本义)。
  • 升级后单据 status 保持 pending_review,仍在下轮扫描范围(level 已达 2,幂等闸门自动跳过);人工处置后自然退出(status ≠ pending_review)。
  • escalation_level 读缺省 0:int((alert["payload"] or {}).get("escalation_level") or 0)。

3.4 定时任务脚本

scripts/cron/escalation_scan.py(结构对齐 scripts/demo/rebuild_alerts.py):

"""RISK-007 处置时效升级扫描(建议 15min 周期;RISK_ESCALATION_SCAN_MINUTES 可配)。
初期独立脚本 + 系统 cron;内嵌 lifespan 归 M4 评估(开发计划挂账 #2)。"""
# ① sys.path 引导项目根(rebuild_alerts 先例)
# ② from app.utils.trace import new_trace; new_trace()   ← 显式生成 trace(无 HTTP 上下文)
# ③ scan_and_escalate() → print JSON 摘要(供 cron 日志/演示走查)
# 审计写库失败:service 内降级 + logger.error 留底(红线 5 口径,不阻塞下次扫描)

3.5 对话 Tool:query_overdue_alerts

def query_overdue_alerts(customer_id, core_ro=None, risk_repo=None, **params):
    """超期 pending 单列表(只读;FR-9 API 变更)。
    hours = params.get("hours")  # 缺省取 settings L1 阈值
    → repo.list_pending_alerts_all() → 过滤 overdue_h >= hours
    → [{alert_id, customer_id, alert_type, created_at, escalation_level(读 payload,缺省0),
        overdue_hours(round 1位), risk_score}],按 overdue_hours 降序(对话线查询置顶语义)
    """
# 注册表条目:requires_customer=False(risk_officer 无绑定客户可查全量,C2 守卫先例)
#             param_whitelist=("hours",), int_bounds={"hours": (1, 720)}

意图词组:("query_overdue_alerts", ("超期", "超时", "逾期", "多久没处理", "处置时效"))——置于 risk 组 alert_query 之前(评审 P2-4:「超时预警」类话术不得误命中台账)。


4. C6 · RISK-008 代理人行为链(FR-10)

4.0 前置改造:交易审计 actor 透传(条件 A 数据源前提)

现状:trade_gateway._audit 写 actor_id="SYSTEM"(B6 挂账注释),条件 A 无法归属发起人。

# trade_gateway.submit_trade 加参数:actor_id: str | None = None
# _audit(..., actor_id=actor_id or "SYSTEM")   ← 缺省 SYSTEM,现有测试/脚本零改动
# app/api/simulate.py:submit_trade(req, ..., actor_id=auth.actor_id)

演示期 risk_demo 账号(roles 含 risk_demo)扮演代理人发起 → audit_log.trade_request 行的 actor_id = STAFF-90001;客户本人发起时 actor_id == customer_id,扫描时排除(防误报,A-12 断言)。

4.1 仓储新增

def list_audit_events(self, event_type: str, since: datetime, actor_id: str | None = None,
                      decision: str | None = None, limit: int = 2000) -> list[dict]:
    """audit_log 滑窗查询(RISK-008 数据源;input_summary 解析回 dict)。
    WHERE event_type=:et AND created_at>=:since [AND actor_id=:aid] [AND decision=:d]
    ORDER BY created_at ASC。条件 B/C 用 actor_id+decision='forbidden';
    条件 A 用 event_type='trade_request' 全量(actor 过滤在 Python 端做,排除 SYSTEM)。"""

def find_agent_behavior_alert(self, actor_id: str, day_start: datetime) -> dict | None:
    """同代理人当日 pending 行为链单(去重锚点)。
    SQL:SELECT * FROM risk_alert WHERE status='pending_review' AND alert_type='pattern'
         AND created_at>=:day_start ORDER BY created_at DESC(演示规模可接受);
    Python 过滤 payload.alert_subtype=='agent_behavior' and payload.actor_id==actor_id。
    不用 payload LIKE:JSON 键值歧义 + 序列化空格差异两坑(挂账 #9 登记性能,一期 Python 过滤最稳)。"""

4.2 agent_behavior_service.py(核心职责)

"""代理人异常行为链识别(C6 · PRD FR-10 / 规则表 v1.1 RISK-008 补充约束)。

数据源(定案口径,input_guard_log 不作为数据源):
  条件 A:audit_log(event_type='trade_request') ∩ core_trade 时间线——**join 键 = trade_id**
          (评审 P2-2 补细):审计行 input_summary.trade_id 归属发起人 actor_id;
          扫描先取 trade_request 行 → 过滤 actor_id=='SYSTEM'(历史无归属)与
          actor_id==customer_id(本人交易)→ 以行内 trade_id 集合从 core_trade
          (list_trades_range 当日窗 + get_trade_by_id 补漏)取 traded_at/trade_type/
          product_id 组装时间线 → 模式判定只在「有归属 actor 的 trade_id 子集」内做
  条件 B:audit_log(event_type='authz', decision='forbidden') 且 input_summary.code=='AUTH_403_SCOPE'
  条件 C:同 B 族,code IN ('AUTH_403_NOT_OWNER','AUTH_403_NOT_ASSIGNED')
  (对话线 blocked 经 record_authz_denial 双写 audit_log,天然同口径可查)
"""

@dataclass(frozen=True)
class BehaviorThresholds:
    a_window_hours: int = 24;  a_count: int = 3
    b_window_hours: int = 72;  b_count: int = 5
    c_window_hours: int = 24;  c_count: int = 10
    # from_settings() 对齐 settings.risk_agent_behavior_*

def detect_hits(now, core, repo, th) -> dict[str, dict[str, Any]]:
    """按代理人分组聚合三条件证据。返回 {actor_id: {"A": [...], "B": [...], "C": [...]}},
    每条证据含 {trace_id, at, customer_id, detail}。纯查询,无副作用(单测友好)。"""

def scan_and_alert(now=None, core_ro=None, risk_repo=None, thresholds=None) -> dict:
    """扫描入口(脚本/测试共用;模式对齐 escalation_service)。
    1. detect_hits 收集证据
    2. 逐 actor:
       existing = repo.find_agent_behavior_alert(actor_id, day_start)
       ├─ 有 → 新子条件才 append_alert_event(triggered_rules 追加 "RISK-008",
       │        payload.events 追加证据、alert_subtype 集合并集——同日多子条件只一张单)
       └─ 无 → 出单:
            alert_type='pattern', triggered_rules=["RISK-008"], risk_score=75
            customer_id = 证据条数众数客户(PRD:涉及客户语义,payload 显式声明)
            payload = {alert_subtype:'agent_behavior', actor_id, actor_type:'agent',
                       subtypes_hit:['A'|'B'|'C'], timeline:[...], customers:[...],
                       evidence_trace_ids:[...], note:'customer_id 为涉及客户非归属客户'}
       L3 不写(代理人画像本期仅 payload 承载,挂账 #3)
    3. 推送 notify_role=["risk_officer","risk_manager"](_publish_alert extra 带 alert_subtype)
    4. 审计 event_type='agent_behavior_detected'(每 actor 每次命中一条;显式 new_trace 由脚本入口给)
    返回 {"scanned_windows": ..., "hits": [...], "created": [...], "appended": [...]}
    """

边界细则:

  • 条件 A 模式判定:归属映射 audit 行 → (actor_id, customer_id, trade_id),trade_id 经 core_trade 取时间线;同 (actor, customer) 组内,对每个 redeem 事件找 2h 内该 customer 的 subscribe 且 product_id 不同 → 计一次诱导调仓;同一 redeem 不重复配对(消费制)。窗口按 core_trade.traded_at(事件时点,非扫描墙钟,与 RISK-004 rebuild 口径一致)。
  • 窗口均为滑动窗口 created_at >= now - window,自然日去重键只用于出单(PRD 明确两套口径并存)。
  • 行为链明细展示脱敏:payload 存 customer_id(内部键不脱敏);Tool 输出层对 display_name 走 utils/desensitize 二次脱敏(PRD 系统动作 6)。

4.3 对话 Tool:query_agent_behavior

def query_agent_behavior(customer_id, core_ro=None, risk_repo=None, **params):
    """代理人行为链查询(只读)。agent_id = params.get("agent_id")(可缺省=全部)。
    rows, _ = repo.list_alerts(alert_type='pattern', status=None, page_size=100)  # 注意返回 tuple(评审 P2-1)
    → Python 过滤 payload.alert_subtype=='agent_behavior'(+ actor_id==agent_id 若传)
    → [{alert_id, actor_id, subtypes_hit, customers(脱敏展示), created_at,
        risk_score, status, evidence_trace_ids[:5]}]
    可见性:对话线准入已限 risk_officer(manager 走 HTTP 台账),Tool 层天然 fail-closed。
    """
# 注册表:requires_customer=False, param_whitelist=("agent_id",), int_bounds={}
# 意图词组:("query_agent_behavior", ("代理人", "行为链", "异常行为", "诱导", "越权记录"))

意图词注意与现有 risk 组内 alert_query("预警")词序:query_agent_behavior 放 alert_query 之前("代理人预警"应命中行为链而非台账);"预警"仍是 alert_query 专属词,无冲突。

4.4 定时脚本

scripts/cron/agent_behavior_scan.py:结构同 escalation_scan.py(sys.path 引导 → new_trace() → scan_and_alert() → JSON 摘要打印)。


5. 配置项清单(settings.py + .env.example,C4~C6 开工时一次性加齐)

# ===== 风控追加 v1.1(FR-8/9/10 · PRD §4A)=====
risk_concentration_threshold: float = 0.80            # FR-8 R4+R5 占比阈值
risk_escalation_scan_minutes: int = 15                # FR-9 扫描周期(cron 侧参考)
risk_escalation_l1_hours: int = 4                     # FR-9 普通单 LEVEL_1
risk_escalation_l2_hours: int = 24                    # FR-9 普通单 LEVEL_2
risk_escalation_aml_l1_hours: int = 1                 # FR-9 AML 通道 L1
risk_escalation_aml_l2_hours: int = 4                 # FR-9 AML 通道 L2
risk_agent_behavior_scan_minutes: int = 30            # FR-10 扫描周期
risk_agent_behavior_a_window_hours: int = 24          # FR-10 条件 A
risk_agent_behavior_a_count: int = 3
risk_agent_behavior_b_window_hours: int = 72          # FR-10 条件 B
risk_agent_behavior_b_count: int = 5
risk_agent_behavior_c_window_hours: int = 24          # FR-10 条件 C
risk_agent_behavior_c_count: int = 10

6. 测试方案(A-10 / A-11 / A-12)

6.1 conftest 新增 fixture(免等待基建)

@pytest.fixture()
def backdated_alert(sqlite_engine):
    """注入 created_at 回拨的 pending 单(C5);teardown 按 alert_id 精确删除。
    用法:alert_id = backdated_alert(alert_id="ALT-TEST-...", customer_id=..., hours_ago=5, payload={...})"""

@pytest.fixture()
def backdated_audit_event(sqlite_engine):
    """注入 created_at 回拨的 audit_log 行(C6);trace_id 用 'TEST-TRACE-' 前缀,
    teardown DELETE WHERE trace_id LIKE 'TEST-TRACE-%'(不污染正常 trace 空间)。"""

真 MySQL 集成测试沿用 risk_demo_env;回拨行 teardown 注意 _cleanup_test_rows 按 created_at >= started_at 清不掉回拨行 → 两个 fixture 必须自带精确清理(alert_id / trace 前缀),这是评审必查点。

6.2 用例清单

文件 覆盖
tests/test_rules.py(扩) rule_concentration:90% R5 命中 / 100% R1 不触发 / 空仓不触发 / 截断视同达标 / 阈值边界(79.9% 不触发 / 80% 触发)
tests/test_engine.py(扩) 引擎合并:RISK-006 与 RISK-001 同单聚合、risk_score=max(60,70)=70、payload.alert_subtype 含 concentration、L3 tag 追加、risk_concentration 审计
tests/test_alert_service.py(扩) append_alert_event extra_subtypes 合并(老单无字段首次创建);P0-1 回归:当日已有 agent_behavior 单后再触发 RISK-006 → 出第二张客户维度单,两单 payload 互不污染;alert_type 合并重算(large_amount 单 + 仅 RISK-006 追加 → 保持 large_amount 不翻转);现有断言回归
tests/test_escalation_service.py(新) 3h59m 不升 / 4h 升 L1 / 24h 升 L2 / AML 1h 短通道(两条断言分开)· 同单重复扫描不重复推送(幂等)· 同客户多单同级别合并一次推送 · 阈值内处置不升级 · 处置后退出扫描 · status 全程 pending_review · notify_role 断言(L1 含 risk_manager / L2 含 compliance)· update_alert_escalation 不碰 handler 列
tests/test_agent_behavior_service.py(新) 条件 A:2 次不触发 / 3 次触发 / 本人交易(actor==customer)不计数 / 跨产品 2h 窗口边界 / SYSTEM actor 排除;条件 B:4/5 边界 + code 精确匹配;条件 C:9/10 边界 + 双 code 并集;同日多子条件只一张单(subtypes_hit 并集);customer_id=众数;payload.actor_id 正确指向代理人
tests/test_chat_tools.py(扩) customer_context 带 concentration_ratio;query_overdue_alerts 边界与缺省 hours;query_agent_behavior 过滤与 agent_id 匹配
tests/test_risk_api.py(扩) risk_manager:GET alerts 200 全量 / handle 403 / suitability/check 403 / aml/scan 403(评审 P2-5:防未来新增端点只依赖矩阵漏加角色校验的回归断言) / chat risk 入口 deny(AUTH_403_ROLE)/ debug 头与 JWT 通道各一条
集成(risk_demo_env) A-10(TRD-TEST- 交易落库后 5s 内出单)/ A-11 / A-12 按验收表逐条

全量回归口径(PRD A-10 预声明 · 评审 P1-2 修订):conftest 增加自动生效的回归隔离 fixture(autouse,作用于全部单测):

@pytest.fixture(autouse=True)
def _disable_concentration_rule(monkeypatch):
    """RISK-006 阈值推到不可达,现有用例断言零改动;RISK-006 专测内再手动改回真实阈值。"""
    monkeypatch.setattr(settings, "risk_concentration_threshold", 1.01)

前置条件:RiskThresholds.from_settings() 必须补读 concentration_threshold 字段(引擎/网关测试全部走 from_settings() 默认路径,漏读则 monkeypatch 失效、425 绿被 RISK-006 打穿)。RISK-006 专属测试显式传 RiskThresholds(...) 或再次 monkeypatch 真实阈值 0.80。


7. 实现优先级与执行顺序

步骤 内容 理由
0 C5 前置联动:seed 两个 manager 账号 + deps 矩阵 + chat 守卫 + risk.py 台账分支 + JWT 手册文档 PRD「C5 开工前执行」;独立可测(不依赖 C4)
1 C4(rules + core_ro + engine + alert_service + customer_context + settings) 改动最小、复用度最高、无新基建;先拿下一个完整验收(A-10)
2 C5(repo 方法 + escalation_service + cron 脚本 + Tool + conftest fixture) 建立定时任务基建与 fixture 模式,C6 直接复用;P0 红线规则
3 C6(trade_gateway actor 透传 + repo 审计查询 + agent_behavior_service + Tool + 脚本) 前置改造最多(actor 链路),基建已由 C5 备好
4 演示 SOP 补章节(risk_demo 扮演代理人步骤)+ 02-redis-keys.md 增补 + MEMORY/TODO 收口 随各任务顺手,最后统一核对挂账 #1~#9

每个任务独立 commit(C4 → feat: T-C4 ...),全量 python -m pytest 绿后再进下一步;C4~C6 全部完成后按开发计划 M3 口径验收 A-10/A-11/A-12 并打 tag。

红线自查(每步提交前过一遍)

  • 不改表结构(alert_type/status 复用 payload 承载;audit_log.event_type 为 VARCHAR 可直接扩)✓
  • 升级标记仅定时任务可写(update_alert_escalation 只有 cron 链路调用;handle API 不触碰 payload)✓
  • 人工处置唯一入口 risk_officer(manager 全链路无处置权)✓
  • 四 Agent 不互调 LLM(C5/C6 定时任务无 LLM;新 Tool 全部只读)✓
  • 审计只 INSERT(三个新 event_type 走 insert_audit_log)✓
  • 客户 Agent 边界不涉及(三个需求全部收口在 risk Agent + 定时任务)✓

已登记挂账(不属本期,编码时勿"顺手"处理)

#2 定时任务内嵌 lifespan、#3 agent_profile_l3 建表、#4 escalated 状态改表、#5 alert_type ENUM 扩展、#7 customer_query 事件类型补建、#8 core_trade actor 字段、#9 payload JSON 检索性能。