Files
group_fqcd_jr/tools/seed_risk_alert_demo_data.py
T

183 lines
7.1 KiB
Python
Raw Normal View History

"""造风控演示数据:让概览 / 查询 / 证据三个只读工具都能返回真实内容。
**为什么需要**:配置与 RBAC 补齐、时区修好之后,`fin_risk_alert` 表是空的 ——
概览只能显示"未闭环预警总量 0",看不出任何真实内容,没法验收。
**造什么**:3 条预警,刻意覆盖三个工具各自的读路径:
1. **高危 + 待处理 + 三条规则 + 完整证据快照** —— 概览的高危计数、查询的高危筛选、
证据工具的证据链,三条路径都靠它。
2. **中危 + 调查中 + 两条规则** —— 让"未闭环"统计里出现"调查中"。
3. **低危 + 已闭环** —— 验证"未闭环"筛选真的把它排除掉(这是上一批修复里
`OPEN_STATUSES` 的用途)。
**字段取值不自行发明**,全部照 `risk_scan_service.py:312-333` 的 `_build_alert` 抄:
状态用 `OPEN_STATUSES = ("待处理", "调查中")`(见 `risk_action_service.py:17`)、
`ack_status` 用"未确认"、`alert_level` 用"高/中/低"。
**两个注意点**:
- `fin_risk_alert.id` **不是自增**(见建表语句),所以这里手工生成 id;
- 时间一律按库内约定存 **UTC naive**(`app/infrastructure/db.py:15-21`)。
幂等:按 `alert_no` 判断,已存在则跳过。
用法:python tools/seed_risk_alert_demo_data.py
"""
import asyncio
import datetime as dt
import json
import sys
from typing import Any
import asyncmy
from app.core.config import get_settings
CUSTOMER_ID = 9001 # 已有的画像客户(fin_customer_profile 里那一条)
HANDLER_ID = 9002 # risk_operator
ID_BASE = 9_000_000_000_000_001
ALERTS: tuple[dict[str, Any], ...] = (
{
"id": ID_BASE,
"alert_no": "ALDEMO0001",
"alert_type": "大额频繁交易",
"alert_level": "高",
"trigger_rule_codes": ["RW-007", "RW-002", "RW-012"],
"evidence_summary": (
"客户近 24 小时内 5 笔申购合计 486,000 元,单笔最大 200,000 元,"
"为本人近 90 日均值的 6.2 倍;收款账户与历史常用账户不一致。"
),
"evidence_snapshot": {
"product_id": 7001,
"product_name": "南方季季盈90天",
"transaction_count": 5,
"total_amount": "486000.00",
"max_amount": "200000.00",
"average_amount": "78387.10",
"ratio": "6.20",
"window_hours": 24,
"customer_age": 34,
"login_ip_regions": ["广东深圳", "香港"],
"device_changed": True,
"open_status": "未闭环",
},
"priority_score": 95,
"event_status": "刚刚发生",
"status": "待处理",
"ack_status": "未确认",
"due_at_offset_minutes": 30,
"is_escalated": 0,
},
{
"id": ID_BASE + 1,
"alert_no": "ALDEMO0002",
"alert_type": "非正常时段大额操作",
"alert_level": "高",
"trigger_rule_codes": ["RW-015", "RW-003"],
"evidence_summary": (
"成交时间对应北京时间 02:17(凌晨),金额 88,000 元;"
"该客户近 30 天无凌晨交易记录,且未匹配到有效定投工单。"
),
"evidence_snapshot": {
"product_id": 7002,
"product_name": "南方稳健增利180天",
"amount": "88000.00",
"beijing_hour": 2,
"utc_confirmed_at": "2026-09-09T18:17:00",
"matched_work_order": None,
"channel": "APP",
"device_changed": False,
"open_status": "未闭环",
},
"priority_score": 88,
"event_status": "盘后预警",
"status": "调查中",
"ack_status": "已确认",
"due_at_offset_minutes": 30,
"is_escalated": 1,
},
{
"id": ID_BASE + 2,
"alert_no": "ALDEMO0003",
"alert_type": "低风险频繁交易初筛",
"alert_level": "低",
"trigger_rule_codes": ["RW-018"],
"evidence_summary": "命中低优先级频繁交易初筛,交易来自已核验的自动定投工单。",
"evidence_snapshot": {
"product_id": 7001,
"product_name": "南方季季盈90天",
"matched_work_order": "WO-DEMO-0007",
"channel": "自动定投",
"open_status": "已闭环",
},
"priority_score": 20,
"event_status": "盘后预警",
"status": "已关闭",
"ack_status": "已确认",
"due_at_offset_minutes": None,
"is_escalated": 0,
"close_reason": "已核验为本人自动定投计划,属正常交易。",
},
)
async def main() -> int:
settings = get_settings()
dsn = settings.mysql_dsn.split("://", 1)[1]
credentials, location = dsn.split("@", 1)
user, password = credentials.split(":", 1)
host_port, database = location.split("/", 1)
host, _, port = host_port.partition(":")
connection = await asyncmy.connect(
host=host, port=int(port or 3306), user=user, password=password, db=database
)
now = dt.datetime.now(dt.UTC).replace(tzinfo=None)
try:
cursor = connection.cursor()
created = 0
for spec in ALERTS:
await cursor.execute(
"SELECT id FROM fin_risk_alert WHERE alert_no = %s", (spec["alert_no"],)
)
if await cursor.fetchone() is not None:
print(f"[跳过] {spec['alert_no']} 已存在")
continue
offset = spec["due_at_offset_minutes"]
await cursor.execute(
"INSERT INTO fin_risk_alert ("
" id, alert_no, customer_id, alert_type, alert_level, trigger_rule_codes,"
" evidence_summary, evidence_snapshot, priority_score, event_status, status,"
" ack_status, handler_id, due_at, is_escalated, close_reason,"
" created_at, updated_at"
") VALUES (%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s,%s)",
(
spec["id"], spec["alert_no"], CUSTOMER_ID, spec["alert_type"],
spec["alert_level"], json.dumps(spec["trigger_rule_codes"], ensure_ascii=False),
spec["evidence_summary"],
json.dumps(spec["evidence_snapshot"], ensure_ascii=False),
spec["priority_score"], spec["event_status"], spec["status"],
spec["ack_status"], HANDLER_ID,
(now + dt.timedelta(minutes=offset)) if offset else None,
spec["is_escalated"], spec.get("close_reason"),
now, now,
),
)
created += 1
print(
f"[新增] {spec['alert_no']} 等级={spec['alert_level']} "
f"状态={spec['status']} 规则={spec['trigger_rule_codes']}"
)
await connection.commit()
await cursor.execute("SELECT COUNT(*) FROM fin_risk_alert")
total = (await cursor.fetchone())[0]
print(f"\n已提交 {created} 条;fin_risk_alert 现有 {total} 行")
return 0
finally:
connection.close()
sys.exit(asyncio.run(main()))