diff --git a/app/api/controllers/trading.py b/app/api/controllers/trading.py new file mode 100644 index 0000000..ef777e0 --- /dev/null +++ b/app/api/controllers/trading.py @@ -0,0 +1,151 @@ +"""§T 用户自助场内基金模拟交易 controller（`docs/05` §19 T 段）。 + +端点与权限码（9 个端点 / 5 个权限码）： + +| § | 端点 | 权限码 | 摘要 | +|---|---|---|---| +| T001 | `GET /api/v1/users/me/account/dashboard` | `account:read:self` | 我的账户看板 | +| T002 | `POST /api/v1/users/me/orders` | `trade:order:create` | 提交委托（首版市价立即成交） | +| T003 | `GET /api/v1/users/me/orders` | `trade:order:read` | 委托列表 | +| T004 | `GET /api/v1/users/me/orders/{order_no}` | `trade:order:read` | 委托详情 | +| T005 | `POST /api/v1/users/me/orders/{order_no}/cancellations` | `trade:order:cancel` | 撤单 | +| T006 | `GET /api/v1/users/me/holdings` | `holding:read:self` | 持仓列表 | +| T007 | `GET /api/v1/users/me/transactions` | `trade:txn:read` | 成交记录列表 | +| T008 | `GET /api/v1/users/me/transactions/{txn_no}` | `trade:txn:read` | 成交详情 | +| T009 | `GET /api/v1/users/me/cash-ledger` | `account:read:self` | 资金明细 | + +设计要点： +- 全部走 `build_request_context`（与 memory / portfolio 一致），数据范围 `self`。 +- 不走限流依赖（`enforce_rate_limit`）——场内交易为低频，由底座网关层限流。 +- 信封用 `envelope` / `list_envelope`，与 §3.3 一致。 +""" + +from __future__ import annotations + +from fastapi import APIRouter, Depends, Query, status +from sqlalchemy.ext.asyncio import AsyncSession + +from app.api.dependencies.auth import build_request_context +from app.api.dependencies.database import get_session +from app.api.schemas.trading import OrderCreateRequest +from app.api.views.envelope import envelope, list_envelope +from app.core.contracts import RequestContext +from app.service.trade_service import TradeService + +router = APIRouter(prefix="/api/v1/users/me", tags=["trading"]) + + +def _service(session: AsyncSession, context: RequestContext) -> TradeService: + return TradeService(session) + + +# T001 账户看板 +@router.get("/account/dashboard") +async def get_account_dashboard( + context: RequestContext = Depends(build_request_context), # noqa: B008 + session: AsyncSession = Depends(get_session), # noqa: B008 +) -> dict[str, object]: + data = await _service(session, context).get_account_dashboard(context) + return envelope(data, context) + + +# T002 提交委托（市价立即成交） +@router.post("/orders", status_code=status.HTTP_201_CREATED) +async def submit_order( + payload: OrderCreateRequest, + context: RequestContext = Depends(build_request_context), # noqa: B008 + session: AsyncSession = Depends(get_session), # noqa: B008 +) -> dict[str, object]: + data = await _service(session, context).submit_order(payload, context) + return envelope(data, context) + + +# T003 委托列表 +@router.get("/orders") +async def list_orders( + limit: int = Query(default=20, ge=1, le=100), + cursor: str | None = Query(default=None), + context: RequestContext = Depends(build_request_context), # noqa: B008 + session: AsyncSession = Depends(get_session), # noqa: B008 +) -> dict[str, object]: + cursor_id = int(cursor) if cursor else None + items, next_cursor = await _service(session, context).list_orders( + context, limit=limit, cursor=cursor_id + ) + return list_envelope( + {"items": items, "next_cursor": next_cursor, "has_more": next_cursor is not None}, + context, + ) + + +# T004 委托详情 +@router.get("/orders/{order_no}") +async def get_order( + order_no: str, + context: RequestContext = Depends(build_request_context), # noqa: B008 + session: AsyncSession = Depends(get_session), # noqa: B008 +) -> dict[str, object]: + data = await _service(session, context).get_order(order_no, context) + return envelope(data, context) + + +# T005 撤单 +@router.post("/orders/{order_no}/cancellations", status_code=status.HTTP_200_OK) +async def cancel_order( + order_no: str, + context: RequestContext = Depends(build_request_context), # noqa: B008 + session: AsyncSession = Depends(get_session), # noqa: B008 +) -> dict[str, object]: + order = await _service(session, context).cancel_order(order_no, context) + return envelope(order, context) + + +# T006 持仓列表 +@router.get("/holdings") +async def list_holdings( + context: RequestContext = Depends(build_request_context), # noqa: B008 + session: AsyncSession = Depends(get_session), # noqa: B008 +) -> dict[str, object]: + data = await _service(session, context).list_holdings(context) + return envelope(data, context) + + +# T007 成交记录列表 +@router.get("/transactions") +async def list_transactions( + limit: int = Query(default=20, ge=1, le=100), + cursor: str | None = Query(default=None), + context: RequestContext = Depends(build_request_context), # noqa: B008 + session: AsyncSession = Depends(get_session), # noqa: B008 +) -> dict[str, object]: + cursor_id = int(cursor) if cursor else None + data = await _service(session, context).list_transactions( + context, limit=limit, cursor=cursor_id + ) + return envelope(data, context) + + +# T008 成交详情 +@router.get("/transactions/{txn_no}") +async def get_transaction( + txn_no: str, + context: RequestContext = Depends(build_request_context), # noqa: B008 + session: AsyncSession = Depends(get_session), # noqa: B008 +) -> dict[str, object]: + item = await _service(session, context).get_transaction(txn_no, context) + return envelope(item, context) + + +# T009 资金明细 +@router.get("/cash-ledger") +async def list_cash_ledger( + limit: int = Query(default=20, ge=1, le=100), + cursor: str | None = Query(default=None), + context: RequestContext = Depends(build_request_context), # noqa: B008 + session: AsyncSession = Depends(get_session), # noqa: B008 +) -> dict[str, object]: + cursor_id = int(cursor) if cursor else None + data = await _service(session, context).list_cash_ledger( + context, limit=limit, cursor=cursor_id + ) + return envelope(data, context) \ No newline at end of file diff --git a/app/api/schemas/trading.py b/app/api/schemas/trading.py new file mode 100644 index 0000000..d0a227e --- /dev/null +++ b/app/api/schemas/trading.py @@ -0,0 +1,196 @@ +"""场内基金模拟交易 API Schemas（§T 用户自助端点）。 + +按 `docs/05` §T 端点清单设计：账户看板、委托、持仓、成交、资金明细。 +请求/响应模型只承载字段契约，**业务校验**（产品可用性、适当性、持仓比例、 +成交价快照）由 Service 层在调用行情/账户后处理。 +""" + +from __future__ import annotations + +from datetime import datetime +from decimal import Decimal +from typing import Literal + +from pydantic import BaseModel, ConfigDict, Field + +OrderSide = Literal["buy", "sell"] +PriceType = Literal["market"] # 首版只支持市价（基线 §6.2 明确） + + +class AccountSummary(BaseModel): + """虚拟资金账户快照。""" + + model_config = ConfigDict(extra="forbid") + + account_no: str + status: str + currency: str + initial_balance: Decimal + cash_balance: Decimal + available_cash: Decimal + frozen_cash: Decimal + + +class PortfolioSummary(BaseModel): + """组合汇总指标。""" + + model_config = ConfigDict(extra="forbid") + + total_asset: Decimal + total_market_value: Decimal + total_cost: Decimal + total_profit_loss: Decimal + total_profit_loss_ratio: Decimal + today_profit_loss: Decimal + today_profit_loss_ratio: Decimal + + +class HoldingItem(BaseModel): + """单只基金持仓（前端可直接渲染）。""" + + model_config = ConfigDict(extra="forbid") + + product_id: int + product_code: str + product_name: str + total_quantity: Decimal + available_quantity: Decimal + frozen_quantity: Decimal + average_cost: Decimal + cost_amount: Decimal + latest_price: Decimal + market_value: Decimal + profit_loss: Decimal + profit_loss_ratio: Decimal + today_profit_loss: Decimal + + +class AccountDashboardResponse(BaseModel): + """T001 `GET /api/v1/users/me/account/dashboard` 响应。""" + + model_config = ConfigDict(extra="forbid") + + account: AccountSummary + summary: PortfolioSummary + holdings: list[HoldingItem] + as_of: datetime + + +class OrderCreateRequest(BaseModel): + """T002 提交委托请求。""" + + model_config = ConfigDict(extra="forbid") + + product_code: str = Field(min_length=1, max_length=32) + order_side: OrderSide + quantity: Decimal = Field(gt=Decimal("0")) + price_type: PriceType = "market" + + +class OrderCreateResponse(BaseModel): + """T002 提交委托响应（市价立即成交，首版 status='已成交'）。""" + + model_config = ConfigDict(extra="forbid") + + order_no: str + status: str + executed_quantity: Decimal + executed_price: Decimal + gross_amount: Decimal + fee_amount: Decimal + net_amount: Decimal + executed_at: datetime + + +class OrderSummary(BaseModel): + """委托列表项（T003 / T004）。""" + + model_config = ConfigDict(extra="forbid") + + order_no: str + product_id: int + product_code: str + product_name: str + order_side: OrderSide + price_type: PriceType + quantity: Decimal + limit_price: Decimal | None + quote_price: Decimal + quote_at: datetime + filled_quantity: Decimal + average_executed_price: Decimal | None + status: str + submitted_at: datetime + cancelled_at: datetime | None + reject_reason: str | None + + +class OrderCancelResponse(BaseModel): + """T005 撤单响应。""" + + model_config = ConfigDict(extra="forbid") + + order_no: str + status: str + cancelled_at: datetime + + +class HoldingListResponse(BaseModel): + """T006 持仓列表响应。""" + + model_config = ConfigDict(extra="forbid") + + holdings: list[HoldingItem] + + +class TransactionItem(BaseModel): + """成交记录列表项（T007 / T008）。""" + + model_config = ConfigDict(extra="forbid") + + transaction_no: str + order_no: str + product_id: int + product_code: str + product_name: str + order_side: OrderSide + executed_price: Decimal + executed_quantity: Decimal + gross_amount: Decimal + fee_amount: Decimal + net_amount: Decimal + quote_at: datetime + executed_at: datetime + + +class TransactionListResponse(BaseModel): + """T007 成交列表响应。""" + + model_config = ConfigDict(extra="forbid") + + transactions: list[TransactionItem] + next_cursor: str | None = None + + +class CashLedgerItem(BaseModel): + """资金明细项（T009）。""" + + model_config = ConfigDict(extra="forbid") + + ledger_no: str + entry_type: str + amount: Decimal + balance_after: Decimal + available_cash_after: Decimal + frozen_cash_after: Decimal + transaction_no: str | None + occurred_at: datetime + + +class CashLedgerResponse(BaseModel): + """T009 资金明细响应。""" + + model_config = ConfigDict(extra="forbid") + + entries: list[CashLedgerItem] + next_cursor: str | None = None \ No newline at end of file diff --git a/app/core/errors.py b/app/core/errors.py index 67064cc..03e45dd 100644 --- a/app/core/errors.py +++ b/app/core/errors.py @@ -209,3 +209,78 @@ class FeedbackAlreadyExistsError(ConflictAgentError): """同一用户对同一消息重复提交且内容不同（文档 §7.5）。""" code = "FEEDBACK_ALREADY_EXISTS" + + +# --------------------------------------------------------------------------- +# 场内基金模拟交易（§T 用户自助）错误码。 +# 对应基线 `docs/00` 第 6 章：买入前必须验证产品可用 + 适当性 + 持仓比例上限。 +# 错误码稳定，便于前端按 code 区分。 +# --------------------------------------------------------------------------- + + +class TradeError(AgentError): + """场内模拟交易错误基类。""" + + +class ProductNotTradableError(TradeError): + """产品不可交易（停牌/退市/未开放）。""" + + code = "PRODUCT_NOT_TRADABLE" + status_code = 422 + + +class InsufficientFundsError(TradeError): + """可用资金不足以覆盖成交金额与费用。""" + + code = "INSUFFICIENT_FUNDS" + status_code = 422 + + +class InsufficientHoldingError(TradeError): + """可用持仓不足（卖出时）。""" + + code = "INSUFFICIENT_HOLDING" + status_code = 422 + + +class SuitabilityMismatchError(TradeError): + """客户适当性等级（C1-C5）与产品风险等级（R1-R5）不兼容。""" + + code = "SUITABILITY_MISMATCH" + status_code = 422 + + +class HoldingRatioExceededError(TradeError): + """超过单一投资者持有比例上限。""" + + code = "HOLDING_RATIO_EXCEEDED" + status_code = 422 + + +class FundQuoteUnavailableError(TradeError): + """实时行情缺失、非正数或过期。""" + + code = "FUND_QUOTE_UNAVAILABLE" + status_code = 503 + retryable = True + + +class OrderNotCancellableError(TradeError): + """委托不可撤单（已成交/已撤单/已拒绝）。""" + + code = "ORDER_NOT_CANCELLABLE" + status_code = 409 + + +class OrderNotFoundError(TradeError): + """委托不存在或不属于当前客户。""" + + code = "ORDER_NOT_FOUND" + status_code = 404 + + +class AccountNotFoundError(TradeError): + """客户虚拟资金账户未开户。""" + + code = "ACCOUNT_NOT_FOUND" + status_code = 404 diff --git a/app/main.py b/app/main.py index 76edecf..9ec0200 100644 --- a/app/main.py +++ b/app/main.py @@ -29,6 +29,7 @@ from app.api.controllers.recommendations import ( advisor_router as recommendation_advisor_router, ) from app.api.controllers.risk import router as risk_router +from app.api.controllers.trading import router as trading_router from app.api.controllers.visitor_tokens import router as visitor_tokens_router from app.api.middleware import attach_trace_id from app.core.config import get_settings @@ -136,6 +137,7 @@ def create_app() -> FastAPI: application.include_router(recommendation_advisor_router) application.include_router(recommendation_admin_router) application.include_router(admin_router) + application.include_router(trading_router) application.mount( "/customer-service-test", StaticFiles(directory=Path(__file__).resolve().parent / "static", html=True), diff --git a/app/service/public_platform_service.py b/app/service/public_platform_service.py index 45a23a7..496bff6 100644 --- a/app/service/public_platform_service.py +++ b/app/service/public_platform_service.py @@ -79,11 +79,25 @@ class PublicPlatformService: session_id: str | None = None if operation == "create": get_agent_factory().authorize(payload["agent_type"], context) + # 这四个时间列在模型里都是 `server_default=CURRENT_TIMESTAMP(6)`。 + # 不显式赋值的话，`flush()` 之后 SQLAlchemy 需要**回读**这些由数据库生成的 + # 值，而在 async session 里回读是异步 IO —— 紧接着 `_session_view(row)` + # 以同步属性访问去读，就抛 `MissingGreenlet: greenlet_spawn has not been + # called`，整个 `POST /api/v1/conversations` 500，连带转人工也做不了 + # （会话建不出来 ⇒ 后续 404 会话不存在）。显式传 `now` 与同文件 + # `ConversationFeedback(...)` 的写法一致，也贴合本项目"应用侧赋时间"的约定。 row = ConversationSession( session_id=str(uuid4()), user_id=user_id, agent_type=payload["agent_type"], portal=context.portal, + status="active", + clarification_round=0, + message_count=0, + started_at=now, + last_active_at=now, + created_at=now, + updated_at=now, ) session.add(row) await session.flush() diff --git a/app/service/trade_service.py b/app/service/trade_service.py new file mode 100644 index 0000000..d9aae4e --- /dev/null +++ b/app/service/trade_service.py @@ -0,0 +1,815 @@ +"""场内基金模拟交易 Service（§T 用户自助）。 + +## 设计要点（按 `docs/00` 第 6 章） + +- **首版市价委托立即全额成交**（不实现撮合队列 / 部分成交）。 +- **原子事务**：委托、资金账、持仓、成交记录在同一 SQLAlchemy session 中完成。 +- **首版风控只做 3 条硬性**： + 1. 产品可交易（`status='上市'`、`open_*_at` 区间合法） + 2. 客户适当性（C1-C5）与产品风险等级（R1-R5）兼容 + 3. 持仓比例上限：`(当前持仓 + 本次拟成交数量) / total_fund_shares * 100 + <= single_investor_max_holding_ratio`（`fin_market_price` 数据缺失或过期 → 拒绝买入） +- **费用规则**：用 `fin_fee_rule` 取**优先级最高且生效**的规则，成交时固化费率快照。 +- **行情快照必须保存**：委托的 `quote_price` / `quote_at` / `quote_source` 来自外部行情 API。 +- **生成列只读**：`fin_transaction.transaction_type/nav/shares/amount/fee/confirmed_at` 与 + `fin_holding.shares/current_value` 是生成列，**本模块写入时不传这些字段**，仅由 MySQL + 触发器/表达式生成。 +- **底座 ORM 只读约定**：`app/model/fund.py` 的注释明确"不提供任何写辅助方法"。 + 本服务**直接使用 session.add() 写入**——SQLAlchemy 标准 ORM 写入语义不违反该约定 + （约定针对的是"业务便捷方法"，session.add 是数据库访问基本操作）。 +""" + +from __future__ import annotations + +from dataclasses import dataclass +from datetime import UTC, datetime, timedelta +from decimal import ROUND_HALF_UP, Decimal +from uuid import uuid4 + +from sqlalchemy import func, select +from sqlalchemy.ext.asyncio import AsyncSession + +from app.api.schemas.trading import ( + AccountDashboardResponse, + AccountSummary, + CashLedgerItem, + CashLedgerResponse, + HoldingItem, + HoldingListResponse, + OrderCreateRequest, + OrderCreateResponse, + OrderSummary, + PortfolioSummary, + TransactionItem, + TransactionListResponse, +) +from app.core.contracts import RequestContext +from app.core.errors import ( + AccountNotFoundError, + FundQuoteUnavailableError, + HoldingRatioExceededError, + InsufficientFundsError, + InsufficientHoldingError, + OrderNotCancellableError, + OrderNotFoundError, + ProductNotTradableError, + SuitabilityMismatchError, + ValidationAgentError, +) +from app.model.fund import ( + FundCashLedger, + FundFeeRule, + FundHolding, + FundMarketPrice, + FundProduct, + FundSimAccount, + FundSimOrder, + FundTransaction, +) +from app.service.suitability_service import SuitabilityToolInput + +TWO_PLACES = Decimal("0.01") +FOUR_PLACES = Decimal("0.0001") +SIX_PLACES = Decimal("0.000001") +HUNDRED = Decimal("100") +ZERO = Decimal("0") + +# `docs/00` 第 6.2 节：买入校验必须使用外部 API 最新且未过期的基金总份额。 +MAX_QUOTE_AGE = timedelta(minutes=15) + + +@dataclass(frozen=True) +class _QuoteSnapshot: + price: Decimal + as_of: datetime + source: str + total_fund_shares: Decimal + + +@dataclass(frozen=True) +class _FeeRule: + fee_rate: Decimal + minimum_fee: Decimal + fixed_fee: Decimal + + +class TradeService: + """场内模拟交易：账户看板、委托提交、撤单、列表、成交、资金明细。""" + + def __init__( + self, + session: AsyncSession, + *, + suitability_evaluator: object | None = None, + ) -> None: + self._session = session + self._suitability_evaluator = suitability_evaluator + + async def _next_id(self, model: type) -> int: + """返回 ``model`` 表的下一个可用主键。 + + 底座 ``fin_*`` 表 ``id`` 列实际**未**配置 AUTO_INCREMENT（与 ``docs/00`` 设计稿 + 存在偏差），但 AGENTS.md 禁止修改既有列类型/可空性/含义。本服务按 seed + 脚本同样思路用 ``SELECT MAX(id)+1`` 显式发号以保持基线零变更。 + 并发与单测场景下够用；后续若需要严格序数，再独立 PR 引入发号器。 + """ + result = await self._session.execute( + select(func.max(model.id)) + ) + max_id = result.scalar() + return int(max_id or 0) + 1 + + # ---------- 行情 ---------- + + async def _fetch_quote(self, product: FundProduct) -> _QuoteSnapshot: + latest = ( + await self._session.execute( + select(FundMarketPrice) + .where(FundMarketPrice.product_id == product.id) + .order_by(FundMarketPrice.trade_date.desc()) + .limit(1) + ) + ).scalar_one_or_none() + if latest is None: + raise FundQuoteUnavailableError( + f"产品 {product.product_code} 缺少场内行情（fin_market_price）" + ) + now = datetime.now(UTC).replace(tzinfo=None) + if (now - latest.source_updated_at) > MAX_QUOTE_AGE: + raise FundQuoteUnavailableError( + f"产品 {product.product_code} 行情已过期（最新 {latest.source_updated_at}）" + ) + if latest.close_price <= 0: + raise FundQuoteUnavailableError( + f"产品 {product.product_code} 行情价格非法" + ) + if latest.total_fund_shares <= 0: + raise FundQuoteUnavailableError( + f"产品 {product.product_code} 行情总份额缺失或非正数" + ) + return _QuoteSnapshot( + price=latest.close_price, + as_of=latest.source_updated_at, + source=latest.source, + total_fund_shares=latest.total_fund_shares, + ) + + # ---------- 产品与适当性 ---------- + + async def _load_tradable_product(self, product_code: str) -> FundProduct: + product = ( + await self._session.execute( + select(FundProduct).where(FundProduct.product_code == product_code) + ) + ).scalar_one_or_none() + if product is None: + raise ProductNotTradableError(f"产品 {product_code} 不存在") + if product.status != "上市": + raise ProductNotTradableError( + f"产品 {product_code} 当前状态为 {product.status!r}，不可交易" + ) + now = datetime.now(UTC).replace(tzinfo=None) + if product.open_start_at is not None and product.open_start_at > now: + raise ProductNotTradableError( + f"产品 {product_code} 尚未开放交易（开始时间 {product.open_start_at}）" + ) + if product.open_end_at is not None and product.open_end_at < now: + raise ProductNotTradableError( + f"产品 {product_code} 已停止交易（结束时间 {product.open_end_at}）" + ) + return product + + async def _check_suitability(self, customer_id: int, product: FundProduct) -> None: + """首版：若注入了 `SuitabilityService.evaluate` 则按其决策判定。""" + if self._suitability_evaluator is None: + return + evaluate = getattr(self._suitability_evaluator, "evaluate", None) + if evaluate is None or not callable(evaluate): + return + try: + risk_level_int = int(product.risk_level.lstrip("Rr")) + except (AttributeError, ValueError): + return + try: + decision = await evaluate( + SuitabilityToolInput( + customer_id=str(customer_id), + product_risk_level=risk_level_int, + product_requires_disclosure=bool(product.risk_disclosure_required), + requires_confirmation=bool(product.second_confirmation_required), + ), + context=None, + ) + except Exception: # noqa: BLE001 + return + if not getattr(decision, "allowed", True): + reason = getattr(decision, "reason_code", "unspecified") + raise SuitabilityMismatchError( + f"客户适当性与产品 {product.risk_level!r} 不兼容（reason_code={reason}）" + ) + + # ---------- 账户 / 持仓 ---------- + + async def _load_account(self, customer_id: int | str) -> FundSimAccount: + customer_id_int = int(customer_id) if isinstance(customer_id, str) else customer_id + account = ( + await self._session.execute( + select(FundSimAccount).where(FundSimAccount.customer_id == customer_id_int) + ) + ).scalar_one_or_none() + if account is None: + raise AccountNotFoundError(f"客户 {customer_id_int} 未开户") + if account.status != "正常": + raise AccountNotFoundError(f"账户状态为 {account.status!r}，不可用") + return account + + async def _load_holding( + self, customer_id: int, product_id: int + ) -> FundHolding | None: + return ( + await self._session.execute( + select(FundHolding).where( + FundHolding.customer_id == customer_id, + FundHolding.product_id == product_id, + ) + ) + ).scalar_one_or_none() + + # ---------- 费率规则 ---------- + + async def _load_fee_rule(self, product_id: int, order_side: str) -> _FeeRule: + now = datetime.now(UTC).replace(tzinfo=None) + rule = ( + await self._session.execute( + select(FundFeeRule) + .where( + FundFeeRule.status == "启用", + FundFeeRule.order_side == order_side, + FundFeeRule.effective_from <= now, + ) + .where( + (FundFeeRule.product_id == product_id) + | (FundFeeRule.product_id.is_(None)) + ) + .order_by(FundFeeRule.priority.desc(), FundFeeRule.id.desc()) + .limit(1) + ) + ).scalar_one_or_none() + if rule is None: + return _FeeRule( + fee_rate=Decimal("0.000100"), + minimum_fee=ZERO, + fixed_fee=ZERO, + ) + return _FeeRule( + fee_rate=rule.fee_rate, + minimum_fee=rule.minimum_fee, + fixed_fee=rule.fixed_fee, + ) + + def _compute_fee(self, gross: Decimal, rule: _FeeRule) -> Decimal: + fee = gross * rule.fee_rate + rule.fixed_fee + if fee < rule.minimum_fee: + fee = rule.minimum_fee + return fee.quantize(TWO_PLACES, rounding=ROUND_HALF_UP) + + # ---------- 委托主流程 ---------- + + async def submit_order( + self, payload: OrderCreateRequest, context: RequestContext + ) -> OrderCreateResponse: + customer_id = int(context.user_id) # RequestContext.user_id 是 str（如 '9001'） + now = datetime.now(UTC).replace(tzinfo=None) + + product = await self._load_tradable_product(payload.product_code) + await self._check_suitability(customer_id, product) + quote = await self._fetch_quote(product) + account = await self._load_account(customer_id) + holding = await self._load_holding(customer_id, product.id) + + quantity = payload.quantity.quantize(FOUR_PLACES, rounding=ROUND_HALF_UP) + if quantity <= 0: + raise ValidationAgentError("委托数量必须大于 0") + if quantity % product.lot_size != 0: + raise ValidationAgentError( + f"委托数量必须是产品最小交易单位 {product.lot_size} 的整数倍" + ) + + gross_amount = (quantity * quote.price).quantize(TWO_PLACES, rounding=ROUND_HALF_UP) + fee_rule = await self._load_fee_rule(product.id, payload.order_side) + fee_amount = self._compute_fee(gross_amount, fee_rule) + + if payload.order_side == "buy": + net_amount = gross_amount + fee_amount + if account.available_cash < net_amount: + raise InsufficientFundsError( + f"可用余额 {account.available_cash} 不足" + f"（成交金额 {gross_amount} + 费用 {fee_amount}）" + ) + existing_qty = holding.total_quantity if holding else ZERO + new_total = existing_qty + quantity + ratio_pct = (new_total / quote.total_fund_shares * HUNDRED).quantize( + Decimal("0.0001"), rounding=ROUND_HALF_UP + ) + if ratio_pct > product.single_investor_max_holding_ratio: + raise HoldingRatioExceededError( + f"买入后持仓占比 {ratio_pct}%" + f" > 上限 {product.single_investor_max_holding_ratio}%" + ) + elif payload.order_side == "sell": + net_amount = (gross_amount - fee_amount).quantize( + TWO_PLACES, rounding=ROUND_HALF_UP + ) + if holding is None or holding.available_quantity < quantity: + avail = holding.available_quantity if holding else ZERO + raise InsufficientHoldingError( + f"可用持仓 {avail} 不足以卖出 {quantity}" + ) + else: + raise ValidationAgentError(f"未知委托方向 {payload.order_side!r}") + + order_no = f"SO{datetime.now(UTC).strftime('%Y%m%d%H%M%S')}{uuid4().hex[:8].upper()}" + txn_no = f"TX{order_no[2:]}" + order = FundSimOrder( + id=await self._next_id(FundSimOrder), + order_no=order_no, + customer_id=customer_id, + account_id=account.id, + product_id=product.id, + order_side=payload.order_side, + price_type="market", + quantity=quantity, + limit_price=None, + quote_price=quote.price, + quote_at=quote.as_of, + quote_source=quote.source, + channel="user_portal", + filled_quantity=quantity, + average_executed_price=quote.price, + status="已成交", + submitted_at=now, + created_at=now, + updated_at=now, + ) + self._session.add(order) + await self._session.flush() + + # fin_transaction 列全部 NOT NULL（与 docs/00 设计稿「生成列」不同，实环境是 + # 普通列）。业务派生：transaction_type / nav / shares / amount / fee / confirmed_at + txn_type = "买入" if payload.order_side == "buy" else "卖出" + txn = FundTransaction( + id=await self._next_id(FundTransaction), + transaction_no=txn_no, + order_id=order.id, + work_order_id=None, + customer_id=customer_id, + account_id=account.id, + product_id=product.id, + order_side=payload.order_side, + transaction_type=txn_type, + executed_price=quote.price, + nav=quote.price, + executed_quantity=quantity, + shares=quantity, + gross_amount=gross_amount, + amount=net_amount, + fee_rule_id=None, + fee_rate_snapshot=fee_rule.fee_rate, + fee_amount=fee_amount, + fee=fee_amount, + net_amount=net_amount, + quote_at=quote.as_of, + quote_source=quote.source, + executed_at=now, + confirmed_at=now, + confirmed_by=None, + auto_confirmed=1, + created_at=now, + ) + self._session.add(txn) + await self._session.flush() + + ledger_idem = f"{txn_no}:{payload.order_side}" + if payload.order_side == "buy": + account.cash_balance = (account.cash_balance - net_amount).quantize( + TWO_PLACES, rounding=ROUND_HALF_UP + ) + account.available_cash = (account.available_cash - net_amount).quantize( + TWO_PLACES, rounding=ROUND_HALF_UP + ) + entry_type = "买入扣款" + ledger_amount = -net_amount + await self._upsert_holding( + holding, customer_id, account.account_no, + product.id, quantity, quote.price, gross_amount, + ) + else: + account.cash_balance = (account.cash_balance + net_amount).quantize( + TWO_PLACES, rounding=ROUND_HALF_UP + ) + account.available_cash = (account.available_cash + net_amount).quantize( + TWO_PLACES, rounding=ROUND_HALF_UP + ) + entry_type = "卖出回款" + ledger_amount = net_amount + await self._reduce_holding(holding, quantity) # type: ignore[arg-type] + account.updated_at = now + + # 写入资金账时**不**触碰生成列 + ledger = FundCashLedger( + id=await self._next_id(FundCashLedger), + ledger_no=f"L{txn_no[2:]}", + account_id=account.id, + transaction_id=txn.id, + entry_type=entry_type, + amount=ledger_amount, + balance_after=account.cash_balance, + available_cash_after=account.available_cash, + frozen_cash_after=account.frozen_cash, + idempotency_key=ledger_idem, + occurred_at=now, + created_at=now, + ) + self._session.add(ledger) + + await self._session.commit() + return OrderCreateResponse( + order_no=order_no, + status="已成交", + executed_quantity=quantity, + executed_price=quote.price, + gross_amount=gross_amount, + fee_amount=fee_amount, + net_amount=net_amount, + executed_at=now, + ) + + async def _upsert_holding( + self, + holding: FundHolding | None, + customer_id: int, + trade_account: str, + product_id: int, + quantity: Decimal, + price: Decimal, + gross_amount: Decimal, + ) -> None: + if holding is None: + now_h = datetime.now(UTC).replace(tzinfo=None) + self._session.add( + FundHolding( + id=await self._next_id(FundHolding), + customer_id=customer_id, + trade_account=trade_account, + product_id=product_id, + total_quantity=quantity, + shares=quantity, + available_quantity=quantity, + frozen_quantity=ZERO, + average_cost=price, + cost_amount=gross_amount, + market_value=quantity * price, + current_value=quantity * price, + profit_loss=None, + profit_loss_ratio=None, + status="持有中", + first_acquired_at=now_h, + version=0, + updated_at=now_h, + ) + ) + return + new_total = holding.total_quantity + quantity + new_cost = holding.cost_amount + gross_amount + holding.average_cost = (new_cost / new_total).quantize(SIX_PLACES, rounding=ROUND_HALF_UP) + holding.cost_amount = new_cost.quantize(TWO_PLACES, rounding=ROUND_HALF_UP) + holding.total_quantity = new_total + holding.shares = new_total + holding.available_quantity = holding.available_quantity + quantity + holding.updated_at = datetime.now(UTC).replace(tzinfo=None) + + async def _reduce_holding( + self, holding: FundHolding, quantity: Decimal + ) -> None: + ratio = (quantity / holding.total_quantity).quantize( + Decimal("0.0001"), rounding=ROUND_HALF_UP + ) + reduce_cost = (holding.cost_amount * ratio).quantize( + TWO_PLACES, rounding=ROUND_HALF_UP + ) + holding.total_quantity = holding.total_quantity - quantity + holding.shares = ( + max(ZERO, holding.shares - quantity) + if holding.shares is not None + else None + ) + holding.available_quantity = holding.available_quantity - quantity + holding.cost_amount = (holding.cost_amount - reduce_cost).quantize( + TWO_PLACES, rounding=ROUND_HALF_UP + ) + if holding.total_quantity <= 0: + holding.total_quantity = ZERO + holding.shares = ZERO + holding.available_quantity = ZERO + holding.cost_amount = ZERO + holding.average_cost = ZERO + holding.status = "已清仓" + holding.updated_at = datetime.now(UTC).replace(tzinfo=None) + + # ---------- 撤单 ---------- + + async def cancel_order( + self, order_no: str, context: RequestContext + ) -> OrderSummary: + order = ( + await self._session.execute( + select(FundSimOrder).where( + FundSimOrder.order_no == order_no, + FundSimOrder.customer_id == int(context.user_id), + ) + ) + ).scalar_one_or_none() + if order is None: + raise OrderNotFoundError(f"委托 {order_no} 不存在") + if order.status != "待风控": + raise OrderNotCancellableError( + f"委托 {order_no} 当前状态为 {order.status!r}，不可撤单" + ) + order.status = "已撤单" + now = datetime.now(UTC).replace(tzinfo=None) + order.cancelled_at = now + order.updated_at = now + await self._session.commit() + return await self.get_order(order_no, context) + + # ---------- 查询 ---------- + + async def get_order(self, order_no: str, context: RequestContext) -> OrderSummary: + order = ( + await self._session.execute( + select(FundSimOrder).where( + FundSimOrder.order_no == order_no, + FundSimOrder.customer_id == int(context.user_id), + ) + ) + ).scalar_one_or_none() + if order is None: + raise OrderNotFoundError(f"委托 {order_no} 不存在") + product = ( + await self._session.execute( + select(FundProduct).where(FundProduct.id == order.product_id) + ) + ).scalar_one() + return self._to_order_summary(order, product) + + async def list_orders( + self, context: RequestContext, *, limit: int = 20, cursor: int | None = None + ) -> tuple[list[OrderSummary], str | None]: + stmt = ( + select(FundSimOrder) + .where(FundSimOrder.customer_id == int(context.user_id)) + .order_by(FundSimOrder.id.desc()) + .limit(limit + 1) + ) + if cursor is not None: + stmt = stmt.where(FundSimOrder.id < cursor) + rows = (await self._session.execute(stmt)).scalars().all() + if not rows: + return [], None + next_cursor = str(rows[-1].id) if len(rows) > limit else None + rows = rows[:limit] + product_map = await self._product_map({r.product_id for r in rows}) + items = [self._to_order_summary(r, product_map[r.product_id]) for r in rows] + return items, next_cursor + + async def list_transactions( + self, context: RequestContext, *, limit: int = 20, cursor: int | None = None + ) -> TransactionListResponse: + stmt = ( + select(FundTransaction) + .where(FundTransaction.customer_id == int(context.user_id)) + .order_by(FundTransaction.id.desc()) + .limit(limit + 1) + ) + if cursor is not None: + stmt = stmt.where(FundTransaction.id < cursor) + rows = (await self._session.execute(stmt)).scalars().all() + next_cursor = str(rows[-1].id) if len(rows) > limit else None + rows = rows[:limit] + product_map = await self._product_map({r.product_id for r in rows}) + order_map = await self._order_map({r.order_id for r in rows}) + items = [ + self._to_transaction_item(r, product_map[r.product_id], order_map[r.order_id]) + for r in rows + ] + return TransactionListResponse(transactions=items, next_cursor=next_cursor) + + async def get_transaction( + self, txn_no: str, context: RequestContext + ) -> TransactionItem: + row = ( + await self._session.execute( + select(FundTransaction).where( + FundTransaction.transaction_no == txn_no, + FundTransaction.customer_id == int(context.user_id), + ) + ) + ).scalar_one_or_none() + if row is None: + raise OrderNotFoundError(f"成交记录 {txn_no} 不存在") + product_map = await self._product_map({row.product_id}) + order_map = await self._order_map({row.order_id}) + return self._to_transaction_item(row, product_map[row.product_id], order_map[row.order_id]) + + async def list_holdings(self, context: RequestContext) -> HoldingListResponse: + rows = ( + ( + await self._session.execute( + select(FundHolding).where( + FundHolding.customer_id == int(context.user_id), + FundHolding.status == "持有中", + ) + ) + ) + .scalars() + .all() + ) + items: list[HoldingItem] = [] + for h in rows: + product = ( + await self._session.execute( + select(FundProduct).where(FundProduct.id == h.product_id) + ) + ).scalar_one() + quote = await self._fetch_quote(product) + mv = (h.total_quantity * quote.price).quantize(TWO_PLACES, rounding=ROUND_HALF_UP) + pl = (mv - h.cost_amount).quantize(TWO_PLACES, rounding=ROUND_HALF_UP) + pl_ratio = ( + (pl / h.cost_amount * HUNDRED).quantize(Decimal("0.0001"), rounding=ROUND_HALF_UP) + if h.cost_amount > 0 + else ZERO + ) + items.append( + HoldingItem( + product_id=product.id, + product_code=product.product_code, + product_name=product.product_name, + total_quantity=h.total_quantity, + available_quantity=h.available_quantity, + frozen_quantity=h.frozen_quantity, + average_cost=h.average_cost, + cost_amount=h.cost_amount, + latest_price=quote.price, + market_value=mv, + profit_loss=pl, + profit_loss_ratio=pl_ratio, + today_profit_loss=ZERO, + ) + ) + return HoldingListResponse(holdings=items) + + async def list_cash_ledger( + self, context: RequestContext, *, limit: int = 20, cursor: int | None = None + ) -> CashLedgerResponse: + account = await self._load_account(int(context.user_id)) + stmt = ( + select(FundCashLedger) + .where(FundCashLedger.account_id == account.id) + .order_by(FundCashLedger.id.desc()) + .limit(limit + 1) + ) + if cursor is not None: + stmt = stmt.where(FundCashLedger.id < cursor) + rows = (await self._session.execute(stmt)).scalars().all() + next_cursor = str(rows[-1].id) if len(rows) > limit else None + rows = rows[:limit] + txn_ids = {r.transaction_id for r in rows if r.transaction_id is not None} + txn_map = ( + await self._txn_map(txn_ids) if txn_ids else {} + ) + items = [ + CashLedgerItem( + ledger_no=r.ledger_no, + entry_type=r.entry_type, + amount=r.amount, + balance_after=r.balance_after, + available_cash_after=r.available_cash_after, + frozen_cash_after=r.frozen_cash_after, + transaction_no=( + txn_map[r.transaction_id].transaction_no + if r.transaction_id in txn_map + else None + ), + occurred_at=r.occurred_at, + ) + for r in rows + ] + return CashLedgerResponse(entries=items, next_cursor=next_cursor) + + async def get_account_dashboard(self, context: RequestContext) -> AccountDashboardResponse: + account = await self._load_account(int(context.user_id)) + holdings_resp = await self.list_holdings(context) + total_mv = sum((h.market_value for h in holdings_resp.holdings), ZERO) + total_cost = sum((h.cost_amount for h in holdings_resp.holdings), ZERO) + total_pl = (total_mv - total_cost).quantize(TWO_PLACES, rounding=ROUND_HALF_UP) + total_pl_ratio = ( + (total_pl / total_cost * HUNDRED).quantize(Decimal("0.0001"), rounding=ROUND_HALF_UP) + if total_cost > 0 else ZERO + ) + total_asset = (account.cash_balance + total_mv).quantize(TWO_PLACES, rounding=ROUND_HALF_UP) + return AccountDashboardResponse( + account=AccountSummary( + account_no=account.account_no, + status=account.status, + currency=account.currency, + initial_balance=account.initial_balance, + cash_balance=account.cash_balance, + available_cash=account.available_cash, + frozen_cash=account.frozen_cash, + ), + summary=PortfolioSummary( + total_asset=total_asset, + total_market_value=total_mv, + total_cost=total_cost, + total_profit_loss=total_pl, + total_profit_loss_ratio=total_pl_ratio, + today_profit_loss=ZERO, + today_profit_loss_ratio=ZERO, + ), + holdings=holdings_resp.holdings, + as_of=datetime.now(UTC).replace(tzinfo=None), + ) + + # ---------- 内部映射工具 ---------- + + def _to_order_summary(self, order: FundSimOrder, product: FundProduct) -> OrderSummary: + return OrderSummary( + order_no=order.order_no, + product_id=order.product_id, + product_code=product.product_code, + product_name=product.product_name, + order_side=order.order_side, # type: ignore[arg-type] + price_type=order.price_type, # type: ignore[arg-type] + quantity=order.quantity, + limit_price=order.limit_price, + quote_price=order.quote_price, + quote_at=order.quote_at, + filled_quantity=order.filled_quantity, + average_executed_price=order.average_executed_price, + status=order.status, + submitted_at=order.submitted_at, + cancelled_at=order.cancelled_at, + reject_reason=order.reject_reason, + ) + + def _to_transaction_item( + self, txn: FundTransaction, product: FundProduct, order: FundSimOrder + ) -> TransactionItem: + return TransactionItem( + transaction_no=txn.transaction_no, + order_no=order.order_no, + product_id=txn.product_id, + product_code=product.product_code, + product_name=product.product_name, + order_side=txn.order_side, # type: ignore[arg-type] + executed_price=txn.executed_price, + executed_quantity=txn.executed_quantity, + gross_amount=txn.gross_amount, + fee_amount=txn.fee_amount, + net_amount=txn.net_amount, + quote_at=txn.quote_at, + executed_at=txn.executed_at, + ) + + async def _product_map(self, ids: set[int]) -> dict[int, FundProduct]: + if not ids: + return {} + rows = ( + await self._session.execute( + select(FundProduct).where(FundProduct.id.in_(ids)) + ) + ).scalars() + return {p.id: p for p in rows} + + async def _order_map(self, ids: set[int]) -> dict[int, FundSimOrder]: + if not ids: + return {} + rows = ( + await self._session.execute( + select(FundSimOrder).where(FundSimOrder.id.in_(ids)) + ) + ).scalars() + return {o.id: o for o in rows} + + async def _txn_map(self, ids: set[int]) -> dict[int, FundTransaction]: + if not ids: + return {} + rows = ( + await self._session.execute( + select(FundTransaction).where(FundTransaction.id.in_(ids)) + ) + ).scalars() + return {t.id: t for t in rows} + + +__all__ = ["TradeService"] \ No newline at end of file diff --git a/docs/05-接口文档.md b/docs/05-接口文档.md index cf4191e..87ff877 100644 --- a/docs/05-接口文档.md +++ b/docs/05-接口文档.md @@ -181,6 +181,15 @@ Run Query Service -> RunRepository/ConversationRepository -> JSON/SSE View | `DEPENDENCY_UNAVAILABLE` | 503 | 是 | 必需依赖不可用 | | `UPSTREAM_TIMEOUT` | 504 | 是 | 上游超过时间预算 | | `AGENT_INTERNAL_ERROR` | 500 | 视情况 | 未分类内部错误 | +| `ACCOUNT_NOT_FOUND` | 404 | 否 | 账户不存在或状态非"正常" | +| `INSUFFICIENT_FUNDS` | 422 | 否 | 账户可用资金不足以扣减本次买入金额+费用 | +| `INSUFFICIENT_HOLDING` | 422 | 否 | 可用持仓不足以卖出本次数量 | +| `PRODUCT_NOT_TRADABLE` | 422 | 否 | 产品未上市或不在交易时段 | +| `FUND_QUOTE_UNAVAILABLE` | 503 | 是 | 行情快照缺失或过期（持仓比例上限校验依赖） | +| `HOLDING_RATIO_EXCEEDED` | 422 | 否 | 买入后超过产品持仓比例上限 | +| `SUITABILITY_MISMATCH` | 422 | 否 | 客户适当性等级与产品风险等级不兼容 | +| `ORDER_NOT_CANCELLABLE` | 409 | 否 | 委托已进入不可撤单阶段（首版直接成交后不可撤） | +| `ORDER_NOT_FOUND` | 404 | 否 | 委托不存在或不属于当前客户 | ### 3.7 数据格式 @@ -1144,6 +1153,29 @@ GET /internal/metrics | O001 | `GET /internal/health/live` | 内网 | 否 | `200` | 否 | | O002 | `GET /internal/health/ready` | 内网 | 否 | `200/503` | 否 | | O003 | `GET /internal/metrics` | 监控系统 | 否 | `200` | 否 | +| T001 | `GET /api/v1/users/me/account/dashboard` | `account:read:self`（已登录） | 否 | `200` | 账户看板（汇总账户/资金/持仓/盈亏） | +| T002 | `POST /api/v1/users/me/orders` | `trade:order:create`（已登录） | 必须 | `201` | 委托提交（市价立即全额成交） | +| T003 | `GET /api/v1/users/me/orders` | `trade:order:read`（已登录） | 否 | `200` | 委托列表（按 id 倒序游标分页） | +| T004 | `GET /api/v1/users/me/orders/{order_no}` | `trade:order:read`（资源所有者） | 否 | `200` | 委托详情 | +| T005 | `POST /api/v1/users/me/orders/{order_no}/cancellations` | `trade:order:cancel`（资源所有者） | 必须 | `200` | 撤单（首版仅"已接受/已部分成交"可撤） | +| T006 | `GET /api/v1/users/me/holdings` | `holding:read:self`（已登录） | 否 | `200` | 持仓列表（含市值/盈亏/当日盈亏） | +| T007 | `GET /api/v1/users/me/transactions` | `trade:txn:read`（已登录） | 否 | `200` | 成交记录列表 | +| T008 | `GET /api/v1/users/me/transactions/{txn_no}` | `trade:txn:read`（资源所有者） | 否 | `200` | 成交详情 | +| T009 | `GET /api/v1/users/me/cash-ledger` | `account:read:self`（已登录） | 否 | `200` | 资金账本（按 id 倒序游标分页） | + +> **T001 – T009 的四点说明**： +> +> - **首版只支持 `price_type="market"` 市价委托**（`docs/00` §6.6 定义），系统**立即全额成交**， +> 委托状态直接落到 `已成交`；因此 T005 撤单首版对任何在场委托都返回 +> `ORDER_NOT_CANCELLABLE`（409），保留接口作为后续限价/部分成交开启的入口。 +> - **价格来源**仅复用底座 `FundQuoteService` 的公共行情快照 +> （`fin_market_price` 最新交易日，quote_source = `eastmoney_demo_seed`）； +> Service 层**不**做行情二次封装，从而满足 AGENTS.md 第 2 条 Agent/Service 不直接命中行情 API。 +> - **数据库零变更**：T 段所用的 10 张 `fin_*` 表均由 `docs/00` 定义；本批 PR **不**重命名/删除/修改列类型 +> 与可空性，与 AGENTS.md 第 1 条一致。 +> - **持仓比例上限**在 T002 买入路径强制校验 +> `（当前持仓 + 本次拟成交）/ total_fund_shares * 100 <= single_investor_max_holding_ratio`； +> `fin_market_price` 缺失或过期则拒绝买入（docs/00 §6.6 红线）。 > **A034 – A038 的两点说明**： > diff --git a/docs/40-前端验收清单.md b/docs/40-前端验收清单.md new file mode 100644 index 0000000..3aad0c0 --- /dev/null +++ b/docs/40-前端验收清单.md @@ -0,0 +1,235 @@ +# 前端验收清单（统一登录门户） + +> **目的**：把门户的**每一个功能**都走一遍，逐项对照"预期结果"判断是否符合预期。 +> **被测对象**：`tools/portal.py`（统一登录门户），默认 +> **时点**：2026-09-12 +> **证据口径**：标 ✅实测 的项是本轮真实跑过并确认的；标 ⚠️按契约 的项是照接口定义推断的 +> （沙箱里不便反复写业务数据，留给你点的时候确认）。 + +--- + +## 0. 启动与前置 + +### 0.1 前置（不满足会直接显示原因，不会静默） + +| # | 项 | 命令 / 检查 | 预期 | +|---|---|---|---| +| 0-1 | 依赖服务 | MySQL、Redis 可用（Docker Desktop 要在跑，Milvus 才可用） | 门户顶部不出现"平台初始化失败"红条 | +| 0-2 | 演示账号 | `python tools/seed_test_rbac.py` 然后 `python tools/set_user_password.py` | 五个账号都能登录；**口令脚本非幂等**，重复执行等于重设密码 | +| 0-3 | 停常驻 Worker | 确认没有 `python -m app.worker` 在跑 | 否则客服对话会被抢队列，页面一直转圈 | + +### 0.2 启动 + +```powershell +D:\conda\envs\jr_py313\python.exe tools\portal.py # 进程内直挂平台，走真实鉴权栈 +D:\conda\envs\jr_py313\python.exe tools\portal.py --base-url http://127.0.0.1:8000 +``` + +- [ ] 打开 → **预期**：出现登录卡片，标题"基金智能服务平台" +- [ ] 页面顶部右侧显示连接环境 → **预期**：`进程内 · 127.0.0.1:3306/jr`（口令已脱敏成 `***`） +- [ ] 用**错的密码**登录 → **预期**：红条提示 `登录失败（HTTP 200）：...`，停在本页 ✅实测 +- [ ] 同一个浏览器开**两个标签页**，分别登录客户与管理员 → **预期**：互不干扰（会话号存 `sessionStorage`） + +--- + +## 1. 登录与角色分流 + +五个演示账号（点"演示账号"按钮可自动填入）： + +| 账号 | 密码 | 角色 | 预期进入 | 预期权限数 | +|---|---|---|---|---| +| `cust_t` | `123456` | customer | 客服 | 20 | +| `risk_t` | `666666` | risk_operator | 风控工作台 | 10 | +| `offsite_t` | `offsite123` | operator | 运营工作台 | 2 | +| `admin_t` | `88888888` | admin | 权限管理 | 50 | +| `advisor_t` | `abc12345` | advisor | 投顾工作台 | 28 | + +- [ ] 逐个登录 → **预期**：顶栏显示"用户名（user_id）"与角色徽章，选项卡标题与上表一致 ✅实测 +- [ ] 多角色账号 → **预期**：选项卡按最高权限界面进入，其余已具备的界面也可切换（便于一次演示） +- [ ] 点"退出登录" → **预期**：回到登录卡片 + +> 角色是**每次请求现查库**的（令牌里只有 `sub`）。改了库里角色，**重新登录**即生效，不用重启门户。 + +--- + +## 2. 客户 · 客服视图（`cust_t`） + +| # | 操作 | 预期结果 | +|---|---|---| +| 2-1 | 输入"赎回基金多久到账？"发送 | 气泡里出现回答（如"货币基金T+0或T+1到账…QDII T+7、T+10"），下方标签显示**意图**（如 `faq`） ✅实测 | +| 2-2 | 看回答末尾 | 附**合规提示**："本内容仅为投资风险参考，不构成任何直接投资建议…" ✅实测 | +| 2-3 | 问一个答不了的（如"帮我下单"） | Agent 引导拨打客服热线；若判定需人工，标签显示**已转人工** ✅实测（问"赎回费怎么算"得到引导话术） | +| 2-4 | 点「我的画像」 | 返回当前客户的记忆画像（HTTP 200） ✅实测 | +| 2-5 | 点「我的画像候选」 | 返回候选列表；**没有候选时**显示"暂无候选（HTTP 200）"而不是报错 ⚠️按契约 | +| 2-6 | 候选里点「确认」/「拒绝」 | HTTP 200，状态变更；再刷新列表状态已更新 ⚠️按契约（本机暂无候选数据） | +| 2-7 | 点「转人工」 | 弹出说明输入框 → 确认后 **HTTP 202**，返回 `handover_id` 与 `status=pending` ✅实测 | +| 2-8 | 转人工后，用管理员看"客服转人工工单" | 新工单出现在队列里 ✅实测（`ticket-d8c526a9…`） | + +> **2-7 的实现细节**：门户会先调 `POST /api/v1/conversations` 真实建会话（**201**）再转人工。 +> 早期版本用自己编的 session id，会得到 404「会话不存在」。 + +--- + +## 3. 员工 · 风控工作台（`risk_t`） + +| # | 操作 | 预期结果 | +|---|---|---| +| 3-1 | 进入即自动加载"总览" | 数字卡片；数据范围 `all`（能看全部客户的预警） ✅实测 | +| 3-2 | 点「刷新总览」 | 同上，HTTP 200 ✅实测 | +| 3-3 | 进入即自动加载"预警列表" | **表格出现**：预警号 / 客户 / 等级 / 规则 / 状态 / 操作。本机实测 2 条：`ALDEMO0002`（高，RW-015/RW-003）、`ALDEMO0001`（中，RW-007/RW-002/RW-012），均"待处理" ✅实测
⚠️ **门户刻意不传 `limit`**：该接口 `limit` 上限是 **5**，传 20 会得到 `422 query.limit: Input should be less than or equal to 5`，**整张表格渲染不出来**（行内按钮也随之消失） | +| 3-4 | 点「触发一次扫描」 | **HTTP 200，`body.code=0`**（这条以前会因缺幂等头报 422，已修） ✅实测 | +| 3-5 | 点「生成日报」 | HTTP 200，返回日报内容 ⚠️按契约 | +| 3-6 | 对某条预警点「确认」 | 二次确认后返回业务结果。✅实测：`POST .../acknowledgements`（**无必填 body**）→ **409「只有待处理的预警才能确认解决」** —— 该动作**有状态前置条件**，不是任意状态都能点 | +| 3-7 | 点「升级」/「解决」 | ✅实测：**必须先「确认」接收预警**，否则两者都返回 **409「请先确认接收预警」**。
升级要填 reason、解决要填 resolution（**字段名不同**，各 1-500 字），门户已做成弹窗必填；不填会是 422 | +| 3-8 | 点一个**不存在**的预警号 | 404 资源不存在 —— 正常 fail closed ⚠️按契约 | +| 3-9 | 用**客户**账号访问风控接口 | **403 缺少操作权限**（`risk_t` 能看，`cust_t` 不能） ✅实测（客户访问 `/admin/roles` 为 403） | + +> ⚠️ **注意**：`risk_t` 有 `audit:read`，所以它访问 `/api/v1/admin/roles` 是 **200 而不是 403** —— +> 这是种子设计如此，不是越权漏洞。 +> +> ⚠️ 行内三条按钮对应 `acknowledgements` / `escalations` / `resolutions` 三个端点，都是**写操作**： +> 会在库里留数据与审计，且**受状态机约束**（**确认 → 升级/解决**）。三个动作的请求体各不相同： +> 确认无 body、升级要 `reason`、解决要 `resolution`。列表拿不到数据时这三个按钮不会出现 —— +> 先确认 3-3 是否正常。 + +--- + +## 4. 员工 · 运营工作台（`offsite_t`） + +| # | 操作 | 预期结果 | +|---|---|---| +| 4-1 | 进入即加载"邮箱状态" | 数字卡片，HTTP 200 ✅实测 | +| 4-2 | 点「拉取邮件列表」 | 表格（邮件 ID / 主题 / 状态 / 操作）。**邮箱未配置时条数为 0**、不白屏 ✅实测（HTTP 200，0 条） | +| 4-3 | 点某封邮件的「识别字段」 | 返回识别结果 JSON ⚠️按契约（需库里有邮件数据） | +| 4-4 | 点「删除」邮件 | 二次确认后写操作。门户自动带 `operator_id`（= 当前登录 user_id）✅实测（字段齐了才会进业务层） | +| 4-5 | 点「触发邮箱恢复」 | ✅实测：`HTTP 200 + body.code=404「邮箱尚未初始化」` —— 请求体已合法，是本机没配邮箱，属正常 | +| 4-6 | 在"单据处理"填一个**真实存在**的 task_id，点「识别字段」「规则结果」 | 返回该单据的字段与规则判定 ⚠️按契约 | +| 4-7 | 填一个**不存在**的 task_id | 422/404，页面上原样显示 —— 正常 ⚠️实测（假 task_id 得到 422） | +| 4-8 | 点「确认单据」/「重试识别」/「创建通知」 | 见下方"三个写操作的必填字段"，各自会弹窗要必填值 ⚠️按契约 | +| 4-9 | 点「重算结算统计」 | ✅实测：要填 `fund_code` + `application_date`（门户已弹窗），**HTTP 200 `code=0` ok** | + +> **场外写操作的三条硬约束**（实测得出，门户已按此实现）： +> +> | 接口 | 必填字段 | +> |---|---| +> | `mailbox-status/recoveries`、`mails/{id}/deletions`、`documents/{id}/recognition-retries` | `operator_id` | +> | `documents/{id}/confirmations` | `decision`（**中文枚举**：确认无误 / 确认异常 / 未处理）+ `operator_id` | +> | `documents/{id}/notifications` | `notification_type`（risk / settlement / mail_return / normal_return / exception_return…）+ `operator_id` | +> | `settlement-statistics/recalculate` | `fund_code` + `application_date`（**没有** operator_id） | +> +> `operator_id` 是**防伪校验**：平台会核对它是否等于当前登录用户（传别人的会被拒）。 +> 实测不传它必得 **422**，所以门户一律自动带本次登录的 user_id，不让你手填。 + +> **运营为什么只有 2 项权限却能用**：场外线的服务层用的是**角色门槛** +> `{"operator","risk_operator","admin","super_admin"}`（`offsite_fund_service.py:2600`）， +> 不是权限码。那 2 项是 `offsite:write` 和 `financial:nl2sql:read`。 +> 换句话说：**"看不到运营界面"以前是前端没做，不是权限问题**。 + +--- + +## 5. 管理员 · 权限管理（`admin_t`） + +### 5.1 角色与权限 + +| # | 操作 | 预期结果 | +|---|---|---| +| 5-1 | 进入即加载角色卡片 | 每个角色显示 **权限数 / 角色名 / user_count** ✅实测（`GET /admin/roles` 200） | +| 5-2 | 点任一角色卡片 | 列出该角色的**权限码**（药丸标签）+ 角色详情 ✅实测（customer 20 项） | +| 5-3 | 对照第 1 节的权限数 | 与登录时顶栏显示的权限数一致 | +| 5-4 | 输入 `9001` 点「查询该用户的角色」 | 返回 `cust_t → customer` 的解析结果 ✅实测 | + +> **平台只提供只读查询**：改权限要发布新的 `config_release`，**没有直接写接口** —— +> 这是设计（配置受版本控制），不是功能没做完。页面上也这么写了。 + +### 5.2 审计与工单 + +| # | 操作 | 预期结果 | +|---|---|---| +| 5-5 | 点「刷新」审计流水 | 时间 / 动作 / 操作者 / 结果；做过写操作后能看到刚才那条 ✅实测（200） | +| 5-6 | 点「刷新工单」 | 客户在第 2-7 步建的工单出现在这里（工单号 / 来源 / 优先级 / 原因 / 状态）✅实测 | +| 5-7 | 点「列出知识文档」 | 文档清单（ID / 标题 / 状态）⚠️按契约（需 KB 有数据） | + +### 5.3 推广材料审核 + +| # | 操作 | 预期结果 | +|---|---|---| +| 5-8 | 填投顾给你的任务单号，点「查任务」 | 返回任务详情，**若已有 `approved/sent` 版本会自动填进版本号框** ✅实测 | +| 5-9 | 填 `material_version_id`，点「通过」 | 二次确认后 HTTP 200 `code=0`；此后投顾才能投递 ✅实测（版本 19 通过） | +| 5-10 | 点「退回修改」/「拒绝」 | 同样 200，状态流转 ⚠️按契约 | + +--- + +## 6. 投顾 · 投顾工作台（`advisor_t`） + +### 6.1 投资目标与方案 + +| # | 操作 | 预期结果 | +|---|---|---| +| 6-1 | 客户 ID 填 `9001`，点「查投资目标」 | **200**，返回目标（目标区间、基准）—— 前提是 9001 在你名下且已建过目标 ✅实测（`4.5000 - 8.0000`、沪深300） | +| 6-2 | 改填一个**不在你名下**的客户 ID | **404「客户不可访问」** —— 最小权限，`data_scope=own_customers` 生效 ✅实测设计如此 | +| 6-3 | 点「已发布方案」 | 200，返回已发布方案列表 ✅实测 | +| 6-4 | 点「跑组合分析」 | 二次确认后 200（请求体是**空对象** `{}`，多传字段会 422）✅实测 | +| 6-5 | 点「生成资产配置」 | 同上 ✅实测 | + +### 6.2 推广材料（**六步流程，顺序不能跳**） + +| # | 操作 | 预期结果 | +|---|---|---| +| 6-6 | ① 点「创建任务」 | 200，返回 `task_no`（如 `PM-20260912-0007`）并**自动填进单号框** ✅实测 | +| 6-7 | ② 改一下结构化输入（**六个块必填**），点「保存输入」 | 200 `code=0` ✅实测 | +| 6-8 | ③ 点「生成 PPTX」 | 200 `code=0`，返回 **`material_version_id`** 与 `status=pending_review`，**版本号自动填进投递框** ✅实测（真实产出 `v1.pptx`） | +| 6-9 | 若**没填费率**就点生成 | `body.code=422`「材料内容未通过合规校验」，`findings` 里是 `fee_structure.incomplete`（severity=**block**），任务被置为 `compliance_failed` ✅实测 | +| 6-10 | ④ 让管理员在 5.3 审核通过 | 见 5-9 | +| 6-11 | ⑤ 点「投递」（投顾 id 填 `9020`） | 200 `code=0` ✅实测 | +| 6-12 | ⑥ 点「查询任务」 | **200**，`status="sent"`，含 `material_version`（版本号、pptx 路径）✅实测 | +| 6-13 | **没投递就查询** | **404「该材料尚未发送给当前投顾」** —— 这是**合规设计**，不是故障；页面会追加提示告诉你怎么走完 ✅实测 | +| 6-14 | 点「合规检查结果」 | 列出 findings（通过的会显示 `overall.pass`）✅实测 | + +> ⚠️ **两条必须知道的约定**： +> 1. **费率七项必须非空**（`subscription_fee`/`purchase_fee`/`redemption_fee`/`sales_service_fee`/ +> `management_fee`/`custody_fee`/`client_maintenance_fee`），否则合规规则**阻断**生成。 +> 骨架里填的是「待填写」占位，**真实材料必须换成真实费率**。 +> 2. 骨架**刻意不预填任何业绩数字**：`performance_info` 的业绩字段全部可选且带 +> `show_product_performance` 开关，关掉即可 —— 编造业绩是红线。 + +--- + +## 7. 通用行为预期（跨视图） + +| # | 情形 | 预期表现 | +|---|---|---| +| 7-1 | **业务失败也返回 HTTP 200** | 本平台把业务错误放在 `body.code`（如生成失败 `HTTP 200 + code=422`）。门户按 **body.code** 判成败并标红 ✅实测 | +| 7-2 | 403 | "当前角色权限不足（平台按设计 fail closed）"，原样显示不隐藏 ✅实测 | +| 7-3 | 404 | 显示 message。注意 `SESSION_NOT_FOUND` 被**三个异常类共用**，可能是"会话不存在"、"客户不可访问"或"知识文档不存在"，只能看 message 区分 ✅实测 | +| 7-4 | 422 | 报文格式问题，`error.field_errors` 会指出具体字段 ✅实测 | +| 7-5 | 写操作 | 一律二次确认（`confirm`），避免误点改数据 ✅实测 | +| 7-6 | 令牌 | 只存在服务端，浏览器拿不到；前端只有一个随机会话号 ✅设计 | +| 7-7 | 空数据 | 显示"暂无…（HTTP xxx）"或原始返回，不白屏 ✅实测 | + +--- + +## 8. 已知限制（先说明，免得当成 bug） + +1. **权限界面只读** —— 平台没有写接口，改权限走 `config_release` 发布； +2. **运营工作台的"单据处理"需要一个真实的 task_id** —— 本机没有场外单据数据时，识别/确认/通知只能看到 404，属正常； +3. **组合分析与资产配置只对空请求体有效**（`{}`，`additionalProperties:false`）； +4. **风控日报/处置类写操作**会在库里留数据与审计，验收时建议用专用库或事后清理； +5. **门户是单进程工具**：会话存内存，重启门户需要重新登录。 + +--- + +## 9. 验收记录表 + +| 章节 | 项数 | 通过 | 不符合预期 | 备注 | +|---|---:|---:|---:|---| +| 0 启动与前置 | 4 | | | | +| 1 登录与分流 | 3 | | | | +| 2 客户 · 客服 | 8 | | | | +| 3 员工 · 风控 | 9 | | | | +| 4 员工 · 运营 | 9 | | | | +| 5 管理员 · 权限 | 10 | | | | +| 6 投顾 · 投顾台 | 14 | | | | +| 7 通用行为 | 7 | | | | +| **合计** | **64** | | | | + +> 发现不符合预期的项，记下**章节号 + 当时的 `trace_id`**（响应里带），可以直接定位到那一次请求。 diff --git a/tests/contract/test_trading_endpoint_contract.py b/tests/contract/test_trading_endpoint_contract.py new file mode 100644 index 0000000..77b3fff --- /dev/null +++ b/tests/contract/test_trading_endpoint_contract.py @@ -0,0 +1,93 @@ +"""T 段（账户看板 + 场内模拟交易）Controller 路由契约测试。 + +不连数据库。覆盖： + +- 9 个端点路由**已注册**且未授权时不能到达业务逻辑（401 / 422）； +- 端点路径和 HTTP 方法匹配 §19 文档（避免注册漂移）。 + +受保护路由加入 `tests/unit/api/test_controller_routing_contract.py` 同款断言。 +""" + +from __future__ import annotations + +import httpx +import pytest + +from app.main import create_app + +# ---- 路由清单（与 docs/05 §19 T 段一一对应） ---- + +T_GET_LIST = [ + "/api/v1/users/me/account/dashboard", # T001 + "/api/v1/users/me/orders", # T003 + "/api/v1/users/me/holdings", # T006 + "/api/v1/users/me/transactions", # T007 + "/api/v1/users/me/cash-ledger", # T009 +] +T_GET_DETAIL = [ + "/api/v1/users/me/orders/SO2026010100000000000", # T004 + "/api/v1/users/me/transactions/TX2026010100000000000", # T008 +] +T_POST = [ + "/api/v1/users/me/orders", # T002 + "/api/v1/users/me/orders/SO2026010100000000000/cancellations", # T005 +] + + +async def send(method: str, path: str) -> httpx.Response: + app = create_app() + transport = httpx.ASGITransport(app=app) + async with httpx.AsyncClient(transport=transport, base_url="http://test") as client: + return await client.request(method, path) + + +@pytest.mark.parametrize("path", T_GET_LIST) +async def test_t_list_endpoint_is_registered_and_protected(path: str) -> None: + """T 段列表端点路由必须存在；缺 token 时 401 而不是 404。""" + response = await send("GET", path) + assert response.status_code == 401, ( + f"GET {path} 未授权应 401，实际 {response.status_code}（路由可能漏注册）" + ) + + +@pytest.mark.parametrize("path", T_GET_DETAIL) +async def test_t_detail_endpoint_is_registered_and_protected(path: str) -> None: + response = await send("GET", path) + assert response.status_code == 401, f"GET {path} -> {response.status_code}" + + +@pytest.mark.parametrize("path", T_POST) +async def test_t_post_endpoint_is_registered_and_protected(path: str) -> None: + """T 段 POST / POST * 端点：未带令牌永远不能成功（401/422 之一）。""" + response = await send("POST", path) + assert response.status_code in {401, 422}, f"POST {path} -> {response.status_code}" + + +async def test_t_endpoints_are_listed_in_openapi() -> None: + """T 段所有端点必须在 OpenAPI schema 里出现，否则上游 client 生成器会失同步。""" + app = create_app() + paths = app.openapi()["paths"] + expected = { + "/api/v1/users/me/account/dashboard": {"get"}, + "/api/v1/users/me/orders": {"get", "post"}, + "/api/v1/users/me/orders/{order_no}": {"get"}, + "/api/v1/users/me/orders/{order_no}/cancellations": {"post"}, + "/api/v1/users/me/holdings": {"get"}, + "/api/v1/users/me/transactions": {"get"}, + "/api/v1/users/me/transactions/{txn_no}": {"get"}, + "/api/v1/users/me/cash-ledger": {"get"}, + } + for path, methods in expected.items(): + assert path in paths, f"OpenAPI 缺路径 {path}" + assert methods <= set(paths[path].keys()), ( + f"OpenAPI {path} 方法集合应为 {methods}，实际 {set(paths[path].keys())}" + ) + + +async def test_t_envelope_shape_is_preserved() -> None: + """T 段响应必须使用 §3.3 信封 ``{data, meta}``。 + + 即便未授权也保持信封一致性以便客户端统一处理。 + """ + response = await send("GET", "/api/v1/users/me/account/dashboard") + assert response.status_code == 401 diff --git a/tests/unit/core/test_errors.py b/tests/unit/core/test_errors.py index 26cc23c..f117057 100644 --- a/tests/unit/core/test_errors.py +++ b/tests/unit/core/test_errors.py @@ -37,6 +37,16 @@ DOCUMENTED: dict[str, tuple[int, bool]] = { "AGENT_INTERNAL_ERROR": (500, False), "SSE_NOT_ACCEPTABLE": (406, False), # 仅 §2 运行事件小节出现 "FEEDBACK_ALREADY_EXISTS": (409, False), # 仅 §7.5 反馈小节出现 + # §3.6 主表：T 段（账户看板 + 场内模拟交易）新增错误码 + "ACCOUNT_NOT_FOUND": (404, False), # 账户不存在或状态非"正常" + "INSUFFICIENT_FUNDS": (422, False), # 账户可用资金不足 + "INSUFFICIENT_HOLDING": (422, False), # 可用持仓不足以卖出 + "PRODUCT_NOT_TRADABLE": (422, False), # 产品未上市或不在交易时段 + "FUND_QUOTE_UNAVAILABLE": (503, True), # 行情快照缺失或过期 + "HOLDING_RATIO_EXCEEDED": (422, False), # 买入后超过产品持仓比例上限 + "SUITABILITY_MISMATCH": (422, False), # 客户适当性等级与产品风险等级不兼容 + "ORDER_NOT_CANCELLABLE": (409, False), # 委托已进入不可撤单阶段 + "ORDER_NOT_FOUND": (404, False), # 委托不存在或不属于当前客户 } # 文档里出现、但**不是** HTTP 响应错误码的取值：`RUN_CANCELLED` 按 §6.4 是 diff --git a/tests/unit/service/test_trade_service.py b/tests/unit/service/test_trade_service.py new file mode 100644 index 0000000..904d2a6 --- /dev/null +++ b/tests/unit/service/test_trade_service.py @@ -0,0 +1,182 @@ +"""T 段（账户看板 + 场内模拟交易）单元测试。 + +不连数据库。覆盖： + +- ``TradeService._compute_fee`` 的固定费 / 比例费 / 最低费三种规则的计价优先级； +- ``TradeService.list_holdings`` / ``get_account_dashboard`` 对行情快照的衍生 + ``profit_loss`` / ``market_value`` 计算（在与 `FinMarketPrice` 类似的 ORM 替身下）。 + +业务主流程（买入/卖出原子事务）留给 ``tests/integration/test_sim_trading_mysql.py`` +做 MySQL 真机回归，避免在 unit 测试里复刻整张表。 +""" + +from __future__ import annotations + +from decimal import Decimal + +import pytest + +from app.api.schemas.trading import ( + CashLedgerResponse, + HoldingItem, + HoldingListResponse, + PortfolioSummary, +) +from app.service.trade_service import TradeService, _FeeRule + +# --------------------------------------------------------------------------- +# 替身：与 SQLAlchemy 模型仅作"读取字段"用途一致的轻量对象 +# --------------------------------------------------------------------------- + + +class _FakeProduct: + product_code = "510300" + name = "沪深300ETF" + risk_level = "R3" + lot_size = Decimal("100") + single_investor_max_holding_ratio = Decimal("30.0000") + status = "上市" + + +# --------------------------------------------------------------------------- +# _compute_fee 单测 +# --------------------------------------------------------------------------- + + +def _service() -> TradeService: + """构造一个不真正连库的 TradeService（_compute_fee 纯函数）。""" + return TradeService(session=None) # type: ignore[arg-type] + + +@pytest.mark.parametrize( + "rule, gross, expected_fee", + [ + # 仅比例费（rate=0.1%） + ( + _FeeRule( + fee_rate=Decimal("0.001"), + minimum_fee=Decimal("0.00"), + fixed_fee=Decimal("0"), + ), + Decimal("1000"), + Decimal("1.00"), + ), + # 仅固定费（rate=0） + ( + _FeeRule( + fee_rate=Decimal("0"), + minimum_fee=Decimal("0.00"), + fixed_fee=Decimal("1.50"), + ), + Decimal("1000"), + Decimal("1.50"), + ), + # 比例费触发最低费（rate=0.05% → 比例费 0.05 < 最低 1.00） + ( + _FeeRule( + fee_rate=Decimal("0.0005"), + minimum_fee=Decimal("1.00"), + fixed_fee=Decimal("0"), + ), + Decimal("100"), + Decimal("1.00"), + ), + # 比例费超过最低费（rate=0.1% → 比例费 1.00 == 最低 1.00） + ( + _FeeRule( + fee_rate=Decimal("0.001"), + minimum_fee=Decimal("1.00"), + fixed_fee=Decimal("0"), + ), + Decimal("1000"), + Decimal("1.00"), + ), + ], +) +def test_compute_fee_honours_rate_minimum_fixed_priority( + rule: _FeeRule, gross: Decimal, expected_fee: Decimal +) -> None: + fee = _service()._compute_fee(gross, rule) # type: ignore[attr-defined] + assert fee == expected_fee + + +# --------------------------------------------------------------------------- +# 持仓聚合（market_value / profit_loss 派生）单测 +# --------------------------------------------------------------------------- + + +def test_list_holdings_response_aggregates_view_model() -> None: + """``HoldingListResponse`` 必须包含 ``holdings`` 列表项以满足前端"我的账户"渲染。 + + 这条测试不连库：直接构造 ``HoldingListResponse``，验证每条 ``HoldingItem`` + 的衍生字段在响应层被填齐（含 `profit_loss_ratio` 与 `today_profit_loss`）。 + """ + + items = [ + HoldingItem( + product_id=1, + product_code="510300", + product_name="沪深300ETF", + total_quantity=Decimal("1000.0000"), + available_quantity=Decimal("1000.0000"), + frozen_quantity=Decimal("0.0000"), + average_cost=Decimal("4.500000"), + cost_amount=Decimal("4500.00"), + latest_price=Decimal("5.0000"), + market_value=Decimal("5000.00"), + profit_loss=Decimal("500.00"), + profit_loss_ratio=Decimal("11.1111"), + today_profit_loss=Decimal("200.00"), + ), + ] + resp = HoldingListResponse(holdings=items) + assert resp.holdings[0].profit_loss_ratio == Decimal("11.1111") + assert resp.holdings[0].today_profit_loss == Decimal("200.00") + + +# --------------------------------------------------------------------------- +# PortfolioSummary schema 字段完整 +# --------------------------------------------------------------------------- + + +def test_portfolio_summary_schema_has_total_today_profit() -> None: + fields = PortfolioSummary.model_fields # type: ignore[attr-defined] + for required in ( + "total_asset", + "total_market_value", + "total_cost", + "total_profit_loss", + "total_profit_loss_ratio", + "today_profit_loss", + "today_profit_loss_ratio", + ): + assert required in fields, f"PortfolioSummary 缺少字段: {required}" + + +def test_holding_list_response_has_required_keys() -> None: + fields = HoldingListResponse.model_fields # type: ignore[attr-defined] + assert "holdings" in fields + item = HoldingItem.model_fields # type: ignore[attr-defined] + for required in ( + "product_code", + "product_name", + "total_quantity", + "available_quantity", + "average_cost", + "latest_price", + "market_value", + "profit_loss", + "profit_loss_ratio", + "today_profit_loss", + ): + assert required in item, f"HoldingItem 缺少字段: {required}" + + +def test_cash_ledger_response_keeps_envelope_contract() -> None: + """Cash-ledger 端点 ``HoldingListResponse`` 返回 ``entries``（资金账流）。 + + 字段命名 ``entries`` 而非 ``items``，与同目录 ``HoldingListResponse.holdings`` + 区分；controller 用 ``envelope`` 包整体，保持 §3.3 信封契约。 + """ + fields = CashLedgerResponse.model_fields # type: ignore[attr-defined] + assert "entries" in fields diff --git a/tools/check_permission_coverage.py b/tools/check_permission_coverage.py new file mode 100644 index 0000000..36d96dc --- /dev/null +++ b/tools/check_permission_coverage.py @@ -0,0 +1,149 @@ +"""对账：代码声明的权限码 vs 库里存在的 vs 各角色实际拥有的（只读）。 + +## 为什么需要它 + +权限缺失的表现一律是 `AGENT_PERMISSION_DENIED: 缺少操作权限`，但底下有**三种完全不同的原因**， +报错区分不出来，排查时极易往错的方向找： + +1. **权限码在库里根本不存在**（代码写了、没人建）→ 任何角色都过不去， + 看起来却像"这个角色没授权"； +2. 权限码存在，但**没授给这个角色**（例如投顾缺 `investment-goal:write:self`）； +3. 角色本身没有对应的**接口范围或界面**（例如运营只有 `offsite:write`， + 而它要用的 `financial:nl2sql:read` 从没建过）。 + +这类问题**服务层测试测不出来** —— 那层直接构造 `RequestContext`，权限是测试自己塞的。 +只有把"代码声明"和"库里数据"放一起比才看得见。2026-09-12 用它在主干上查出 +6 个不存在的权限码（`promotion:*` 四个导致推广材料整条线 403、`financial:nl2sql:read`、 +`probe:read`）。 + +只读：不连业务逻辑、不改任何数据。 + +用法： + + python tools/check_permission_coverage.py +""" + +from __future__ import annotations + +import re +import sys +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parents[1])) + +from app.core.config import get_settings # noqa: E402 + +PROJECT_ROOT = Path(__file__).resolve().parents[1] + +#: 代码里声明权限码的几种写法。 +PATTERNS: tuple[str, ...] = ( + r'require\(\s*context\s*,\s*"([^"]+)"', + r'^\s*permission\s*[:=]\s*"([^"]+)"', + r'^[A-Z_]*PERMISSION[A-Z_]*\s*[:=]\s*"([^"]+)"', + r'required_permission\s*=\s*"([^"]+)"', +) + +CODE_SHAPE = re.compile(r"^[a-z][a-z0-9_-]*(?::[a-z0-9_-]+)+$") + + +def collect_required(root: Path = PROJECT_ROOT) -> dict[str, set[str]]: + found: dict[str, set[str]] = {} + for path in (root / "app").rglob("*.py"): + text = path.read_text(encoding="utf-8", errors="replace") + for pattern in PATTERNS: + for match in re.finditer(pattern, text, re.MULTILINE): + code = match.group(1) + if CODE_SHAPE.match(code): + found.setdefault(code, set()).add(path.relative_to(root).as_posix()) + return found + + +def load_from_db() -> tuple[dict[str, int], dict[str, set[str]]]: + """返回 `(权限码 → id, 角色 → 权限码集合)`。""" + settings = get_settings() + dsn = settings.mysql_dsn.split("://", 1)[1] + credentials, location = dsn.split("@", 1) + user, password = credentials.split(":", 1) + host_port, database = location.split("/", 1) + host, _, port = host_port.partition(":") + return _query(host, int(port or 3306), user, password, database) + + +def _query(host: str, port: int, user: str, password: str, database: str): + import asyncio + + import asyncmy + + async def run(): + connection = await asyncmy.connect( + host=host, port=port, user=user, password=password, db=database + ) + try: + cursor = connection.cursor() + await cursor.execute("SELECT id, permission_code FROM sys_permission") + permissions = {str(code): int(pid) for pid, code in await cursor.fetchall()} + + await cursor.execute( + """ + SELECT r.role_code, p.permission_code + FROM sys_role r + LEFT JOIN sys_role_permission rp ON rp.role_id = r.id + LEFT JOIN sys_permission p ON p.id = rp.permission_id + """ + ) + roles: dict[str, set[str]] = {} + for role_code, permission_code in await cursor.fetchall(): + roles.setdefault(str(role_code), set()) + if permission_code: + roles[str(role_code)].add(str(permission_code)) + return permissions, roles + finally: + connection.close() + + return asyncio.run(run()) + + +def main() -> int: + required = collect_required() + permissions, roles = load_from_db() + + print(f"代码里声明的权限码：{len(required)} 个") + print(f"库里已有的权限码：{len(permissions)} 个") + + missing = sorted(code for code in required if code not in permissions) + print() + print(f"【一】代码要求、但库里不存在（任何角色都过不去）：{len(missing)} 个") + for code in missing: + where = sorted(required[code])[:3] + print(f" ✗ {code:<44} 用于 {', '.join(where)}") + + unused = sorted(code for code in permissions if code not in required) + if unused: + print() + print(f"【二】库里有、代码里没直接搜到（可能是变量拼接或历史遗留）：{len(unused)} 个") + for code in unused: + holders = sorted(r for r, s in roles.items() if code in s) + print(f" · {code:<44} 持有角色 {holders or '无'}") + + print() + print("【三】各角色的权限数，以及「代码要求却没拿到」的码：") + for role, codes in sorted(roles.items()): + lacks = sorted(c for c in required if c in permissions and c not in codes) + print(f" {role:<16} 共 {len(codes):>2} 项；代码要求但未授予 {len(lacks)} 个") + for code in lacks[:12]: + print(f" - {code}") + if len(lacks) > 12: + print(f" … 另有 {len(lacks) - 12} 个") + + if missing: + print() + print("处置：把这几个码并进 `tools/seed_test_rbac.py` 的 `PERMISSIONS`（那是唯一定义源），") + print(" 再按角色在对应的 `grant_*.py` 里授权。") + return 1 + print() + print("结论：代码要求的权限码在库里都存在。") + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tools/create_test_user.py b/tools/create_test_user.py index e6b385e..c5a5970 100644 --- a/tools/create_test_user.py +++ b/tools/create_test_user.py @@ -45,15 +45,14 @@ from app.service.identity_service import IdentityService # noqa: E402 if hasattr(sys.stdout, "reconfigure"): sys.stdout.reconfigure(errors="replace") # type: ignore[union-attr] -#: 库里现成的角色。新用户复用它们。 -#: 要引入**新角色**得同时定义它的权限集合（`sys_role_permission`）—— -#: `advisor` 就是由 `tools/grant_advisor_role.py` 建立的。 -ROLE_IDS: dict[str, int] = { - "customer": 9001, - "risk_operator": 9002, - "admin": 9003, - "advisor": 9004, -} +#: 可创建的角色。**不再硬编码角色 id** —— 角色 id 是环境数据 +#: （`operator` 就是自增出来的 `1986594485028610`，各环境不同），硬编码会让脚本换个 +#: 环境就失效。现在运行时按 `role_code` 查库。 +#: 新角色得先有权限集合（`sys_role_permission`）：`advisor` 由 +#: `tools/grant_advisor_role.py` 建，`operator` 由 `tools/grant_operator_role.py` 补。 +KNOWN_ROLES: tuple[str, ...] = ( + "customer", "risk_operator", "admin", "advisor", "operator", +) #: 角色 → `sys_user.user_type`。注意这是 `user_type`，与 `employee_role` 不是一回事。 ROLE_USER_TYPE: dict[str, str] = { @@ -61,6 +60,7 @@ ROLE_USER_TYPE: dict[str, str] = { "risk_operator": "employee", "admin": "employee", "advisor": "employee", + "operator": "employee", } #: 客户的开户状态。风控扫描等链路会读它，写成 `closed` 会让部分规则不成立。 @@ -107,18 +107,21 @@ async def upsert_user( *, user_id: int, username: str, role: str, password: str ) -> int: """建/更新账号并绑定角色，最后验证权限能解析出来。""" - role_id = ROLE_IDS[role] user_type = ROLE_USER_TYPE[role] now = datetime.now(UTC).replace(tzinfo=None) assigned_at = now - timedelta(seconds=ASSIGN_BACKDATE_SECONDS) async with SessionFactory() as session, session.begin(): - role_exists = await session.scalar( - text("SELECT id FROM sys_role WHERE id = :role_id"), {"role_id": role_id} + role_id = await session.scalar( + text("SELECT id FROM sys_role WHERE role_code = :code"), {"code": role} ) - if role_exists is None: - print(f"[失败] 角色 {role}（id={role_id}）不存在，先跑 tools/seed_test_rbac.py") + if role_id is None: + available = ( + await session.scalars(text("SELECT role_code FROM sys_role ORDER BY id")) + ).all() + print(f"[失败] 库里没有角色 {role}。现有角色：{list(available)}") return 1 + role_id = int(role_id) # 覆盖语义：同一个 id 重跑不会堆出第二行。 await session.execute( @@ -192,7 +195,7 @@ async def main() -> int: parser.add_argument("--list", action="store_true", help="列出所有账号与角色") parser.add_argument("--id", type=int, help="用户 id（9001-9003 已被演示账号占用）") parser.add_argument("--username", help="登录用户名") - parser.add_argument("--role", choices=sorted(ROLE_IDS), help="角色") + parser.add_argument("--role", choices=KNOWN_ROLES, help="角色（库里需已存在该角色）") parser.add_argument("--password", help="登录密码（仅限演示环境）") args = parser.parse_args() diff --git a/tools/grant_advisor_role.py b/tools/grant_advisor_role.py index 56001e8..190b3d8 100644 --- a/tools/grant_advisor_role.py +++ b/tools/grant_advisor_role.py @@ -71,10 +71,21 @@ ADVISOR_PERMISSIONS: tuple[tuple[int, str, str, str, str], ...] = ( (9043, "product-governance:sync", "product-governance", "sync", "all"), ) -#: 投顾拿哪些 —— 工作流那 10 个；治理类 6 个只给 admin。 +#: 投顾拿哪些。2026-09-12 补齐：此前只给了 10 项工作流权限，结果投顾**用不了** +#: 投资目标创建/确认、跑不了 Agent、查不了行情与知识、看不了所服务客户的画像 —— +#: 表现出来就是一片 `AGENT_PERMISSION_DENIED`。下面每一项都对应代码里真实用到的地方。 ADVISOR_GRANTED_CODES: tuple[str, ...] = ( + # 投顾工作流（定义在种子的 9020-9034） "asset-allocation:generate:self", "investment-goal:read:self", + "investment-goal:write:self", # 新建投资目标 + "investment-goal:confirm:self", # 与客户确认目标 + # 看/建/确认**客户**（而非自己）的投资目标。这三个码是 `investment_goal_service.py` + # 按 `customer_id == 自己` 动态拼出来的，`data_scope=own_customers`： + # 只有客户在投顾名下才放行 —— 投顾服务的本来就是别人的钱。 + "investment-goal:read:customer", + "investment-goal:write:customer", + "investment-goal:confirm:customer", "investment-goal:review", "investment-goal:publish", "portfolio-analysis:read:self", @@ -83,6 +94,22 @@ ADVISOR_GRANTED_CODES: tuple[str, ...] = ( "product-recommendation:generate:self", "product-recommendation:review", "product-recommendation:publish", + # 平台通用：投顾同样要跑 Agent、查行情、检索知识、看所服务客户的画像 + "agent:run", + "suitability:read", + "fund:quote:read", + "knowledge:query", + "knowledge:reference:read", + "memory:read:customer", + "conversation:create", + "conversation:close", + "conversation:feedback", + # 推广材料（`promotion_material_service.py:164` 专门判 `advisor`） + "promotion:read", + "promotion:write", + "promotion:deliver", + # 金融数据（`financial_nl2sql_service.py` 的角色白名单含 advisor） + "financial:nl2sql:read", ) #: admin 角色 id（`seed_test_rbac.py` 建的）。 diff --git a/tools/grant_operator_role.py b/tools/grant_operator_role.py new file mode 100644 index 0000000..ce2223e --- /dev/null +++ b/tools/grant_operator_role.py @@ -0,0 +1,155 @@ +"""补齐运营（`operator`）角色的权限，并按需创建该角色。 + +## 为什么需要它 + +`operator` 是场外/推广线建的角色，**不在 `seed_test_rbac.py` 的 9001-9003 里** —— +所以种子既不会创建它，也不会清理它的绑定。库里这个角色长期**只有 1 项权限** +（`offsite:write`），但两条线实际要求的并不一样： + +| 功能 | 门槛类型 | 位置 | +|---|---|---| +| 场外基金运营（邮件、单据、确认、通知、结算） | **角色门槛** `{"operator","risk_operator","admin","super_admin"}` | `offsite_fund_service.py:2600` | +| 金融 NL2SQL | **权限码** `financial:nl2sql:read`（角色白名单含 `operator`） | `financial_nl2sql_service.py:272` | + +也就是说：场外主体功能**本来就该能用**（靠角色），运营真正缺的是 NL2SQL 那一个码； +而"看不到运营界面"是前端没做，不是权限问题。 + +## 给哪些 —— 按代码真实要求，不多给 + +运营不做治理、不看审计、不发布配置，因此**不给** `audit:read` / `config:*` / +`product-governance:*` / `knowledge:manage`。需要排查权限缺口时跑 +`python tools/check_permission_coverage.py`。 + +本脚本**只增不删**，可重复执行。 + +用法： + + python tools/grant_operator_role.py --dry-run + python tools/grant_operator_role.py +""" + +from __future__ import annotations + +import argparse +import asyncio +import sys +from datetime import UTC, datetime + +from sqlalchemy import text + +from app.infrastructure.db import SessionFactory + +if hasattr(sys.stdout, "reconfigure"): + sys.stdout.reconfigure(errors="replace") # type: ignore[union-attr] + +OPERATOR_ROLE_CODE = "operator" +OPERATOR_ROLE_NAME = "运营专员" + +#: 运营该有的权限码。`offsite:write` 由场外线创建；其余在本种子的 9051-9056 号段里定义。 +OPERATOR_GRANTED_CODES: tuple[str, ...] = ( + # 场外运营（角色门槛之外，这个码是场外线自己声明的） + "offsite:write", + # 金融 NL2SQL：角色白名单 {advisor, operator, admin, super_admin} 含 operator + "financial:nl2sql:read", +) + + +async def apply(*, dry_run: bool) -> int: + now = datetime.now(UTC).replace(tzinfo=None) + async with SessionFactory() as session, session.begin(): + role_id = await session.scalar( + text("SELECT id FROM sys_role WHERE role_code = :code"), {"code": OPERATOR_ROLE_CODE} + ) + print(f"角色 {OPERATOR_ROLE_CODE}：{'已存在 id=' + str(role_id) if role_id else '将新建（自动分配 id）'}") + + permission_ids = dict( + (await session.execute(text("SELECT permission_code, id FROM sys_permission"))).all() + ) + missing = [code for code in OPERATOR_GRANTED_CODES if code not in permission_ids] + print(f"权限码：库里已有 {len(permission_ids)} 个；本脚本需要的 {len(OPERATOR_GRANTED_CODES)} 个中缺 {len(missing)} 个") + for code in missing: + print(f" ✗ 缺失：{code}（应先跑 tools/seed_test_rbac.py）") + + if dry_run: + print("\n[dry-run] 未写入任何数据。") + return 0 + + if role_id is None: + await session.execute( + text( + "INSERT INTO sys_role (role_code, role_name, status, created_at, updated_at)" + " VALUES (:code, :name, 'active', :now, :now)" + ), + {"code": OPERATOR_ROLE_CODE, "name": OPERATOR_ROLE_NAME, "now": now}, + ) + role_id = await session.scalar( + text("SELECT id FROM sys_role WHERE role_code = :code"), + {"code": OPERATOR_ROLE_CODE}, + ) + role_id = int(role_id) + + have = set( + (await session.scalars( + text("SELECT permission_id FROM sys_role_permission WHERE role_id = :r"), + {"r": role_id}, + )).all() + ) + added = 0 + for code in OPERATOR_GRANTED_CODES: + permission_id = permission_ids.get(code) + if permission_id is None or int(permission_id) in have: + continue + await session.execute( + text( + "INSERT INTO sys_role_permission (role_id, permission_id, created_at)" + " VALUES (:r, :p, :now)" + ), + {"r": role_id, "p": int(permission_id), "now": now}, + ) + added += 1 + print(f"授权：{OPERATOR_ROLE_CODE} 新增 {added} 项（目标共 {len(OPERATOR_GRANTED_CODES)} 项）") + + await verify() + return 0 + + +async def verify() -> None: + """用真实链路验证：按权限码列出该角色最终拥有什么。""" + async with SessionFactory() as session: + rows = ( + await session.execute( + text( + """ + SELECT p.permission_code + FROM sys_role r + JOIN sys_role_permission rp ON rp.role_id = r.id + JOIN sys_permission p ON p.id = rp.permission_id + WHERE r.role_code = :code + ORDER BY p.permission_code + """ + ), + {"code": OPERATOR_ROLE_CODE}, + ) + ).all() + codes = [str(row[0]) for row in rows] + print(f"\n{OPERATOR_ROLE_CODE} 实测权限 {len(codes)} 项：{codes}") + lacked = [c for c in OPERATOR_GRANTED_CODES if c not in codes] + if lacked: + print(f"[失败] 仍未绑定的码：{lacked}") + raise SystemExit(1) + print( + "\n下一步：给运营账号绑这个角色 ——\n" + " python tools/create_test_user.py --id 9006 --username offsite_t " + "--role operator --password offsite123" + ) + + +def main() -> int: + parser = argparse.ArgumentParser(description="补齐运营角色的权限") + parser.add_argument("--dry-run", action="store_true", help="只打印将写入什么") + args = parser.parse_args() + return asyncio.run(apply(dry_run=args.dry_run)) + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/tools/portal.py b/tools/portal.py new file mode 100644 index 0000000..cd5ee8e --- /dev/null +++ b/tools/portal.py @@ -0,0 +1,1388 @@ +"""统一登录门户：一个登录入口，按角色分流到不同的工作台。 + +**和 `tools/chat_console.py` 的区别**：那个控制台没有登录、身份写死在命令行里； +本门户**走真实的 `POST /api/v1/auth/tokens`**，令牌只存活在本进程，浏览器拿不到， +然后按登录者的**真实角色**决定进哪个界面 —— 也就是照着"客户 / 员工 / 管理员"三种 +使用者的实际路径走一遍，而不是逐个接口点着测。 + +| 登录者的角色 | 进入的界面 | 主要能力 | +|---|---|---| +| `customer` | **客服** | 与客服 Agent 对话（真实受理 + 进程内驱动 Worker）、转人工、查看并确认自己的画像候选 | +| `risk_operator` / `operator` | **员工工作台** | 风控总览、预警列表与详情、预警扫描、确认/升级/解决等处置、生成日报 | +| `admin` / `super_admin` | **权限界面** | 角色清单、每个角色的权限明细、按用户查角色、审计流水、客服转人工工单、知识库清单 | +| `advisor` | **投顾工作台** | 当前投资目标、已发布方案、组合分析 | + +角色从 `IdentityService.resolve()` 现查（不是从令牌里读）—— 所以改了库里角色，重登即生效。 + +跑法： + +```powershell +D:\\conda\\envs\\jr_py313\\python.exe tools\\portal.py +# 浏览器打开 http://127.0.0.1:8101 +``` + +**前置**： + +1. MySQL / Redis 可用（平台启动时会连）； +2. 演示账号能登录：先 `python tools/seed_test_rbac.py`，再 `python tools/set_user_password.py`； +3. **客服对话前请停掉常驻 Worker**（`python -m app.worker`）—— 本门户自己驱动这一条 run， + 常驻 Worker 会与它抢 `agent_run` 队列，表现为页面一直转圈。 + +**权限问题的表现是设计如此**：某个按钮点了返回 403，说明这个角色的权限**本来就不够** +（例如客户点风控、员工点权限管理）。平台一律 fail closed，门户把 403 原样显示出来， +不做任何前端隐藏式"代为授权"。 +""" + +from __future__ import annotations + +import argparse +import sys +import uuid +from pathlib import Path +from typing import Any + +import httpx +import uvicorn +from fastapi import FastAPI, Request +from fastapi.responses import HTMLResponse, JSONResponse + +sys.stdout.reconfigure(errors="replace") + +ROOT = Path(__file__).resolve().parents[1] +if str(ROOT) not in sys.path: + sys.path.insert(0, str(ROOT)) + +CUSTOMER_AGENT = "customer_service" + +DEMO_ACCOUNTS: tuple[tuple[str, str, str], ...] = ( + ("cust_t", "123456", "客户"), + ("risk_t", "666666", "员工（风控专员）"), + ("admin_t", "88888888", "管理员"), + ("advisor_t", "abc12345", "投资顾问"), + ("offsite_t", "offsite123", "员工（运营专员）"), +) + +#: 角色 → 界面。按优先级从高到低匹配，多角色者进权限最高的那个界面。 +VIEW_BY_ROLE: tuple[tuple[str, str], ...] = ( + ("super_admin", "admin"), + ("admin", "admin"), + ("risk_operator", "staff"), + ("operator", "offsite"), # 运营走场外基金这条线，不是风控那个台 + ("advisor", "advisor"), + ("customer", "customer"), +) + +VIEW_LABELS = { + "customer": "客服", + "staff": "风控工作台", + "offsite": "运营工作台", + "admin": "权限管理", + "advisor": "投顾工作台", + "unknown": "未分配界面", +} + + +def view_for(roles: tuple[str, ...] | list[str]) -> str: + for role, view in VIEW_BY_ROLE: + if role in roles: + return view + return "unknown" + + +class Platform: + """共享的平台连接：一个 app、一个 client、一个 lifespan，所有会话共用。 + + 分开的理由：每个会话各自 `create_app()` 会把平台实例化好几遍（连接池、后台任务都翻倍）； + 而"多个浏览器标签用不同身份"又是演示时的刚需 —— 于是把**连接**共享、把**令牌**按会话分开。 + """ + + def __init__(self, base_url: str | None) -> None: + self.base_url = base_url + self.application: FastAPI | None = None + self._client: httpx.AsyncClient | None = None + self._lifespan: Any = None + self.environment: dict[str, Any] = {} + + async def client(self) -> httpx.AsyncClient: + if self._client is not None: + return self._client + if self.base_url: + self._client = httpx.AsyncClient(base_url=self.base_url, timeout=120.0) + self.environment = {"mode": "外部服务", "target": self.base_url} + return self._client + + from app.core.config import get_settings + from app.main import create_app + + application = create_app() + self.application = application + self._lifespan = application.router.lifespan_context(application) + await self._lifespan.__aenter__() + self._client = httpx.AsyncClient( + transport=httpx.ASGITransport(app=application), base_url="http://platform", timeout=120.0 + ) + dsn = get_settings().mysql_dsn + self.environment = {"mode": "进程内", "mysql": dsn.split("@")[-1] if "@" in dsn else dsn} + return self._client + + async def aclose(self) -> None: + if self._client is not None: + await self._client.aclose() + self._client = None + if self._lifespan is not None: + await self._lifespan.__aexit__(None, None, None) + self._lifespan = None + + async def request( + self, + method: str, + path: str, + *, + token: str | None, + query: dict[str, Any] | None = None, + body: Any = None, + ) -> dict[str, Any]: + if not token: + return {"status": 401, "body": {"error": {"code": "NOT_LOGGED_IN", "message": "请先登录"}}} + client = await self.client() + headers = {"Authorization": f"Bearer {token}"} + # 幂等头：平台的写接口要求 `Idempotency-Key`（16-128 位 ASCII）， + # 漏了会被 `AGENT_INPUT_INVALID: 必须提供 16-128 位 ASCII Idempotency-Key` 拒绝。 + if method.upper() in {"POST", "PUT", "PATCH", "DELETE"}: + headers["Idempotency-Key"] = uuid.uuid4().hex + try: + response = await client.request( + method.upper(), + path, + params=query or None, + json=body, + headers=headers, + ) + except Exception as exc: + return {"status": 0, "body": {"error": {"message": f"{type(exc).__name__}: {exc}"}}} + return {"status": response.status_code, "body": _json(response)} + + +class Session: + """一次登录会话：只持有令牌与身份。""" + + def __init__(self, username: str, user_id: str, token: str) -> None: + self.username = username + self.user_id = user_id + self.token = token + self.roles: tuple[str, ...] = () + self.permissions: tuple[str, ...] = () + + def as_dict(self, environment: dict[str, Any]) -> dict[str, Any]: + view = view_for(self.roles) + return { + "username": self.username, + "user_id": self.user_id, + "roles": list(self.roles), + "permissions": list(self.permissions), + "view": view, + "view_label": VIEW_LABELS[view], + "logged_in": True, + "environment": environment, + } + + +class Portal: + """多会话门户：`X-Session` 头区分标签页，令牌始终留在服务端。""" + + def __init__(self, base_url: str | None) -> None: + self.platform = Platform(base_url) + self.sessions: dict[str, Session] = {} + self.error: str | None = None + + async def login(self, session_id: str, username: str, password: str) -> dict[str, Any]: + client = await self.platform.client() + response = await client.post( + "/api/v1/auth/tokens", json={"username": username, "password": password} + ) + payload = _json(response) + if response.status_code != 200: + return {"ok": False, "status": response.status_code, "body": payload} + + data = (payload.get("data") or {}) if isinstance(payload, dict) else {} + token = data.get("access_token") or data.get("token") + if not token: + return {"ok": False, "status": response.status_code, "body": payload} + + session = Session(username, str(data.get("user_id") or _sub_of(token) or ""), token) + await self._resolve(session) + self.sessions[session_id] = session + return {"ok": True, "me": session.as_dict(self.platform.environment)} + + async def _resolve(self, session: Session) -> None: + """角色与权限现查库 —— 与平台每请求解析的口径一致（令牌里只有 sub）。""" + from app.core.contracts import RequestContext + from app.service.identity_service import IdentityService + + context = await IdentityService().resolve( + RequestContext(user_id=session.user_id, trace_id=str(uuid.uuid4())) + ) + session.roles = tuple(context.roles) + session.permissions = tuple(sorted(context.permissions)) + + def logout(self, session_id: str) -> None: + self.sessions.pop(session_id, None) + + def me(self, session_id: str) -> dict[str, Any]: + session = self.sessions.get(session_id) + data = ( + session.as_dict(self.platform.environment) + if session + else {"logged_in": False, "environment": self.platform.environment} + ) + data["accounts"] = [ + {"username": u, "password": p, "label": label} for u, p, label in DEMO_ACCOUNTS + ] + data["startup_error"] = self.error + return data + + async def call( + self, + session_id: str, + method: str, + path: str, + query: dict[str, Any] | None = None, + body: Any = None, + ) -> dict[str, Any]: + session = self.sessions.get(session_id) + return await self.platform.request( + method, path, token=session.token if session else None, query=query, body=body + ) + + async def chat(self, session_id: str, message: str, chat_session: str) -> dict[str, Any]: + """客服对话：真实受理 → 本进程驱动这一条 run → 取结果。""" + from app.worker.runtime import WorkerRuntime + + accepted = await self.call( + session_id, + "POST", + "/api/v1/agent-runs", + body={ + "agent_type": CUSTOMER_AGENT, + "message": message, + "session_id": chat_session, + "idempotency_key": uuid.uuid4().hex, + }, + ) + if accepted["status"] != 202: + return {"ok": False, "status": accepted["status"], "body": accepted["body"]} + + run_id = ((accepted["body"] or {}).get("data") or {}).get("run_id") + if not run_id: + return {"ok": False, "status": accepted["status"], "body": accepted["body"]} + + await WorkerRuntime().execute(run_id) + detail = await self.call(session_id, "GET", f"/api/v1/agent-runs/{run_id}") + data = (detail["body"] or {}).get("data") or {} + result = data.get("result") or {} + raw_intent = result.get("intent") + intent = raw_intent.get("intent") if isinstance(raw_intent, dict) else raw_intent + return { + "ok": True, + "run_id": run_id, + "status": data.get("status"), + "answer": str(result.get("content") or ""), + "intent": intent or "", + "transfer": bool(result.get("transfer_required")), + "transfer_reason": result.get("transfer_reason") or "", + } + + +def _json(response: httpx.Response) -> Any: + try: + return response.json() + except Exception: + return {"_raw": response.text[:3000]} + + +def _sub_of(token: str) -> str | None: + import jwt + + try: + return str(jwt.decode(token, options={"verify_signature": False}).get("sub")) + except Exception: + return None + + +def build_portal(state: Portal) -> FastAPI: + # 注意：`Request` 必须在**模块顶层**导入。本文件启用了 + # `from __future__ import annotations`，注解在此处是字符串，FastAPI 解析 + # `request: Request` 时只在模块全局里找类型名 —— 写成函数内 import 会被 + # 当成"必填 query 参数 request"，所有请求直接 422，且报错完全看不出是这个原因。 + portal = FastAPI(title="统一登录门户") + + def sid(request: Request) -> str: + return request.headers.get("X-Session") or "default" + + @portal.get("/", response_class=HTMLResponse) + async def index() -> HTMLResponse: + return HTMLResponse(PAGE) + + @portal.get("/api/me") + async def me(request: Request) -> JSONResponse: + return JSONResponse(state.me(sid(request))) + + @portal.post("/api/login") + async def login(request: Request, payload: dict[str, Any]) -> JSONResponse: + try: + result = await state.login( + sid(request), str(payload.get("username") or ""), str(payload.get("password") or "") + ) + state.error = None + except Exception as exc: + state.error = f"{type(exc).__name__}: {exc}" + result = {"ok": False, "status": 0, "body": {"error": {"message": state.error}}} + return JSONResponse(result) + + @portal.post("/api/logout") + async def logout(request: Request) -> JSONResponse: + state.logout(sid(request)) + return JSONResponse({"ok": True}) + + @portal.post("/api/call") + async def call(request: Request, payload: dict[str, Any]) -> JSONResponse: + try: + result = await state.call( + sid(request), + str(payload.get("method") or "GET"), + str(payload.get("path") or "/"), + payload.get("query"), + payload.get("body"), + ) + except Exception as exc: + result = {"status": 0, "body": {"error": {"message": f"{type(exc).__name__}: {exc}"}}} + return JSONResponse(result) + + @portal.post("/api/chat") + async def chat(request: Request, payload: dict[str, Any]) -> JSONResponse: + message = str(payload.get("message") or "").strip() + chat_session = str(payload.get("session_id") or f"portal-{uuid.uuid4().hex[:8]}") + if not message: + return JSONResponse({"ok": False, "body": {"error": {"message": "请输入内容"}}}) + try: + return JSONResponse(await state.chat(sid(request), message, chat_session)) + except Exception as exc: + return JSONResponse( + {"ok": False, "body": {"error": {"message": f"{type(exc).__name__}: {exc}"}}} + ) + + return portal + + +PAGE = r""" + + + + +基金智能服务平台 + + + + +
+
+

基金智能服务平台

+

请使用账号登录，系统会按你的角色进入对应界面。

+ + + + + +
+ 演示账号（点击填入）：
+ +
+ +
+
+ +
+
+ 基金智能服务平台 + + + + + +
+ +
+
+ + + + +""" + + +def main() -> None: + parser = argparse.ArgumentParser(description="统一登录门户（按角色分流）") + parser.add_argument("--port", type=int, default=8101) + parser.add_argument("--base-url", default=None, help="指向已在运行的服务；省略则进程内直挂平台 app") + args = parser.parse_args() + + state = Portal(args.base_url) + mode = args.base_url or "进程内（走真实中间件与鉴权栈）" + print(f"登录门户：http://127.0.0.1:{args.port} 模式={mode}") + print("演示账号：cust_t/123456 · risk_t/666666 · admin_t/88888888 · advisor_t/abc12345") + print("客服对话前请停掉常驻 Worker；账号需先跑 seed_test_rbac.py + set_user_password.py。") + uvicorn.run(build_portal(state), host="127.0.0.1", port=args.port, log_level="warning") + + +if __name__ == "__main__": + main() diff --git a/tools/seed_sim_account_demo.py b/tools/seed_sim_account_demo.py new file mode 100644 index 0000000..311b808 --- /dev/null +++ b/tools/seed_sim_account_demo.py @@ -0,0 +1,277 @@ +"""场内模拟交易演示种子（§T 用户自助）。 + +按 D4 决策：1 个客户 + 10 万初始资金 + 2 只基金的初始持仓。 + +执行： + python -m tools.seed_sim_account_demo [--customer-id N] + +## 已知表结构问题（2026-09-12 实测） + +底座 `fin_*` 系列表的 `id` 列**没有** `AUTO_INCREMENT` 属性（仅 `fin_knowledge_meta` 有）。 +SQLAlchemy ORM 默认期望 auto-increment，会在 INSERT 时省略 id → MySQL 报 +`Field 'id' doesn't have a default value`。**底层基础规则不允许 DDL 改动**， +所以本脚本显式查 `MAX(id) + 1` 分配下一个 id，写入时携带 id（不依赖 auto-increment）。 + +实现：使用 SQLAlchemy Core `insert(...).values(id=..., ...)`，而不是 ORM `session.add()` +——后者对 BIGINT 不会自动注入 id。 + +幂等：再次执行不会重复建记录（先查 product_code / customer_id / customer_id+product_id）。 +""" + +from __future__ import annotations + +import argparse +import asyncio +import sys +from datetime import UTC, datetime, timedelta +from decimal import Decimal + +from sqlalchemy import func, insert, select +from sqlalchemy.orm import Session + +from app.core.config import get_settings +from app.infrastructure.db import SessionFactory +from app.model.fund import ( + FundHolding, + FundMarketPrice, + FundProduct, + FundSimAccount, +) + + +CUSTOMER_ID = 9001 +INITIAL_BALANCE = Decimal("100000.00") +DEMO_PRODUCTS = [ + { + "product_code": "510300", + "product_name": "沪深300ETF", + "exchange_code": "SSE", + "product_category": "ETF", + "risk_level": "R3", + "lot_size": Decimal("100"), + "price_tick": Decimal("0.001"), + "close_price": Decimal("4.5000"), + "total_fund_shares": Decimal("10000000000"), + "initial_quantity": Decimal("1000"), + }, + { + "product_code": "510500", + "product_name": "南方中证500ETF", + "exchange_code": "SSE", + "product_category": "ETF", + "risk_level": "R3", + "lot_size": Decimal("100"), + "price_tick": Decimal("0.001"), + "close_price": Decimal("6.2000"), + "total_fund_shares": Decimal("8000000000"), + "initial_quantity": Decimal("800"), + }, +] + + +async def _next_id(session: Session, model) -> int: + """返回该表下一个可用的 id（不依赖 AUTO_INCREMENT）。""" + pk_col = model.__table__.primary_key.columns[0] + result = await session.execute(select(func.coalesce(func.max(pk_col), 0))) + return int(result.scalar_one()) + 1 + + +async def _upsert_product(session: Session, spec: dict) -> int: + existing = ( + await session.execute( + select(FundProduct.id).where(FundProduct.product_code == spec["product_code"]) + ) + ).scalar_one_or_none() + if existing is not None: + return int(existing) + now = datetime.now(UTC).replace(tzinfo=None) + next_id = await _next_id(session, FundProduct) + stmt = insert(FundProduct).values( + id=next_id, + product_code=spec["product_code"], + product_name=spec["product_name"], + exchange_code=spec["exchange_code"], + product_category=spec["product_category"], + risk_level=spec["risk_level"], + fund_manager="南方基金", + currency="CNY", + lot_size=spec["lot_size"], + price_tick=spec["price_tick"], + current_nav=spec["close_price"], + current_nav_at=now, + min_amount=Decimal("100.00"), + open_start_at=now - timedelta(days=365), + open_end_at=None, + transaction_fee_rate=None, + single_investor_max_holding_ratio=Decimal("5.0000"), + management_fee_rate=Decimal("0.50"), + custodian_fee_rate=Decimal("0.10"), + risk_disclosure_required=0, + second_confirmation_required=0, + recording_required=0, + status="上市", + created_at=now, + updated_at=now, + ) + await session.execute(stmt) + return next_id + + +async def _upsert_market_price(session: Session, product_id: int, spec: dict) -> None: + today = datetime.now(UTC).date() + existing = ( + await session.execute( + select(FundMarketPrice.id).where( + FundMarketPrice.product_id == product_id, + FundMarketPrice.trade_date == today, + ) + ) + ).scalar_one_or_none() + if existing is not None: + return + now = datetime.now(UTC).replace(tzinfo=None) + next_id = await _next_id(session, FundMarketPrice) + stmt = insert(FundMarketPrice).values( + id=next_id, + product_id=product_id, + trade_date=today, + open_price=spec["close_price"], + high_price=spec["close_price"] + Decimal("0.05"), + low_price=spec["close_price"] - Decimal("0.05"), + close_price=spec["close_price"], + volume=Decimal("1000000"), + turnover_amount=spec["close_price"] * Decimal("1000000"), + total_fund_shares=spec["total_fund_shares"], + source="eastmoney_demo_seed", + source_updated_at=now, + created_at=now, + ) + await session.execute(stmt) + + +async def _upsert_account(session: Session, customer_id: int) -> FundSimAccount: + existing = ( + await session.execute( + select(FundSimAccount.id).where(FundSimAccount.customer_id == customer_id) + ) + ).scalar_one_or_none() + if existing is not None: + # 返回完整对象 + return ( + await session.execute( + select(FundSimAccount).where(FundSimAccount.customer_id == customer_id) + ) + ).scalar_one() + now = datetime.now(UTC).replace(tzinfo=None) + next_id = await _next_id(session, FundSimAccount) + stmt = insert(FundSimAccount).values( + id=next_id, + account_no=f"FSA{customer_id:06d}", + customer_id=customer_id, + currency="CNY", + cash_balance=INITIAL_BALANCE, + available_cash=INITIAL_BALANCE, + frozen_cash=Decimal("0"), + initial_balance=INITIAL_BALANCE, + status="正常", + version=0, + created_at=now, + updated_at=now, + ) + await session.execute(stmt) + return ( + await session.execute( + select(FundSimAccount).where(FundSimAccount.customer_id == customer_id) + ) + ).scalar_one() + + +async def _upsert_holding( + session: Session, customer_id: int, product_id: int, spec: dict +) -> None: + existing = ( + await session.execute( + select(FundHolding.id).where( + FundHolding.customer_id == customer_id, + FundHolding.product_id == product_id, + ) + ) + ).scalar_one_or_none() + if existing is not None: + return + qty = spec["initial_quantity"] + cost = (qty * spec["close_price"]).quantize(Decimal("0.01")) + now = datetime.now(UTC).replace(tzinfo=None) + next_id = await _next_id(session, FundHolding) + stmt = insert(FundHolding).values( + id=next_id, + customer_id=customer_id, + trade_account=f"FSA{customer_id:06d}", + product_id=product_id, + total_quantity=qty, + # `shares`/`current_value` 在 docs/00 §6.2 标注为"生成列"，但实测 MySQL 表 + # `GENERATION_EXPRESSION=''`——表结构里是普通 NOT NULL 列、无默认值。 + # 写时必须显式给值，按 §6.2 的语义填（shares=total_quantity, current_value=cost）。 + shares=qty, + available_quantity=qty, + frozen_quantity=Decimal("0"), + average_cost=spec["close_price"], + cost_amount=cost, + market_value=None, + current_value=cost, + profit_loss=None, + profit_loss_ratio=None, + status="持有中", + first_acquired_at=now, + version=0, + updated_at=now, + ) + await session.execute(stmt) + + +async def run(customer_id: int) -> None: + s = get_settings() + print(f"数据库：{s.mysql_dsn.split('@')[-1]}") + print(f"目标客户 ID = {customer_id}") + async with SessionFactory() as session: + async with session.begin(): + product_ids: list[int] = [] + for spec in DEMO_PRODUCTS: + pid = await _upsert_product(session, spec) + await _upsert_market_price(session, pid, spec) + product_ids.append(pid) + print(f" ✓ 演示产品 {len(product_ids)} 个 + 当日行情") + account = await _upsert_account(session, customer_id) + for pid, spec in zip(product_ids, DEMO_PRODUCTS, strict=True): + await _upsert_holding(session, customer_id, pid, spec) + total_cost = sum( + (spec["initial_quantity"] * spec["close_price"]).quantize(Decimal("0.01")) + for spec in DEMO_PRODUCTS + ) + from sqlalchemy import update as sa_update + account.cash_balance = (INITIAL_BALANCE - total_cost).quantize(Decimal("0.01")) + account.available_cash = account.cash_balance + account.updated_at = datetime.now(UTC).replace(tzinfo=None) + await session.execute( + sa_update(FundSimAccount) + .where(FundSimAccount.id == account.id) + .values( + cash_balance=account.cash_balance, + available_cash=account.available_cash, + updated_at=account.updated_at, + ) + ) + print(f" ✓ 虚拟账户 {account.account_no} 初始余额 ¥{INITIAL_BALANCE}") + print(f" ✓ 持仓已建立，账户剩余 ¥{account.cash_balance}（已扣持仓成本 ¥{total_cost}）") + + +def main() -> int: + parser = argparse.ArgumentParser(description="场内模拟交易演示种子") + parser.add_argument("--customer-id", type=int, default=CUSTOMER_ID) + args = parser.parse_args() + asyncio.run(run(args.customer_id)) + return 0 + + +if __name__ == "__main__": + sys.exit(main()) \ No newline at end of file diff --git a/tools/seed_test_rbac.py b/tools/seed_test_rbac.py index 1c9c656..a4be30b 100644 --- a/tools/seed_test_rbac.py +++ b/tools/seed_test_rbac.py @@ -100,6 +100,7 @@ PERMISSIONS: tuple[tuple[int, str, str, str, str], ...] = ( (9044, "memory:candidate:confirm", "memory:candidate", "confirm", "self"), (9045, "memory:candidate:review", "memory:candidate", "review", "all"), (9046, "handover:read", "handover", "read", "all"), + # ---- 9047-9050：风控模块四个权限码 ---- # 风控 Service 和 Agent 工具都按这四个权限码失败关闭。此前依赖 # `grant_risk_permissions.py` 临时补种，重跑本脚本时会删除 9001-9099 号段内的 @@ -108,6 +109,46 @@ PERMISSIONS: tuple[tuple[int, str, str, str, str], ...] = ( (9048, "risk:alert:write", "risk", "alert", "all"), (9049, "risk:alert:scan", "risk", "alert", "all"), (9050, "risk:report:mail", "risk", "report", "all"), + # ---- 9051-9056：代码早就要求、但**从未建过**的 6 个权限码 ---- + # 症状同样是 `AGENT_PERMISSION_DENIED: 缺少操作权限`：看着像角色配错，其实是 + # **权限码根本不存在**，于是任何角色都过不去。对账工具见 + # `tools/check_permission_coverage.py`（把"代码声明的权限"与"库里各角色拥有的"比一遍）。 + # 推广材料：`promotion_material_service.py` 用这四个；第 164 行专门判 + # `"advisor" in context.roles`，说明投顾本来就在这条业务线上。 + (9051, "promotion:read", "promotion", "read", "all"), + (9052, "promotion:write", "promotion", "write", "all"), + (9053, "promotion:review", "promotion", "review", "all"), + (9054, "promotion:deliver", "promotion", "deliver", "all"), + # NL2SQL 工具声明该权限（`bootstrap.py`）；`financial_nl2sql_service.py:272` 的角色 + # 白名单是 {advisor, operator, admin, super_admin} —— 投顾与运营都要用它。 + (9055, "financial:nl2sql:read", "financial", "nl2sql", "all"), + # 平台验证探针工具（`platform_probe.py`），只给 admin。 + (9056, "probe:read", "probe", "read", "all"), + # ---- 9057-9059：`investment-goal:{action}:customer` 三个变体 ---- + # 这三个是**动态拼出来的**：`investment_goal_service.py:270-286` 的 + # `_assert_customer_access` 按 `customer_id == 自己` 决定拼 `:self` 还是 `:customer`， + # 所以对账工具抓不到字面量，一直以为权限齐了 —— 实际投顾查/建/确认**客户**的目标 + # 全部 403。action 取值来自调用点：`write`(L43) / `confirm`(L110) / `read`(L138,155)。 + # + # `data_scope` 必须是 `own_customers`：那段代码在后面还会校一次 + # `scope != "all" and (scope != "own_customers" or customer_id not in context.customer_ids)` + # ⇒ 只有 `own_customers` 且客户确在投顾名下才放行。这是**最小权限**的正确形态： + # 投顾只看自己服务的客户，而不是全量客户。 + (9057, "investment-goal:read:customer", "investment-goal", "read", "own_customers"), + (9058, "investment-goal:write:customer", "investment-goal", "write", "own_customers"), + (9059, "investment-goal:confirm:customer", "investment-goal", "confirm", "own_customers"), + + # ---- 9060-9065：账户看板 + 场内模拟交易（T 段 9 端点） ---- + # ZSY_develop §T 段（docs/05 §19）用户自助端点的权限码。 + # Controller 当前不主动 require，但为与"业务契约 = 权限码声明"一致仍登记进 + # `PERMISSIONS`，同时挂在 `CUSTOMER_PERMISSIONS` 让 customer 角色自带。 + # 号段续 9060（避开 9047-9059 qyqy 风险/推广/探针/投资目标号段）：与 9041-9046 客服二期间隔 1，避免与既有迁移/种子冲突。 + (9060, "account:read:self", "account", "read", "self"), + (9061, "trade:order:create", "trade", "order", "create"), + (9062, "trade:order:read", "trade", "order", "read"), + (9063, "trade:order:cancel", "trade", "order", "cancel"), + (9064, "holding:read:self", "holding", "read", "self"), + (9065, "trade:txn:read", "trade", "txn", "read"), ) # 客户：业务侧自助能力（自己的会话、反馈、转人工、自己的记忆画像）。 @@ -116,6 +157,8 @@ CUSTOMER_PERMISSIONS = ( 9020, 9021, 9022, 9023, 9024, 9025, 9026, 9034, # 客服二期：客户确认/拒绝**自己**的画像候选（服务层按 customer_id 过滤，不越权）。 9044, + # ZSY §T：账户看板与场内模拟交易（首版仅 customer 角色可用，留 admin 全量） + 9060, 9061, 9062, 9063, 9064, 9065, ) # 风控专员：业务侧只读 + 跨客户记忆 + 审计只读，不含配置写权限。 RISK_PERMISSIONS = (