From ea091c041b6fa30f58083a19c1b2c995e3a70efb Mon Sep 17 00:00:00 2001 From: zhangyongcai <17382809141.@163.com> Date: Thu, 10 Sep 2026 23:21:30 +0800 Subject: [PATCH] =?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 --- .gitignore | 10 ++- api/router.py | 6 +- api/routers/account.py | 31 +++++++++ api/routers/holdings.py | 20 ++++++ api/routers/purchase.py | 23 +++++++ api/routers/redeem.py | 22 +++++++ main.py | 6 +- model/fin_account.py | 33 ++++++++++ model/fin_customer_profile.py | 34 ++++++++++ model/fin_holdings.py | 29 ++++++++ model/fin_product.py | 32 +++++++++ repositories/fin_account.py | 97 +++++++++++++++++++++++++++ repositories/fin_customer_profile.py | 21 ++++++ repositories/fin_holdings.py | 75 +++++++++++++++++++++ repositories/fin_product.py | 7 ++ requirements.txt | 24 +++---- schemas/account.py | 34 ++++++++++ schemas/holdings.py | 30 +++++++++ schemas/purchase.py | 24 +++++++ schemas/redeem.py | 24 +++++++ service/account.py | 80 ++++++++++++++++++++++ service/holdings.py | 37 +++++++++++ service/purchase.py | 99 ++++++++++++++++++++++++++++ service/redeem.py | 80 ++++++++++++++++++++++ sql/schema.sql | 39 +++++++++-- 25 files changed, 897 insertions(+), 20 deletions(-) create mode 100644 api/routers/account.py create mode 100644 api/routers/holdings.py create mode 100644 api/routers/purchase.py create mode 100644 api/routers/redeem.py create mode 100644 model/fin_account.py create mode 100644 model/fin_customer_profile.py create mode 100644 model/fin_holdings.py create mode 100644 model/fin_product.py create mode 100644 repositories/fin_account.py create mode 100644 repositories/fin_customer_profile.py create mode 100644 repositories/fin_holdings.py create mode 100644 repositories/fin_product.py create mode 100644 schemas/account.py create mode 100644 schemas/holdings.py create mode 100644 schemas/purchase.py create mode 100644 schemas/redeem.py create mode 100644 service/account.py create mode 100644 service/holdings.py create mode 100644 service/purchase.py create mode 100644 service/redeem.py diff --git a/.gitignore b/.gitignore index 2eea525..12e23d4 100644 --- a/.gitignore +++ b/.gitignore @@ -1 +1,9 @@ -.env \ No newline at end of file +.env + +# Python +__pycache__/ +*.py[cod] + +# 虚拟环境 +.venv/ +venv/ diff --git a/api/router.py b/api/router.py index f199e06..9161688 100644 --- a/api/router.py +++ b/api/router.py @@ -3,7 +3,11 @@ """ from fastapi import APIRouter -from api.routers import auth +from api.routers import account, auth, holdings, purchase, redeem api_router = APIRouter() api_router.include_router(auth.router, prefix="/api", tags=["认证"]) +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=["交易"]) diff --git a/api/routers/account.py b/api/routers/account.py new file mode 100644 index 0000000..cb1f628 --- /dev/null +++ b/api/routers/account.py @@ -0,0 +1,31 @@ +"""资金账户路由：余额查询、余额加减（业务在 service/account.py，路由只做编排）。""" +from fastapi import APIRouter, Depends +from sqlalchemy.ext.asyncio import AsyncSession + +from api.deps import get_current_user +from config.deps import get_db +from model.sys_user import SysUser +from schemas.account import AdjustReq +from service import account as account_service +from utils.response import success + +router = APIRouter() + + +@router.get("/account/balance", summary="查询当前用户余额") +async def balance( + user: SysUser = Depends(get_current_user), + db: AsyncSession = Depends(get_db), +): + result = await account_service.get_balance(db, user) + return success(result.model_dump(mode="json")) + + +@router.post("/account/adjust", summary="调整余额（add=充值 / sub=提现）") +async def adjust( + req: AdjustReq, + user: SysUser = Depends(get_current_user), + db: AsyncSession = Depends(get_db), +): + result = await account_service.adjust_balance(db, user, req.direction, req.amount) + return success(result.model_dump(mode="json")) diff --git a/api/routers/holdings.py b/api/routers/holdings.py new file mode 100644 index 0000000..2ae4ac9 --- /dev/null +++ b/api/routers/holdings.py @@ -0,0 +1,20 @@ +"""持仓路由：查询当前用户持仓（业务在 service/holdings.py，路由只做编排）。""" +from fastapi import APIRouter, Depends +from sqlalchemy.ext.asyncio import AsyncSession + +from api.deps import get_current_user +from config.deps import get_db +from model.sys_user import SysUser +from service import holdings as holdings_service +from utils.response import success + +router = APIRouter() + + +@router.get("/holdings", summary="查询当前用户持仓（持有中）") +async def list_holdings( + user: SysUser = Depends(get_current_user), + db: AsyncSession = Depends(get_db), +): + result = await holdings_service.get_holdings(db, user) + return success([h.model_dump(mode="json") for h in result]) diff --git a/api/routers/purchase.py b/api/routers/purchase.py new file mode 100644 index 0000000..944155b --- /dev/null +++ b/api/routers/purchase.py @@ -0,0 +1,23 @@ + +"""申购路由：申购基金产品（业务在 service/purchase.py，路由只做编排）。""" +from fastapi import APIRouter, Depends +from sqlalchemy.ext.asyncio import AsyncSession + +from api.deps import get_current_user +from config.deps import get_db +from model.sys_user import SysUser +from schemas.purchase import PurchaseReq +from service import purchase as purchase_service +from utils.response import success + +router = APIRouter() + + +@router.post("/purchase", summary="申购基金产品") +async def purchase( + req: PurchaseReq, + user: SysUser = Depends(get_current_user), + db: AsyncSession = Depends(get_db), +): + result = await purchase_service.purchase(db, user, req.product_id, req.amount) + return success(result.model_dump(mode="json")) diff --git a/api/routers/redeem.py b/api/routers/redeem.py new file mode 100644 index 0000000..768ce79 --- /dev/null +++ b/api/routers/redeem.py @@ -0,0 +1,22 @@ +"""赎回路由：赎回基金产品（业务在 service/redeem.py，路由只做编排）。""" +from fastapi import APIRouter, Depends +from sqlalchemy.ext.asyncio import AsyncSession + +from api.deps import get_current_user +from config.deps import get_db +from model.sys_user import SysUser +from schemas.redeem import RedeemReq +from service import redeem as redeem_service +from utils.response import success + +router = APIRouter() + + +@router.post("/redeem", summary="赎回基金产品") +async def redeem( + req: RedeemReq, + user: SysUser = Depends(get_current_user), + db: AsyncSession = Depends(get_db), +): + result = await redeem_service.redeem(db, user, req.product_id, req.shares) + return success(result.model_dump(mode="json")) diff --git a/main.py b/main.py index b7e6be0..1df473c 100644 --- a/main.py +++ b/main.py @@ -1,6 +1,7 @@ """应用入口：四库异步生命周期 + 中间件 + 全局异常 + 路由装配。""" from contextlib import asynccontextmanager +import uvicorn from fastapi import FastAPI from api.router import api_router @@ -28,4 +29,7 @@ app.include_router(api_router) @app.get("/") async def root(): - return {"message": "智能公募基金系统 API", "docs": "/docs"} \ No newline at end of file + return {"message": "智能公募基金系统 API", "docs": "/docs"} + +if __name__ == '__main__': + uvicorn.run(app, host="127.0.0.1", port=8001) \ No newline at end of file diff --git a/model/fin_account.py b/model/fin_account.py new file mode 100644 index 0000000..207d923 --- /dev/null +++ b/model/fin_account.py @@ -0,0 +1,33 @@ +"""fin_account 客户资金账户表 ORM 模型（现金余额，一人一户）。 + +balance 为账户总余额（含冻结部分），可用余额 = balance - frozen_amount，不落库。 +""" +from __future__ import annotations + +from datetime import datetime +from decimal import Decimal + +from sqlalchemy import BigInteger, DateTime, Integer, Numeric, String, func +from sqlalchemy.orm import Mapped, mapped_column + +from model.base import Base + + +class FinAccount(Base): + __tablename__ = "fin_account" + __table_args__ = {"comment": "客户资金账户表（现金余额，申购扣款/赎回入账的账务载体）"} + + customer_id: Mapped[int] = mapped_column( + BigInteger, primary_key=True, autoincrement=False + ) + balance: Mapped[Decimal] = mapped_column(Numeric(18, 2), server_default="0") + frozen_amount: Mapped[Decimal] = mapped_column(Numeric(18, 2), server_default="0") + currency: Mapped[str] = mapped_column(String(8), server_default="CNY") + status: Mapped[str] = mapped_column(String(16), server_default="正常") + version: Mapped[int] = mapped_column(Integer, server_default="0") + 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/fin_customer_profile.py b/model/fin_customer_profile.py new file mode 100644 index 0000000..7f71983 --- /dev/null +++ b/model/fin_customer_profile.py @@ -0,0 +1,34 @@ +"""fin_customer_profile 客户画像主表 ORM 模型。""" +from __future__ import annotations + +from datetime import datetime +from decimal import Decimal + +from sqlalchemy import JSON, BigInteger, DateTime, Integer, Numeric, String, func +from sqlalchemy.orm import Mapped, mapped_column + +from model.base import Base + + +class FinCustomerProfile(Base): + __tablename__ = "fin_customer_profile" + __table_args__ = {"comment": "客户画像主表（RAG/GraphRAG 个性化推荐核心上下文）"} + + customer_id: Mapped[int] = mapped_column( + BigInteger, primary_key=True, autoincrement=False + ) + risk_level: Mapped[str | None] = mapped_column(String(16)) + risk_score: Mapped[int | None] = mapped_column(Integer) + investment_experience: Mapped[str | None] = mapped_column(String(16)) + annual_income_range: Mapped[str | None] = mapped_column(String(32)) + total_assets: Mapped[Decimal | None] = mapped_column(Numeric(18, 2)) + asset_allocation: Mapped[dict | None] = mapped_column(JSON) + product_preference: Mapped[dict | None] = mapped_column(JSON) + customer_level: Mapped[str | None] = mapped_column(String(16)) + confidence_score: Mapped[Decimal] = mapped_column( + Numeric(5, 2), server_default="0.50" + ) + profile_version: Mapped[int] = mapped_column(Integer, server_default="1") + update_time: Mapped[datetime] = mapped_column( + DateTime, server_default=func.now(), onupdate=func.now() + ) diff --git a/model/fin_holdings.py b/model/fin_holdings.py new file mode 100644 index 0000000..b312c0e --- /dev/null +++ b/model/fin_holdings.py @@ -0,0 +1,29 @@ +"""fin_holdings 持仓表 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 FinHoldings(Base): + __tablename__ = "fin_holdings" + __table_args__ = {"comment": "持仓表（当前/历史持仓快照）"} + + id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True) + customer_id: Mapped[int] = mapped_column(BigInteger) + product_id: Mapped[int] = mapped_column(BigInteger) + shares: Mapped[Decimal] = mapped_column(Numeric(18, 4), server_default="0") + cost_amount: Mapped[Decimal] = mapped_column(Numeric(18, 2), server_default="0") + current_value: Mapped[Decimal] = mapped_column(Numeric(18, 2), server_default="0") + profit_loss: Mapped[Decimal] = mapped_column(Numeric(18, 2), server_default="0") + profit_ratio: Mapped[Decimal] = mapped_column(Numeric(8, 4), server_default="0") + status: Mapped[str] = mapped_column(String(16), 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/fin_product.py b/model/fin_product.py new file mode 100644 index 0000000..f324ba9 --- /dev/null +++ b/model/fin_product.py @@ -0,0 +1,32 @@ +"""fin_product 公募基金产品表 ORM 模型。""" +from __future__ import annotations + +from datetime import date, datetime +from decimal import Decimal + +from sqlalchemy import BigInteger, Date, DateTime, Integer, Numeric, String, func +from sqlalchemy.orm import Mapped, mapped_column + +from model.base import Base + + +class FinProduct(Base): + __tablename__ = "fin_product" + __table_args__ = {"comment": "公募基金产品表"} + + id: Mapped[int] = mapped_column(BigInteger, primary_key=True, autoincrement=True) + product_code: Mapped[str] = mapped_column(String(32), unique=True) + product_name: Mapped[str] = mapped_column(String(128)) + product_type: Mapped[str] = mapped_column(String(32)) + risk_level: Mapped[str] = mapped_column(String(8)) + expected_return: Mapped[Decimal | None] = mapped_column(Numeric(7, 4)) + nav: Mapped[Decimal | None] = mapped_column(Numeric(12, 6)) + nav_date: Mapped[date | None] = mapped_column(Date) + fee_rate: Mapped[Decimal] = mapped_column(Numeric(6, 4), server_default="0.0000") + term_days: Mapped[int] = mapped_column(Integer, server_default="0") + fund_manager: Mapped[str | None] = mapped_column(String(64)) + status: Mapped[str] = mapped_column(String(16), 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/repositories/fin_account.py b/repositories/fin_account.py new file mode 100644 index 0000000..831d943 --- /dev/null +++ b/repositories/fin_account.py @@ -0,0 +1,97 @@ +"""fin_account 仓储：按客户 ID 取资金账户 + 余额加减（原子 UPDATE）。 + +注：本表主键为 customer_id（非 id），故不复用 BaseRepository.delete/count 中的 id 约定。 +余额增减用原子 UPDATE（balance = balance ± delta）防并发丢更新，不依赖乐观锁重试。 +""" +from __future__ import annotations + +from decimal import Decimal + +from sqlalchemy import select, update + +from model.fin_account import FinAccount +from repositories.base import BaseRepository + + +class FinAccountRepo(BaseRepository): + model = FinAccount + + async def get_by_customer_id(self, customer_id: int) -> FinAccount | None: + return await self.db.scalar( + select(FinAccount).where(FinAccount.customer_id == customer_id) + ) + + async def create(self, customer_id: int, balance: Decimal) -> FinAccount: + """开资金户（充值时账户不存在则自动开户入账）。""" + account = FinAccount(customer_id=customer_id, balance=balance) + self.db.add(account) + await self.db.commit() + await self.db.refresh(account) + return account + + async def add_balance(self, customer_id: int, delta: Decimal) -> FinAccount: + """入账：balance += delta，原子自增后回读最新余额。""" + await self.db.execute( + update(FinAccount) + .where(FinAccount.customer_id == customer_id) + .values( + balance=FinAccount.balance + delta, + version=FinAccount.version + 1, + ) + ) + await self.db.commit() + return await self.get_by_customer_id(customer_id) + + async def subtract_balance( + self, customer_id: int, delta: Decimal + ) -> FinAccount | None: + """出账：balance -= delta，可用余额（balance - frozen_amount）不足时返回 None。""" + result = await self.db.execute( + update(FinAccount) + .where( + FinAccount.customer_id == customer_id, + FinAccount.balance - FinAccount.frozen_amount >= delta, + ) + .values( + balance=FinAccount.balance - delta, + version=FinAccount.version + 1, + ) + ) + await self.db.commit() + if result.rowcount == 0: + return None + return await self.get_by_customer_id(customer_id) + + async def deduct_balance(self, customer_id: int, delta: Decimal) -> bool: + """申购事务内扣款：balance -= delta（可用余额不足则不动）。 + + 不 commit，由 service 层事务统一提交，保证「扣款 + 加仓」原子性。 + 可用余额（balance - frozen_amount）不足时返回 False。 + """ + result = await self.db.execute( + update(FinAccount) + .where( + FinAccount.customer_id == customer_id, + FinAccount.balance - FinAccount.frozen_amount >= delta, + ) + .values( + balance=FinAccount.balance - delta, + version=FinAccount.version + 1, + ) + ) + return result.rowcount > 0 + + async def credit_balance(self, customer_id: int, delta: Decimal) -> bool: + """赎回事务内入账：balance += delta。 + + 不 commit，由 service 层事务统一提交，保证「减仓 + 入账」原子性。 + """ + result = await self.db.execute( + update(FinAccount) + .where(FinAccount.customer_id == customer_id) + .values( + balance=FinAccount.balance + delta, + version=FinAccount.version + 1, + ) + ) + return result.rowcount > 0 diff --git a/repositories/fin_customer_profile.py b/repositories/fin_customer_profile.py new file mode 100644 index 0000000..d168254 --- /dev/null +++ b/repositories/fin_customer_profile.py @@ -0,0 +1,21 @@ +"""fin_customer_profile 画像仓储：按客户 ID 取画像。 + +注：本表主键为 customer_id（非 id），故不复用 BaseRepository.get 的 id 约定。 +""" +from __future__ import annotations + +from sqlalchemy import select + +from model.fin_customer_profile import FinCustomerProfile +from repositories.base import BaseRepository + + +class FinCustomerProfileRepo(BaseRepository): + model = FinCustomerProfile + + async def get_by_customer_id(self, customer_id: int) -> FinCustomerProfile | None: + return await self.db.scalar( + select(FinCustomerProfile).where( + FinCustomerProfile.customer_id == customer_id + ) + ) diff --git a/repositories/fin_holdings.py b/repositories/fin_holdings.py new file mode 100644 index 0000000..253a2a9 --- /dev/null +++ b/repositories/fin_holdings.py @@ -0,0 +1,75 @@ +"""fin_holdings 持仓仓储：按客户（+状态）查持仓、申购加仓 upsert、赎回减仓。""" +from __future__ import annotations + +from decimal import Decimal + +from sqlalchemy import case, select, update +from sqlalchemy.dialects.mysql import insert as mysql_insert + +from model.fin_holdings import FinHoldings +from repositories.base import BaseRepository + + +class FinHoldingsRepo(BaseRepository): + model = FinHoldings + + async def list_by_customer( + self, customer_id: int, status: str | None = None + ) -> list[FinHoldings]: + """按客户 ID 查持仓，可按状态过滤（status=None 表示不过滤）。""" + stmt = select(FinHoldings).where(FinHoldings.customer_id == customer_id) + if status is not None: + stmt = stmt.where(FinHoldings.status == status) + stmt = stmt.order_by(FinHoldings.id) + return list((await self.db.scalars(stmt)).all()) + + async def get_by_customer_product( + self, customer_id: int, product_id: int + ) -> FinHoldings | None: + return await self.db.scalar( + select(FinHoldings).where( + FinHoldings.customer_id == customer_id, + FinHoldings.product_id == product_id, + ) + ) + + async def upsert( + self, customer_id: int, product_id: int, add_shares: Decimal, add_cost: Decimal + ) -> None: + """申购加仓：有则加份额/成本，无则新增（靠 uk_customer_product 唯一键）。不 commit。""" + stmt = mysql_insert(FinHoldings).values( + customer_id=customer_id, + product_id=product_id, + shares=add_shares, + cost_amount=add_cost, + ) + stmt = stmt.on_duplicate_key_update( + shares=FinHoldings.shares + add_shares, + cost_amount=FinHoldings.cost_amount + add_cost, + ) + await self.db.execute(stmt) + + async def redeem( + self, customer_id: int, product_id: int, redeem_shares: Decimal + ) -> bool: + """赎回减仓：shares -= redeem_shares，份额归 0 时状态置'已清仓'。 + + 持仓份额不足（含无持仓、shares=0）时不动作，返回 False。 + 不 commit，由 service 层事务统一提交，保证「减仓 + 入账」原子性。 + """ + result = await self.db.execute( + update(FinHoldings) + .where( + FinHoldings.customer_id == customer_id, + FinHoldings.product_id == product_id, + FinHoldings.shares >= redeem_shares, + ) + .values( + shares=FinHoldings.shares - redeem_shares, + status=case( + (FinHoldings.shares - redeem_shares == 0, "已清仓"), + else_=FinHoldings.status, + ), + ) + ) + return result.rowcount > 0 diff --git a/repositories/fin_product.py b/repositories/fin_product.py new file mode 100644 index 0000000..1b17e52 --- /dev/null +++ b/repositories/fin_product.py @@ -0,0 +1,7 @@ +"""fin_product 产品仓储：申购按主键取产品（复用 BaseRepository.get）。""" +from model.fin_product import FinProduct +from repositories.base import BaseRepository + + +class FinProductRepo(BaseRepository): + model = FinProduct diff --git a/requirements.txt b/requirements.txt index 55494c7..0edbc0c 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1,12 +1,12 @@ -fastapi -uvicorn -httpx -pyjwt -pydantic -pydantic-settings -sqlalchemy -aiomysql -redis -pymilvus -neo4j -python-dotenv \ No newline at end of file +fastapi~=0.141.1 +sqlalchemy~=2.0.52 +aiomysql~=0.2.0 +cryptography~=44.0.0 +httpx~=0.28.1 +pydantic~=2.13.5 +starlette~=1.6.0 +neo4j~=6.3.0 +redis~=8.1.0 +pymilvus~=3.0.1 +pydantic-settings~=2.15.0 +pyjwt~=2.13.0 \ No newline at end of file diff --git a/schemas/account.py b/schemas/account.py new file mode 100644 index 0000000..cfaa2b1 --- /dev/null +++ b/schemas/account.py @@ -0,0 +1,34 @@ +"""资金账户相关 DTO。 + +金额统一序列化为两位小数字符串（如 "48000.00"），避免 JSON number 的浮点精度隐患。 +""" +from datetime import datetime +from decimal import Decimal +from typing import Literal + +from pydantic import BaseModel, Field, field_serializer + + +class BalanceResp(BaseModel): + """用户余额响应体。available_balance = balance - frozen_amount，由 service 派生。""" + + customer_id: int + balance: Decimal + available_balance: Decimal + frozen_amount: Decimal + currency: str + status: str + update_time: datetime | None = None + + @field_serializer("balance", "available_balance", "frozen_amount") + def _fmt_money(self, value: Decimal) -> str: + return f"{value:.2f}" + + +class AdjustReq(BaseModel): + """加减余额入参。amount 为正数（元），direction 决定加/减。""" + + direction: Literal["add", "sub"] = Field( + ..., description="add=入账(充值)，sub=出账(提现)" + ) + amount: Decimal = Field(gt=0) diff --git a/schemas/holdings.py b/schemas/holdings.py new file mode 100644 index 0000000..437dd70 --- /dev/null +++ b/schemas/holdings.py @@ -0,0 +1,30 @@ +"""持仓相关 DTO。 + +金额字段序列化为字符串，避免 JSON number 浮点精度隐患； +份额 / 盈亏比例保留 4 位，金额保留 2 位。 +""" +from decimal import Decimal + +from pydantic import BaseModel, field_serializer + + +class HoldingResp(BaseModel): + """单条持仓记录。""" + + id: int + customer_id: int + product_id: int + shares: Decimal + cost_amount: Decimal + current_value: Decimal + profit_loss: Decimal + profit_ratio: Decimal + status: str + + @field_serializer("shares", "profit_ratio") + def _fmt_ratio(self, value: Decimal) -> str: + return f"{value:.4f}" + + @field_serializer("cost_amount", "current_value", "profit_loss") + def _fmt_money(self, value: Decimal) -> str: + return f"{value:.2f}" diff --git a/schemas/purchase.py b/schemas/purchase.py new file mode 100644 index 0000000..2cd5680 --- /dev/null +++ b/schemas/purchase.py @@ -0,0 +1,24 @@ +"""申购相关 DTO。""" +from decimal import Decimal + +from pydantic import BaseModel, Field, field_serializer + +from schemas.holdings import HoldingResp + + +class PurchaseReq(BaseModel): + """申购入参。amount 为正数（元），按净值折算份额。""" + + product_id: int + amount: Decimal = Field(gt=0) + + +class PurchaseResp(BaseModel): + """申购结果：最新余额 + 该产品最新持仓。""" + + balance: Decimal + holding: HoldingResp + + @field_serializer("balance") + def _fmt_balance(self, value: Decimal) -> str: + return f"{value:.2f}" diff --git a/schemas/redeem.py b/schemas/redeem.py new file mode 100644 index 0000000..b80071c --- /dev/null +++ b/schemas/redeem.py @@ -0,0 +1,24 @@ +"""赎回相关 DTO。""" +from decimal import Decimal + +from pydantic import BaseModel, Field, field_serializer + +from schemas.holdings import HoldingResp + + +class RedeemReq(BaseModel): + """赎回入参。shares 为正数（份额），按净值折算入账金额。""" + + product_id: int + shares: Decimal = Field(gt=0) + + +class RedeemResp(BaseModel): + """赎回结果：最新余额 + 该产品最新持仓。""" + + balance: Decimal + holding: HoldingResp + + @field_serializer("balance") + def _fmt_balance(self, value: Decimal) -> str: + return f"{value:.2f}" diff --git a/service/account.py b/service/account.py new file mode 100644 index 0000000..60d0469 --- /dev/null +++ b/service/account.py @@ -0,0 +1,80 @@ +"""资金账户服务：余额查询 + 余额加减（路由层只编排，不碰数据/逻辑）。""" +from __future__ import annotations + +from decimal import Decimal + +from sqlalchemy.ext.asyncio import AsyncSession + +from model.fin_account import FinAccount +from model.sys_user import SysUser +from repositories.fin_account import FinAccountRepo +from schemas.account import BalanceResp +from utils.exceptions import ForbiddenError, NotFoundError, ParamError + +_ZERO = Decimal("0.00") +_DEFAULT_CURRENCY = "CNY" + + +def _build_balance_resp(account: FinAccount) -> BalanceResp: + return BalanceResp( + customer_id=account.customer_id, + balance=account.balance, + available_balance=account.balance - account.frozen_amount, + frozen_amount=account.frozen_amount, + currency=account.currency, + status=account.status, + update_time=account.update_time, + ) + + +async def get_balance(db: AsyncSession, user: SysUser) -> BalanceResp: + """查询当前用户现金余额。 + + - 仅客户账号可查（员工共用 sys_user，但无资金账户）； + - 未开资金户时按零余额返回，不报错。 + """ + if user.user_type != "CUSTOMER": + raise ForbiddenError("仅客户账号可查询资金余额") + + account = await FinAccountRepo(db).get_by_customer_id(user.id) + if account is None: + return BalanceResp( + customer_id=user.id, + balance=_ZERO, + available_balance=_ZERO, + frozen_amount=_ZERO, + currency=_DEFAULT_CURRENCY, + status="正常", + ) + + return _build_balance_resp(account) + + +async def adjust_balance( + db: AsyncSession, user: SysUser, direction: str, amount: Decimal +) -> BalanceResp: + """加减余额：direction=add 入账 / sub 出账。 + + - add：balance += amount，未开资金户时自动开户入账； + - sub：balance -= amount，可用余额（balance - frozen_amount）不足时报错。 + """ + if user.user_type != "CUSTOMER": + raise ForbiddenError("仅客户账号可调整余额") + + repo = FinAccountRepo(db) + account = await repo.get_by_customer_id(user.id) + + if direction == "add": + if account is None: + account = await repo.create(user.id, amount) + else: + account = await repo.add_balance(user.id, amount) + else: # sub + if account is None: + raise NotFoundError("资金账户不存在") + updated = await repo.subtract_balance(user.id, amount) + if updated is None: + raise ParamError("可用余额不足") + account = updated + + return _build_balance_resp(account) diff --git a/service/holdings.py b/service/holdings.py new file mode 100644 index 0000000..8b16926 --- /dev/null +++ b/service/holdings.py @@ -0,0 +1,37 @@ +"""持仓服务：查询当前客户持仓（路由层只编排，不碰数据/逻辑）。""" +from __future__ import annotations + +from sqlalchemy.ext.asyncio import AsyncSession + +from model.sys_user import SysUser +from repositories.fin_holdings import FinHoldingsRepo +from schemas.holdings import HoldingResp +from utils.exceptions import ForbiddenError + +_HOLDING_STATUS = "持有中" + + +async def get_holdings(db: AsyncSession, user: SysUser) -> list[HoldingResp]: + """查询当前客户的在持持仓（status=持有中）。 + + - 仅客户账号可查（员工共用 sys_user，但无持仓）； + - 无持仓返回空列表，不报错。 + """ + if user.user_type != "CUSTOMER": + raise ForbiddenError("仅客户账号可查询持仓") + + holdings = await FinHoldingsRepo(db).list_by_customer(user.id, _HOLDING_STATUS) + return [ + HoldingResp( + id=h.id, + customer_id=h.customer_id, + product_id=h.product_id, + shares=h.shares, + cost_amount=h.cost_amount, + current_value=h.current_value, + profit_loss=h.profit_loss, + profit_ratio=h.profit_ratio, + status=h.status, + ) + for h in holdings + ] diff --git a/service/purchase.py b/service/purchase.py new file mode 100644 index 0000000..fa44f49 --- /dev/null +++ b/service/purchase.py @@ -0,0 +1,99 @@ +"""申购服务：风险匹配校验 + 余额扣减 + 持仓加仓（单事务原子）。""" +from __future__ import annotations + +from decimal import ROUND_HALF_UP, Decimal + +from sqlalchemy.ext.asyncio import AsyncSession + +from model.sys_user import SysUser +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 schemas.holdings import HoldingResp +from schemas.purchase import PurchaseResp +from utils.exceptions import ( + ForbiddenError, + NotFoundError, + NotSuitableError, + ParamError, +) + +_MONEY = Decimal("0.01") +_SHARES = Decimal("0.0001") + +# 风险等级 → 序号。兼容两套口径：R1~R5 与 保守~激进（同一映射）。 +_RISK_RANK = { + "R1": 1, "R2": 2, "R3": 3, "R4": 4, "R5": 5, + "保守": 1, "稳健": 2, "平衡": 3, "进取": 4, "激进": 5, +} + + +def _risk_rank(level: str | None) -> int | None: + """客户画像 / 产品的风险等级统一转序号；未知返回 None。""" + if not level: + return None + return _RISK_RANK.get(level.strip()) + + +async def purchase( + db: AsyncSession, user: SysUser, product_id: int, amount: Decimal +) -> PurchaseResp: + """申购基金：校验通过后扣减余额并加仓，全程单事务。 + + - 仅客户可申购； + - 产品须在售且净值非空； + - 客户风险等级序号 >= 产品风险等级序号，否则 1005 拦截； + - 余额不足拦截；扣款 + 加仓要么都成、要么都回滚。 + """ + if user.user_type != "CUSTOMER": + raise ForbiddenError("仅客户账号可申购") + + amount = amount.quantize(_MONEY, rounding=ROUND_HALF_UP) + + product = await FinProductRepo(db).get(product_id) + if product is None: + raise NotFoundError("产品不存在") + if product.status != "在售": + raise ParamError("产品不在售") + if product.nav is None: + raise ParamError("产品暂无净值，无法申购") + + profile = await FinCustomerProfileRepo(db).get_by_customer_id(user.id) + customer_rank = _risk_rank(profile.risk_level if profile else None) + if customer_rank is None: + raise NotSuitableError("客户无风险等级，无法申购") + product_rank = _risk_rank(product.risk_level) + if product_rank is None or customer_rank < product_rank: + raise NotSuitableError() + + shares = (amount / product.nav).quantize(_SHARES, rounding=ROUND_HALF_UP) + + 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) + await db.commit() + 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 new file mode 100644 index 0000000..97af761 --- /dev/null +++ b/service/redeem.py @@ -0,0 +1,80 @@ +"""赎回服务：校验持仓 → 减仓 + 余额入账（单事务原子）。""" +from __future__ import annotations + +from decimal import ROUND_HALF_UP, Decimal + +from sqlalchemy.ext.asyncio import AsyncSession + +from model.sys_user import SysUser +from repositories.fin_account import FinAccountRepo +from repositories.fin_holdings import FinHoldingsRepo +from repositories.fin_product import FinProductRepo +from schemas.holdings import HoldingResp +from schemas.redeem import RedeemResp +from utils.exceptions import ForbiddenError, NotFoundError, ParamError + +_MONEY = Decimal("0.01") +_SHARES = Decimal("0.0001") + + +async def redeem( + db: AsyncSession, user: SysUser, product_id: int, shares: Decimal +) -> RedeemResp: + """赎回基金：校验通过后减仓并按净值入账，全程单事务。 + + - 仅客户可赎回； + - 产品须存在且净值非空； + - 持仓须存在且份额 > 0，赎回份额不得超过持仓份额； + - 减仓（归 0 时置'已清仓'）+ 入账要么都成、要么都回滚。 + """ + if user.user_type != "CUSTOMER": + raise ForbiddenError("仅客户账号可赎回") + + shares = shares.quantize(_SHARES, rounding=ROUND_HALF_UP) + if shares <= 0: + raise ParamError("赎回份额须大于 0") + + product = await FinProductRepo(db).get(product_id) + if product is None: + raise NotFoundError("产品不存在") + if product.nav is None: + raise ParamError("产品暂无净值，无法赎回") + + account_repo = FinAccountRepo(db) + holdings_repo = FinHoldingsRepo(db) + + holding = await holdings_repo.get_by_customer_product(user.id, product_id) + if holding is None or holding.shares <= 0: + raise ParamError("无可赎回份额") + if holding.shares < shares: + raise ParamError("可赎回份额不足") + + credited = (shares * product.nav).quantize(_MONEY, rounding=ROUND_HALF_UP) + if credited <= 0: + raise ParamError("赎回金额过低") + + try: + if not await holdings_repo.redeem(user.id, product_id, shares): + raise ParamError("可赎回份额不足") + await account_repo.credit_balance(user.id, credited) + await db.commit() + 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, + ), + ) diff --git a/sql/schema.sql b/sql/schema.sql index f2d3f68..c78b503 100644 --- a/sql/schema.sql +++ b/sql/schema.sql @@ -59,7 +59,6 @@ CREATE TABLE IF NOT EXISTS fin_product ( expected_return DECIMAL(7,4) NULL COMMENT '预期年化收益率（%）', nav DECIMAL(12,6) NULL COMMENT '最新单位净值', nav_date DATE NULL COMMENT '净值日期', - min_amount DECIMAL(16,2) NOT NULL DEFAULT 10.00 COMMENT '起投金额（元）', fee_rate DECIMAL(6,4) NOT NULL DEFAULT 0.0000 COMMENT '申购费率（%）', term_days INT NOT NULL DEFAULT 0 COMMENT '产品期限天数，0=活期（开放申赎）', fund_manager VARCHAR(64) NULL COMMENT '基金经理', @@ -267,9 +266,14 @@ CREATE TABLE IF NOT EXISTS conversation_archive ( content MEDIUMTEXT NULL COMMENT '对话内容', tool_calls JSON NULL COMMENT '工具调用记录 [{"tool":"nl2sql",...}]', create_time DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, + trace_id VARCHAR(64) NULL COMMENT '请求链路追踪ID', KEY idx_session (session_id), KEY idx_user_time (user_id, create_time), - KEY idx_agent (agent_type) + KEY idx_agent (agent_type), + message_id VARCHAR(64) NULL COMMENT '消息唯一ID，用于归档幂等', + agent_run_id VARCHAR(64) NULL COMMENT 'Agent运行ID，用于调用链路追踪', + UNIQUE KEY uk_session_message (session_id, message_id), + KEY idx_agent_run (agent_run_id) ) COMMENT='会话归档表（审计回溯 + Agent 持续学习素材，归档前脱敏）'; -- --------------------------------------------------------------------- @@ -337,7 +341,7 @@ CREATE TABLE IF NOT EXISTS audit_log ( target VARCHAR(128) NULL COMMENT '操作对象（单号/ID）', detail TEXT NULL COMMENT '详情（JSON 字符串）', ip VARCHAR(64) NULL, - trace_id VARCHAR(32) NULL COMMENT '链路号', + trace_id VARCHAR(64) NULL COMMENT '链路号', status VARCHAR(8) NOT NULL DEFAULT '成功' COMMENT '成功/失败', create_time DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, KEY idx_user (user_id), @@ -406,7 +410,16 @@ CREATE TABLE IF NOT EXISTS memory_unit ( status VARCHAR(16) NOT NULL DEFAULT 'active' COMMENT 'active/demoted/archived/deleted', valid_until DATE NULL COMMENT '有效期', KEY idx_customer_status (customer_id, status), - KEY idx_tag (tag) + KEY idx_tag (tag), + session_id VARCHAR(64) NULL COMMENT '记忆来源会话', + agent_run_id VARCHAR(64) NULL COMMENT '产生该记忆的Agent运行ID', + evidence_ref VARCHAR(128) NULL COMMENT '证据引用', + milvus_id VARCHAR(128) NULL COMMENT 'Milvus向量ID', + graph_node_id VARCHAR(128) NULL COMMENT 'Neo4j节点或关系ID', + valid_from DATETIME NULL COMMENT '记忆生效时间', + last_verified_at DATETIME NULL COMMENT '最近确认时间', + KEY idx_customer_tag_status (customer_id, tag, status), + KEY idx_agent_run (agent_run_id); ) COMMENT='记忆单元表（三层记忆中期主体，向量镜像在 Milvus customer_memory）'; -- --------------------------------------------------------------------- @@ -437,4 +450,20 @@ CREATE TABLE IF NOT EXISTS portfolio_benchmark ( create_time DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, update_time DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, UNIQUE KEY uk_risk_level (risk_level) -) COMMENT='组合基准配置表（投顾Agent 再平衡参照，运营可调整）'; \ No newline at end of file +) COMMENT='组合基准配置表（投顾Agent 再平衡参照，运营可调整）'; + +-- --------------------------------------------------------------------- +-- 25 客户资金账户表（现金账户，一人一户） +-- --------------------------------------------------------------------- +CREATE TABLE IF NOT EXISTS fin_account ( + customer_id BIGINT UNSIGNED NOT NULL COMMENT '客户ID，一对一关联 sys_user.id', + balance DECIMAL(18,2) NOT NULL DEFAULT 0 COMMENT '账户总余额（元），含冻结部分；可用余额=balance-frozen_amount', + frozen_amount DECIMAL(18,2) NOT NULL DEFAULT 0 COMMENT '冻结金额（元），申购在途/待确认订单占用', + currency VARCHAR(8) NOT NULL DEFAULT 'CNY' COMMENT '币种，当前仅 CNY', + status VARCHAR(16) NOT NULL DEFAULT '正常' COMMENT '账户状态：正常/冻结/销户', + version INT NOT NULL DEFAULT 0 COMMENT '乐观锁版本号，扣款/入账 CAS 用', + create_time DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP, + update_time DATETIME NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, + PRIMARY KEY (customer_id), + KEY idx_status (status) +) COMMENT='客户资金账户表（现金余额，申购扣款/赎回入账的账务载体）';