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, - ), - )