From 98751d592fae3ca7ea034457f5cf5eadb559ec78 Mon Sep 17 00:00:00 2001 From: zhangyongcai <17382809141.@163.com> Date: Sat, 12 Sep 2026 20:59:10 +0800 Subject: [PATCH 1/3] =?UTF-8?q?feat:=E6=96=B0=E5=A2=9E=E6=8E=A5=E5=8F=A3?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- api/deps.py | 9 +++ api/router.py | 4 +- model/biz_work_order.py | 34 +++++------ repositories/sys_user.py | 13 +++- schemas/purchase.py | 16 +++-- schemas/redeem.py | 16 +++-- service/purchase.py | 122 +++++++++++++++++++++++++++++-------- service/redeem.py | 128 +++++++++++++++++++++++++++++++-------- 8 files changed, 258 insertions(+), 84 deletions(-) diff --git a/api/deps.py b/api/deps.py index 71c0db7..c0fe658 100644 --- a/api/deps.py +++ b/api/deps.py @@ -44,3 +44,12 @@ async def require_customer(user: SysUser = Depends(get_current_user)) -> SysUser if user.user_type != "CUSTOMER": raise ForbiddenError("仅客户用户可以访问 client_agent") return user + + +async def require_risk_officer(user: SysUser = Depends(get_current_user)) -> SysUser: + """仅允许管理员或风控专员执行风控处置/工单操作。""" + if user.user_type == "ADMIN": + return user + if user.user_type != "EMPLOYEE" or user.employee_role != "风控专员": + raise ForbiddenError("仅风控专员可执行风控处置") + return user diff --git a/api/router.py b/api/router.py index bd25ed8..90d5018 100644 --- a/api/router.py +++ b/api/router.py @@ -5,7 +5,7 @@ from fastapi import APIRouter from api.chat import client_agent, customer_agent, knowledge from api.routers import product, questionnaire -from api.routers import account, auth, holdings, purchase, redeem +from api.routers import account, auth, holdings, purchase, redeem, risk, work_order api_router = APIRouter() api_router.include_router(auth.router, prefix="/api", tags=["认证"]) @@ -13,6 +13,8 @@ api_router.include_router(account.router, prefix="/api", tags=["资金账户"]) api_router.include_router(holdings.router, prefix="/api", tags=["持仓"]) api_router.include_router(purchase.router, prefix="/api", tags=["交易"]) api_router.include_router(redeem.router, prefix="/api", tags=["交易"]) +api_router.include_router(risk.router, prefix="/api", tags=["风控"]) +api_router.include_router(work_order.router, prefix="/api", tags=["工单"]) api_router.include_router(customer_agent.router, prefix="/api/agent/customer", tags=["客服Agent"]) api_router.include_router(client_agent.router, prefix="/api/agent/client", tags=["ClientAgent"]) api_router.include_router(knowledge.router, prefix="/api/knowledge", tags=["知识库"]) diff --git a/model/biz_work_order.py b/model/biz_work_order.py index 10a50d8..e60004f 100644 --- a/model/biz_work_order.py +++ b/model/biz_work_order.py @@ -1,5 +1,4 @@ -"""biz_work_order 工单表 ORM 模型。""" - +"""biz_work_order 通用业务工单 ORM 模型（拦截/冻结后开启的独立审批链）。""" from __future__ import annotations from datetime import datetime @@ -12,26 +11,21 @@ from model.base import Base class BizWorkOrder(Base): - """客户工单状态记录。""" - __tablename__ = "biz_work_order" - __table_args__ = {"comment": "通用业务工单表"} + __table_args__ = {"comment": "通用业务工单表（投标顾签约、可疑交易上报等流程载体）"} id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True) - work_order_no: Mapped[str] = mapped_column(String(32), nullable=False) - order_type: Mapped[str] = mapped_column(String(32), nullable=False) - sub_type: Mapped[str | None] = mapped_column(String(32), nullable=True) - customer_id: Mapped[int | None] = mapped_column(BigInteger, nullable=True) - submitter_id: Mapped[int | None] = mapped_column(BigInteger, nullable=True) - handler_id: Mapped[int | None] = mapped_column(BigInteger, nullable=True) - current_node: Mapped[str] = mapped_column(String(32), nullable=False) - priority: Mapped[str] = mapped_column(String(8), nullable=False) - status: Mapped[str] = mapped_column(String(16), nullable=False) - biz_content: Mapped[dict[str, Any] | None] = mapped_column(JSON, nullable=True) - create_time: Mapped[datetime] = mapped_column( - DateTime, server_default=func.now(), nullable=False - ) + work_order_no: Mapped[str] = mapped_column(String(32), unique=True) + order_type: Mapped[str] = mapped_column(String(32)) + sub_type: Mapped[str | None] = mapped_column(String(32)) + customer_id: Mapped[int | None] = mapped_column(BigInteger) + submitter_id: Mapped[int | None] = mapped_column(BigInteger) + handler_id: Mapped[int | None] = mapped_column(BigInteger) + current_node: Mapped[str] = mapped_column(String(32), server_default="初审") + priority: Mapped[str] = mapped_column(String(8), server_default="普通") + status: Mapped[str] = mapped_column(String(16), server_default="待处理") + biz_content: Mapped[dict[str, Any] | None] = mapped_column(JSON) + create_time: Mapped[datetime] = mapped_column(DateTime, server_default=func.now()) update_time: Mapped[datetime] = mapped_column( - DateTime, server_default=func.now(), onupdate=func.now(), nullable=False + DateTime, server_default=func.now(), onupdate=func.now() ) - diff --git a/repositories/sys_user.py b/repositories/sys_user.py index 5304dc5..214631f 100644 --- a/repositories/sys_user.py +++ b/repositories/sys_user.py @@ -1,7 +1,7 @@ -"""sys_user 仓储：认证用按用户名/主键取用户。""" +"""sys_user 仓储：认证用按用户名/主键取用户 + 账号状态更新。""" from __future__ import annotations -from sqlalchemy import select +from sqlalchemy import select, update from model.sys_user import SysUser from repositories.base import BaseRepository @@ -14,4 +14,11 @@ class SysUserRepo(BaseRepository): return await self.db.scalar(select(SysUser).where(SysUser.username == username)) async def get_by_phone(self, phone: str) -> SysUser | None: - return await self.db.scalar(select(SysUser).where(SysUser.phone == phone)) \ No newline at end of file + return await self.db.scalar(select(SysUser).where(SysUser.phone == phone)) + + async def update_status(self, user_id: int, status: str) -> int: + """更新账号状态（如风控冻结客户）。不 commit，由 service 层事务统一提交。""" + result = await self.db.execute( + update(SysUser).where(SysUser.id == user_id).values(status=status) + ) + return result.rowcount diff --git a/schemas/purchase.py b/schemas/purchase.py index 2cd5680..0ea0834 100644 --- a/schemas/purchase.py +++ b/schemas/purchase.py @@ -14,11 +14,17 @@ class PurchaseReq(BaseModel): class PurchaseResp(BaseModel): - """申购结果：最新余额 + 该产品最新持仓。""" + """申购结果：order_no + status（已确认 / 风控挂起）。 - balance: Decimal - holding: HoldingResp + 已确认时返回最新余额 + 该产品持仓；风控挂起时账户/持仓不变，仅返回预警关联。 + """ + + order_no: str + status: str + balance: Decimal | None = None + holding: HoldingResp | None = None + alert_id: int | None = None @field_serializer("balance") - def _fmt_balance(self, value: Decimal) -> str: - return f"{value:.2f}" + def _fmt_balance(self, value: Decimal | None) -> str | None: + return f"{value:.2f}" if value is not None else None diff --git a/schemas/redeem.py b/schemas/redeem.py index b80071c..4897868 100644 --- a/schemas/redeem.py +++ b/schemas/redeem.py @@ -14,11 +14,17 @@ class RedeemReq(BaseModel): class RedeemResp(BaseModel): - """赎回结果：最新余额 + 该产品最新持仓。""" + """赎回结果：order_no + status（已确认 / 风控挂起）。 - balance: Decimal - holding: HoldingResp + 已确认时返回最新余额 + 该产品持仓；风控挂起时账户/持仓不变，仅返回预警关联。 + """ + + order_no: str + status: str + balance: Decimal | None = None + holding: HoldingResp | None = None + alert_id: int | None = None @field_serializer("balance") - def _fmt_balance(self, value: Decimal) -> str: - return f"{value:.2f}" + def _fmt_balance(self, value: Decimal | None) -> str | None: + return f"{value:.2f}" if value is not None else None diff --git a/service/purchase.py b/service/purchase.py index fa44f49..0933779 100644 --- a/service/purchase.py +++ b/service/purchase.py @@ -1,23 +1,31 @@ -"""申购服务：风险匹配校验 + 余额扣减 + 持仓加仓（单事务原子）。""" +"""申购服务：适当性校验 → 建单(待确认) → 风控检测 → 落账/挂起（单事务原子）。""" from __future__ import annotations +from datetime import datetime from decimal import ROUND_HALF_UP, Decimal from sqlalchemy.ext.asyncio import AsyncSession +from model.fin_risk_alert import FinRiskAlert +from model.fin_transaction import FinTransaction from model.sys_user import SysUser +from model.trade_order import TradeOrder from repositories.fin_account import FinAccountRepo from repositories.fin_customer_profile import FinCustomerProfileRepo from repositories.fin_holdings import FinHoldingsRepo from repositories.fin_product import FinProductRepo +from repositories.trade_order import TradeOrderRepo from schemas.holdings import HoldingResp from schemas.purchase import PurchaseResp +from service.risk.engine import RiskEngine, summarize +from service.risk.settle import settle from utils.exceptions import ( ForbiddenError, NotFoundError, NotSuitableError, ParamError, ) +from utils.order_no import gen_order_no _MONEY = Decimal("0.01") _SHARES = Decimal("0.0001") @@ -36,15 +44,30 @@ def _risk_rank(level: str | None) -> int | None: return _RISK_RANK.get(level.strip()) +def _holding_resp(holding) -> HoldingResp: + return HoldingResp( + id=holding.id, + customer_id=holding.customer_id, + product_id=holding.product_id, + shares=holding.shares, + cost_amount=holding.cost_amount, + current_value=holding.current_value, + profit_loss=holding.profit_loss, + profit_ratio=holding.profit_ratio, + status=holding.status, + ) + + async def purchase( db: AsyncSession, user: SysUser, product_id: int, amount: Decimal ) -> PurchaseResp: - """申购基金：校验通过后扣减余额并加仓，全程单事务。 + """申购基金：适当性校验 → 建单 → 风控检测 → 落账/挂起，全程单事务。 - 仅客户可申购； - 产品须在售且净值非空； - - 客户风险等级序号 >= 产品风险等级序号，否则 1005 拦截； - - 余额不足拦截；扣款 + 加仓要么都成、要么都回滚。 + - 适当性不匹配（1005）在建单前直接拒绝，不生成订单； + - 未命中规则：落账 + 写成交流水，订单「已确认」； + - 命中规则：生成预警，订单「风控挂起」，账户与持仓不变。 """ if user.user_type != "CUSTOMER": raise ForbiddenError("仅客户账号可申购") @@ -69,31 +92,82 @@ async def purchase( shares = (amount / product.nav).quantize(_SHARES, rounding=ROUND_HALF_UP) + now = datetime.now() + order = TradeOrder( + order_no=gen_order_no("PO"), + customer_id=user.id, + product_id=product_id, + advisor_id=None, + order_type="申购", + amount=amount, + shares=shares, + nav=product.nav, + fee=Decimal("0"), + status="待确认", + create_time=now, + ) + db.add(order) + await db.flush() # 生成 order.id，供流水/预警关联 + + hits = await RiskEngine(db).detect(order, user) account_repo = FinAccountRepo(db) holdings_repo = FinHoldingsRepo(db) try: - if not await account_repo.deduct_balance(user.id, amount): - raise ParamError("可用余额不足") - await holdings_repo.upsert(user.id, product_id, shares, amount) + if not hits: + # 未命中：落账 + 写流水 + 订单确认 + await settle(db, order) + db.add( + FinTransaction( + transaction_no=gen_order_no("TR"), + order_id=order.id, + customer_id=user.id, + product_id=product_id, + operator_id=None, + transaction_type="申购", + amount=amount, + shares=shares, + nav=product.nav, + fee=Decimal("0"), + status="已确认", + create_time=now, + ) + ) + await TradeOrderRepo(db).update_status( + order.id, status="已确认", confirm_time=now + ) + await db.commit() + account = await account_repo.get_by_customer_id(user.id) + holding = await holdings_repo.get_by_customer_product(user.id, product_id) + return PurchaseResp( + order_no=order.order_no, + status="已确认", + balance=account.balance, + holding=_holding_resp(holding), + ) + + # 命中：生成预警 + 订单挂起 + summary = summarize(hits) + alert = FinRiskAlert( + customer_id=user.id, + order_id=order.id, + alert_type=summary.alert_type, + alert_level=summary.alert_level, + trigger_detail=summary.trigger_detail, + transaction_ids=summary.transaction_ids or None, + confidence=summary.confidence, + status="未处理", + create_time=now, + ) + db.add(alert) + await db.flush() # 生成 alert.id，供订单回填 risk_alert_id + await TradeOrderRepo(db).update_status( + order.id, status="风控挂起", risk_alert_id=alert.id + ) await db.commit() + return PurchaseResp( + order_no=order.order_no, status="风控挂起", alert_id=alert.id + ) except Exception: await db.rollback() raise - - account = await account_repo.get_by_customer_id(user.id) - holding = await holdings_repo.get_by_customer_product(user.id, product_id) - return PurchaseResp( - balance=account.balance, - holding=HoldingResp( - id=holding.id, - customer_id=holding.customer_id, - product_id=holding.product_id, - shares=holding.shares, - cost_amount=holding.cost_amount, - current_value=holding.current_value, - profit_loss=holding.profit_loss, - profit_ratio=holding.profit_ratio, - status=holding.status, - ), - ) diff --git a/service/redeem.py b/service/redeem.py index 97af761..22f0c5e 100644 --- a/service/redeem.py +++ b/service/redeem.py @@ -1,31 +1,53 @@ -"""赎回服务：校验持仓 → 减仓 + 余额入账（单事务原子）。""" +"""赎回服务：校验持仓 → 建单(待确认) → 风控检测 → 落账/挂起（单事务原子）。""" from __future__ import annotations +from datetime import datetime from decimal import ROUND_HALF_UP, Decimal from sqlalchemy.ext.asyncio import AsyncSession +from model.fin_risk_alert import FinRiskAlert +from model.fin_transaction import FinTransaction from model.sys_user import SysUser +from model.trade_order import TradeOrder from repositories.fin_account import FinAccountRepo from repositories.fin_holdings import FinHoldingsRepo from repositories.fin_product import FinProductRepo +from repositories.trade_order import TradeOrderRepo from schemas.holdings import HoldingResp from schemas.redeem import RedeemResp +from service.risk.engine import RiskEngine, summarize +from service.risk.settle import settle from utils.exceptions import ForbiddenError, NotFoundError, ParamError +from utils.order_no import gen_order_no _MONEY = Decimal("0.01") _SHARES = Decimal("0.0001") +def _holding_resp(holding) -> HoldingResp: + return HoldingResp( + id=holding.id, + customer_id=holding.customer_id, + product_id=holding.product_id, + shares=holding.shares, + cost_amount=holding.cost_amount, + current_value=holding.current_value, + profit_loss=holding.profit_loss, + profit_ratio=holding.profit_ratio, + status=holding.status, + ) + + async def redeem( db: AsyncSession, user: SysUser, product_id: int, shares: Decimal ) -> RedeemResp: - """赎回基金：校验通过后减仓并按净值入账，全程单事务。 + """赎回基金：校验持仓 → 建单 → 风控检测 → 落账/挂起，全程单事务。 - 仅客户可赎回； - - 产品须存在且净值非空； - - 持仓须存在且份额 > 0，赎回份额不得超过持仓份额； - - 减仓（归 0 时置'已清仓'）+ 入账要么都成、要么都回滚。 + - 产品须存在且净值非空，持仓须足额； + - 未命中规则：减仓 + 入账 + 写流水，订单「已确认」； + - 命中规则：生成预警，订单「风控挂起」，账户与持仓不变。 """ if user.user_type != "CUSTOMER": raise ForbiddenError("仅客户账号可赎回") @@ -40,8 +62,8 @@ async def redeem( if product.nav is None: raise ParamError("产品暂无净值，无法赎回") - account_repo = FinAccountRepo(db) holdings_repo = FinHoldingsRepo(db) + account_repo = FinAccountRepo(db) holding = await holdings_repo.get_by_customer_product(user.id, product_id) if holding is None or holding.shares <= 0: @@ -53,28 +75,82 @@ async def redeem( if credited <= 0: raise ParamError("赎回金额过低") + now = datetime.now() + order = TradeOrder( + order_no=gen_order_no("PO"), + customer_id=user.id, + product_id=product_id, + advisor_id=None, + order_type="赎回", + amount=credited, + shares=shares, + nav=product.nav, + fee=Decimal("0"), + status="待确认", + create_time=now, + ) + db.add(order) + await db.flush() # 生成 order.id，供流水/预警关联 + + hits = await RiskEngine(db).detect(order, user) + try: - if not await holdings_repo.redeem(user.id, product_id, shares): - raise ParamError("可赎回份额不足") - await account_repo.credit_balance(user.id, credited) + if not hits: + # 未命中：减仓 + 入账 + 写流水 + 订单确认 + await settle(db, order) + db.add( + FinTransaction( + transaction_no=gen_order_no("TR"), + order_id=order.id, + customer_id=user.id, + product_id=product_id, + operator_id=None, + transaction_type="赎回", + amount=credited, + shares=shares, + nav=product.nav, + fee=Decimal("0"), + status="已确认", + create_time=now, + ) + ) + await TradeOrderRepo(db).update_status( + order.id, status="已确认", confirm_time=now + ) + await db.commit() + account = await account_repo.get_by_customer_id(user.id) + current_holding = await holdings_repo.get_by_customer_product( + user.id, product_id + ) + return RedeemResp( + order_no=order.order_no, + status="已确认", + balance=account.balance, + holding=_holding_resp(current_holding), + ) + + # 命中：生成预警 + 订单挂起 + summary = summarize(hits) + alert = FinRiskAlert( + customer_id=user.id, + order_id=order.id, + alert_type=summary.alert_type, + alert_level=summary.alert_level, + trigger_detail=summary.trigger_detail, + transaction_ids=summary.transaction_ids or None, + confidence=summary.confidence, + status="未处理", + create_time=now, + ) + db.add(alert) + await db.flush() # 生成 alert.id，供订单回填 risk_alert_id + await TradeOrderRepo(db).update_status( + order.id, status="风控挂起", risk_alert_id=alert.id + ) await db.commit() + return RedeemResp( + order_no=order.order_no, status="风控挂起", alert_id=alert.id + ) except Exception: await db.rollback() raise - - account = await account_repo.get_by_customer_id(user.id) - holding = await holdings_repo.get_by_customer_product(user.id, product_id) - return RedeemResp( - balance=account.balance, - holding=HoldingResp( - id=holding.id, - customer_id=holding.customer_id, - product_id=holding.product_id, - shares=holding.shares, - cost_amount=holding.cost_amount, - current_value=holding.current_value, - profit_loss=holding.profit_loss, - profit_ratio=holding.profit_ratio, - status=holding.status, - ), - ) -- 2.54.0 From 8b6893af18e77c7e7a409ee072328e19e02b3d97 Mon Sep 17 00:00:00 2001 From: zhangyongcai <17382809141.@163.com> Date: Sat, 12 Sep 2026 21:08:20 +0800 Subject: [PATCH 2/3] =?UTF-8?q?feat:=E9=A3=8E=E6=8E=A7Agent=EF=BC=88?= =?UTF-8?q?=E8=A7=84=E5=88=99=E5=BC=95=E6=93=8E+=E5=89=8D=E7=BD=AE?= =?UTF-8?q?=E6=8B=A6=E6=88=AA+=E4=BA=BA=E5=B7=A5=E5=A4=84=E7=BD=AE+?= =?UTF-8?q?=E5=B7=A5=E5=8D=95=E9=97=AD=E7=8E=AF=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- api/routers/risk.py | 51 +++++ api/routers/work_order.py | 62 +++++ model/audit_log.py | 26 +++ model/fin_risk_alert.py | 30 +++ model/fin_transaction.py | 29 +++ model/risk_rule.py | 29 +++ model/sys_message.py | 23 ++ model/trade_order.py | 31 +++ repositories/biz_work_order.py | 51 +++++ repositories/fin_risk_alert.py | 62 +++++ repositories/fin_transaction.py | 38 ++++ repositories/risk_rule.py | 20 ++ repositories/trade_order.py | 38 ++++ schemas/risk.py | 29 +++ schemas/work_order.py | 38 ++++ service/risk/__init__.py | 1 + service/risk/engine.py | 392 ++++++++++++++++++++++++++++++++ service/risk/handle.py | 258 +++++++++++++++++++++ service/risk/settle.py | 38 ++++ service/work_order.py | 142 ++++++++++++ utils/order_no.py | 15 ++ 21 files changed, 1403 insertions(+) create mode 100644 api/routers/risk.py create mode 100644 api/routers/work_order.py create mode 100644 model/audit_log.py create mode 100644 model/fin_risk_alert.py create mode 100644 model/fin_transaction.py create mode 100644 model/risk_rule.py create mode 100644 model/sys_message.py create mode 100644 model/trade_order.py create mode 100644 repositories/biz_work_order.py create mode 100644 repositories/fin_risk_alert.py create mode 100644 repositories/fin_transaction.py create mode 100644 repositories/risk_rule.py create mode 100644 repositories/trade_order.py create mode 100644 schemas/risk.py create mode 100644 schemas/work_order.py create mode 100644 service/risk/__init__.py create mode 100644 service/risk/engine.py create mode 100644 service/risk/handle.py create mode 100644 service/risk/settle.py create mode 100644 service/work_order.py create mode 100644 utils/order_no.py diff --git a/api/routers/risk.py b/api/routers/risk.py new file mode 100644 index 0000000..20ac821 --- /dev/null +++ b/api/routers/risk.py @@ -0,0 +1,51 @@ +"""风控处置路由：预警列表 + 放行/拦截/冻结（仅风控专员）。""" +from fastapi import APIRouter, Depends +from sqlalchemy.ext.asyncio import AsyncSession + +from api.deps import require_risk_officer +from config.deps import get_db +from model.sys_user import SysUser +from service.risk import handle as risk_handle +from utils.response import success + +router = APIRouter(prefix="/risk", tags=["风控"]) + + +@router.get("/alert/list", summary="预警列表") +async def list_alerts( + status: str | None = None, + user: SysUser = Depends(require_risk_officer), + db: AsyncSession = Depends(get_db), +): + alerts = await risk_handle.list_alerts(db, status) + return success([a.model_dump(mode="json") for a in alerts]) + + +@router.post("/alert/{alert_id}/release", summary="放行") +async def release( + alert_id: int, + user: SysUser = Depends(require_risk_officer), + db: AsyncSession = Depends(get_db), +): + result = await risk_handle.release(db, user, alert_id) + return success(result) + + +@router.post("/alert/{alert_id}/block", summary="拦截") +async def block( + alert_id: int, + user: SysUser = Depends(require_risk_officer), + db: AsyncSession = Depends(get_db), +): + result = await risk_handle.block(db, user, alert_id) + return success(result) + + +@router.post("/alert/{alert_id}/freeze", summary="冻结") +async def freeze( + alert_id: int, + user: SysUser = Depends(require_risk_officer), + db: AsyncSession = Depends(get_db), +): + result = await risk_handle.freeze(db, user, alert_id) + return success(result) diff --git a/api/routers/work_order.py b/api/routers/work_order.py new file mode 100644 index 0000000..4ad730d --- /dev/null +++ b/api/routers/work_order.py @@ -0,0 +1,62 @@ +"""业务工单路由：列表 / 详情 / 认领 / 提交审核 / 复核（仅风控专员）。""" +from fastapi import APIRouter, Depends +from sqlalchemy.ext.asyncio import AsyncSession + +from api.deps import require_risk_officer +from config.deps import get_db +from model.sys_user import SysUser +from schemas.work_order import WorkOrderIdReq, WorkOrderReviewReq +from service import work_order as work_order_service +from utils.response import success + +router = APIRouter(prefix="/work-order", tags=["工单"]) + + +@router.get("/list", summary="工单列表") +async def list_work_orders( + status: str | None = None, + user: SysUser = Depends(require_risk_officer), + db: AsyncSession = Depends(get_db), +): + orders = await work_order_service.list_work_orders(db, status) + return success([o.model_dump(mode="json") for o in orders]) + + +@router.get("/{work_order_id}", summary="工单详情") +async def get_work_order( + work_order_id: int, + user: SysUser = Depends(require_risk_officer), + db: AsyncSession = Depends(get_db), +): + wo = await work_order_service.get_work_order(db, work_order_id) + return success(wo.model_dump(mode="json")) + + +@router.post("/claim", summary="认领") +async def claim( + req: WorkOrderIdReq, + user: SysUser = Depends(require_risk_officer), + db: AsyncSession = Depends(get_db), +): + result = await work_order_service.claim(db, user, req.work_order_id) + return success(result) + + +@router.post("/submit-review", summary="提交审核") +async def submit_review( + req: WorkOrderIdReq, + user: SysUser = Depends(require_risk_officer), + db: AsyncSession = Depends(get_db), +): + result = await work_order_service.submit_review(db, user, req.work_order_id) + return success(result) + + +@router.post("/review", summary="复核") +async def review( + req: WorkOrderReviewReq, + user: SysUser = Depends(require_risk_officer), + db: AsyncSession = Depends(get_db), +): + result = await work_order_service.review(db, user, req.work_order_id, req.approve, req.comment) + return success(result) diff --git a/model/audit_log.py b/model/audit_log.py new file mode 100644 index 0000000..c51f80c --- /dev/null +++ b/model/audit_log.py @@ -0,0 +1,26 @@ +"""audit_log 操作审计日志 ORM 模型（谁在何时做了什么，企业级合规留痕）。""" +from __future__ import annotations + +from datetime import datetime + +from sqlalchemy import BigInteger, DateTime, String, Text, func +from sqlalchemy.orm import Mapped, mapped_column + +from model.base import Base + + +class AuditLog(Base): + __tablename__ = "audit_log" + __table_args__ = {"comment": "操作审计日志表（企业级合规留痕）"} + + id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True) + user_id: Mapped[int | None] = mapped_column(BigInteger) + username: Mapped[str | None] = mapped_column(String(64)) + module: Mapped[str] = mapped_column(String(32)) + action: Mapped[str] = mapped_column(String(64)) + target: Mapped[str | None] = mapped_column(String(128)) + detail: Mapped[str | None] = mapped_column(Text) + ip: Mapped[str | None] = mapped_column(String(64)) + trace_id: Mapped[str | None] = mapped_column(String(32)) + status: Mapped[str] = mapped_column(String(8), server_default="成功") + create_time: Mapped[datetime] = mapped_column(DateTime, server_default=func.now()) diff --git a/model/fin_risk_alert.py b/model/fin_risk_alert.py new file mode 100644 index 0000000..239362f --- /dev/null +++ b/model/fin_risk_alert.py @@ -0,0 +1,30 @@ +"""fin_risk_alert 风控预警 ORM 模型（命中记录 + 人工处置留痕）。""" +from __future__ import annotations + +from datetime import datetime +from decimal import Decimal +from typing import Any + +from sqlalchemy import BigInteger, DateTime, JSON, Numeric, String, Text, func +from sqlalchemy.orm import Mapped, mapped_column + +from model.base import Base + + +class FinRiskAlert(Base): + __tablename__ = "fin_risk_alert" + __table_args__ = {"comment": "风控预警表（阻断后由风控专员在控制台处置）"} + + id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True) + customer_id: Mapped[int] = mapped_column(BigInteger) + order_id: Mapped[int | None] = mapped_column(BigInteger) + alert_type: Mapped[str] = mapped_column(String(32)) + alert_level: Mapped[str] = mapped_column(String(8)) + trigger_detail: Mapped[str | None] = mapped_column(Text) + transaction_ids: Mapped[list | None] = mapped_column(JSON) + confidence: Mapped[Decimal] = mapped_column(Numeric(5, 2), server_default="0.50") + status: Mapped[str] = mapped_column(String(16), server_default="未处理") + handler_id: Mapped[int | None] = mapped_column(BigInteger) + handle_result: Mapped[str | None] = mapped_column(Text) + handle_time: Mapped[datetime | None] = mapped_column(DateTime) + create_time: Mapped[datetime] = mapped_column(DateTime, server_default=func.now()) diff --git a/model/fin_transaction.py b/model/fin_transaction.py new file mode 100644 index 0000000..24ed795 --- /dev/null +++ b/model/fin_transaction.py @@ -0,0 +1,29 @@ +"""fin_transaction 成交流水 ORM 模型（成交后落库的账务凭证）。""" +from __future__ import annotations + +from datetime import datetime +from decimal import Decimal + +from sqlalchemy import BigInteger, DateTime, Numeric, String, func +from sqlalchemy.orm import Mapped, mapped_column + +from model.base import Base + + +class FinTransaction(Base): + __tablename__ = "fin_transaction" + __table_args__ = {"comment": "成交流水表（成交后落库）"} + + id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True) + transaction_no: Mapped[str] = mapped_column(String(32), unique=True) + order_id: Mapped[int | None] = mapped_column(BigInteger) + customer_id: Mapped[int] = mapped_column(BigInteger) + product_id: Mapped[int] = mapped_column(BigInteger) + operator_id: Mapped[int | None] = mapped_column(BigInteger) + transaction_type: Mapped[str] = mapped_column(String(16)) + amount: Mapped[Decimal] = mapped_column(Numeric(18, 2)) + shares: Mapped[Decimal | None] = mapped_column(Numeric(18, 4)) + nav: Mapped[Decimal | None] = mapped_column(Numeric(12, 6)) + fee: Mapped[Decimal] = mapped_column(Numeric(12, 2), server_default="0") + status: Mapped[str] = mapped_column(String(16), server_default="已确认") + create_time: Mapped[datetime] = mapped_column(DateTime, server_default=func.now()) diff --git a/model/risk_rule.py b/model/risk_rule.py new file mode 100644 index 0000000..6cbd585 --- /dev/null +++ b/model/risk_rule.py @@ -0,0 +1,29 @@ +"""risk_rule 风控规则 ORM 模型（规则引擎表驱动配置：存参数不存逻辑）。""" +from __future__ import annotations + +from datetime import datetime +from decimal import Decimal +from typing import Any + +from sqlalchemy import BigInteger, DateTime, JSON, Numeric, String, func +from sqlalchemy.orm import Mapped, mapped_column + +from model.base import Base + + +class RiskRule(Base): + __tablename__ = "risk_rule" + __table_args__ = {"comment": "风控规则表（20 条反洗钱/异常交易规则，驱动风控Agent）"} + + id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True) + rule_id: Mapped[str] = mapped_column(String(16), unique=True) + rule_name: Mapped[str] = mapped_column(String(64)) + trigger_condition: Mapped[str | None] = mapped_column(String(512)) + threshold: Mapped[dict[str, Any] | None] = mapped_column(JSON) + risk_level: Mapped[str] = mapped_column(String(8), server_default="低") + weight: Mapped[Decimal] = mapped_column(Numeric(4, 2), server_default="1.00") + status: Mapped[str] = mapped_column(String(8), server_default="启用") + create_time: Mapped[datetime] = mapped_column(DateTime, server_default=func.now()) + update_time: Mapped[datetime] = mapped_column( + DateTime, server_default=func.now(), onupdate=func.now() + ) diff --git a/model/sys_message.py b/model/sys_message.py new file mode 100644 index 0000000..52d6cc0 --- /dev/null +++ b/model/sys_message.py @@ -0,0 +1,23 @@ +"""sys_message 站内信 ORM 模型（风控拦截/冻结后触达客户的通道）。""" +from __future__ import annotations + +from datetime import datetime + +from sqlalchemy import BigInteger, DateTime, Integer, String, func +from sqlalchemy.orm import Mapped, mapped_column + +from model.base import Base + + +class SysMessage(Base): + __tablename__ = "sys_message" + __table_args__ = {"comment": "站内信/消息中心表（投顾触达客户通道）"} + + id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True) + user_id: Mapped[int] = mapped_column(BigInteger) + msg_type: Mapped[str] = mapped_column(String(32)) + title: Mapped[str] = mapped_column(String(128)) + content: Mapped[str | None] = mapped_column(String(512)) + biz_id: Mapped[str | None] = mapped_column(String(64)) + is_read: Mapped[int] = mapped_column(Integer, server_default="0") + create_time: Mapped[datetime] = mapped_column(DateTime, server_default=func.now()) diff --git a/model/trade_order.py b/model/trade_order.py new file mode 100644 index 0000000..a29e392 --- /dev/null +++ b/model/trade_order.py @@ -0,0 +1,31 @@ +"""trade_order 交易申请单 ORM 模型（风控阻断闭环主表，状态机单一来源）。""" +from __future__ import annotations + +from datetime import datetime +from decimal import Decimal + +from sqlalchemy import BigInteger, DateTime, Numeric, String, func +from sqlalchemy.orm import Mapped, mapped_column + +from model.base import Base + + +class TradeOrder(Base): + __tablename__ = "trade_order" + __table_args__ = {"comment": "交易申请单（风控阻断闭环主表，状态机单一来源）"} + + id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True) + order_no: Mapped[str] = mapped_column(String(32), unique=True) + customer_id: Mapped[int] = mapped_column(BigInteger) + product_id: Mapped[int] = mapped_column(BigInteger) + advisor_id: Mapped[int | None] = mapped_column(BigInteger) + order_type: Mapped[str] = mapped_column(String(16)) + amount: Mapped[Decimal | None] = mapped_column(Numeric(18, 2)) + shares: Mapped[Decimal | None] = mapped_column(Numeric(18, 4)) + nav: Mapped[Decimal | None] = mapped_column(Numeric(12, 6)) + fee: Mapped[Decimal | None] = mapped_column(Numeric(12, 2)) + status: Mapped[str] = mapped_column(String(16), server_default="待确认") + risk_alert_id: Mapped[int | None] = mapped_column(BigInteger) + cancel_reason: Mapped[str | None] = mapped_column(String(255)) + create_time: Mapped[datetime] = mapped_column(DateTime, server_default=func.now()) + confirm_time: Mapped[datetime | None] = mapped_column(DateTime) diff --git a/repositories/biz_work_order.py b/repositories/biz_work_order.py new file mode 100644 index 0000000..aa645e6 --- /dev/null +++ b/repositories/biz_work_order.py @@ -0,0 +1,51 @@ +"""biz_work_order 工单仓储：查工单 + 流转条件更新（防并发，不 commit）。""" +from __future__ import annotations + +from sqlalchemy import select, update + +from model.biz_work_order import BizWorkOrder +from repositories.base import BaseRepository + + +class BizWorkOrderRepo(BaseRepository): + model = BizWorkOrder + + async def get_by_work_order_no(self, work_order_no: str) -> BizWorkOrder | None: + return await self.db.scalar( + select(BizWorkOrder).where(BizWorkOrder.work_order_no == work_order_no) + ) + + async def list_with_filter( + self, + *, + handler_id: int | None = None, + status: str | None = None, + customer_id: int | None = None, + ) -> list[BizWorkOrder]: + stmt = select(BizWorkOrder) + if handler_id is not None: + stmt = stmt.where(BizWorkOrder.handler_id == handler_id) + if status is not None: + stmt = stmt.where(BizWorkOrder.status == status) + if customer_id is not None: + stmt = stmt.where(BizWorkOrder.customer_id == customer_id) + stmt = stmt.order_by(BizWorkOrder.id.desc()) + return list((await self.db.scalars(stmt)).all()) + + async def conditional_transition( + self, work_order_id: int, *, from_status: str, to_status: str, **fields + ) -> bool: + """工单流转：条件更新 WHERE id=? AND status=from_status。 + + 不 commit，由 service 层事务统一提交；rowcount==0 表示状态已变（并发冲突）， + 返回 False 供上层提示，避免跳步流转。 + """ + result = await self.db.execute( + update(BizWorkOrder) + .where( + BizWorkOrder.id == work_order_id, + BizWorkOrder.status == from_status, + ) + .values(status=to_status, **fields) + ) + return result.rowcount > 0 diff --git a/repositories/fin_risk_alert.py b/repositories/fin_risk_alert.py new file mode 100644 index 0000000..78233c5 --- /dev/null +++ b/repositories/fin_risk_alert.py @@ -0,0 +1,62 @@ +"""fin_risk_alert 仓储：按状态查预警 + 处置条件更新（防并发，不 commit）。""" +from __future__ import annotations + +from datetime import datetime + +from sqlalchemy import select, update + +from model.fin_risk_alert import FinRiskAlert +from repositories.base import BaseRepository + + +class FinRiskAlertRepo(BaseRepository): + model = FinRiskAlert + + async def list_by_status( + self, status: str | None = None, customer_id: int | None = None + ) -> list[FinRiskAlert]: + stmt = select(FinRiskAlert) + if status is not None: + stmt = stmt.where(FinRiskAlert.status == status) + if customer_id is not None: + stmt = stmt.where(FinRiskAlert.customer_id == customer_id) + stmt = stmt.order_by(FinRiskAlert.id.desc()) + return list((await self.db.scalars(stmt)).all()) + + async def conditional_handle( + self, + alert_id: int, + *, + handler_id: int, + new_status: str, + handle_result: str, + handle_time: datetime, + ) -> bool: + """处置预警：条件更新 WHERE id=? AND status='未处理'。 + + 不 commit，由 service 层事务统一提交。rowcount==0 表示已被他人处置， + 返回 False 供上层提示「已被处理」，避免重复落账/工单/通知。 + """ + result = await self.db.execute( + update(FinRiskAlert) + .where( + FinRiskAlert.id == alert_id, + FinRiskAlert.status == "未处理", + ) + .values( + status=new_status, + handler_id=handler_id, + handle_result=handle_result, + handle_time=handle_time, + ) + ) + return result.rowcount > 0 + + async def update_status(self, alert_id: int, *, status: str, **fields) -> int: + """无条件更新预警状态（放行失败回写「已确认」等场景）。不 commit。""" + result = await self.db.execute( + update(FinRiskAlert) + .where(FinRiskAlert.id == alert_id) + .values(status=status, **fields) + ) + return result.rowcount diff --git a/repositories/fin_transaction.py b/repositories/fin_transaction.py new file mode 100644 index 0000000..488f0c8 --- /dev/null +++ b/repositories/fin_transaction.py @@ -0,0 +1,38 @@ +"""fin_transaction 成交流水仓储：按客户查历史流水（规则引擎聚合规则用）。""" +from __future__ import annotations + +from datetime import datetime + +from sqlalchemy import func, select + +from model.fin_transaction import FinTransaction +from repositories.base import BaseRepository + + +class FinTransactionRepo(BaseRepository): + model = FinTransaction + + async def list_since( + self, customer_id: int, since: datetime + ) -> list[FinTransaction]: + """查该客户 since 之后的所有成交流水，按成交时间升序。 + + 供规则引擎聚合规则（R005~R013 等）在内存中做窗口统计。 + """ + stmt = ( + select(FinTransaction) + .where( + FinTransaction.customer_id == customer_id, + FinTransaction.create_time >= since, + ) + .order_by(FinTransaction.create_time, FinTransaction.id) + ) + return list((await self.db.scalars(stmt)).all()) + + async def get_last_transaction_time(self, customer_id: int) -> datetime | None: + """该客户最近一笔成交时间（R016 休眠账户激活判定用）。""" + return await self.db.scalar( + select(func.max(FinTransaction.create_time)).where( + FinTransaction.customer_id == customer_id + ) + ) diff --git a/repositories/risk_rule.py b/repositories/risk_rule.py new file mode 100644 index 0000000..6178be1 --- /dev/null +++ b/repositories/risk_rule.py @@ -0,0 +1,20 @@ +"""risk_rule 仓储：读启用中的风控规则（规则引擎表驱动数据源）。""" +from __future__ import annotations + +from sqlalchemy import select + +from model.risk_rule import RiskRule +from repositories.base import BaseRepository + + +class RiskRuleRepo(BaseRepository): + model = RiskRule + + async def list_enabled(self) -> list[RiskRule]: + """加载启用中的规则，按 rule_id 升序，供规则引擎逐条路由判断。""" + stmt = ( + select(RiskRule) + .where(RiskRule.status == "启用") + .order_by(RiskRule.rule_id) + ) + return list((await self.db.scalars(stmt)).all()) diff --git a/repositories/trade_order.py b/repositories/trade_order.py new file mode 100644 index 0000000..ca0791f --- /dev/null +++ b/repositories/trade_order.py @@ -0,0 +1,38 @@ +"""trade_order 仓储：按单号/客户查申请单 + 状态流转（条件更新，不 commit）。""" +from __future__ import annotations + +from sqlalchemy import select, update + +from model.trade_order import TradeOrder +from repositories.base import BaseRepository + + +class TradeOrderRepo(BaseRepository): + model = TradeOrder + + async def get_by_order_no(self, order_no: str) -> TradeOrder | None: + return await self.db.scalar( + select(TradeOrder).where(TradeOrder.order_no == order_no) + ) + + async def list_by_customer( + self, customer_id: int, status: str | None = None + ) -> list[TradeOrder]: + """按客户查申请单，可按状态过滤，按创建时间倒序。""" + stmt = select(TradeOrder).where(TradeOrder.customer_id == customer_id) + if status is not None: + stmt = stmt.where(TradeOrder.status == status) + stmt = stmt.order_by(TradeOrder.id.desc()) + return list((await self.db.scalars(stmt)).all()) + + async def update_status(self, order_id: int, *, status: str, **fields) -> int: + """更新订单状态（可附带 risk_alert_id / confirm_time / cancel_reason 等）。 + + 不 commit，由 service 层事务统一提交；返回 rowcount（0 表示订单不存在）。 + """ + result = await self.db.execute( + update(TradeOrder) + .where(TradeOrder.id == order_id) + .values(status=status, **fields) + ) + return result.rowcount diff --git a/schemas/risk.py b/schemas/risk.py new file mode 100644 index 0000000..f1bd1fb --- /dev/null +++ b/schemas/risk.py @@ -0,0 +1,29 @@ +"""风控预警相关 DTO。""" +from __future__ import annotations + +from datetime import datetime +from decimal import Decimal + +from pydantic import BaseModel, field_serializer + + +class RiskAlertResp(BaseModel): + """单条风控预警（列表 / 详情共用）。""" + + id: int + customer_id: int + order_id: int | None + alert_type: str + alert_level: str + trigger_detail: str | None + transaction_ids: list | None + confidence: Decimal + status: str + handler_id: int | None + handle_result: str | None + handle_time: datetime | None + create_time: datetime + + @field_serializer("confidence") + def _fmt_confidence(self, value: Decimal) -> str: + return f"{value:.2f}" diff --git a/schemas/work_order.py b/schemas/work_order.py new file mode 100644 index 0000000..0ec651c --- /dev/null +++ b/schemas/work_order.py @@ -0,0 +1,38 @@ +"""业务工单相关 DTO。""" +from __future__ import annotations + +from datetime import datetime + +from pydantic import BaseModel + + +class WorkOrderResp(BaseModel): + """工单详情（列表 / 详情共用）。""" + + id: int + work_order_no: str + order_type: str + sub_type: str | None + customer_id: int | None + submitter_id: int | None + handler_id: int | None + current_node: str + priority: str + status: str + biz_content: dict | None + create_time: datetime + update_time: datetime + + +class WorkOrderIdReq(BaseModel): + """认领 / 提交审核 入参。""" + + work_order_id: int + + +class WorkOrderReviewReq(BaseModel): + """复核入参：approve=True 复核通过，False 复核驳回。""" + + work_order_id: int + approve: bool + comment: str | None = None diff --git a/service/risk/__init__.py b/service/risk/__init__.py new file mode 100644 index 0000000..33cebce --- /dev/null +++ b/service/risk/__init__.py @@ -0,0 +1 @@ +"""风控服务包：规则引擎（engine）与人工处置（handle）。""" diff --git a/service/risk/engine.py b/service/risk/engine.py new file mode 100644 index 0000000..57b83d3 --- /dev/null +++ b/service/risk/engine.py @@ -0,0 +1,392 @@ +"""风控规则引擎（纯代码，不调用 LLM）。 + +读 risk_rule 表（表驱动参数）→ 按 rule_id 路由到判断函数 → 传入当前订单 + +客户画像/账户/注册时间 + 历史成交流水 → 输出命中规则列表。 + +设计约定： +- 聚合规则统计「历史成交流水」（当前订单尚未成交，不计入历史），阈值即历史累计； + 当前这笔的金额/类型由单笔规则（R001~R004、R015~R018）单独判断。 +- 阈值来自 risk_rule.threshold（JSON），代码只存判断逻辑，参数可运营调整。 +""" +from __future__ import annotations + +from collections import defaultdict +from dataclasses import dataclass +from datetime import datetime, timedelta +from decimal import Decimal +from typing import Any + +from sqlalchemy.ext.asyncio import AsyncSession + +from model.fin_customer_profile import FinCustomerProfile +from model.fin_holdings import FinHoldings +from model.fin_product import FinProduct +from model.fin_transaction import FinTransaction +from model.risk_rule import RiskRule +from model.sys_user import SysUser +from model.trade_order import TradeOrder +from repositories.fin_customer_profile import FinCustomerProfileRepo +from repositories.fin_holdings import FinHoldingsRepo +from repositories.fin_product import FinProductRepo +from repositories.fin_transaction import FinTransactionRepo +from repositories.risk_rule import RiskRuleRepo + +# 风险等级 → 序号（兼容 R1~R5 与 保守~激进 两套口径）。 +_RISK_RANK = { + "R1": 1, "R2": 2, "R3": 3, "R4": 4, "R5": 5, + "保守": 1, "稳健": 2, "平衡": 3, "进取": 4, "激进": 5, +} +_LEVEL_RANK = {"低": 1, "中": 2, "高": 3} + + +def _risk_rank(level: str | None) -> int | None: + if not level: + return None + return _RISK_RANK.get(level.strip()) + + +def _int(threshold: dict[str, Any] | None, key: str, default: int) -> int: + if not threshold or key not in threshold: + return default + return int(threshold[key]) + + +def _dec(threshold: dict[str, Any] | None, key: str, default: Decimal) -> Decimal: + if not threshold or key not in threshold: + return default + return Decimal(str(threshold[key])) + + +@dataclass +class RuleHit: + """单条规则命中结果。""" + + rule: RiskRule + detail: str + transaction_ids: list[int] + + +@dataclass +class AlertSummary: + """命中规则列表聚合出的预警摘要。""" + + alert_type: str + alert_level: str + trigger_detail: str + transaction_ids: list[int] + confidence: Decimal + + +@dataclass +class _Ctx: + """一次检测预查的上下文（避免每条规则重复查库）。""" + + product: FinProduct | None + profile: FinCustomerProfile | None + holdings: FinHoldings | None + history: list[FinTransaction] + last_tx_time: datetime | None + now: datetime + + +def summarize(hits: list[RuleHit]) -> AlertSummary: + """聚合命中列表：级别取最高档，置信度按权重合成，详情拼接全部命中规则。 + + confidence = 1 − Π(1 − weight_i)。 + """ + if not hits: + raise ValueError("无命中规则，无法聚合预警") + top = max(hits, key=lambda h: _LEVEL_RANK[h.rule.risk_level]) + conf = Decimal("1") + for h in hits: + conf *= Decimal("1") - h.rule.weight + confidence = (Decimal("1") - conf).quantize(Decimal("0.01")) + detail = "；".join( + f"{h.rule.rule_id} {h.rule.rule_name}：{h.detail}" for h in hits + ) + tx_ids: list[int] = [] + seen: set[int] = set() + for h in hits: + for tid in h.transaction_ids: + if tid not in seen: + seen.add(tid) + tx_ids.append(tid) + return AlertSummary( + alert_type=top.rule.rule_name, + alert_level=top.rule.risk_level, + trigger_detail=detail, + transaction_ids=tx_ids, + confidence=confidence, + ) + + +class RiskEngine: + """规则引擎：detect(order, customer) -> 命中规则列表。""" + + def __init__(self, db: AsyncSession): + self.db = db + + async def detect(self, order: TradeOrder, customer: SysUser) -> list[RuleHit]: + rules = await RiskRuleRepo(self.db).list_enabled() + now = datetime.now() + history = await FinTransactionRepo(self.db).list_since( + customer.id, now - timedelta(days=90) + ) + ctx = _Ctx( + product=await FinProductRepo(self.db).get(order.product_id), + profile=await FinCustomerProfileRepo(self.db).get_by_customer_id(customer.id), + holdings=await FinHoldingsRepo(self.db).get_by_customer_product( + customer.id, order.product_id + ), + history=history, + last_tx_time=await FinTransactionRepo(self.db).get_last_transaction_time( + customer.id + ), + now=now, + ) + hits: list[RuleHit] = [] + for rule in rules: + hit = await self._dispatch(rule, order, customer, ctx) + if hit is not None: + hits.append(hit) + return hits + + async def _dispatch( + self, rule: RiskRule, order: TradeOrder, customer: SysUser, ctx: _Ctx + ) -> RuleHit | None: + fn = getattr(self, f"_check_{rule.rule_id.lower()}", None) + if fn is None: + return None + return await fn(rule, order, customer, ctx) + + # ------------------------------------------------------------------ A 类单笔 + + async def _check_r001(self, rule, order, customer, ctx) -> RuleHit | None: + if order.order_type != "申购" or order.amount is None: + return None + threshold = _dec(rule.threshold, "amount", Decimal("1000000")) + if order.amount >= threshold: + return RuleHit(rule, f"单笔申购金额 {order.amount} ≥ {threshold}", []) + return None + + async def _check_r002(self, rule, order, customer, ctx) -> RuleHit | None: + if order.order_type != "赎回" or order.amount is None: + return None + threshold = _dec(rule.threshold, "amount", Decimal("1000000")) + if order.amount >= threshold: + return RuleHit(rule, f"单笔赎回金额 {order.amount} ≥ {threshold}", []) + return None + + async def _check_r003(self, rule, order, customer, ctx) -> RuleHit | None: + if order.order_type != "申购" or order.amount is None: + return None + threshold = _dec(rule.threshold, "amount", Decimal("500000")) + if order.amount >= threshold: + return RuleHit(rule, f"单笔申购金额 {order.amount} ≥ {threshold}", []) + return None + + async def _check_r004(self, rule, order, customer, ctx) -> RuleHit | None: + if order.order_type != "赎回" or order.amount is None: + return None + threshold = _dec(rule.threshold, "amount", Decimal("500000")) + if order.amount >= threshold: + return RuleHit(rule, f"单笔赎回金额 {order.amount} ≥ {threshold}", []) + return None + + async def _check_r014(self, rule, order, customer, ctx) -> RuleHit | None: + if order.order_type != "申购": + return None + customer_rank = _risk_rank(ctx.profile.risk_level if ctx.profile else None) + product_rank = _risk_rank(ctx.product.risk_level if ctx.product else None) + if customer_rank is not None and product_rank is not None and customer_rank < product_rank: + return RuleHit( + rule, + f"客户风险等级（{ctx.profile.risk_level}）低于产品（{ctx.product.risk_level}）", + [], + ) + return None + + async def _check_r020(self, rule, order, customer, ctx) -> RuleHit | None: + start = (rule.threshold or {}).get("start", "00:00") + end = (rule.threshold or {}).get("end", "06:00") + start_hour = int(start.split(":")[0]) + end_hour = int(end.split(":")[0]) + if start_hour <= ctx.now.hour < end_hour: + return RuleHit(rule, f"下单时间 {ctx.now:%H:%M} 落在 {start}–{end}", []) + return None + + # ------------------------------------------------------------------ B 类聚合 + + async def _check_r005(self, rule, order, customer, ctx) -> RuleHit | None: + days = _int(rule.threshold, "days", 7) + count = _int(rule.threshold, "count", 10) + recent = [t for t in ctx.history if t.create_time >= ctx.now - timedelta(days=days)] + if len(recent) >= count: + return RuleHit(rule, f"近{days}天申赎 {len(recent)} 笔 ≥ {count}", [t.id for t in recent]) + return None + + async def _check_r006(self, rule, order, customer, ctx) -> RuleHit | None: + days = _int(rule.threshold, "days", 1) + count = _int(rule.threshold, "count", 5) + recent = [t for t in ctx.history if t.create_time >= ctx.now - timedelta(days=days)] + if len(recent) >= count: + return RuleHit(rule, f"近{days}天申赎 {len(recent)} 笔 ≥ {count}", [t.id for t in recent]) + return None + + async def _check_r007(self, rule, order, customer, ctx) -> RuleHit | None: + if order.order_type != "申购": + return None + days = _int(rule.threshold, "days", 3) + count = _int(rule.threshold, "count", 5) + max_amount = _dec(rule.threshold, "max_amount", Decimal("10000")) + small = [ + t for t in ctx.history + if t.create_time >= ctx.now - timedelta(days=days) + and t.transaction_type == "申购" + and t.amount < max_amount + ] + if len(small) >= count: + return RuleHit(rule, f"近{days}天小额申购 {len(small)} 笔 ≥ {count}", [t.id for t in small]) + return None + + async def _check_r008(self, rule, order, customer, ctx) -> RuleHit | None: + if order.order_type != "赎回": + return None + days = _int(rule.threshold, "days", 7) + bought = [ + t for t in ctx.history + if t.product_id == order.product_id + and t.transaction_type == "申购" + and t.create_time >= ctx.now - timedelta(days=days) + ] + if bought: + first = min(t.create_time for t in bought) + return RuleHit(rule, f"同产品 {first:%Y-%m-%d} 申购后 {days} 天内赎回", [t.id for t in bought]) + return None + + async def _check_r009(self, rule, order, customer, ctx) -> RuleHit | None: + if order.order_type != "赎回": + return None + days = _int(rule.threshold, "days", 3) + bought = [ + t for t in ctx.history + if t.product_id == order.product_id + and t.transaction_type == "申购" + and t.create_time >= ctx.now - timedelta(days=days) + ] + if bought and order.shares is not None: + bought_shares = sum( + ((t.shares or Decimal("0")) for t in bought), Decimal("0") + ) + if order.shares >= bought_shares: + return RuleHit(rule, f"同产品近{days}天申购 {bought_shares} 份后全额赎回", [t.id for t in bought]) + return None + + async def _check_r010(self, rule, order, customer, ctx) -> RuleHit | None: + if order.order_type != "赎回": + return None + hours = _int(rule.threshold, "hours", 24) + bought = [ + t for t in ctx.history + if t.transaction_type == "申购" + and t.create_time >= ctx.now - timedelta(hours=hours) + ] + if bought: + return RuleHit(rule, f"入账后 {hours} 小时内赎回", [t.id for t in bought]) + return None + + async def _check_r011(self, rule, order, customer, ctx) -> RuleHit | None: + days = _int(rule.threshold, "days", 7) + amount = _dec(rule.threshold, "amount", Decimal("2000000")) + recent = [t for t in ctx.history if t.create_time >= ctx.now - timedelta(days=days)] + total = sum((t.amount for t in recent), Decimal("0")) + if total >= amount: + return RuleHit(rule, f"近{days}天累计申赎 {total} ≥ {amount}", [t.id for t in recent]) + return None + + async def _check_r012(self, rule, order, customer, ctx) -> RuleHit | None: + days = _int(rule.threshold, "days", 30) + amount = _dec(rule.threshold, "amount", Decimal("5000000")) + recent = [t for t in ctx.history if t.create_time >= ctx.now - timedelta(days=days)] + total = sum((t.amount for t in recent), Decimal("0")) + if total >= amount: + return RuleHit(rule, f"近{days}天累计申赎 {total} ≥ {amount}", [t.id for t in recent]) + return None + + async def _check_r013(self, rule, order, customer, ctx) -> RuleHit | None: + if order.amount is None: + return None + days = _int(rule.threshold, "days", 1) + total_th = _dec(rule.threshold, "total", Decimal("1000000")) + each_lt = _dec(rule.threshold, "each_lt", Decimal("500000")) + today = [t for t in ctx.history if t.create_time >= ctx.now - timedelta(days=days)] + amounts = [t.amount for t in today] + [order.amount] + if ( + len(amounts) >= 2 + and all(a < each_lt for a in amounts) + and sum(amounts, Decimal("0")) >= total_th + ): + return RuleHit( + rule, + f"当日 {len(amounts)} 笔合计 {sum(amounts, Decimal('0'))} ≥ {total_th}，单笔均 < {each_lt}", + [t.id for t in today], + ) + return None + + async def _check_r018(self, rule, order, customer, ctx) -> RuleHit | None: + if order.order_type != "赎回" or order.amount is None: + return None + amount = _dec(rule.threshold, "amount", Decimal("100000")) + holding = ctx.holdings + if ( + holding is not None + and holding.shares > 0 + and order.shares is not None + and order.shares >= holding.shares + and order.amount >= amount + ): + return RuleHit(rule, f"赎回 {order.shares} 份清仓且金额 {order.amount} ≥ {amount}", []) + return None + + async def _check_r019(self, rule, order, customer, ctx) -> RuleHit | None: + days = _int(rule.threshold, "days", 7) + by_product: dict[int, dict[str, bool]] = defaultdict( + lambda: {"申购": False, "赎回": False} + ) + for t in ctx.history: + if t.create_time >= ctx.now - timedelta(days=days): + by_product[t.product_id][t.transaction_type] = True + mutual = [pid for pid, types in by_product.items() if types["申购"] and types["赎回"]] + if len(mutual) >= 2: + return RuleHit(rule, f"近{days}天 {len(mutual)} 个产品互买互卖", []) + return None + + # ------------------------------------------------------------------ C 类客户 + + async def _check_r015(self, rule, order, customer, ctx) -> RuleHit | None: + if order.amount is None: + return None + days = _int(rule.threshold, "days", 30) + amount = _dec(rule.threshold, "amount", Decimal("500000")) + registered_days = (ctx.now - customer.create_time).days + if registered_days < days and order.amount >= amount: + return RuleHit(rule, f"注册 {registered_days} 天（< {days}）且金额 {order.amount} ≥ {amount}", []) + return None + + async def _check_r016(self, rule, order, customer, ctx) -> RuleHit | None: + if order.amount is None: + return None + days = _int(rule.threshold, "days", 90) + amount = _dec(rule.threshold, "amount", Decimal("100000")) + dormant = ctx.last_tx_time is None or ctx.last_tx_time < ctx.now - timedelta(days=days) + if dormant and order.amount >= amount: + return RuleHit(rule, f"{days} 天无交易后首笔 {order.amount} ≥ {amount}", []) + return None + + async def _check_r017(self, rule, order, customer, ctx) -> RuleHit | None: + if order.amount is None: + return None + amount = _dec(rule.threshold, "amount", Decimal("100000")) + if order.amount >= amount and order.amount % Decimal("10000") == 0: + return RuleHit(rule, f"金额 {order.amount} 为整万且 ≥ {amount}", []) + return None diff --git a/service/risk/handle.py b/service/risk/handle.py new file mode 100644 index 0000000..340948b --- /dev/null +++ b/service/risk/handle.py @@ -0,0 +1,258 @@ +"""风控人工处置：放行 / 拦截 / 冻结（挂起自动，处置人工）。 + +处置接口的 handler_id 一律取自当前登录用户（后端注入），不信任前端； +用条件更新 WHERE status='未处理' 防并发重复处置； +处置与工单流转均写 audit_log 留痕。 +""" +from __future__ import annotations + +from datetime import datetime +from decimal import Decimal + +from sqlalchemy.ext.asyncio import AsyncSession + +from model.audit_log import AuditLog +from model.biz_work_order import BizWorkOrder +from model.fin_risk_alert import FinRiskAlert +from model.fin_transaction import FinTransaction +from model.sys_message import SysMessage +from model.sys_user import SysUser +from model.trade_order import TradeOrder +from repositories.fin_risk_alert import FinRiskAlertRepo +from repositories.sys_user import SysUserRepo +from repositories.trade_order import TradeOrderRepo +from schemas.risk import RiskAlertResp +from service.risk.settle import settle +from utils.exceptions import NotFoundError, ParamError +from utils.order_no import gen_order_no + +# 预警级别 → 工单优先级 +_LEVEL_PRIORITY = {"低": "普通", "中": "紧急", "高": "特急"} + + +def _audit(db: AsyncSession, user: SysUser, module: str, action: str, target, detail: str) -> None: + db.add( + AuditLog( + user_id=user.id, + username=user.real_name or user.username, + module=module, + action=action, + target=str(target), + detail=detail, + status="成功", + ) + ) + + +def _build_work_order( + handler: SysUser, alert: FinRiskAlert, order: TradeOrder, order_type: str +) -> BizWorkOrder: + """拦截/冻结时生成工单：预填类型、客户、提交人、优先级、业务内容。""" + return BizWorkOrder( + work_order_no=gen_order_no("WO"), + order_type=order_type, + sub_type=alert.alert_type, + customer_id=order.customer_id, + submitter_id=handler.id, + handler_id=None, + current_node="初审", + priority=_LEVEL_PRIORITY.get(alert.alert_level, "普通"), + status="待处理", + biz_content={ + "order_no": order.order_no, + "alert_id": alert.id, + "alert_type": alert.alert_type, + "alert_level": alert.alert_level, + "trigger_detail": alert.trigger_detail, + "confidence": str(alert.confidence), + }, + ) + + +def _alert_resp(alert: FinRiskAlert) -> RiskAlertResp: + return RiskAlertResp( + id=alert.id, + customer_id=alert.customer_id, + order_id=alert.order_id, + alert_type=alert.alert_type, + alert_level=alert.alert_level, + trigger_detail=alert.trigger_detail, + transaction_ids=alert.transaction_ids, + confidence=alert.confidence, + status=alert.status, + handler_id=alert.handler_id, + handle_result=alert.handle_result, + handle_time=alert.handle_time, + create_time=alert.create_time, + ) + + +async def _get_pending_alert(db: AsyncSession, alert_id: int) -> tuple[FinRiskAlert, TradeOrder]: + alert = await FinRiskAlertRepo(db).get(alert_id) + if alert is None: + raise NotFoundError("预警不存在") + order = await TradeOrderRepo(db).get(alert.order_id) + if order is None: + raise NotFoundError("关联订单不存在") + if order.status != "风控挂起": + raise ParamError("订单状态异常，无法处置") + return alert, order + + +async def release(db: AsyncSession, handler: SysUser, alert_id: int) -> dict: + """放行：订单已确认 + 落账 + 写流水；预警已排除。放行失败（余额不足）订单转失败。""" + alert_repo = FinRiskAlertRepo(db) + order_repo = TradeOrderRepo(db) + alert, order = await _get_pending_alert(db, alert_id) + now = datetime.now() + + try: + if not await alert_repo.conditional_handle( + alert_id, + handler_id=handler.id, + new_status="已排除", + handle_result="放行", + handle_time=now, + ): + raise ParamError("该预警已被处理") + + try: + await settle(db, order) + except ParamError as e: + # 放行失败：订单转失败，预警回写已确认 + await order_repo.update_status( + order.id, status="失败", cancel_reason=f"放行失败：{e.message}" + ) + await alert_repo.update_status( + alert_id, status="已确认", handle_result=f"放行失败：{e.message}" + ) + _audit(db, handler, "risk", "release_failed", alert_id, f"放行失败：{e.message}") + await db.commit() + return { + "alert_id": alert_id, + "status": "已确认", + "order_status": "失败", + "message": "放行失败，订单已作废", + } + + db.add( + FinTransaction( + transaction_no=gen_order_no("TR"), + order_id=order.id, + customer_id=order.customer_id, + product_id=order.product_id, + operator_id=handler.id, + transaction_type=order.order_type, + amount=order.amount, + shares=order.shares, + nav=order.nav, + fee=order.fee or Decimal("0"), + status="已确认", + create_time=now, + ) + ) + await order_repo.update_status(order.id, status="已确认", confirm_time=now) + _audit(db, handler, "risk", "release", alert_id, f"放行订单 {order.order_no}") + await db.commit() + return { + "alert_id": alert_id, + "status": "已排除", + "order_status": "已确认", + "order_no": order.order_no, + } + except Exception: + await db.rollback() + raise + + +async def block(db: AsyncSession, handler: SysUser, alert_id: int) -> dict: + """拦截：订单失败 + 站内信通知；预警已确认 + 生成「可疑交易上报」工单。""" + alert_repo = FinRiskAlertRepo(db) + order_repo = TradeOrderRepo(db) + alert, order = await _get_pending_alert(db, alert_id) + now = datetime.now() + + try: + if not await alert_repo.conditional_handle( + alert_id, + handler_id=handler.id, + new_status="已确认", + handle_result="拦截", + handle_time=now, + ): + raise ParamError("该预警已被处理") + + await order_repo.update_status(order.id, status="失败", cancel_reason="风控拦截") + db.add( + SysMessage( + user_id=order.customer_id, + msg_type="风控", + title="交易被拦截", + content=f"您的{order.order_type}订单 {order.order_no} 因风险控制被拦截，如有疑问请联系客服。", + biz_id=order.order_no, + is_read=0, + create_time=now, + ) + ) + db.add(_build_work_order(handler, alert, order, "可疑交易上报")) + _audit(db, handler, "risk", "block", alert_id, f"拦截订单 {order.order_no}") + await db.commit() + return { + "alert_id": alert_id, + "status": "已确认", + "order_status": "失败", + "order_no": order.order_no, + } + except Exception: + await db.rollback() + raise + + +async def freeze(db: AsyncSession, handler: SysUser, alert_id: int) -> dict: + """冻结：订单失败 + 客户冻结 + 站内信通知；预警已确认 + 生成「冻结复核」工单。""" + alert_repo = FinRiskAlertRepo(db) + order_repo = TradeOrderRepo(db) + alert, order = await _get_pending_alert(db, alert_id) + now = datetime.now() + + try: + if not await alert_repo.conditional_handle( + alert_id, + handler_id=handler.id, + new_status="已确认", + handle_result="冻结", + handle_time=now, + ): + raise ParamError("该预警已被处理") + + await order_repo.update_status(order.id, status="失败", cancel_reason="风控冻结") + await SysUserRepo(db).update_status(order.customer_id, "冻结") + db.add( + SysMessage( + user_id=order.customer_id, + msg_type="风控", + title="账户已被冻结", + content="您的账户因风险控制已被冻结，请联系客服处理。", + biz_id=order.order_no, + is_read=0, + create_time=now, + ) + ) + db.add(_build_work_order(handler, alert, order, "冻结复核")) + _audit(db, handler, "risk", "freeze", alert_id, f"冻结客户 {order.customer_id}，订单 {order.order_no}") + await db.commit() + return { + "alert_id": alert_id, + "status": "已确认", + "order_status": "失败", + "customer_status": "冻结", + "order_no": order.order_no, + } + except Exception: + await db.rollback() + raise + + +async def list_alerts(db: AsyncSession, status: str | None = None) -> list[RiskAlertResp]: + alerts = await FinRiskAlertRepo(db).list_by_status(status) + return [_alert_resp(a) for a in alerts] diff --git a/service/risk/settle.py b/service/risk/settle.py new file mode 100644 index 0000000..38192a4 --- /dev/null +++ b/service/risk/settle.py @@ -0,0 +1,38 @@ +"""落账逻辑：按订单类型复用现有扣款+加仓 / 减仓+入账仓储方法。 + +供两处调用：未命中分支（申购/赎回接口）与放行处置（release）。不 commit， +由外层事务统一提交，保证「落账 + 写流水 / 改订单状态」原子性。 +""" +from __future__ import annotations + +from sqlalchemy.ext.asyncio import AsyncSession + +from model.trade_order import TradeOrder +from repositories.fin_account import FinAccountRepo +from repositories.fin_holdings import FinHoldingsRepo +from utils.exceptions import ParamError + + +async def settle(db: AsyncSession, order: TradeOrder) -> None: + """执行落账（申购扣款+加仓 / 赎回减仓+入账）。不 commit。 + + - 申购：余额不足返回 False 时抛 ParamError； + - 赎回：份额不足返回 False 时抛 ParamError。 + """ + account_repo = FinAccountRepo(db) + holdings_repo = FinHoldingsRepo(db) + + if order.order_type == "申购": + if not await account_repo.deduct_balance(order.customer_id, order.amount): + raise ParamError("可用余额不足") + await holdings_repo.upsert( + order.customer_id, order.product_id, order.shares, order.amount + ) + elif order.order_type == "赎回": + if not await holdings_repo.redeem( + order.customer_id, order.product_id, order.shares + ): + raise ParamError("可赎回份额不足") + await account_repo.credit_balance(order.customer_id, order.amount) + else: + raise ParamError(f"未知订单类型：{order.order_type}") diff --git a/service/work_order.py b/service/work_order.py new file mode 100644 index 0000000..6bf18a8 --- /dev/null +++ b/service/work_order.py @@ -0,0 +1,142 @@ +"""业务工单链：认领 → 提交审核 → 复核（软校验职责分离，流转写审计）。 + +状态机：待处理 ─认领─► 处理中 ─提交审核─► 待审核 ─复核─► 已完成 / 已驳回（终态）。 +流转均用条件更新防并发跳步，每次流转写 audit_log。 +""" +from __future__ import annotations + +from datetime import datetime + +from sqlalchemy.ext.asyncio import AsyncSession + +from model.audit_log import AuditLog +from model.biz_work_order import BizWorkOrder +from model.sys_user import SysUser +from repositories.biz_work_order import BizWorkOrderRepo +from schemas.work_order import WorkOrderResp +from utils.exceptions import NotFoundError, ParamError + + +def _audit(db: AsyncSession, user: SysUser, module: str, action: str, target, detail: str) -> None: + db.add( + AuditLog( + user_id=user.id, + username=user.real_name or user.username, + module=module, + action=action, + target=str(target), + detail=detail, + status="成功", + ) + ) + + +def _resp(wo: BizWorkOrder) -> WorkOrderResp: + return WorkOrderResp( + id=wo.id, + work_order_no=wo.work_order_no, + order_type=wo.order_type, + sub_type=wo.sub_type, + customer_id=wo.customer_id, + submitter_id=wo.submitter_id, + handler_id=wo.handler_id, + current_node=wo.current_node, + priority=wo.priority, + status=wo.status, + biz_content=wo.biz_content, + create_time=wo.create_time, + update_time=wo.update_time, + ) + + +async def list_work_orders(db: AsyncSession, status: str | None = None) -> list[WorkOrderResp]: + orders = await BizWorkOrderRepo(db).list_with_filter(status=status) + return [_resp(o) for o in orders] + + +async def get_work_order(db: AsyncSession, work_order_id: int) -> WorkOrderResp: + wo = await BizWorkOrderRepo(db).get(work_order_id) + if wo is None: + raise NotFoundError("工单不存在") + return _resp(wo) + + +async def claim(db: AsyncSession, user: SysUser, work_order_id: int) -> dict: + """认领：待处理 → 处理中，handler_id 记为当前专员。""" + wo_repo = BizWorkOrderRepo(db) + wo = await wo_repo.get(work_order_id) + if wo is None: + raise NotFoundError("工单不存在") + try: + if not await wo_repo.conditional_transition( + work_order_id, + from_status="待处理", + to_status="处理中", + handler_id=user.id, + update_time=datetime.now(), + ): + raise ParamError("工单已被认领或状态已变更") + _audit(db, user, "work_order", "claim", work_order_id, f"认领工单 {wo.work_order_no}：待处理 → 处理中") + await db.commit() + return {"work_order_id": work_order_id, "status": "处理中", "work_order_no": wo.work_order_no} + except Exception: + await db.rollback() + raise + + +async def submit_review(db: AsyncSession, user: SysUser, work_order_id: int) -> dict: + """提交审核：处理中 → 待审核。""" + wo_repo = BizWorkOrderRepo(db) + wo = await wo_repo.get(work_order_id) + if wo is None: + raise NotFoundError("工单不存在") + try: + if not await wo_repo.conditional_transition( + work_order_id, + from_status="处理中", + to_status="待审核", + current_node="复核", + update_time=datetime.now(), + ): + raise ParamError("工单状态已变更，无法提交审核") + _audit(db, user, "work_order", "submit_review", work_order_id, f"提交审核工单 {wo.work_order_no}：处理中 → 待审核") + await db.commit() + return {"work_order_id": work_order_id, "status": "待审核", "work_order_no": wo.work_order_no} + except Exception: + await db.rollback() + raise + + +async def review( + db: AsyncSession, user: SysUser, work_order_id: int, approve: bool, comment: str | None = None +) -> dict: + """复核：待审核 → 已完成 / 已驳回（终态）。复核人与处理人为同一人时软校验仅提示。""" + wo_repo = BizWorkOrderRepo(db) + wo = await wo_repo.get(work_order_id) + if wo is None: + raise NotFoundError("工单不存在") + to_status = "已完成" if approve else "已驳回" + warning = "复核人与处理人为同一人" if wo.handler_id == user.id else None + try: + if not await wo_repo.conditional_transition( + work_order_id, + from_status="待审核", + to_status=to_status, + current_node="完成", + update_time=datetime.now(), + ): + raise ParamError("工单状态已变更，无法复核") + detail = f"复核工单 {wo.work_order_no}：待审核 → {to_status}" + if comment: + detail += f"（{comment}）" + _audit(db, user, "work_order", "review", work_order_id, detail) + await db.commit() + return { + "work_order_id": work_order_id, + "status": to_status, + "work_order_no": wo.work_order_no, + "warning": warning, + } + except Exception: + await db.rollback() + raise diff --git a/utils/order_no.py b/utils/order_no.py new file mode 100644 index 0000000..721dc2b --- /dev/null +++ b/utils/order_no.py @@ -0,0 +1,15 @@ +"""单号生成：前缀 + 日期 + 随机序列（唯一性由 DB 唯一键兜底）。""" +from __future__ import annotations + +import secrets +from datetime import datetime + + +def gen_order_no(prefix: str = "PO") -> str: + """生成全局唯一单号，如 PO20260912A1B2C3D4。 + + - PO：交易申请单（trade_order.order_no） + - TR：成交流水（fin_transaction.transaction_no） + - WO：业务工单（biz_work_order.work_order_no） + """ + return f"{prefix}{datetime.now():%Y%m%d}{secrets.token_hex(4).upper()}" -- 2.54.0 From 69db5c46bd60d1ed894e92e969b39fb0eecaa9f0 Mon Sep 17 00:00:00 2001 From: zhangyongcai <17382809141.@163.com> Date: Sat, 12 Sep 2026 21:15:42 +0800 Subject: [PATCH 3/3] =?UTF-8?q?docs:=E8=A1=A5=E5=85=85=E9=A3=8E=E6=8E=A7Ag?= =?UTF-8?q?ent=E9=9C=80=E6=B1=82=E6=96=87=E6=A1=A3=E4=B8=8E=E8=A7=84?= =?UTF-8?q?=E5=88=99=E5=88=9D=E5=A7=8B=E5=8C=96SQL?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 新增《风控Agent需求文档》：规则引擎、前置拦截、人工处置、工单闭环 - 新增 sql/risk_rule_seed.sql：20条风控规则种子数据（INSERT IGNORE） --- sql/risk_rule_seed.sql | 33 ++++ 风控Agent需求文档.md | 341 +++++++++++++++++++++++++++++++++++++++++ 2 files changed, 374 insertions(+) create mode 100644 sql/risk_rule_seed.sql create mode 100644 风控Agent需求文档.md diff --git a/sql/risk_rule_seed.sql b/sql/risk_rule_seed.sql new file mode 100644 index 0000000..cc58fe2 --- /dev/null +++ b/sql/risk_rule_seed.sql @@ -0,0 +1,33 @@ +-- ===================================================================== +-- 风控规则初始化（20 条，见《风控Agent需求文档》§7） +-- 执行：mysql --default-character-set=utf8mb4 -u -p < sql/risk_rule_seed.sql +-- 说明：INSERT IGNORE 依赖 uk_rule_id 唯一键，重复执行不报错、不覆盖已有运营调整。 +-- ===================================================================== +SET NAMES utf8mb4; + +INSERT IGNORE INTO risk_rule + (rule_id, rule_name, trigger_condition, threshold, risk_level, weight, status) +VALUES + -- A 类 · 单笔规则 + ('R001', '大额申购-红', '单笔申购金额 ≥ 100万', '{"amount":1000000}', '高', 0.90, '启用'), + ('R002', '大额赎回-红', '单笔赎回金额 ≥ 100万', '{"amount":1000000}', '高', 0.90, '启用'), + ('R003', '大额申购-黄', '单笔申购金额 ≥ 50万', '{"amount":500000}', '中', 0.70, '启用'), + ('R004', '大额赎回-黄', '单笔赎回金额 ≥ 50万', '{"amount":500000}', '中', 0.70, '启用'), + ('R014', '适当性异常', '客户风险等级 < 产品风险等级', '{}', '高', 0.95, '启用'), + ('R020', '夜间交易', '下单时间 00:00–06:00', '{"start":"00:00","end":"06:00"}', '低', 0.35, '启用'), + -- B 类 · 聚合规则 + ('R005', '高频交易', '近 7 天申赎 ≥ 10 笔', '{"days":7,"count":10}', '中', 0.65, '启用'), + ('R006', '当日密集交易', '近 1 天申赎 ≥ 5 笔', '{"days":1,"count":5}', '中', 0.60, '启用'), + ('R007', '密集小额申购', '近 3 天 ≥ 5 笔、每笔 < 1万', '{"days":3,"count":5,"max_amount":10000}', '低', 0.50, '启用'), + ('R008', '短炒赎回', '同产品申购后 7 天内赎回', '{"days":7}', '中', 0.65, '启用'), + ('R009', '快进快出', '同产品申购后 3 天内全额赎回', '{"days":3}', '高', 0.85, '启用'), + ('R010', '资金快进快出', '入账后 24 小时内赎回', '{"hours":24}', '高', 0.90, '启用'), + ('R011', '短期累计大额', '近 7 天累计申赎 ≥ 200万', '{"days":7,"amount":2000000}', '高', 0.85, '启用'), + ('R012', '中期累计大额', '近 30 天累计申赎 ≥ 500万', '{"days":30,"amount":5000000}', '中', 0.65, '启用'), + ('R013', '拆分规避大额', '同 1 天多笔合计 ≥ 100万、单笔 < 50万', '{"days":1,"total":1000000,"each_lt":500000}', '中', 0.70, '启用'), + ('R018', '全额清仓赎回', '赎回后该产品持仓归 0 且金额 ≥ 10万', '{"amount":100000}', '低', 0.45, '启用'), + ('R019', '对倒交易', '近 7 天多产品互买互卖', '{"days":7}', '中', 0.70, '停用'), + -- C 类 · 客户维度规则 + ('R015', '新客户大额', '注册 < 30 天且单笔 ≥ 50万', '{"days":30,"amount":500000}', '中', 0.65, '启用'), + ('R016', '休眠账户激活', '90 天无交易后首笔 ≥ 10万', '{"days":90,"amount":100000}', '低', 0.55, '启用'), + ('R017', '整数金额', '金额为整万且 ≥ 10万', '{"amount":100000}', '低', 0.40, '启用'); diff --git a/风控Agent需求文档.md b/风控Agent需求文档.md new file mode 100644 index 0000000..b9bdef0 --- /dev/null +++ b/风控Agent需求文档.md @@ -0,0 +1,341 @@ +# 风控 Agent 需求文档 + +> 版本：v1.0 ｜ 日期：2026-09-12 ｜ 状态：待评审 +> 依据：《开发计划.md》§2.5 风控控制台、§5 风控Agent、Phase 4 +> 关联：《风控Agent设计方案.md》 + +--- + +## 1. 背景与目标 + +为系统接入**风控 Agent**：每一笔申购/赎回在下单落账前，先经过风控规则引擎检测，命中规则的订单被「挂起」等待人工处置，未命中的正常成交。 + +**核心目标**：形成「检测 → 阻断 → 人工处置 → 回写」的合规闭环。 + +**合规铁律**：机器只负责告警 + 阻断；放行/拦截/冻结的处置权永远在人工。 + +--- + +## 2. 术语与核心概念 + +| 术语 | 表 | 定位 | +|---|---|---| +| 交易申请单 | `trade_order` | 风控**拦截对象**，状态机主表（事前） | +| 成交流水 | `fin_transaction` | 账务凭证，只含真正成交的单（事后） | +| 风控预警 | `fin_risk_alert` | 命中记录 + 人工处置留痕 | +| 风控规则 | `risk_rule` | 规则引擎的表驱动配置（参数） | +| 业务工单 | `biz_work_order` | 拦截/冻结后开启的独立审批链 | + +**两张链的关系**：预警链（`fin_risk_alert`）在人工处置后结束；工单链（`biz_work_order`）只在「拦截/冻结」时**额外开启**，是处置的下游产物。 + +--- + +## 3. 现状与差距 + +- 当前 `service/purchase.py` / `service/redeem.py` 为**即时成交**（直接扣款+加仓 / 减仓+入账），不产生任何交易留痕。 +- `trade_order` / `fin_transaction` / `fin_risk_alert` / `risk_rule` / `biz_work_order` 五张表**只有 DDL，无 model/repo/schema 实现**。 +- 结论：风控 Agent 当前无数据可检测，需先补齐交易留痕。 + +--- + +## 4. 核心需求 + +### FR1 · 前置拦截（交易留痕） + +申购/赎回由「即时成交」改为「先建单 → 检测 → 通过才落账」。 + +``` +请求（申购/赎回） + ↓ +FastAPI 鉴权（get_current_user）→ 获取当前用户 + ↓ +适当性校验（硬拦截，申购专用）： + 客户风险等级 ≥ 产品风险等级，否则 1005 直接拒绝，不生成订单 + ↓ +创建 trade_order，status = '待确认' + - 生成唯一 order_no（前缀+日期+序列，如 PO20260912xxxx） + - 填入 order_type / amount / shares / nav（申请口径） + - ★ 不碰账户余额、不碰持仓 + ↓ +同步进入风控规则引擎检测（FR2） + ↓ +├─ 未命中 ──► status='已确认' + 落账（扣款+加仓 / 减仓+入账） +│ + 写 fin_transaction + confirm_time=now +│ +└─ 命中 ────► 生成 fin_risk_alert（未处理） + + trade_order.status='风控挂起' + risk_alert_id 关联 + 账户、持仓保持不动，等待人工处置（FR4） +``` + +**关键约束**： +- 建单 → 检测 → 落账/挂起 **在单个事务内原子完成**（无外部调用，全为 DB + 纯计算）。 +- 落账逻辑**复用**现有 `deduct_balance`/`holdings.upsert`、`holdings.redeem`/`credit_balance`，不重写。 + +### FR2 · 规则引擎检测（纯代码，不用大模型） + +- **同步执行**：在下单请求内当场完成，结果驱动状态流转。 +- **纯规则引擎**：阈值比较 + 历史流水统计，全程 `if/else + SQL`，**不调用 LLM**。 +- **表驱动**：`risk_rule` 表存参数（阈值/级别/权重/启停），代码存判断逻辑。 + +检测输入：当前 `trade_order` + 该客户历史流水 + 客户画像/账户/注册时间。 +检测输出：命中规则列表（规则 ID + 规则名 + 命中详情）。 + +### FR3 · 预警生成 + +- **聚合粒度**：一笔订单命中多条规则，只生成**一条**预警；`alert_type` 取主规则，`trigger_detail` 记录全部命中规则，`alert_level` 取最高档。 +- **分级**：级别取命中规则中 `risk_level` 最高档（低→蓝 / 中→黄 / 高→红）。 +- **置信度**：`confidence = 1 − Π(1 − weight_i)`（命中规则越多、权重越高，置信度越高）。 + +### FR4 · 人工处置（挂起自动，处置人工） + +三个处置动作均为**风控专员手动**，系统绝不自动放行/自动作废： + +| 动作 | 订单 | 预警 | 客户 | 工单 | +|---|---|---|---|---| +| 放行 | `已确认` + 落账 | `已排除` | 正常成交 | 无 | +| 拦截 | `失败`（作废） | `已确认` | 站内信通知 | 生成「可疑交易上报」 | +| 冻结 | `失败`（作废） | `已确认` | `sys_user.status=冻结` + 站内信通知 | 生成「冻结复核」 | + +**约束**： +- 处置接口的 `handler_id` **由后端从当前登录用户获取**，不信任前端传入。 +- 处置接口需校验角色：风控专员（`user_type=EMPLOYEE` 且 `employee_role=风控专员`，或 `ADMIN`）。 +- 处置结果回写订单与预警表（`handler_id`/`handle_result`/`handle_time`），全程留痕可追责。 +- **放行时重新校验余额**：挂起期间余额可能变化，申购放行若余额不足则放行失败、订单转「失败」。 + +### FR5 · 工单链 + +- **生成时机**：仅在「拦截」或「冻结」处置时自动创建，系统自动挂起阶段不产生工单。 +- **创建方式**：系统自动 INSERT，预填 `order_type`（可疑交易上报/冻结复核）、`customer_id`、`submitter_id`（当前专员）、`priority`（按预警级别）、`biz_content`（打包命中规则+订单号+预警 id）；`status` 默认「待处理」、`current_node` 默认「初审」。 +- **流转（后端接口驱动）**： + +``` + 认领 提交审核 复核通过 + 待处理 ──────────► 处理中 ──────────► 待审核 ──────────► 已完成 + (handler_id=自己) │ + └──────► 已驳回（复核驳回，终态） +``` + +- **职责分离**：复核人 ≠ 处理人，先做**软校验**（仅提示，不强制阻断）。 +- **终点**：工单「已完成」即项目内终点，**不上报人行**（监管报送为系统外动作，不在本项目范围）。 +- **每次流转写 `audit_log`**（谁、何时、从什么态改到什么态）。 + +--- + +## 5. 状态机 + +### 5.1 `trade_order` 订单状态机 + +``` +待确认 ──(未命中)──────► 已确认 ──► 落账 + 写 fin_transaction + │ + └──(命中)──────────► 风控挂起 + │ + ├──(放行)──► 已确认 ──► 落账 + 写 fin_transaction + ├──(拦截)──► 失败（作废）+ 通知客户 + 生成工单 + └──(冻结)──► 失败（作废）+ 客户冻结 + 通知客户 + 生成工单 +``` + +### 5.2 `fin_risk_alert` 预警状态机（无「处理中」中间态） + +``` +未处理 ──(放行)────► 已排除 + │ + ├──(拦截)──────► 已确认 + └──(冻结)──────► 已确认 +``` + +**防并发**：处置接口用条件更新 `WHERE id=? AND status='未处理'`，更新不到（rowcount=0）即视为已被他人处置，返回提示，不重复落账/工单/通知。 + +### 5.3 `biz_work_order` 工单状态机 + +``` +待处理 ──认领──► 处理中 ──提交审核──► 待审核 ──复核通过──► 已完成 + └──复核驳回──► 已驳回（终态） +``` + +--- + +## 6. 数据表设计 + +> 五张表 DDL 已在 `sql/schema.sql`，本次需补 ORM model + repository + schema。 + +### 6.1 `trade_order` 字段映射 + +| 字段 | 来源 | +|---|---| +| `order_no` | 生成唯一单号 `PO+日期+序列` | +| `customer_id` | 当前用户 id | +| `product_id` | 入参 product_id | +| `advisor_id` | 客户自购为 null | +| `order_type` | 申购 / 赎回 | +| `amount` / `shares` | 申购金额+折算份额；赎回份额+折算金额 | +| `nav` | 产品当前净值 | +| `fee` | 0（暂无手续费） | +| `status` | 待确认 / 风控挂起 / 已确认 / 失败 | +| `risk_alert_id` | 挂起时关联预警 id | +| `confirm_time` | 确认时间 | + +### 6.2 `fin_transaction` 成交流水 + +仅在**风控通过、落账成交**时写一笔：`transaction_no`、`order_id`（关联申请单）、`customer_id`、`product_id`、`transaction_type`（申购/赎回）、`amount`、`shares`、`nav`、`fee`、`status='已确认'`。 + +### 6.3 `fin_risk_alert` 预警 + +| 字段 | 填什么 | +|---|---| +| `customer_id` | 交易客户 | +| `order_id` | 触发预警的申请单 | +| `alert_type` | 主命中规则名 | +| `alert_level` | 低(蓝)/中(黄)/高(红)，取命中规则最高档 | +| `trigger_detail` | 全部命中规则 + 触发详情 | +| `transaction_ids` | 关联历史交易 ID 列表（聚合规则用） | +| `confidence` | 预警置信度 | +| `status` | 未处理 → 已确认 / 已排除 | +| `handler_id` / `handle_result` / `handle_time` | 处置留痕 | + +### 6.4 `risk_rule` 规则（表驱动） + +20 条规则配置见 §7，字段：`rule_id` / `rule_name` / `trigger_condition` / `threshold`(JSON) / `risk_level` / `weight` / `status`。 + +### 6.5 `biz_work_order` 工单 + +见 §4 FR5，`work_order_no` 生成 `WO+日期+序列`。 + +--- + +## 7. 规则清单（20 条） + +### A 类 · 单笔规则（只看当前订单） + +| 编号 | 规则名 | 触发条件 | 阈值 JSON | 级别 | 权重 | +|---|---|---|---|---|---| +| R001 | 大额申购-红 | 单笔申购金额 ≥ 100万 | `{"amount":1000000}` | 高 | 0.90 | +| R002 | 大额赎回-红 | 单笔赎回金额 ≥ 100万 | `{"amount":1000000}` | 高 | 0.90 | +| R003 | 大额申购-黄 | 单笔申购金额 ≥ 50万 | `{"amount":500000}` | 中 | 0.70 | +| R004 | 大额赎回-黄 | 单笔赎回金额 ≥ 50万 | `{"amount":500000}` | 中 | 0.70 | +| R014 | 适当性异常 | 客户风险等级 < 产品风险等级 | `{}` | 高 | 0.95 | +| R020 | 夜间交易 | 下单时间 00:00–06:00 | `{"start":"00:00","end":"06:00"}` | 低 | 0.35 | + +### B 类 · 聚合规则（查客户历史流水） + +| 编号 | 规则名 | 触发条件 | 阈值 JSON | 级别 | 权重 | +|---|---|---|---|---|---| +| R005 | 高频交易 | 近 7 天申赎 ≥ 10 笔 | `{"days":7,"count":10}` | 中 | 0.65 | +| R006 | 当日密集交易 | 近 1 天申赎 ≥ 5 笔 | `{"days":1,"count":5}` | 中 | 0.60 | +| R007 | 密集小额申购 | 近 3 天 ≥ 5 笔、每笔 < 1万 | `{"days":3,"count":5,"max_amount":10000}` | 低 | 0.50 | +| R008 | 短炒赎回 | 同产品申购后 7 天内赎回 | `{"days":7}` | 中 | 0.65 | +| R009 | 快进快出 | 同产品申购后 3 天内全额赎回 | `{"days":3}` | 高 | 0.85 | +| R010 | 资金快进快出 | 入账后 24 小时内赎回 | `{"hours":24}` | 高 | 0.90 | +| R011 | 短期累计大额 | 近 7 天累计申赎 ≥ 200万 | `{"days":7,"amount":2000000}` | 高 | 0.85 | +| R012 | 中期累计大额 | 近 30 天累计申赎 ≥ 500万 | `{"days":30,"amount":5000000}` | 中 | 0.65 | +| R013 | 拆分规避大额 | 同 1 天多笔合计 ≥ 100万、单笔 < 50万 | `{"days":1,"total":1000000,"each_lt":500000}` | 中 | 0.70 | +| R018 | 全额清仓赎回 | 赎回后该产品持仓归 0 且金额 ≥ 10万 | `{"amount":100000}` | 低 | 0.45 | +| R019 | 对倒交易 | 近 7 天多产品互买互卖 | `{"days":7}` | 中 | 0.70 | + +### C 类 · 客户维度规则（看画像/账户/注册时间） + +| 编号 | 规则名 | 触发条件 | 阈值 JSON | 级别 | 权重 | +|---|---|---|---|---|---| +| R015 | 新客户大额 | 注册 < 30 天且单笔 ≥ 50万 | `{"days":30,"amount":500000}` | 中 | 0.65 | +| R016 | 休眠账户激活 | 90 天无交易后首笔 ≥ 10万 | `{"days":90,"amount":100000}` | 低 | 0.55 | +| R017 | 整数金额 | 金额为整万且 ≥ 10万 | `{"amount":100000}` | 低 | 0.40 | + +**统计**：高 6 条、中 8 条、低 6 条。 + +**规则启用状态**：R019 对倒交易初始 `status=停用`（实现最复杂、Mock 下难触发），其余 19 条默认启用。 + +--- + +## 8. 接口清单 + +### 8.1 交易接口（改造） + +| 接口 | 说明 | +|---|---| +| `POST /api/purchase` | 改造为前置拦截：适当性校验 → 建单 → 检测 → 落账/挂起 | +| `POST /api/redeem` | 同上（赎回无适当性校验） | + +### 8.2 处置接口（新增，风控专员） + +| 接口 | 动作 | 副作用 | +|---|---|---| +| `POST /api/risk/alert/{id}/release` | 放行 | 订单已确认 + 落账；预警已排除 | +| `POST /api/risk/alert/{id}/block` | 拦截 | 订单失败 + 通知；预警已确认 + 生成工单 | +| `POST /api/risk/alert/{id}/freeze` | 冻结 | 订单失败 + 客户冻结 + 通知；预警已确认 + 生成工单 | + +### 8.3 工单接口（新增） + +| 接口 | 动作 | 状态变化 | +|---|---|---| +| `GET /api/work-order/list` | 列表 | — | +| `GET /api/work-order/{id}` | 详情 | — | +| `POST /api/work-order/claim` | 认领 | 待处理 → 处理中 | +| `POST /api/work-order/submit-review` | 提交审核 | 处理中 → 待审核 | +| `POST /api/work-order/review` | 复核 | 待审核 → 已完成 / 已驳回 | + +### 8.4 规则引擎（内部服务，非 HTTP 接口） + +- `detect(order, customer) -> 命中规则列表`：加载启用规则 → 按 `rule_id` 路由到判断函数 → 传入交易数据 + `threshold` 执行 → 汇总命中。 + +--- + +## 9. 关键设计约束（汇总） + +1. **单事务原子**：建单 → 检测 → 落账/挂起在单个事务内完成。 +2. **落账复用**：复用现有扣款/加仓、减仓/入账逻辑，不重写。 +3. **纯代码检测**：规则匹配用 `if/else + SQL`，不用 LLM。 +4. **人工处置**：挂起自动，放行/拦截/冻结人工，系统不自动处置。 +5. **防并发**：处置接口条件更新 `status='未处理'`，防重复处置。 +6. **handler_id 后端取**：不信任前端传入；处置/工单接口均校验角色权限。 +7. **放行重新校验余额**：防挂起期间余额变化导致落账失败。 +8. **预警聚合**：一笔订单一条预警；级别取最高；置信度按权重合成。 +9. **工单后置**：仅拦截/冻结时生成，为处置的下游产物。 +10. **审计留痕**：处置与工单流转均写 `audit_log`。 + +--- + +## 10. 实施分期 + +``` +Step 1 建 5 张表的 model + repo + schema +Step 2 改 purchase/redeem：适当性校验 → 建单(待确认) → 调风控引擎 +Step 3 写规则引擎（读 risk_rule + 查历史流水，输出命中列表） +Step 4 命中分支：挂起 + 生成预警；未命中：落账 + 写流水 +Step 5 处置接口（放行/拦截/冻结）+ 权限 + 防并发 + 审计 +Step 6 工单链（自动建单 + claim/review 接口） ← 可后置 +``` + +--- + +## 11. 验收标准 + +1. **正常申购/赎回**：未命中规则 → 订单「已确认」、账户/持仓变动、`fin_transaction` 生成一条成交流水。 +2. **命中挂起**：大额申购（如 ≥100万）→ 订单「风控挂起」、生成一条 `fin_risk_alert`（未处理）、账户与持仓不变。 +3. **放行**：风控专员放行 → 订单「已确认」+ 落账、预警「已排除」。 +4. **拦截**：风控专员拦截 → 订单「失败」、预警「已确认」、生成「可疑交易上报」工单、客户收到站内信。 +5. **冻结**：风控专员冻结 → 订单「失败」、预警「已确认」、客户 `sys_user.status=冻结`、生成「冻结复核」工单。 +6. **防并发**：同一预警被两次处置，第二次返回"已被处理"，不重复落账/工单/通知。 +7. **权限**：非风控专员调用处置接口被拒绝。 +8. **工单链**：拦截/冻结生成的工单能走完「待处理 → 处理中 → 待审核 → 已完成/已驳回」。 +9. **留痕**：处置与工单流转均有 `audit_log` 记录（谁、何时、从什么态到什么态）。 + +--- + +## 12. 已确认决策清单 + +| # | 决策点 | 结论 | +|---|---|---| +| 1 | 风控时机 | 前置拦截（先建单 → 检测 → 通过才落账） | +| 2 | 检测方式 | 同步 + 纯规则引擎，不用 LLM | +| 3 | 落表 | `trade_order` 为风控主表，`fin_transaction` 为成交凭证 | +| 4 | 预警粒度 | 一笔订单聚合一条预警 | +| 5 | 分级置信度 | 级别取最高；`confidence = 1 − Π(1 − weight_i)` | +| 6 | 处置方式 | 挂起自动，放行/拦截/冻结人工 | +| 7 | 预警中间态 | 不加「处理中」，用条件更新防并发 | +| 8 | 工单时机 | 仅拦截/冻结时生成，人工处置之后 | +| 9 | 工单复核分离 | 软校验（仅提示），不强制 | +| 10 | 工单驳回 | 已驳回为终态，不回流 | +| 11 | 工单终点 | 「已完成」即终点，不上报人行 | +| 12 | R019 对倒交易 | 初始停用 | +| 13 | R014 适当性异常 | 保留作兜底 | +| 14 | 单号生成 | 前缀+日期+序列（`PO...` / `WO...`） | -- 2.54.0