702 lines
30 KiB
Python
702 lines
30 KiB
Python
"""风控规则扫描与预警生成 Service。
|
||
|
||
扫描只读取交易、资金、持仓、客户、产品和工单事实；写入仅限预警和审计。
|
||
规则扫描不负责通知外发，也不在 Web 进程中启动定时任务。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import json
|
||
import logging
|
||
from datetime import UTC, date, datetime, timedelta
|
||
from decimal import Decimal
|
||
from typing import Any
|
||
from uuid import uuid4
|
||
|
||
from sqlalchemy import Select, func, select
|
||
from sqlalchemy.ext.asyncio import AsyncSession
|
||
|
||
from app.core.config import get_settings
|
||
from app.core.contracts import RequestContext
|
||
from app.core.errors import AgentError, ConflictAgentError
|
||
from app.core.timeutil import local_date, local_hour
|
||
from app.model.audit import InteractionAudit
|
||
from app.model.fund import (
|
||
FundCapitalFlow,
|
||
FundCustomerProfile,
|
||
FundHolding,
|
||
FundProduct,
|
||
FundRiskAlert,
|
||
FundRiskNotification,
|
||
FundTransaction,
|
||
)
|
||
from app.model.risk import RiskLoginRecord, RiskUser, RiskWorkOrder
|
||
from app.service.authorization_service import AuthorizationService
|
||
from app.service.risk_notification_service import RiskNotificationService
|
||
from app.service.risk_smtp_mail_service import (
|
||
RiskSmtpConfigurationError,
|
||
RiskSmtpMailService,
|
||
RiskSmtpSendError,
|
||
)
|
||
|
||
HIGH_RISK = "高"
|
||
MEDIUM_RISK = "中"
|
||
LOW_RISK = "低"
|
||
RISK_ORDER = {LOW_RISK: 0, MEDIUM_RISK: 1, HIGH_RISK: 2}
|
||
# 《个人投资者适当性管理指南》第十五条豁免规则：这两组越级购买是**允许**的，
|
||
# 但单只产品持仓有额度上限。键是（客户等级, 产品等级），值是"单只持仓 / 总资产"上限。
|
||
EXEMPTION_LIMITS: dict[tuple[str, str], Decimal] = {
|
||
("C3", "R4"): Decimal("0.20"),
|
||
("C4", "R5"): Decimal("0.10"),
|
||
}
|
||
|
||
|
||
def _exemption_exceeded(exemption: dict[str, Any]) -> bool:
|
||
"""豁免额度是否被突破。占比算不出来时**不算突破**（取向见 `_exemption_state`）。"""
|
||
ratio = exemption.get("exemption_ratio")
|
||
limit = exemption.get("exemption_limit")
|
||
if ratio is None or limit is None:
|
||
return False
|
||
return Decimal(ratio) > Decimal(limit)
|
||
|
||
|
||
def _suitability_summary(
|
||
customer_level: str,
|
||
product_level: str,
|
||
missing_trace: bool,
|
||
exemption: dict[str, Any],
|
||
) -> str:
|
||
parts = [f"{customer_level} 客户购买 {product_level} 产品"]
|
||
if _exemption_exceeded(exemption):
|
||
parts.append(
|
||
f"单只持仓占比 {Decimal(exemption['exemption_ratio']):.2%} "
|
||
f"超过第十五条豁免上限 {Decimal(exemption['exemption_limit']):.0%}"
|
||
)
|
||
if missing_trace:
|
||
parts.append("交易留痕不完整")
|
||
return "，".join(parts) + "。"
|
||
_scan_lock = asyncio.Lock()
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
class RiskScanBusyError(ConflictAgentError):
|
||
"""同一进程内已有扫描任务执行。"""
|
||
|
||
|
||
class RiskScanError(AgentError):
|
||
"""规则扫描失败。"""
|
||
|
||
code = "AGENT_INTERNAL_ERROR"
|
||
status_code = 500
|
||
|
||
|
||
class RiskRuleEngine:
|
||
"""五条当前启用风控规则的只读判定和预警构造。"""
|
||
|
||
def __init__(self, session: AsyncSession) -> None:
|
||
self.session = session
|
||
self._handler_id: int | None = None
|
||
|
||
async def refresh_alerts(self) -> list[FundRiskAlert]:
|
||
self._handler_id = await self._risk_operator_id()
|
||
alerts: list[FundRiskAlert] = []
|
||
alerts.extend(await self._fast_in_fast_out())
|
||
alerts.extend(await self._suitability_mismatch())
|
||
alerts.extend(await self._elderly_redemption())
|
||
alerts.extend(await self._low_risk_night_trade())
|
||
alerts.extend(await self._auto_investment_false_positive())
|
||
alerts = self._merge_same_transaction_alerts(alerts)
|
||
for alert in alerts:
|
||
self.session.add(alert)
|
||
self.session.add(InteractionAudit(
|
||
actor_type="system",
|
||
target_customer_id=alert.customer_id,
|
||
portal="api",
|
||
action_type="risk_alert_created",
|
||
detail={"alert_no": alert.alert_no, "rule_codes": alert.trigger_rule_codes},
|
||
created_at=datetime.now(UTC).replace(tzinfo=None),
|
||
))
|
||
await self.session.flush()
|
||
return alerts
|
||
|
||
async def _fast_in_fast_out(self) -> list[FundRiskAlert]:
|
||
alerts: list[FundRiskAlert] = []
|
||
transactions = await self._scalars(
|
||
select(FundTransaction).where(FundTransaction.transaction_type == "赎回")
|
||
)
|
||
for transaction in transactions:
|
||
if transaction.confirmed_at is None or transaction.amount is None:
|
||
continue
|
||
capital = await self.session.scalar(
|
||
select(FundCapitalFlow)
|
||
.where(
|
||
FundCapitalFlow.customer_id == transaction.customer_id,
|
||
FundCapitalFlow.flow_type == "入金",
|
||
FundCapitalFlow.status == "成功",
|
||
FundCapitalFlow.settled_at.is_not(None),
|
||
FundCapitalFlow.settled_at <= transaction.confirmed_at,
|
||
FundCapitalFlow.settled_at >= transaction.confirmed_at - timedelta(days=3),
|
||
)
|
||
.order_by(FundCapitalFlow.settled_at.desc())
|
||
.limit(1)
|
||
)
|
||
if capital is None or capital.amount <= 0:
|
||
continue
|
||
ratio = Decimal(transaction.amount) / Decimal(capital.amount)
|
||
if ratio >= Decimal("0.8") and transaction.amount >= 500000 and not await self._exists(
|
||
transaction.id, "RW-003"
|
||
):
|
||
alerts.append(self._build_alert(
|
||
transaction=transaction,
|
||
alert_type="大额快进快出",
|
||
level=HIGH_RISK,
|
||
rules=["RW-003"],
|
||
summary=(
|
||
f"3 日内入金 {capital.amount:.0f} 元后赎回 "
|
||
f"{transaction.amount:.0f} 元，赎回比例 {ratio:.2%}。"
|
||
),
|
||
priority=98,
|
||
event_status="正在发生",
|
||
evidence={
|
||
"product_id": transaction.product_id,
|
||
"capital_flow_id": capital.id,
|
||
"flow_no": capital.flow_no,
|
||
"ratio": str(ratio),
|
||
},
|
||
))
|
||
return alerts
|
||
|
||
async def _suitability_mismatch(self) -> list[FundRiskAlert]:
|
||
alerts: list[FundRiskAlert] = []
|
||
transactions = await self._scalars(
|
||
select(FundTransaction).where(FundTransaction.transaction_type == "申购")
|
||
)
|
||
for transaction in transactions:
|
||
customer = await self.session.get(RiskUser, transaction.customer_id)
|
||
product = await self.session.get(FundProduct, transaction.product_id)
|
||
work_order = (
|
||
await self.session.get(RiskWorkOrder, transaction.work_order_id)
|
||
if transaction.work_order_id is not None
|
||
else None
|
||
)
|
||
if (
|
||
customer is None
|
||
or product is None
|
||
or not customer.investor_type
|
||
or await self._exists(transaction.id, "RW-007")
|
||
):
|
||
continue
|
||
product_level = _level_value(product.risk_level, "R")
|
||
customer_level = _level_value(customer.investor_type, "C")
|
||
if product_level is None or customer_level is None:
|
||
# 等级字段脏（不是 R1-R5 / C1-C5）。**跳过这一条**，而不是让整个扫描挂掉：
|
||
# 原先 `_level_value` 直接 `int(...)`，一条脏数据抛 ValueError 会冒到
|
||
# `scan()` 的兜底 → 整批 rollback，前面规则扫出来的预警全部白做。
|
||
# 数据脏属于运维问题，不该升级成"整个风控停摆"。
|
||
logger.warning(
|
||
"跳过等级字段异常的记录：transaction_id=%s risk_level=%r investor_type=%r",
|
||
transaction.id,
|
||
product.risk_level,
|
||
customer.investor_type,
|
||
)
|
||
continue
|
||
gap = product_level - customer_level
|
||
exemption = await self._exemption_state(transaction, customer, product)
|
||
missing_trace = (
|
||
(
|
||
product.risk_disclosure_required
|
||
and (
|
||
work_order is None
|
||
or work_order.risk_disclosure_ack_at is None
|
||
)
|
||
)
|
||
or (
|
||
product.second_confirmation_required
|
||
and (
|
||
work_order is None
|
||
or work_order.second_confirmation_at is None
|
||
)
|
||
)
|
||
or (
|
||
product.recording_required
|
||
and (work_order is None or not work_order.recording_reference)
|
||
)
|
||
)
|
||
exceeded = _exemption_exceeded(exemption)
|
||
missing_trace = bool(missing_trace)
|
||
if gap > 0 and (missing_trace or exceeded):
|
||
level = HIGH_RISK if gap >= 2 else MEDIUM_RISK
|
||
alerts.append(self._build_alert(
|
||
transaction=transaction,
|
||
alert_type="适当性错配",
|
||
level=level,
|
||
rules=["RW-007"],
|
||
summary=_suitability_summary(
|
||
customer.investor_type or "",
|
||
product.risk_level,
|
||
missing_trace,
|
||
exemption,
|
||
),
|
||
priority=90 if level == HIGH_RISK else 70,
|
||
event_status="刚刚发生",
|
||
evidence={
|
||
"product_id": product.id,
|
||
"level_gap": gap,
|
||
"work_order_id": transaction.work_order_id,
|
||
**exemption,
|
||
},
|
||
))
|
||
return alerts
|
||
|
||
async def _exemption_state(
|
||
self,
|
||
transaction: FundTransaction,
|
||
customer: RiskUser,
|
||
product: FundProduct,
|
||
) -> dict[str, Any]:
|
||
"""按第十五条核算豁免额度，返回要写进证据快照的字段。
|
||
|
||
政策原文：C3 买 R4 需签《产品风险超越投资者风险承受能力揭示书》且**单只 R4 持仓
|
||
不超过总资产 20%**；C4 买 R5 同理，上限 10%。所以越级购买本身不是违规，
|
||
**超出额度**才是 —— 原先扫描侧只看"留痕是否齐全"，完全没有额度这一维。
|
||
|
||
数据缺失时的取向：**不把"算不出来"当成"超限"**。实测本项目库里
|
||
`fin_customer_profile.total_asset` 全为 0、`fin_holding` 为空 —— 画像与持仓由本
|
||
项目之外的流程写入（与 `behavior_score` 同源），拿 0 去算会让每一笔 C3→R4 都变成
|
||
"超限"，等于把豁免规则变成新的误报源。因此这里只如实写出缺失，
|
||
由研判侧提示人工确认。
|
||
"""
|
||
limit = EXEMPTION_LIMITS.get((customer.investor_type or "", product.risk_level))
|
||
if limit is None:
|
||
return {}
|
||
total_asset = await self.session.scalar(
|
||
select(FundCustomerProfile.total_asset).where(
|
||
FundCustomerProfile.customer_id == transaction.customer_id
|
||
)
|
||
)
|
||
holding_value = await self.session.scalar(
|
||
select(FundHolding.current_value)
|
||
.where(
|
||
FundHolding.customer_id == transaction.customer_id,
|
||
FundHolding.product_id == transaction.product_id,
|
||
)
|
||
.order_by(FundHolding.id.desc())
|
||
.limit(1)
|
||
)
|
||
state: dict[str, Any] = {"exemption_limit": str(limit)}
|
||
if (
|
||
total_asset is None
|
||
or holding_value is None
|
||
or Decimal(total_asset) <= 0
|
||
):
|
||
state["exemption_ratio"] = None
|
||
state["exemption_data_missing"] = True
|
||
return state
|
||
state["exemption_ratio"] = str(Decimal(holding_value) / Decimal(total_asset))
|
||
state["exemption_data_missing"] = False
|
||
state["holding_value"] = str(holding_value)
|
||
state["total_asset"] = str(total_asset)
|
||
return state
|
||
|
||
async def _elderly_redemption(self) -> list[FundRiskAlert]:
|
||
alerts: list[FundRiskAlert] = []
|
||
transactions = await self._scalars(
|
||
select(FundTransaction).where(FundTransaction.transaction_type == "赎回")
|
||
)
|
||
for transaction in transactions:
|
||
if transaction.confirmed_at is None or transaction.amount is None:
|
||
continue
|
||
profile = await self.session.get(FundCustomerProfile, transaction.customer_id)
|
||
age = _age(profile.birth_date) if profile else 0
|
||
if profile is None or age < 65 or transaction.amount < 300000:
|
||
continue
|
||
average = await self.session.scalar(
|
||
select(func.avg(FundTransaction.amount)).where(
|
||
FundTransaction.customer_id == transaction.customer_id,
|
||
FundTransaction.id != transaction.id,
|
||
FundTransaction.confirmed_at >= transaction.confirmed_at - timedelta(days=365),
|
||
)
|
||
)
|
||
# `average <= 0` 也必须挡住：均值为 0 时 `amount < 0*3` 恒为 False，
|
||
# 会一路走到下面算 `amount / average` 直接除零。
|
||
average_amount = Decimal(average) if average is not None else None
|
||
if (
|
||
average_amount is None
|
||
or average_amount <= 0
|
||
or transaction.amount < average_amount * 3
|
||
):
|
||
continue
|
||
login = await self.session.scalar(
|
||
select(RiskLoginRecord)
|
||
.where(
|
||
RiskLoginRecord.user_id == transaction.customer_id,
|
||
RiskLoginRecord.login_result == "成功",
|
||
RiskLoginRecord.login_at <= transaction.confirmed_at,
|
||
)
|
||
.order_by(RiskLoginRecord.login_at.desc())
|
||
.limit(1)
|
||
)
|
||
if login is not None and not login.is_common_device and not await self._exists(
|
||
transaction.id, "RW-012"
|
||
):
|
||
alerts.append(self._build_alert(
|
||
transaction=transaction,
|
||
alert_type="老年客户异常大额赎回",
|
||
level=HIGH_RISK,
|
||
rules=["RW-012"],
|
||
summary=(
|
||
f"{age} 岁客户赎回 {transaction.amount:.0f} 元，"
|
||
"超过历史均值且使用非常用设备。"
|
||
),
|
||
priority=96,
|
||
event_status="正在发生",
|
||
evidence={
|
||
"product_id": transaction.product_id,
|
||
"age": age,
|
||
"device_id": login.device_id,
|
||
# 把生成条件用到的均值也写进快照。原先这里只有 age / device_id，
|
||
# 研判侧要复核"是否真达到 3 倍历史均值"却无从下手，只能跳过这一条、
|
||
# 直接让"非常用设备"定案（docs/25 P2：研判漏阈值条件）。
|
||
# 写 ratio 与 RW-003 的快照风格保持一致。
|
||
"average_amount": str(average_amount),
|
||
"ratio": str(transaction.amount / average_amount),
|
||
},
|
||
))
|
||
return alerts
|
||
|
||
async def _low_risk_night_trade(self) -> list[FundRiskAlert]:
|
||
alerts: list[FundRiskAlert] = []
|
||
for transaction in await self._scalars(select(FundTransaction)):
|
||
if (
|
||
transaction.confirmed_at is not None
|
||
# 必须按**北京时间**判断"凌晨"：库里存 UTC，直接取 .hour 会让
|
||
# [0,6) UTC 变成北京 08:00–14:00，整条规则判的是上午（docs/25 P1 #6）。
|
||
and 0 <= local_hour(transaction.confirmed_at) < 6
|
||
and transaction.amount is not None
|
||
and transaction.amount <= 10000
|
||
and not await self._exists(transaction.id, "RW-015")
|
||
):
|
||
alerts.append(self._build_alert(
|
||
transaction=transaction,
|
||
alert_type="非正常时段小额操作",
|
||
level=LOW_RISK,
|
||
rules=["RW-015"],
|
||
summary=f"凌晨时段发生 {transaction.amount:.0f} 元交易，金额较小。",
|
||
priority=20,
|
||
event_status="盘后预警",
|
||
evidence={
|
||
"product_id": transaction.product_id,
|
||
"hour": transaction.confirmed_at.hour,
|
||
},
|
||
))
|
||
return alerts
|
||
|
||
async def _auto_investment_false_positive(self) -> list[FundRiskAlert]:
|
||
alerts: list[FundRiskAlert] = []
|
||
transactions = await self._scalars(
|
||
select(FundTransaction).where(FundTransaction.work_order_id.is_not(None))
|
||
)
|
||
for transaction in transactions:
|
||
work_order = await self.session.get(RiskWorkOrder, transaction.work_order_id)
|
||
if (
|
||
work_order is not None
|
||
and work_order.channel in {"定投", "自动定投"}
|
||
and not await self._exists(transaction.id, "RW-018")
|
||
):
|
||
alerts.append(self._build_alert(
|
||
transaction=transaction,
|
||
alert_type="频繁交易初筛",
|
||
level=LOW_RISK,
|
||
rules=["RW-018"],
|
||
summary="近 30 天交易频率较高，但交易来自有效定投工单。",
|
||
priority=10,
|
||
event_status="盘后预警",
|
||
evidence={
|
||
"product_id": transaction.product_id,
|
||
"work_order_no": work_order.work_order_no,
|
||
"channel": work_order.channel,
|
||
},
|
||
))
|
||
return alerts
|
||
|
||
def _build_alert(
|
||
self,
|
||
*,
|
||
transaction: FundTransaction,
|
||
alert_type: str,
|
||
level: str,
|
||
rules: list[str],
|
||
summary: str,
|
||
priority: int,
|
||
event_status: str,
|
||
evidence: dict[str, object],
|
||
) -> FundRiskAlert:
|
||
now = datetime.now(UTC).replace(tzinfo=None)
|
||
return FundRiskAlert(
|
||
id=_new_alert_id(),
|
||
alert_no=f"AL{uuid4().hex[:12].upper()}",
|
||
customer_id=transaction.customer_id,
|
||
related_transaction_id=transaction.id,
|
||
related_order_id=transaction.order_id,
|
||
related_work_order_id=transaction.work_order_id,
|
||
alert_type=alert_type,
|
||
alert_level=level,
|
||
trigger_rule_codes=rules,
|
||
evidence_summary=summary,
|
||
evidence_snapshot=evidence,
|
||
priority_score=priority,
|
||
event_status=event_status,
|
||
status="待处理",
|
||
ack_status="未确认",
|
||
handler_id=self._handler_id,
|
||
due_at=now + timedelta(minutes=30) if level == HIGH_RISK else None,
|
||
is_escalated=0,
|
||
created_at=now,
|
||
updated_at=now,
|
||
)
|
||
|
||
async def _exists(self, transaction_id: int, rule_code: str) -> bool:
|
||
return await self.session.scalar(
|
||
select(FundRiskAlert.id).where(
|
||
FundRiskAlert.related_transaction_id == transaction_id,
|
||
# 同 risk_repository：`.contains()` 会被编译成 `LIKE`，
|
||
# 对 JSON 数组列永远不匹配，去重就形同失效。
|
||
func.json_contains(FundRiskAlert.trigger_rule_codes, json.dumps(rule_code)),
|
||
)
|
||
) is not None
|
||
|
||
async def _risk_operator_id(self) -> int | None:
|
||
value = await self.session.scalar(
|
||
select(RiskUser.id)
|
||
.where(RiskUser.employee_role == "risk_operator", RiskUser.status == "正常")
|
||
.order_by(RiskUser.id.asc())
|
||
.limit(1)
|
||
)
|
||
return int(value) if value is not None else None
|
||
|
||
async def _scalars(self, statement: Select[Any]) -> list[Any]:
|
||
result = await self.session.scalars(statement)
|
||
return list(result.all())
|
||
|
||
@staticmethod
|
||
def _merge_same_transaction_alerts(alerts: list[FundRiskAlert]) -> list[FundRiskAlert]:
|
||
grouped: dict[int | None, list[FundRiskAlert]] = {}
|
||
for alert in alerts:
|
||
grouped.setdefault(alert.related_transaction_id, []).append(alert)
|
||
merged: list[FundRiskAlert] = []
|
||
for transaction_id, items in grouped.items():
|
||
if transaction_id is None or len(items) == 1:
|
||
merged.extend(items)
|
||
continue
|
||
primary = max(items, key=lambda item: item.priority_score)
|
||
primary.trigger_rule_codes = list(
|
||
dict.fromkeys(code for item in items for code in item.trigger_rule_codes)
|
||
)
|
||
primary.evidence_summary = "；".join(item.evidence_summary for item in items)
|
||
primary.evidence_snapshot = {
|
||
"product_id": primary.evidence_snapshot.get("product_id"),
|
||
"merged_alerts": [
|
||
{"alert_type": item.alert_type, "evidence": item.evidence_snapshot}
|
||
for item in items
|
||
],
|
||
}
|
||
primary.alert_level = max(
|
||
(item.alert_level for item in items),
|
||
key=RISK_ORDER.__getitem__,
|
||
)
|
||
primary.priority_score = max(item.priority_score for item in items)
|
||
merged.append(primary)
|
||
return merged
|
||
|
||
|
||
class RiskScanService:
|
||
def __init__(
|
||
self,
|
||
session: AsyncSession,
|
||
*,
|
||
rule_engine: RiskRuleEngine | None = None,
|
||
notification_service: RiskNotificationService | None = None,
|
||
notification_enabled: bool = True,
|
||
notification_email: str | None = None,
|
||
mail_enabled: bool = False,
|
||
mail_dry_run: bool = False,
|
||
smtp_mail_service: RiskSmtpMailService | None = None,
|
||
) -> None:
|
||
self.session = session
|
||
self.rule_engine = rule_engine or RiskRuleEngine(session)
|
||
self.notification_service = notification_service or RiskNotificationService(session)
|
||
self.notification_enabled = notification_enabled
|
||
self.notification_email = notification_email
|
||
self.mail_enabled = mail_enabled
|
||
self.mail_dry_run = mail_dry_run
|
||
self.smtp_mail_service = smtp_mail_service or RiskSmtpMailService()
|
||
|
||
@classmethod
|
||
def from_settings(
|
||
cls,
|
||
session: AsyncSession,
|
||
*,
|
||
rule_engine: RiskRuleEngine | None = None,
|
||
notification_service: RiskNotificationService | None = None,
|
||
) -> RiskScanService:
|
||
"""从环境配置创建扫描服务，供手工扫描和定时扫描共用。"""
|
||
settings = get_settings()
|
||
recipients = _parse_recipients(settings.risk_alert_mail_recipients)
|
||
notification_email = recipients[0] if recipients else None
|
||
if settings.risk_alert_mail_enabled and not notification_email:
|
||
logger.warning("高风险邮件通知已启用，但未配置收件人")
|
||
return cls(
|
||
session,
|
||
rule_engine=rule_engine,
|
||
notification_service=notification_service,
|
||
notification_email=notification_email,
|
||
mail_enabled=settings.risk_alert_mail_enabled and bool(notification_email),
|
||
mail_dry_run=settings.risk_alert_mail_dry_run,
|
||
)
|
||
|
||
async def scan(self, context: RequestContext) -> dict[str, int | str]:
|
||
await AuthorizationService.require(context, "risk:alert:scan")
|
||
if _scan_lock.locked():
|
||
raise RiskScanBusyError("规则扫描正在执行，请稍后重试")
|
||
async with _scan_lock:
|
||
try:
|
||
alerts = await self.rule_engine.refresh_alerts()
|
||
notification_count, notification_failure = await self._create_notifications(alerts)
|
||
await self.session.commit()
|
||
except Exception as error:
|
||
await self.session.rollback()
|
||
raise RiskScanError("规则扫描失败") from error
|
||
result: dict[str, int | str] = {
|
||
"message": "规则扫描完成",
|
||
"created_count": len(alerts),
|
||
"high_risk_count": sum(alert.alert_level == HIGH_RISK for alert in alerts),
|
||
"notification_count": notification_count,
|
||
}
|
||
if notification_failure:
|
||
# 通知失败**不回滚**预警（预警已经生成、比通知重要），但必须让调用方看见。
|
||
# 原先这里无处可查：外面只拿到 notification_count=0，分不清"这批预警本来
|
||
# 就不用通知"和"高风险通知创建失败了"——而后者意味着处置链路的第一环断了，
|
||
# 扫描却报"完成"。
|
||
result["notification_failure"] = notification_failure
|
||
result["message"] = "规则扫描完成，但高风险通知创建失败"
|
||
return result
|
||
|
||
async def _create_notifications(self, alerts: list[FundRiskAlert]) -> tuple[int, str]:
|
||
"""为高风险预警创建通知记录，返回 `(创建条数, 失败原因)`。
|
||
|
||
**失败原因必须交给调用方。** 原先这里失败时 `return 0`，而"无需通知"（没有高风险
|
||
预警、或通知功能关闭）也返回 0 —— 外面根本分不清两者。风控里这个区别很要紧：
|
||
通知没发出去等于处置链路的第一环断了，而扫描依旧报"完成"。
|
||
|
||
失败**不回滚**预警本身：预警已经生成、比通知重要，不该因为通知写失败就丢掉。
|
||
"""
|
||
if not self.notification_enabled:
|
||
return 0, ""
|
||
high_risk = [alert for alert in alerts if alert.alert_level == HIGH_RISK]
|
||
if not high_risk:
|
||
return 0, ""
|
||
try:
|
||
async with self.session.begin_nested():
|
||
count = 0
|
||
for alert in high_risk:
|
||
title = f"高风险预警：{alert.alert_type}"
|
||
self.notification_service.create_in_app(
|
||
alert,
|
||
receiver_user_id=alert.handler_id,
|
||
title=title,
|
||
content=alert.evidence_summary,
|
||
)
|
||
count += 1
|
||
if self.notification_email:
|
||
notification = self.notification_service.create_mail_record(
|
||
alert,
|
||
receiver_email=self.notification_email,
|
||
title=title,
|
||
content=alert.evidence_summary,
|
||
mail_enabled=self.mail_enabled,
|
||
)
|
||
count += 1
|
||
if self.mail_enabled:
|
||
await self._deliver_mail(
|
||
alert,
|
||
notification,
|
||
title,
|
||
)
|
||
return count, ""
|
||
except Exception as error:
|
||
logger.exception("高风险通知记录创建失败，预警扫描继续提交")
|
||
return 0, f"{type(error).__name__}: {error}"[:200]
|
||
|
||
async def _deliver_mail(
|
||
self,
|
||
alert: FundRiskAlert,
|
||
notification: FundRiskNotification,
|
||
title: str,
|
||
) -> None:
|
||
"""发送高风险通知邮件，失败只回写通知记录，不影响预警落库。"""
|
||
if self.mail_dry_run:
|
||
self.notification_service.mark_mail_failed(
|
||
notification,
|
||
"邮件发送处于 dry-run 模式",
|
||
)
|
||
return
|
||
try:
|
||
result = await self.smtp_mail_service.send(
|
||
[self.notification_email or ""],
|
||
title,
|
||
f"预警编号：{alert.alert_no}；{alert.evidence_summary}",
|
||
)
|
||
except (RiskSmtpConfigurationError, RiskSmtpSendError) as error:
|
||
logger.warning("高风险预警邮件发送失败：alert_no=%s error=%s", alert.alert_no, error)
|
||
self.notification_service.mark_mail_failed(notification, str(error))
|
||
except Exception:
|
||
logger.exception("高风险预警邮件发送发生未预期异常：alert_no=%s", alert.alert_no)
|
||
self.notification_service.mark_mail_failed(notification, "邮件发送发生未预期异常")
|
||
else:
|
||
if result.get("status") == "sent":
|
||
self.notification_service.mark_mail_sent(notification)
|
||
else:
|
||
self.notification_service.mark_mail_failed(notification, "邮件发送未完成")
|
||
|
||
|
||
def _new_alert_id() -> int:
|
||
"""生成非零 63 位正整数；预警表主键不自增。"""
|
||
return uuid4().int & ((1 << 63) - 1) or 1
|
||
|
||
|
||
def _age(birth_date: date | None) -> int:
|
||
if birth_date is None:
|
||
return 0
|
||
# 用**北京时间**的今天：原先用 UTC 日期，而研判侧用服务器 date.today()，
|
||
# 两处口径不一致会让同一客户在生日边界上差一岁（docs/25 P1 #6）。
|
||
today = local_date(datetime.now(UTC))
|
||
return today.year - birth_date.year - (
|
||
(today.month, today.day) < (birth_date.month, birth_date.day)
|
||
)
|
||
|
||
|
||
def _level_value(level: str, prefix: str) -> int | None:
|
||
"""把 "R2" / "C3" 解析成 2 / 3；格式不符返回 None，**不再抛异常**。
|
||
|
||
原先写的是 `int(level.replace(prefix, ""))`：一条脏数据（等级字段写了「中风险」之类）
|
||
就会抛 ValueError，一路冒到 `scan()` 的兜底 → **整批 rollback**，这次扫描前面已经
|
||
生成的预警全部作废。数据脏是运维问题，不该升级成"整个风控停摆"。
|
||
|
||
用 `removeprefix` 而不是 `replace`：只去掉开头那一个前缀字符，
|
||
`"R2R"` 这种脏值不会被错当成 2。（没有前缀的 `"2"` 仍能解析——历史数据可能不带前缀。）
|
||
"""
|
||
digits = level.strip().upper().removeprefix(prefix.upper())
|
||
return int(digits) if digits.isdigit() else None
|
||
|
||
|
||
def _parse_recipients(value: str | None) -> list[str]:
|
||
"""解析逗号或分号分隔的邮件收件人配置。"""
|
||
if not value:
|
||
return []
|
||
return [
|
||
recipient.strip()
|
||
for recipient in value.replace(";", ",").split(",")
|
||
if recipient.strip()
|
||
]
|