"""风控 Agent 的只读工具处理器。""" from __future__ import annotations from collections import Counter from typing import Any from app.core.contracts import RequestContext from app.core.risk_contracts import RiskAlertEvidenceQuery, RiskAlertQuery from app.infrastructure.db import SessionFactory from app.repository.fund_query_repository import PageRequest from app.repository.risk_repository import RiskRepository from app.service.risk_judgement_service import ( assess_alert_detail, assess_alert_list_item, ) from app.service.risk_query_service import scope_from_context FIELD_MEANINGS: dict[str, str] = { "alert_no": "预警编号", "customer_id": "客户编号", "customer_no": "客户编号", "customer_name": "客户姓名（脱敏）", "product_code": "产品代码", "product_name": "产品名称", "alert_type": "预警类型", "risk_level": "风险等级", "rule_codes": "命中规则编号", "triggered_rules": "命中规则编号", "evidence_summary": "证据摘要", "event_status": "事件状态", "status": "业务状态", "ack_status": "预警确认状态", "created_at": "创建时间", "updated_at": "更新时间", "due_at": "处理截止时间", "ack_at": "确认时间", "escalated_at": "升级时间", "priority_score": "优先级分数", "is_escalated": "是否已升级", "evidence_archived": "证据是否已归档", "close_reason": "关闭原因", "disposition_hint": "列表级只读研判建议", "disposition_assessment": "详情级只读研判草案", "name": "客户姓名（脱敏）", "birth_date": "出生日期", "occupation": "职业", "mobile_masked": "手机号（脱敏）", "investor_type": "投资者风险承受等级", "investment_horizon": "投资期限", "preferred_asset_class": "偏好资产类别", "trading_frequency": "交易频率", "total_asset": "总资产", "behavior_score": "行为分", "risk_tags": "风险标签", "trade_account": "交易账号", "product_category": "产品类别", "risk_disclosure_required": "是否要求风险揭示", "second_confirmation_required": "是否要求二次确认", "recording_required": "是否要求双录", "transaction_no": "交易编号", "order_side": "买卖方向", "transaction_type": "交易类型", "amount": "金额", "channel": "交易渠道", "trade_status": "交易状态", "risk_disclosure_signed": "风险揭示签署状态", "second_confirmation": "二次确认状态", "confirmed_at": "确认时间", "executed_at": "成交时间", "flow_no": "资金流水编号", "flow_type": "资金流水类型", "settled_at": "到账时间", "occurred_at": "发生时间", "source_type": "资金来源", "match_status": "资金匹配状态", "shares": "持有份额", "cost_amount": "持仓成本", "current_value": "当前市值", "profit_loss": "浮动盈亏", "profit_loss_ratio": "盈亏比例", "holding_days": "持有天数", "holding_ratio": "持仓占资产比例", "login_at": "登录时间", "login_result": "登录结果", "ip_region": "登录地区", "device_id": "登录设备标识", "is_common_device": "是否常用设备", "failure_reason": "失败原因", "notification_no": "通知编号", "send_status": "发送状态", "receiver_email": "接收邮箱", "sent_at": "发送时间", "handler_id": "处理人编号", "related_transaction_id": "关联交易编号", "related_work_order_id": "关联工单编号", "evidence_truncated": "被截断的证据类型", "data_truncated": "数据是否被截断", "truncated": "结果是否被截断", } def with_field_meanings(value: dict[str, Any]) -> dict[str, Any]: """附加当前结果实际出现字段的中文含义，供模型转换后回答。""" names: set[str] = set() def collect(item: object) -> None: if isinstance(item, dict): for key, child in item.items(): names.add(str(key)) collect(child) elif isinstance(item, (list, tuple)): for child in item: collect(child) collect(value) meanings = { name: FIELD_MEANINGS[name] for name in sorted(names) if name in FIELD_MEANINGS } return {**value, "field_meanings": meanings} async def search_risk_alerts_tool( arguments: RiskAlertQuery, context: RequestContext, ) -> dict[str, Any]: async with SessionFactory() as session: repository = RiskRepository(session, scope=scope_from_context(context)) items: list[dict[str, Any]] = [] offset = 0 while True: page = await repository.list_alerts( customer_no=arguments.customer_no, product_code=arguments.product_code, product_name=arguments.product_name, risk_level=arguments.risk_level, rule_code=arguments.rule_code, start_time=arguments.start_time, end_time=arguments.end_time, page=PageRequest(limit=10, offset=offset), ) for item in page.items: record = item.to_dict() record["disposition_hint"] = assess_alert_list_item(record) items.append(record) if page.next_offset is None: return build_alert_search_result( items, arguments.model_dump(mode="json", exclude_none=True), ) offset = page.next_offset def build_alert_search_result( items: list[dict[str, Any]], filters: dict[str, Any] | None = None, ) -> dict[str, Any]: """构建完整分组汇总和精简预警明细。""" compact_items = [_compact_alert_item(item) for item in items] rule_counts: Counter[str] = Counter() disposition_counts: Counter[str] = Counter() for item in compact_items: rule_counts.update(item.get("rule_codes") or ()) hint = item.get("disposition_hint") or {} if isinstance(hint, dict) and hint.get("verdict"): disposition_counts[str(hint["verdict"])] += 1 return with_field_meanings({ "total": len(compact_items), "filters": filters or {}, "summary": { "customer_groups": _customer_groups(compact_items), "product_groups": _product_groups(compact_items), "risk_level_counts": dict(sorted(Counter( str(item.get("risk_level") or "未知") for item in compact_items ).items())), "rule_counts": dict(sorted(rule_counts.items())), "disposition_counts": dict(sorted(disposition_counts.items())), "complete": True, }, "items": compact_items, }) def _compact_alert_item(item: dict[str, Any]) -> dict[str, Any]: fields = ( "alert_no", "customer_no", "customer_name", "product_code", "product_name", "alert_type", "risk_level", "rule_codes", "evidence_summary", "status", "event_status", "created_at", "disposition_hint", ) return {field: item.get(field) for field in fields if field in item} def _customer_groups(items: list[dict[str, Any]]) -> list[dict[str, Any]]: groups: dict[str, dict[str, Any]] = {} for item in items: customer_no = item.get("customer_no") if not customer_no: continue key = str(customer_no) group = groups.setdefault( key, { "customer_no": key, "customer_name": item.get("customer_name"), "alert_count": 0, "risk_levels": set(), }, ) group["alert_count"] += 1 if item.get("risk_level"): group["risk_levels"].add(str(item["risk_level"])) return [ { **group, "risk_levels": sorted(group["risk_levels"]), } for group in sorted( groups.values(), key=lambda value: (-value["alert_count"], value["customer_no"]), ) ] def _product_groups(items: list[dict[str, Any]]) -> list[dict[str, Any]]: groups: dict[str, dict[str, Any]] = {} for item in items: product_code = item.get("product_code") product_name = item.get("product_name") key = str(product_code or product_name or "") if not key: continue group = groups.setdefault( key, { "product_code": product_code, "product_name": product_name, "alert_count": 0, "risk_levels": set(), }, ) group["alert_count"] += 1 if item.get("risk_level"): group["risk_levels"].add(str(item["risk_level"])) return [ { **group, "risk_levels": sorted(group["risk_levels"]), } for group in sorted( groups.values(), key=lambda value: ( -value["alert_count"], str(value["product_code"] or value["product_name"]), ), ) ] async def get_risk_overview_tool( arguments: RiskAlertQuery, context: RequestContext, ) -> dict[str, Any]: del arguments async with SessionFactory() as session: data = await RiskRepository(session, scope=scope_from_context(context)).overview() return with_field_meanings(data) async def get_alert_evidence_tool( arguments: RiskAlertEvidenceQuery, context: RequestContext, ) -> dict[str, Any] | None: async with SessionFactory() as session: record = await RiskRepository( session, scope=scope_from_context(context), ).get_alert_detail(arguments.alert_no) if record is None: return None detail = record.to_dict() detail["disposition_assessment"] = assess_alert_detail(detail) return with_field_meanings(detail)