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/2] =?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/2] =?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