1013 lines
36 KiB
Python
1013 lines
36 KiB
Python
"""风控领域只读查询 Repository。
|
|
|
|
本模块只提供查询能力，不修改预警、客户、交易、资金、持仓或工单数据。
|
|
简单单表查询优先复用 `FundQueryRepository`；需要客户身份、登录记录、工单联表
|
|
或预警详情聚合时，由本模块实现。
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
from collections.abc import Callable
|
|
from dataclasses import dataclass
|
|
from datetime import UTC, date, datetime
|
|
from decimal import Decimal
|
|
from types import MappingProxyType
|
|
from typing import Any
|
|
|
|
from sqlalchemy import ColumnElement, Select, and_, case, false, func, or_, select
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.model.fund import (
|
|
FundCapitalFlow,
|
|
FundCustomerProfile,
|
|
FundHolding,
|
|
FundProduct,
|
|
FundRiskAlert,
|
|
FundRiskAssessment,
|
|
FundRiskNotification,
|
|
FundTransaction,
|
|
)
|
|
from app.model.risk import RiskLoginRecord, RiskUser, RiskWorkOrder
|
|
from app.repository.fund_query_repository import (
|
|
AnyOf,
|
|
CustomerScope,
|
|
FieldFilter,
|
|
FundFilterOperator,
|
|
FundPage,
|
|
FundQueryRepository,
|
|
FundQuerySpec,
|
|
FundRecord,
|
|
PageRequest,
|
|
)
|
|
|
|
OPEN_STATUSES = ("待处理", "调查中")
|
|
LOW_RISK = "低"
|
|
MEDIUM_RISK = "中"
|
|
HIGH_RISK = "高"
|
|
BEHAVIOR_SCORE_RANGES = {
|
|
"normal": (16, 20),
|
|
"slight": (11, 15),
|
|
"attention": (6, 10),
|
|
"high": (1, 5),
|
|
"immediate": (0, 0),
|
|
}
|
|
|
|
__all__ = ["RiskReportSnapshot", "RiskRepository"]
|
|
|
|
|
|
@dataclass(frozen=True, slots=True)
|
|
class RiskReportSnapshot:
|
|
daily: tuple[FundRecord, ...]
|
|
unresolved: tuple[FundRecord, ...]
|
|
false_positive: tuple[FundRecord, ...]
|
|
dispositions: tuple[FundRecord, ...]
|
|
|
|
|
|
class RiskRepository:
|
|
"""风控只读查询入口；未提供合法客户范围时默认拒绝。"""
|
|
|
|
def __init__(self, session: AsyncSession, *, scope: CustomerScope | None = None) -> None:
|
|
self.session = session
|
|
self.scope = scope
|
|
|
|
async def overview(self) -> dict[str, Any]:
|
|
"""返回未闭环预警概览和重点预警。"""
|
|
scope_condition = self._scope_condition(
|
|
FundRiskAlert.customer_id,
|
|
FundCustomerProfile.trade_account,
|
|
)
|
|
open_conditions = [FundRiskAlert.status.in_(OPEN_STATUSES)]
|
|
if scope_condition is not None:
|
|
open_conditions.append(scope_condition)
|
|
|
|
rows = (
|
|
await self.session.execute(
|
|
select(FundRiskAlert.alert_level, func.count(FundRiskAlert.id))
|
|
.where(*open_conditions)
|
|
.group_by(FundRiskAlert.alert_level)
|
|
)
|
|
).all()
|
|
levels = {str(level): int(count or 0) for level, count in rows}
|
|
pending = int(
|
|
await self.session.scalar(
|
|
select(func.count(FundRiskAlert.id)).where(
|
|
*open_conditions,
|
|
FundRiskAlert.status == "待处理",
|
|
)
|
|
)
|
|
or 0
|
|
)
|
|
overdue = int(
|
|
await self.session.scalar(
|
|
select(func.count(FundRiskAlert.id)).where(
|
|
*open_conditions,
|
|
FundRiskAlert.due_at.is_not(None),
|
|
FundRiskAlert.due_at <= _utc_now_naive(),
|
|
)
|
|
)
|
|
or 0
|
|
)
|
|
high_priority = await self.list_alerts(
|
|
risk_level=HIGH_RISK,
|
|
page=PageRequest(limit=3),
|
|
)
|
|
return {
|
|
"total": sum(levels.values()),
|
|
"levels": {
|
|
LOW_RISK: levels.get(LOW_RISK, 0),
|
|
MEDIUM_RISK: levels.get(MEDIUM_RISK, 0),
|
|
HIGH_RISK: levels.get(HIGH_RISK, 0),
|
|
},
|
|
"pending": pending,
|
|
"overdue": overdue,
|
|
"high_priority": high_priority.items,
|
|
}
|
|
|
|
async def list_alerts(
|
|
self,
|
|
*,
|
|
keyword: str | None = None,
|
|
customer_no: str | None = None,
|
|
product_code: str | None = None,
|
|
product_name: str | None = None,
|
|
risk_level: str | None = None,
|
|
rule_code: str | None = None,
|
|
start_time: datetime | None = None,
|
|
end_time: datetime | None = None,
|
|
open_only: bool = True,
|
|
page: PageRequest | None = None,
|
|
) -> FundPage:
|
|
"""查询预警列表，默认只返回未闭环预警。"""
|
|
request = page or PageRequest()
|
|
statement = self._alert_list_statement(
|
|
keyword=keyword,
|
|
customer_no=customer_no,
|
|
product_code=product_code,
|
|
product_name=product_name,
|
|
risk_level=risk_level,
|
|
rule_code=rule_code,
|
|
start_time=start_time,
|
|
end_time=end_time,
|
|
open_only=open_only,
|
|
)
|
|
rows = (
|
|
await self.session.execute(
|
|
statement.order_by(
|
|
case(
|
|
(FundRiskAlert.alert_level == HIGH_RISK, 0),
|
|
(FundRiskAlert.alert_level == MEDIUM_RISK, 1),
|
|
(FundRiskAlert.alert_level == LOW_RISK, 2),
|
|
else_=3,
|
|
),
|
|
FundRiskAlert.created_at.desc(),
|
|
FundRiskAlert.id.desc(),
|
|
)
|
|
.limit(request.limit + 1)
|
|
.offset(request.offset)
|
|
)
|
|
).all()
|
|
has_more = len(rows) > request.limit
|
|
records = tuple(
|
|
FundRecord(
|
|
entity="risk_alert",
|
|
values=MappingProxyType(
|
|
self._alert_row(
|
|
alert,
|
|
customer_no_value,
|
|
name,
|
|
product_code_value,
|
|
product_name_value,
|
|
)
|
|
),
|
|
)
|
|
for (
|
|
alert,
|
|
customer_no_value,
|
|
name,
|
|
product_code_value,
|
|
product_name_value,
|
|
) in rows[: request.limit]
|
|
)
|
|
return FundPage(
|
|
entity="risk_alert",
|
|
items=records,
|
|
limit=request.limit,
|
|
offset=request.offset,
|
|
next_offset=request.offset + request.limit if has_more else None,
|
|
)
|
|
|
|
async def get_alert_detail(self, alert_no: str) -> FundRecord | None:
|
|
"""聚合指定预警的客户、交易、产品、工单和证据快照。"""
|
|
statement = select(FundRiskAlert).where(FundRiskAlert.alert_no == alert_no)
|
|
scope_condition = self._scope_condition(
|
|
FundRiskAlert.customer_id,
|
|
FundCustomerProfile.trade_account,
|
|
)
|
|
if scope_condition is not None:
|
|
statement = statement.where(scope_condition)
|
|
alert = await self.session.scalar(statement)
|
|
if alert is None:
|
|
return None
|
|
|
|
profile = await self.session.scalar(
|
|
select(FundCustomerProfile).where(
|
|
FundCustomerProfile.customer_id == alert.customer_id
|
|
)
|
|
)
|
|
customer = await self.session.scalar(
|
|
select(RiskUser).where(RiskUser.id == alert.customer_id)
|
|
)
|
|
transaction = (
|
|
await self.session.scalar(
|
|
select(FundTransaction).where(
|
|
FundTransaction.id == alert.related_transaction_id
|
|
)
|
|
)
|
|
if alert.related_transaction_id is not None
|
|
else None
|
|
)
|
|
snapshot = alert.evidence_snapshot if isinstance(alert.evidence_snapshot, dict) else {}
|
|
product_id = (
|
|
transaction.product_id
|
|
if transaction is not None
|
|
else snapshot.get("product_id")
|
|
)
|
|
product = (
|
|
await self.session.scalar(
|
|
select(FundProduct).where(FundProduct.id == int(product_id))
|
|
)
|
|
if product_id is not None
|
|
else None
|
|
)
|
|
work_order = (
|
|
await self.session.scalar(
|
|
select(RiskWorkOrder).where(
|
|
RiskWorkOrder.id == alert.related_work_order_id
|
|
)
|
|
)
|
|
if alert.related_work_order_id is not None
|
|
else None
|
|
)
|
|
capital_flows = list(
|
|
await self.session.scalars(
|
|
select(FundCapitalFlow)
|
|
.where(FundCapitalFlow.customer_id == alert.customer_id)
|
|
.order_by(
|
|
FundCapitalFlow.settled_at.desc(),
|
|
FundCapitalFlow.id.desc(),
|
|
)
|
|
)
|
|
)
|
|
holdings = list(
|
|
await self.session.scalars(
|
|
select(FundHolding)
|
|
.where(FundHolding.customer_id == alert.customer_id)
|
|
.order_by(FundHolding.id.desc())
|
|
)
|
|
)
|
|
login_records = list(
|
|
await self.session.scalars(
|
|
select(RiskLoginRecord)
|
|
.where(RiskLoginRecord.user_id == alert.customer_id)
|
|
.order_by(RiskLoginRecord.login_at.desc(), RiskLoginRecord.id.desc())
|
|
)
|
|
)
|
|
values = {
|
|
"alert": self._alert_row(alert, customer.user_no if customer else None,
|
|
profile.real_name if profile else None,
|
|
product.product_code if product else None,
|
|
product.product_name if product else None),
|
|
"customer": self._profile_row(profile),
|
|
"transaction": self._model_values(transaction),
|
|
"product": self._model_values(product),
|
|
"work_order": self._model_values(work_order),
|
|
"capital_flows": [
|
|
self._model_values(item) for item in capital_flows
|
|
],
|
|
"holdings": [
|
|
self._model_values(item) for item in holdings
|
|
],
|
|
"login_records": [
|
|
self._model_values(item) for item in login_records
|
|
],
|
|
"evidence_snapshot": snapshot,
|
|
}
|
|
return FundRecord(entity="risk_alert_detail", values=MappingProxyType(values))
|
|
|
|
async def daily_report_snapshot(
|
|
self,
|
|
day_start: datetime,
|
|
next_day: datetime,
|
|
) -> RiskReportSnapshot:
|
|
"""一次读取日报所需的四组预警，历史未闭环不应用分页。"""
|
|
async def load(*conditions: Any) -> tuple[FundRecord, ...]:
|
|
statement = select(FundRiskAlert).where(*conditions)
|
|
scope = self._scope_condition(
|
|
FundRiskAlert.customer_id,
|
|
FundCustomerProfile.trade_account,
|
|
)
|
|
if scope is not None:
|
|
statement = statement.where(scope)
|
|
rows = (await self.session.scalars(
|
|
statement.order_by(FundRiskAlert.id.asc())
|
|
)).all()
|
|
return tuple(
|
|
FundRecord(
|
|
entity="risk_alert",
|
|
values=MappingProxyType(
|
|
self._alert_row(item, None, None, None, None)
|
|
),
|
|
)
|
|
for item in rows
|
|
)
|
|
|
|
daily = await load(
|
|
FundRiskAlert.created_at >= day_start,
|
|
FundRiskAlert.created_at < next_day,
|
|
)
|
|
unresolved = await load(FundRiskAlert.status.in_(OPEN_STATUSES))
|
|
false_positive = await load(
|
|
FundRiskAlert.status == "已排除",
|
|
FundRiskAlert.closed_at >= day_start,
|
|
FundRiskAlert.closed_at < next_day,
|
|
)
|
|
dispositions = await load(or_(
|
|
and_(FundRiskAlert.ack_at >= day_start, FundRiskAlert.ack_at < next_day),
|
|
and_(FundRiskAlert.closed_at >= day_start, FundRiskAlert.closed_at < next_day),
|
|
and_(FundRiskAlert.escalated_at >= day_start, FundRiskAlert.escalated_at < next_day),
|
|
and_(
|
|
FundRiskAlert.status == "调查中",
|
|
FundRiskAlert.updated_at >= day_start,
|
|
FundRiskAlert.updated_at < next_day,
|
|
),
|
|
))
|
|
return RiskReportSnapshot(
|
|
daily=daily,
|
|
unresolved=unresolved,
|
|
false_positive=false_positive,
|
|
dispositions=dispositions,
|
|
)
|
|
|
|
async def list_customers(
|
|
self,
|
|
*,
|
|
keyword: str | None = None,
|
|
behavior_level: str | None = None,
|
|
page: PageRequest | None = None,
|
|
) -> FundPage:
|
|
"""客户画像、行为分和最新风险测评分页查询。"""
|
|
request = page or PageRequest()
|
|
latest = (
|
|
select(
|
|
FundRiskAssessment.customer_id,
|
|
func.max(FundRiskAssessment.assessed_at).label("assessed_at"),
|
|
)
|
|
.group_by(FundRiskAssessment.customer_id)
|
|
.subquery()
|
|
)
|
|
statement = (
|
|
select(RiskUser, FundCustomerProfile, FundRiskAssessment)
|
|
.join(FundCustomerProfile, FundCustomerProfile.customer_id == RiskUser.id)
|
|
.outerjoin(latest, latest.c.customer_id == RiskUser.id)
|
|
.outerjoin(
|
|
FundRiskAssessment,
|
|
(FundRiskAssessment.customer_id == RiskUser.id)
|
|
& (FundRiskAssessment.assessed_at == latest.c.assessed_at),
|
|
)
|
|
.where(RiskUser.user_type == "CUSTOMER")
|
|
)
|
|
conditions: list[ColumnElement[bool]] = []
|
|
scope_condition = self._scope_condition(
|
|
RiskUser.id,
|
|
FundCustomerProfile.trade_account,
|
|
)
|
|
if scope_condition is not None:
|
|
conditions.append(scope_condition)
|
|
if keyword and keyword.strip():
|
|
like = f"%{keyword.strip()}%"
|
|
conditions.append(
|
|
or_(
|
|
RiskUser.user_no.like(like),
|
|
RiskUser.username.like(like),
|
|
FundCustomerProfile.real_name.like(like),
|
|
)
|
|
)
|
|
if behavior_level:
|
|
minimum, maximum = BEHAVIOR_SCORE_RANGES.get(behavior_level, (0, 20))
|
|
conditions.append(FundCustomerProfile.behavior_score.between(minimum, maximum))
|
|
return await self._page(
|
|
statement.where(*conditions).order_by(RiskUser.id.asc()),
|
|
request,
|
|
"risk_customer",
|
|
self._customer_row,
|
|
)
|
|
|
|
async def list_products(
|
|
self,
|
|
*,
|
|
keyword: str | None = None,
|
|
page: PageRequest | None = None,
|
|
) -> FundPage:
|
|
"""产品为公共数据，直接复用 FundQueryRepository。"""
|
|
spec = None
|
|
if keyword and keyword.strip():
|
|
like = f"%{keyword.strip()}%"
|
|
spec = FundQuerySpec.where(
|
|
AnyOf(
|
|
(
|
|
FieldFilter("product_code", like, FundFilterOperator.LIKE),
|
|
FieldFilter("product_name", like, FundFilterOperator.LIKE),
|
|
)
|
|
)
|
|
)
|
|
return await FundQueryRepository(
|
|
self.session,
|
|
scope=CustomerScope.unrestricted(),
|
|
).products(spec=spec, page=page)
|
|
|
|
async def list_transactions(
|
|
self,
|
|
*,
|
|
keyword: str | None = None,
|
|
start_time: datetime | None = None,
|
|
end_time: datetime | None = None,
|
|
page: PageRequest | None = None,
|
|
) -> FundPage:
|
|
"""交易流水查询，并补充客户、产品、渠道和风险留痕。"""
|
|
request = page or PageRequest()
|
|
statement = (
|
|
select(
|
|
FundTransaction,
|
|
RiskUser.user_no,
|
|
FundProduct.product_code,
|
|
FundProduct.product_name,
|
|
RiskWorkOrder,
|
|
)
|
|
.join(RiskUser, RiskUser.id == FundTransaction.customer_id)
|
|
.join(
|
|
FundCustomerProfile,
|
|
FundCustomerProfile.customer_id == FundTransaction.customer_id,
|
|
)
|
|
.join(FundProduct, FundProduct.id == FundTransaction.product_id)
|
|
.outerjoin(RiskWorkOrder, RiskWorkOrder.id == FundTransaction.work_order_id)
|
|
)
|
|
conditions = self._conditions_with_scope(
|
|
FundTransaction.customer_id,
|
|
FundCustomerProfile.trade_account,
|
|
)
|
|
if keyword and keyword.strip():
|
|
like = f"%{keyword.strip()}%"
|
|
conditions.append(
|
|
or_(
|
|
FundTransaction.transaction_no.like(like),
|
|
RiskUser.user_no.like(like),
|
|
FundProduct.product_code.like(like),
|
|
FundProduct.product_name.like(like),
|
|
)
|
|
)
|
|
if start_time is not None:
|
|
conditions.append(FundTransaction.executed_at >= start_time)
|
|
if end_time is not None:
|
|
conditions.append(FundTransaction.executed_at <= end_time)
|
|
return await self._page(
|
|
statement.where(*conditions).order_by(
|
|
FundTransaction.executed_at.desc(),
|
|
FundTransaction.id.desc(),
|
|
),
|
|
request,
|
|
"risk_transaction",
|
|
self._transaction_row,
|
|
)
|
|
|
|
async def list_capital_flows(
|
|
self,
|
|
*,
|
|
keyword: str | None = None,
|
|
start_time: datetime | None = None,
|
|
end_time: datetime | None = None,
|
|
page: PageRequest | None = None,
|
|
) -> FundPage:
|
|
"""资金流水查询。"""
|
|
request = page or PageRequest()
|
|
statement = (
|
|
select(FundCapitalFlow, RiskUser.user_no)
|
|
.join(RiskUser, RiskUser.id == FundCapitalFlow.customer_id)
|
|
.join(
|
|
FundCustomerProfile,
|
|
FundCustomerProfile.customer_id == FundCapitalFlow.customer_id,
|
|
)
|
|
)
|
|
conditions = self._conditions_with_scope(
|
|
FundCapitalFlow.customer_id,
|
|
FundCustomerProfile.trade_account,
|
|
)
|
|
if keyword and keyword.strip():
|
|
like = f"%{keyword.strip()}%"
|
|
conditions.append(
|
|
or_(
|
|
FundCapitalFlow.flow_no.like(like),
|
|
RiskUser.user_no.like(like),
|
|
)
|
|
)
|
|
if start_time is not None:
|
|
conditions.append(FundCapitalFlow.occurred_at >= start_time)
|
|
if end_time is not None:
|
|
conditions.append(FundCapitalFlow.occurred_at <= end_time)
|
|
return await self._page(
|
|
statement.where(*conditions).order_by(
|
|
FundCapitalFlow.created_at.desc(),
|
|
FundCapitalFlow.id.desc(),
|
|
),
|
|
request,
|
|
"risk_capital_flow",
|
|
self._capital_flow_row,
|
|
)
|
|
|
|
async def list_holdings(
|
|
self,
|
|
*,
|
|
keyword: str | None = None,
|
|
page: PageRequest | None = None,
|
|
) -> FundPage:
|
|
"""持仓查询，并计算持有天数和持仓占比。"""
|
|
request = page or PageRequest()
|
|
statement = (
|
|
select(
|
|
FundHolding,
|
|
RiskUser.user_no,
|
|
FundProduct.product_code,
|
|
FundProduct.product_name,
|
|
FundCustomerProfile.total_asset,
|
|
)
|
|
.join(RiskUser, RiskUser.id == FundHolding.customer_id)
|
|
.join(FundCustomerProfile, FundCustomerProfile.customer_id == FundHolding.customer_id)
|
|
.join(FundProduct, FundProduct.id == FundHolding.product_id)
|
|
)
|
|
conditions = self._conditions_with_scope(
|
|
FundHolding.customer_id,
|
|
FundCustomerProfile.trade_account,
|
|
)
|
|
if keyword and keyword.strip():
|
|
like = f"%{keyword.strip()}%"
|
|
conditions.append(
|
|
or_(
|
|
RiskUser.user_no.like(like),
|
|
FundProduct.product_code.like(like),
|
|
FundProduct.product_name.like(like),
|
|
)
|
|
)
|
|
return await self._page(
|
|
statement.where(*conditions).order_by(
|
|
FundHolding.updated_at.desc(),
|
|
FundHolding.id.desc(),
|
|
),
|
|
request,
|
|
"risk_holding",
|
|
self._holding_row,
|
|
)
|
|
|
|
async def list_login_records(
|
|
self,
|
|
*,
|
|
keyword: str | None = None,
|
|
start_time: datetime | None = None,
|
|
end_time: datetime | None = None,
|
|
page: PageRequest | None = None,
|
|
) -> FundPage:
|
|
"""登录记录查询。"""
|
|
request = page or PageRequest()
|
|
statement = (
|
|
select(RiskLoginRecord, RiskUser.user_no)
|
|
.join(RiskUser, RiskUser.id == RiskLoginRecord.user_id)
|
|
.join(FundCustomerProfile, FundCustomerProfile.customer_id == RiskLoginRecord.user_id)
|
|
)
|
|
conditions = self._conditions_with_scope(
|
|
RiskLoginRecord.user_id,
|
|
FundCustomerProfile.trade_account,
|
|
)
|
|
if keyword and keyword.strip():
|
|
like = f"%{keyword.strip()}%"
|
|
conditions.append(
|
|
or_(
|
|
RiskUser.user_no.like(like),
|
|
RiskLoginRecord.device_id.like(like),
|
|
RiskLoginRecord.ip_region.like(like),
|
|
)
|
|
)
|
|
if start_time is not None:
|
|
conditions.append(RiskLoginRecord.login_at >= start_time)
|
|
if end_time is not None:
|
|
conditions.append(RiskLoginRecord.login_at <= end_time)
|
|
return await self._page(
|
|
statement.where(*conditions).order_by(
|
|
RiskLoginRecord.login_at.desc(),
|
|
RiskLoginRecord.id.desc(),
|
|
),
|
|
request,
|
|
"risk_login_record",
|
|
self._login_record_row,
|
|
)
|
|
|
|
async def list_notifications(
|
|
self,
|
|
*,
|
|
keyword: str | None = None,
|
|
send_status: str | None = None,
|
|
start_time: datetime | None = None,
|
|
end_time: datetime | None = None,
|
|
page: PageRequest | None = None,
|
|
) -> FundPage:
|
|
"""通知记录查询，并返回预警编号。"""
|
|
request = page or PageRequest()
|
|
statement = (
|
|
select(
|
|
FundRiskNotification,
|
|
FundRiskAlert.alert_no,
|
|
RiskUser.user_no,
|
|
)
|
|
.join(FundRiskAlert, FundRiskAlert.id == FundRiskNotification.alert_id)
|
|
.join(RiskUser, RiskUser.id == FundRiskAlert.customer_id)
|
|
.join(
|
|
FundCustomerProfile,
|
|
FundCustomerProfile.customer_id == FundRiskAlert.customer_id,
|
|
)
|
|
)
|
|
conditions = self._conditions_with_scope(
|
|
FundRiskAlert.customer_id,
|
|
FundCustomerProfile.trade_account,
|
|
)
|
|
if keyword and keyword.strip():
|
|
like = f"%{keyword.strip()}%"
|
|
conditions.append(
|
|
or_(
|
|
FundRiskNotification.notification_no.like(like),
|
|
FundRiskAlert.alert_no.like(like),
|
|
FundRiskNotification.title.like(like),
|
|
FundRiskNotification.send_status.like(like),
|
|
FundRiskNotification.receiver_email.like(like),
|
|
)
|
|
)
|
|
if send_status:
|
|
conditions.append(FundRiskNotification.send_status == send_status)
|
|
if start_time is not None:
|
|
conditions.append(FundRiskNotification.created_at >= start_time)
|
|
if end_time is not None:
|
|
conditions.append(FundRiskNotification.created_at <= end_time)
|
|
return await self._page(
|
|
statement.where(*conditions).order_by(
|
|
FundRiskNotification.created_at.desc(),
|
|
FundRiskNotification.id.desc(),
|
|
),
|
|
request,
|
|
"risk_notification",
|
|
self._notification_row,
|
|
)
|
|
|
|
def _alert_list_statement(
|
|
self,
|
|
*,
|
|
keyword: str | None,
|
|
customer_no: str | None,
|
|
product_code: str | None,
|
|
product_name: str | None,
|
|
risk_level: str | None,
|
|
rule_code: str | None,
|
|
start_time: datetime | None,
|
|
end_time: datetime | None,
|
|
open_only: bool,
|
|
) -> Select[Any]:
|
|
statement = (
|
|
select(
|
|
FundRiskAlert,
|
|
RiskUser.user_no,
|
|
FundCustomerProfile.real_name,
|
|
FundProduct.product_code,
|
|
FundProduct.product_name,
|
|
)
|
|
.outerjoin(RiskUser, RiskUser.id == FundRiskAlert.customer_id)
|
|
.outerjoin(
|
|
FundCustomerProfile,
|
|
FundCustomerProfile.customer_id == FundRiskAlert.customer_id,
|
|
)
|
|
.outerjoin(
|
|
FundTransaction,
|
|
FundTransaction.id == FundRiskAlert.related_transaction_id,
|
|
)
|
|
.outerjoin(FundProduct, FundProduct.id == FundTransaction.product_id)
|
|
)
|
|
conditions: list[ColumnElement[bool]] = []
|
|
if open_only:
|
|
conditions.append(FundRiskAlert.status.in_(OPEN_STATUSES))
|
|
if keyword and keyword.strip():
|
|
like = f"%{keyword.strip()}%"
|
|
conditions.append(
|
|
or_(
|
|
FundRiskAlert.alert_no.like(like),
|
|
FundRiskAlert.alert_type.like(like),
|
|
FundRiskAlert.evidence_summary.like(like),
|
|
RiskUser.user_no.like(like),
|
|
FundCustomerProfile.real_name.like(like),
|
|
)
|
|
)
|
|
if customer_no:
|
|
conditions.append(RiskUser.user_no == customer_no)
|
|
if product_code:
|
|
conditions.append(FundProduct.product_code == product_code)
|
|
if product_name:
|
|
conditions.append(FundProduct.product_name.like(f"%{product_name}%"))
|
|
if risk_level:
|
|
conditions.append(FundRiskAlert.alert_level == risk_level)
|
|
if rule_code:
|
|
conditions.append(FundRiskAlert.trigger_rule_codes.contains([rule_code]))
|
|
if start_time is not None:
|
|
conditions.append(FundRiskAlert.created_at >= start_time)
|
|
if end_time is not None:
|
|
conditions.append(FundRiskAlert.created_at <= end_time)
|
|
scope_condition = self._scope_condition(
|
|
FundRiskAlert.customer_id,
|
|
FundCustomerProfile.trade_account,
|
|
)
|
|
if scope_condition is not None:
|
|
conditions.append(scope_condition)
|
|
return statement.where(*conditions)
|
|
|
|
def _scope_condition(
|
|
self,
|
|
customer_column: Any,
|
|
trade_account_column: Any | None = None,
|
|
) -> Any:
|
|
if self.scope is None or self.scope.is_denied:
|
|
return false()
|
|
conditions = []
|
|
if self.scope.customer_ids is not None:
|
|
conditions.append(customer_column.in_(sorted(self.scope.customer_ids)))
|
|
if self.scope.trade_accounts is not None:
|
|
if trade_account_column is None:
|
|
return false()
|
|
conditions.append(
|
|
customer_column.in_(
|
|
select(FundCustomerProfile.customer_id).where(
|
|
trade_account_column.in_(sorted(self.scope.trade_accounts))
|
|
)
|
|
)
|
|
)
|
|
if not conditions:
|
|
return None
|
|
return and_(*conditions) if len(conditions) > 1 else conditions[0]
|
|
|
|
def _conditions_with_scope(
|
|
self,
|
|
customer_column: Any,
|
|
trade_account_column: Any | None = None,
|
|
) -> list[Any]:
|
|
condition = self._scope_condition(customer_column, trade_account_column)
|
|
return [] if condition is None else [condition]
|
|
|
|
async def _page(
|
|
self,
|
|
statement: Select[Any],
|
|
page: PageRequest,
|
|
entity: str,
|
|
row_builder: Callable[..., dict[str, Any]],
|
|
) -> FundPage:
|
|
rows = (
|
|
await self.session.execute(
|
|
statement.limit(page.limit + 1).offset(page.offset)
|
|
)
|
|
).all()
|
|
has_more = len(rows) > page.limit
|
|
records = tuple(
|
|
FundRecord(
|
|
entity=entity,
|
|
values=MappingProxyType(row_builder(*row)),
|
|
)
|
|
for row in rows[: page.limit]
|
|
)
|
|
return FundPage(
|
|
entity=entity,
|
|
items=records,
|
|
limit=page.limit,
|
|
offset=page.offset,
|
|
next_offset=page.offset + page.limit if has_more else None,
|
|
)
|
|
|
|
@staticmethod
|
|
def _customer_row(
|
|
user: RiskUser,
|
|
profile: FundCustomerProfile,
|
|
assessment: FundRiskAssessment | None,
|
|
) -> dict[str, Any]:
|
|
return {
|
|
"customer_id": str(user.id),
|
|
"customer_no": user.user_no,
|
|
"username": user.username,
|
|
"name": mask_name(profile.real_name),
|
|
"age": _age(profile.birth_date),
|
|
"occupation": profile.occupation,
|
|
"mobile_masked": profile.mobile_masked,
|
|
"total_asset": profile.total_asset,
|
|
"customer_tier": user.customer_tier,
|
|
"risk_level": profile.investor_type,
|
|
"risk_score": assessment.total_score if assessment else None,
|
|
"assessment_date": assessment.assessed_at if assessment else None,
|
|
"assessment_valid_until": assessment.valid_until if assessment else None,
|
|
"assessment_expired": bool(
|
|
assessment
|
|
and assessment.valid_until
|
|
and assessment.valid_until < _utc_now_naive()
|
|
),
|
|
"behavior_score": profile.behavior_score,
|
|
"risk_tags": profile.risk_tags or [],
|
|
"opened_at": profile.opened_at,
|
|
"status": user.status,
|
|
}
|
|
|
|
@staticmethod
|
|
def _transaction_row(
|
|
transaction: FundTransaction,
|
|
customer_no: str,
|
|
product_code: str,
|
|
product_name: str,
|
|
work_order: RiskWorkOrder | None,
|
|
) -> dict[str, Any]:
|
|
return {
|
|
"transaction_no": transaction.transaction_no,
|
|
"customer_no": customer_no,
|
|
"product_code": product_code,
|
|
"product_name": product_name,
|
|
"transaction_type": transaction.transaction_type,
|
|
"amount": transaction.amount,
|
|
"channel": work_order.channel if work_order else None,
|
|
"trade_status": work_order.status if work_order else "已确认",
|
|
"risk_disclosure_signed": bool(
|
|
work_order and work_order.risk_disclosure_ack_at
|
|
),
|
|
"second_confirmation": bool(
|
|
work_order and work_order.second_confirmation_at
|
|
),
|
|
"recording_id": work_order.recording_reference if work_order else None,
|
|
"work_order_no": work_order.work_order_no if work_order else None,
|
|
"confirmed_at": transaction.confirmed_at,
|
|
"executed_at": transaction.executed_at,
|
|
}
|
|
|
|
@staticmethod
|
|
def _capital_flow_row(
|
|
flow: FundCapitalFlow,
|
|
customer_no: str,
|
|
) -> dict[str, Any]:
|
|
return {
|
|
"flow_no": flow.flow_no,
|
|
"customer_no": customer_no,
|
|
"flow_type": flow.flow_type,
|
|
"amount": flow.amount,
|
|
"status": flow.status,
|
|
"settled_at": flow.settled_at,
|
|
"occurred_at": flow.occurred_at,
|
|
"source_type": flow.source_type,
|
|
"match_status": flow.match_status,
|
|
}
|
|
|
|
@staticmethod
|
|
def _holding_row(
|
|
holding: FundHolding,
|
|
customer_no: str,
|
|
product_code: str,
|
|
product_name: str,
|
|
total_asset: Decimal | None,
|
|
) -> dict[str, Any]:
|
|
current_value = holding.current_value or holding.market_value or Decimal("0")
|
|
holding_days = None
|
|
if holding.first_acquired_at is not None:
|
|
holding_days = (datetime.now(UTC).date() - holding.first_acquired_at.date()).days
|
|
holding_ratio = None
|
|
if total_asset is not None and total_asset != 0:
|
|
holding_ratio = (current_value / total_asset).quantize(Decimal("0.0001"))
|
|
return {
|
|
"customer_no": customer_no,
|
|
"product_code": product_code,
|
|
"product_name": product_name,
|
|
"shares": holding.shares or holding.total_quantity,
|
|
"cost_amount": holding.cost_amount,
|
|
"current_value": current_value,
|
|
"profit_loss": holding.profit_loss,
|
|
"holding_days": holding_days,
|
|
"holding_ratio": holding_ratio,
|
|
}
|
|
|
|
@staticmethod
|
|
def _login_record_row(
|
|
record: RiskLoginRecord,
|
|
customer_no: str,
|
|
) -> dict[str, Any]:
|
|
return {
|
|
"id": str(record.id),
|
|
"customer_no": customer_no,
|
|
"login_at": record.login_at,
|
|
"login_result": record.login_result,
|
|
"ip_region": record.ip_region,
|
|
"device_id": record.device_id,
|
|
"is_common_device": bool(record.is_common_device),
|
|
"failure_reason": record.failure_reason,
|
|
}
|
|
|
|
@staticmethod
|
|
def _notification_row(
|
|
notification: FundRiskNotification,
|
|
alert_no: str,
|
|
customer_no: str,
|
|
) -> dict[str, Any]:
|
|
return {
|
|
"notification_id": notification.notification_no,
|
|
"notification_no": notification.notification_no,
|
|
"alert_no": alert_no,
|
|
"customer_no": customer_no,
|
|
"channel": notification.channel,
|
|
"title": notification.title,
|
|
"send_status": notification.send_status,
|
|
"receiver_email": notification.receiver_email,
|
|
"send_time": notification.sent_at,
|
|
}
|
|
|
|
@staticmethod
|
|
def _alert_row(
|
|
alert: FundRiskAlert,
|
|
customer_no: str | None,
|
|
real_name: str | None,
|
|
product_code: str | None,
|
|
product_name: str | None,
|
|
) -> dict[str, Any]:
|
|
snapshot = alert.evidence_snapshot if isinstance(alert.evidence_snapshot, dict) else {}
|
|
return {
|
|
"alert_no": alert.alert_no,
|
|
"customer_id": str(alert.customer_id),
|
|
"customer_no": customer_no,
|
|
"customer_name": mask_name(real_name),
|
|
"product_code": product_code,
|
|
"product_name": product_name,
|
|
"alert_type": alert.alert_type,
|
|
"risk_level": alert.alert_level,
|
|
"rule_codes": tuple(alert.trigger_rule_codes or []),
|
|
"evidence_summary": alert.evidence_summary,
|
|
"evidence_snapshot": snapshot,
|
|
"priority_score": alert.priority_score,
|
|
"event_status": alert.event_status,
|
|
"status": alert.status,
|
|
"ack_status": alert.ack_status,
|
|
"ack_at": alert.ack_at,
|
|
"due_at": alert.due_at,
|
|
"is_escalated": bool(alert.is_escalated),
|
|
"escalated_at": alert.escalated_at,
|
|
"evidence_archived": bool(snapshot.get("evidence_archive")),
|
|
"created_at": alert.created_at,
|
|
"updated_at": alert.updated_at,
|
|
}
|
|
|
|
@staticmethod
|
|
def _profile_row(profile: FundCustomerProfile | None) -> dict[str, Any] | None:
|
|
if profile is None:
|
|
return None
|
|
return {
|
|
"customer_id": str(profile.customer_id),
|
|
"name": mask_name(profile.real_name),
|
|
"birth_date": profile.birth_date,
|
|
"occupation": profile.occupation,
|
|
"mobile_masked": profile.mobile_masked,
|
|
"investor_type": profile.investor_type,
|
|
"investment_horizon": profile.investment_horizon,
|
|
"trading_frequency": profile.trading_frequency,
|
|
"total_asset": profile.total_asset,
|
|
"behavior_score": profile.behavior_score,
|
|
"risk_tags": profile.risk_tags or [],
|
|
"updated_at": profile.updated_at,
|
|
}
|
|
|
|
@staticmethod
|
|
def _model_values(model: Any | None) -> dict[str, Any] | None:
|
|
if model is None:
|
|
return None
|
|
return {
|
|
column.key: getattr(model, column.key)
|
|
for column in model.__table__.columns
|
|
}
|
|
|
|
|
|
def mask_name(name: str | None) -> str | None:
|
|
"""只保留姓名首字，其余使用 `*`。"""
|
|
if not name:
|
|
return name
|
|
return name[0] + "*" * max(len(name) - 1, 1)
|
|
|
|
|
|
def _age(birth_date: date | None) -> int | None:
|
|
if birth_date is None:
|
|
return None
|
|
today = datetime.now(UTC).date()
|
|
return today.year - birth_date.year - (
|
|
(today.month, today.day) < (birth_date.month, birth_date.day)
|
|
)
|
|
|
|
|
|
def _utc_now_naive() -> datetime:
|
|
return datetime.now(UTC).replace(tzinfo=None)
|