diff --git a/app/api/controllers/onboarding.py b/app/api/controllers/onboarding.py new file mode 100644 index 0000000..f6d70b0 --- /dev/null +++ b/app/api/controllers/onboarding.py @@ -0,0 +1,27 @@ +"""Mandatory customer onboarding endpoints.""" + +from fastapi import APIRouter, Depends, Header, status + +from app.api.dependencies.auth import build_request_context +from app.api.schemas.risk_questionnaire import RiskQuestionnaireSubmission +from app.core.contracts import RequestContext +from app.service.risk_questionnaire_service import RiskQuestionnaireService + +router = APIRouter(prefix="/api/v1/onboarding", tags=["customer-onboarding"]) + + +@router.get("/risk-questionnaire") +async def get_risk_questionnaire( + context: RequestContext = Depends(build_request_context), # noqa: B008 +) -> dict[str, object]: + return await RiskQuestionnaireService().questionnaire(context) + + +@router.post("/risk-questionnaire/submissions", status_code=status.HTTP_201_CREATED) +async def submit_risk_questionnaire( + payload: RiskQuestionnaireSubmission, + context: RequestContext = Depends(build_request_context), # noqa: B008 + key: str | None = Header(default=None, alias="Idempotency-Key"), +) -> dict[str, object]: + return await RiskQuestionnaireService().submit(payload, context, key) + diff --git a/app/api/dependencies/auth.py b/app/api/dependencies/auth.py index 8595847..0547a99 100644 --- a/app/api/dependencies/auth.py +++ b/app/api/dependencies/auth.py @@ -54,4 +54,11 @@ async def build_request_context( # 令牌非法、账号停用、角色读取失败一律按 401 处理，形态完全一致。 raise unauthorized() from exc request.state.request_context = context + if not request.url.path.startswith("/api/v1/onboarding/"): + from app.service.risk_questionnaire_service import RiskQuestionnaireService + + if await RiskQuestionnaireService().is_required(context): + from app.core.errors import OnboardingRequiredError + + raise OnboardingRequiredError("请先完成开户风险测评问卷") return context diff --git a/app/api/schemas/risk_questionnaire.py b/app/api/schemas/risk_questionnaire.py new file mode 100644 index 0000000..5dab314 --- /dev/null +++ b/app/api/schemas/risk_questionnaire.py @@ -0,0 +1,6 @@ +"""HTTP DTOs for customer opening risk questionnaires.""" + +from app.core.risk_questionnaire_contracts import RiskQuestionnaireSubmission + +__all__ = ["RiskQuestionnaireSubmission"] + diff --git a/app/core/errors.py b/app/core/errors.py index 1a5ac77..67064cc 100644 --- a/app/core/errors.py +++ b/app/core/errors.py @@ -53,6 +53,10 @@ class ForbiddenAgentError(AgentError): status_code = 403 +class OnboardingRequiredError(ForbiddenAgentError): + """开户前置条件未满足，沿用文档登记的权限错误码。""" + + class AgentPermissionDeniedError(ForbiddenAgentError): """`AGENT_PERMISSION_DENIED` 的语义化别名，供新代码使用。""" diff --git a/app/core/risk_questionnaire_contracts.py b/app/core/risk_questionnaire_contracts.py new file mode 100644 index 0000000..8bfa9d0 --- /dev/null +++ b/app/core/risk_questionnaire_contracts.py @@ -0,0 +1,89 @@ +"""Customer-facing opening questionnaire contract without scoring details.""" + +from typing import Any, Literal + +from pydantic import BaseModel, ConfigDict, model_validator + +QUESTIONNAIRE_VERSION = "opening-risk-v1" + +QUESTIONS: tuple[dict[str, Any], ...] = ( + {"id": "q1", "title": "您的职业是？", "options": [ + "经济和金融行业人员", "党政群机关、社会组织工作人员", "事业单位工作人员", + "法律、社会和宗教人员", "教科文卫专业人员", "社会生产服务和生活服务人员", + "企业单位工作人员", "生产制造及有关人员", "工程技术人员", + "农林牧渔水利生产及辅助人员", "军人", "学生", "无业", "个体户", "自由职业者", + "不便分类的其他从业人员", + ]}, + {"id": "q2", "title": "您的学历是？", "options": ["硕士及以上", "本科", "大专", "高中及以下"]}, + {"id": "q3", "title": "您的主要收入来源是？", "options": [ + "工资、劳务报酬", "生产经营所得", "出租、出售房地产等非金融性资产收入", + "利息、股息、转让等金融资产收入", "无固定收入", + ]}, + {"id": "q4", "title": "您的家庭可支配年收入为（折合人民币）：", "options": [ + "500 万元以上", "100-500 万元", "50-100 万元", "50 万元以下", + ]}, + {"id": "q5", "title": ( + "在您每年的家庭可支配收入中，可用于金融投资（储蓄存款除外）的比例为：" + ), "options": [ + "大于 50%", "25% 至 50%", "10% 至 25%", "小于 10%", + ]}, + {"id": "q6", "title": "您是否有尚未清偿的债务：", "options": [ + "没有", "有，占资产比例不超过 10%", "有，占资产比例不超过 20%", + "有，占资产比例不超过 50%", "有，占资产比例超过 50%", + ]}, + {"id": "q7", "title": "您的投资经历：", "options": [ + "大部分投资于股票、外汇、期货等", "大部分投资于基金、股票、信托产品等", + "资产均衡分布于存款、国债、银行理财产品、股票、基金等", "大部分投资于存款、国债等", + "没有证券期货投资知识或金融投资经验", + ]}, + {"id": "q8", "title": "您有多少年投资基金、股票、信托或金融衍生产品等风险投资品的经验：", + "options": [ + "5 年以上", "2 至 5 年", "少于 2 年", "没有经验", + ]}, + {"id": "q9", "title": "您计划投资多久：", "options": [ + "5 年以上", "3 至 5 年", "1 至 3 年", "1 年以下", + ]}, + {"id": "q10", "title": "您打算重点投资于哪些种类的产品：", "options": [ + "A、债券、货币市场基金、债券基金等固定收益类投资品种", + "B、股票、混合型基金、股票型基金等权益类投资品种及 A 选项中的投资品种", + "C、期货、期权等金融衍生品及 B 选项中的投资品种", + "D、C 选项中的投资品种及其他产品或者服务", + ]}, + {"id": "q11", "title": "以下哪项描述最符合您的投资态度：", "options": [ + "希望取得高收益，能够接受长期波动，包括本金亏损", + "寻求资金的较高收益，愿意为此承担有限的本金亏损", + "稳健投资，愿意接受短期亏损，但无法接受可能出现的大幅波动", + "保守投资，能够容忍少量本金亏损，愿意承担一定幅度的收益波动", + "厌恶风险，不愿承受任何投资损失，追求稳定回报", + ]}, + {"id": "q12", "title": ( + "产品 A 预期收益 10%、损失较小；产品 B 预期收益 30%、亏损较大。您会怎么支配投资：" + ), "options": [ + "全部投资于 B", "大部分投资于 B", "大部分投资于 A", "全部投资于 A", + ]}, + {"id": "q13", "title": "您认为自己能承受的最大投资损失是多少：", "options": [ + "可超过 50%", "30%-50%", "10%-30%", "10% 以内", + ]}, +) +QUESTION_OPTION_COUNTS = {str(item["id"]): len(item["options"]) for item in QUESTIONS} +DECLARATION = ( + "本人保证提供的信息真实、准确、完整，知晓并确认信息发生重要变化、可能影响投资者分类的，" + "应当及时更新并告知平台。" +) + + +class RiskQuestionnaireSubmission(BaseModel): + model_config = ConfigDict(extra="forbid", frozen=True) + + answers: dict[str, int] + declaration_accepted: Literal[True] + + @model_validator(mode="after") + def validate_answers(self) -> "RiskQuestionnaireSubmission": + if set(self.answers) != set(QUESTION_OPTION_COUNTS): + raise ValueError("answers must include every questionnaire question exactly once") + for question_id, option_id in self.answers.items(): + if not 1 <= option_id <= QUESTION_OPTION_COUNTS[question_id]: + raise ValueError(f"invalid option for {question_id}") + return self + diff --git a/app/main.py b/app/main.py index a8ed6dc..94d932a 100644 --- a/app/main.py +++ b/app/main.py @@ -6,6 +6,7 @@ from app.api.controllers.agent_runs import router as agent_runs_router from app.api.controllers.conversations import router as conversations_router from app.api.controllers.health import router as health_router from app.api.controllers.knowledge import router as knowledge_router +from app.api.controllers.onboarding import router as onboarding_router from app.api.controllers.public_platform import router as public_platform_router from app.api.middleware import attach_trace_id from app.core.config import get_settings @@ -45,6 +46,7 @@ def create_app() -> FastAPI: application.include_router(public_platform_router) application.include_router(knowledge_router) application.include_router(health_router) + application.include_router(onboarding_router) application.include_router(admin_router) return application diff --git a/app/model/profile_tag.py b/app/model/profile_tag.py new file mode 100644 index 0000000..060d390 --- /dev/null +++ b/app/model/profile_tag.py @@ -0,0 +1,58 @@ +"""Append-only evidence and review records for internal customer-profile tags.""" + +from datetime import datetime +from decimal import Decimal +from typing import Any + +from sqlalchemy import JSON, BigInteger, DateTime, Numeric, String +from sqlalchemy.orm import Mapped, mapped_column + +from app.model.base import Base + + +class AdvisorProfileDriftReview(Base): + """A candidate profile held for review when its tags materially drift.""" + + __tablename__ = "advisor_profile_drift_review" + + id: Mapped[int] = mapped_column(BigInteger, primary_key=True) + drift_no: Mapped[str] = mapped_column(String(36), nullable=False, unique=True) + customer_id: Mapped[int] = mapped_column(BigInteger, nullable=False) + source_assessment_id: Mapped[int] = mapped_column(BigInteger, nullable=False) + candidate_profile_uuid: Mapped[str] = mapped_column(String(36), nullable=False, unique=True) + candidate_profile_version: Mapped[int] = mapped_column(BigInteger, nullable=False) + candidate_snapshot: Mapped[dict[str, Any]] = mapped_column(JSON, nullable=False) + candidate_generation_basis: Mapped[dict[str, Any]] = mapped_column(JSON, nullable=False) + changed_tags: Mapped[list[dict[str, Any]]] = mapped_column(JSON, nullable=False) + status: Mapped[str] = mapped_column(String(16), nullable=False) + reviewer_user_id: Mapped[int | None] = mapped_column(BigInteger) + reviewed_at: Mapped[datetime | None] = mapped_column(DateTime) + review_comment: Mapped[str | None] = mapped_column(String(1000)) + created_at: Mapped[datetime] = mapped_column(DateTime, nullable=False) + updated_at: Mapped[datetime] = mapped_column(DateTime, nullable=False) + + +class AdvisorProfileTag(Base): + """A tag value with evidence lineage. Exactly one active row exists per tag key.""" + + __tablename__ = "advisor_profile_tag" + + id: Mapped[int] = mapped_column(BigInteger, primary_key=True) + tag_uuid: Mapped[str] = mapped_column(String(36), nullable=False, unique=True) + customer_id: Mapped[int] = mapped_column(BigInteger, nullable=False) + tag_key: Mapped[str] = mapped_column(String(64), nullable=False) + tag_value: Mapped[Any] = mapped_column(JSON, nullable=False) + tag_value_hash: Mapped[str] = mapped_column(String(64), nullable=False) + confidence: Mapped[Decimal] = mapped_column(Numeric(5, 4), nullable=False) + source_type: Mapped[str] = mapped_column(String(32), nullable=False) + source_reference: Mapped[str] = mapped_column(String(128), nullable=False) + source_confidence: Mapped[Decimal] = mapped_column(Numeric(5, 4), nullable=False) + profile_version: Mapped[int] = mapped_column(BigInteger, nullable=False) + drift_review_id: Mapped[int | None] = mapped_column(BigInteger) + previous_tag_id: Mapped[int | None] = mapped_column(BigInteger) + drift_reason: Mapped[str | None] = mapped_column(String(32)) + status: Mapped[str] = mapped_column(String(16), nullable=False) + active_customer_tag: Mapped[str | None] = mapped_column(String(160), unique=True) + created_at: Mapped[datetime] = mapped_column(DateTime, nullable=False) + updated_at: Mapped[datetime] = mapped_column(DateTime, nullable=False) + diff --git a/app/model/risk_questionnaire.py b/app/model/risk_questionnaire.py new file mode 100644 index 0000000..6bd96bb --- /dev/null +++ b/app/model/risk_questionnaire.py @@ -0,0 +1,30 @@ +"""Additive projection model for opening-risk questionnaire profiles.""" + +from datetime import datetime +from typing import Any + +from sqlalchemy import JSON, BigInteger, DateTime, String +from sqlalchemy.orm import Mapped, mapped_column + +from app.model.base import Base +from app.model.fund import FundRiskAssessment as RiskAssessment + + +class ProfileSnapshot(Base): + __tablename__ = "profile_snapshots" + + id: Mapped[int] = mapped_column(BigInteger, primary_key=True) + profile_uuid: Mapped[str | None] = mapped_column(String(36), unique=True) + customer_id: Mapped[int] = mapped_column(BigInteger, nullable=False) + version: Mapped[int] = mapped_column(BigInteger, nullable=False) + snapshot: Mapped[dict[str, Any]] = mapped_column(JSON, nullable=False) + generation_basis: Mapped[dict[str, Any] | None] = mapped_column(JSON) + snapshot_hash: Mapped[str | None] = mapped_column(String(64)) + is_current: Mapped[bool] = mapped_column(nullable=False, default=False) + current_customer_id: Mapped[int | None] = mapped_column(BigInteger) + generated_at: Mapped[datetime | None] = mapped_column(DateTime) + created_at: Mapped[datetime] = mapped_column(DateTime, nullable=False) + updated_at: Mapped[datetime] = mapped_column(DateTime, nullable=False) + + +__all__ = ["ProfileSnapshot", "RiskAssessment"] diff --git a/app/repository/risk_questionnaire_repository.py b/app/repository/risk_questionnaire_repository.py new file mode 100644 index 0000000..3372601 --- /dev/null +++ b/app/repository/risk_questionnaire_repository.py @@ -0,0 +1,171 @@ +"""Repository for formal risk assessments and internal profile projections.""" + +from datetime import datetime +from typing import cast + +from sqlalchemy import func, select, update +from sqlalchemy.ext.asyncio import AsyncSession + +from app.model.fund import FundRiskAssessment as RiskAssessment +from app.model.memory import MemorySyncOutbox +from app.model.profile_tag import AdvisorProfileDriftReview, AdvisorProfileTag +from app.model.risk_questionnaire import ProfileSnapshot + + +class RiskQuestionnaireRepository: + def __init__(self, session: AsyncSession) -> None: + self.session = session + + async def has_valid_assessment(self, customer_id: int, now: datetime) -> bool: + return await self.session.scalar( + select(RiskAssessment.id) + .where(RiskAssessment.customer_id == customer_id, RiskAssessment.valid_until > now) + .limit(1) + ) is not None + + async def assessments_since(self, customer_id: int, since: datetime) -> int: + value = await self.session.scalar( + select(func.count()) + .select_from(RiskAssessment) + .where(RiskAssessment.customer_id == customer_id, RiskAssessment.assessed_at >= since) + ) + return int(value or 0) + + async def next_profile_version(self, customer_id: int) -> int: + value = await self.session.scalar( + select(func.coalesce(func.max(ProfileSnapshot.version), 0)).where( + ProfileSnapshot.customer_id == customer_id + ) + ) + return int(value or 0) + 1 + + async def deactivate_current_profile(self, customer_id: int, now: datetime) -> None: + await self.session.execute( + update(ProfileSnapshot) + .where(ProfileSnapshot.customer_id == customer_id, ProfileSnapshot.is_current.is_(True)) + .values(is_current=False, updated_at=now) + ) + + def add_assessment(self, assessment: RiskAssessment) -> None: + self.session.add(assessment) + + def add_profile(self, profile: ProfileSnapshot) -> None: + self.session.add(profile) + + def add_sync_event(self, event: MemorySyncOutbox) -> None: + self.session.add(event) + + async def latest_assessment(self, customer_id: int) -> RiskAssessment | None: + return cast(RiskAssessment | None, await self.session.scalar( + select(RiskAssessment) + .where(RiskAssessment.customer_id == customer_id) + .order_by(RiskAssessment.assessed_at.desc(), RiskAssessment.id.desc()) + .limit(1) + )) + + async def current_profile(self, customer_id: int) -> ProfileSnapshot | None: + return cast(ProfileSnapshot | None, await self.session.scalar( + select(ProfileSnapshot) + .where( + ProfileSnapshot.customer_id == customer_id, + ProfileSnapshot.is_current.is_(True), + ) + .order_by(ProfileSnapshot.version.desc(), ProfileSnapshot.id.desc()) + .limit(1) + )) + + async def active_tags( + self, customer_id: int, *, lock: bool = False + ) -> list[AdvisorProfileTag]: + statement = ( + select(AdvisorProfileTag) + .where( + AdvisorProfileTag.customer_id == customer_id, + AdvisorProfileTag.status == "active", + ) + .order_by(AdvisorProfileTag.tag_key.asc(), AdvisorProfileTag.id.desc()) + ) + if lock: + statement = statement.with_for_update() + return list(await self.session.scalars(statement)) + + async def tags(self, customer_id: int, *, limit: int = 100) -> list[AdvisorProfileTag]: + return list(await self.session.scalars( + select(AdvisorProfileTag) + .where(AdvisorProfileTag.customer_id == customer_id) + .order_by(AdvisorProfileTag.tag_key.asc(), AdvisorProfileTag.id.desc()) + .limit(limit) + )) + + async def pending_drift_review( + self, customer_id: int, *, lock: bool = False + ) -> AdvisorProfileDriftReview | None: + statement = ( + select(AdvisorProfileDriftReview) + .where( + AdvisorProfileDriftReview.customer_id == customer_id, + AdvisorProfileDriftReview.status == "pending_review", + ) + .order_by( + AdvisorProfileDriftReview.created_at.desc(), + AdvisorProfileDriftReview.id.desc(), + ) + .limit(1) + ) + if lock: + statement = statement.with_for_update() + return cast(AdvisorProfileDriftReview | None, await self.session.scalar(statement)) + + async def drift_review( + self, review_id: int, *, lock: bool = False + ) -> AdvisorProfileDriftReview | None: + statement = select(AdvisorProfileDriftReview).where( + AdvisorProfileDriftReview.id == review_id + ) + if lock: + statement = statement.with_for_update() + return cast(AdvisorProfileDriftReview | None, await self.session.scalar(statement)) + + async def pending_reviews(self, *, limit: int) -> list[AdvisorProfileDriftReview]: + return list(await self.session.scalars( + select(AdvisorProfileDriftReview) + .where(AdvisorProfileDriftReview.status == "pending_review") + .order_by( + AdvisorProfileDriftReview.created_at.asc(), + AdvisorProfileDriftReview.id.asc(), + ) + .limit(limit) + )) + + async def tags_for_review( + self, review_id: int, *, lock: bool = False + ) -> list[AdvisorProfileTag]: + statement = ( + select(AdvisorProfileTag) + .where(AdvisorProfileTag.drift_review_id == review_id) + .order_by(AdvisorProfileTag.tag_key.asc(), AdvisorProfileTag.id.asc()) + ) + if lock: + statement = statement.with_for_update() + return list(await self.session.scalars(statement)) + + async def supersede_active_tags( + self, customer_id: int, tag_keys: tuple[str, ...], now: datetime + ) -> None: + if not tag_keys: + return + await self.session.execute( + update(AdvisorProfileTag) + .where( + AdvisorProfileTag.customer_id == customer_id, + AdvisorProfileTag.tag_key.in_(tag_keys), + AdvisorProfileTag.status == "active", + ) + .values(status="superseded", active_customer_tag=None, updated_at=now) + ) + + def add_tag(self, tag: AdvisorProfileTag) -> None: + self.session.add(tag) + + def add_drift_review(self, review: AdvisorProfileDriftReview) -> None: + self.session.add(review) diff --git a/app/service/risk_questionnaire_service.py b/app/service/risk_questionnaire_service.py new file mode 100644 index 0000000..e0c212c --- /dev/null +++ b/app/service/risk_questionnaire_service.py @@ -0,0 +1,421 @@ +"""Opening risk-questionnaire application service with server-only scoring.""" + +from dataclasses import dataclass +from datetime import UTC, datetime, timedelta +from decimal import Decimal +from typing import Any +from uuid import uuid4 + +from sqlalchemy.ext.asyncio import AsyncSession + +from app.core.contracts import RequestContext +from app.core.errors import ForbiddenAgentError, InvalidStateError +from app.core.risk_questionnaire_contracts import ( + DECLARATION, + QUESTIONNAIRE_VERSION, + QUESTIONS, + RiskQuestionnaireSubmission, +) +from app.infrastructure.db import SessionFactory +from app.model.audit import InteractionAudit +from app.model.fund import FundRiskAssessment as RiskAssessment +from app.model.memory import MemorySyncOutbox +from app.model.profile_tag import AdvisorProfileDriftReview, AdvisorProfileTag +from app.model.risk_questionnaire import ProfileSnapshot +from app.repository.risk_questionnaire_repository import RiskQuestionnaireRepository +from app.service.api_transaction_service import ApiTransactionService, digest + +# This versioned mapping is deliberately server-side. It is never included in customer responses. +_SCORE_RULES: dict[str, dict[int, int]] = { + "q1": {1: 5, 2: 3, 3: 3, 4: 4, 5: 4, 6: 2, 7: 3, 8: 2, 9: 4, 10: 2, 11: 3, + 12: 1, 13: 1, 14: 3, 15: 3, 16: 2}, + "q2": {1: 4, 2: 3, 3: 2, 4: 1}, + "q3": {1: 3, 2: 4, 3: 2, 4: 5, 5: 1}, + "q4": {1: 4, 2: 3, 3: 2, 4: 1}, + "q5": {1: 4, 2: 3, 3: 2, 4: 1}, + "q6": {1: 5, 2: 4, 3: 3, 4: 2, 5: 1}, + "q7": {1: 5, 2: 4, 3: 3, 4: 2, 5: 1}, + "q8": {1: 4, 2: 3, 3: 2, 4: 1}, + "q9": {1: 4, 2: 3, 3: 2, 4: 1}, + "q10": {1: 1, 2: 2, 3: 3, 4: 4}, + "q11": {1: 5, 2: 4, 3: 3, 4: 2, 5: 1}, + "q12": {1: 4, 2: 3, 3: 2, 4: 1}, + "q13": {1: 4, 2: 3, 3: 2, 4: 1}, +} +_RISK_BANDS = ((22, "C1", "谨慎型"), (31, "C2", "稳健型"), (40, "C3", "平衡型"), + (49, "C4", "成长型"), (57, "C5", "进取型")) +_HORIZONS = {1: "5年以上", 2: "3至5年", 3: "1至3年", 4: "1年以下"} +_ASSET_PREFERENCES = { + 1: ["固定收益类"], + 2: ["固定收益类", "权益类"], + 3: ["固定收益类", "权益类", "衍生品"], + 4: ["固定收益类", "权益类", "衍生品", "其他"], +} + + +@dataclass(frozen=True) +class _ScoreResult: + total: int + risk_level: str + risk_profile: str + + +class RiskQuestionnaireService: + async def questionnaire(self, context: RequestContext) -> dict[str, object]: + self._require_customer(context) + required = await self.is_required(context) + return { + "data": { + "required": required, + "questionnaire": { + "version": QUESTIONNAIRE_VERSION, + "questions": QUESTIONS, + "declaration": DECLARATION, + } if required else None, + }, + "meta": {"trace_id": context.trace_id}, + } + + async def is_required(self, context: RequestContext) -> bool: + if "customer" not in context.roles: + return False + async with SessionFactory() as session: + return not await RiskQuestionnaireRepository(session).has_valid_assessment( + int(context.user_id), _utc_now() + ) + + async def submit( + self, payload: RiskQuestionnaireSubmission, context: RequestContext, key: str | None + ) -> dict[str, object]: + self._require_customer(context) + + async def operation(session: AsyncSession) -> dict[str, Any]: + now = _utc_now() + customer_id = int(context.user_id) + repository = RiskQuestionnaireRepository(session) + if await repository.pending_drift_review(customer_id, lock=True) is not None: + raise InvalidStateError("画像标签漂移正在复核，暂不能提交新的风险测评") + await self._validate_submission_rate(repository, customer_id, now) + score = self.score(payload.answers) + assessment_id = _identifier() + valid_until = _next_year(now) + assessment = RiskAssessment( + id=assessment_id, + customer_id=customer_id, + questionnaire_version=QUESTIONNAIRE_VERSION, + answers=dict(payload.answers), + total_score=score.total, + investor_type=score.risk_level, + assessed_at=now, + valid_until=valid_until, + created_at=now, + ) + version = await repository.next_profile_version(customer_id) + profile_uuid = str(uuid4()) + snapshot = self._profile_snapshot(score, payload.answers, valid_until) + repository.add_assessment(assessment) + basis = { + "source": "fin_risk_assessment", + "assessment_id": str(assessment_id), + "questionnaire_version": QUESTIONNAIRE_VERSION, + } + active_tags = await repository.active_tags(customer_id, lock=True) + candidates = self._profile_tag_candidates(score, payload.answers, assessment_id) + changes = self._drift_changes(active_tags, candidates) + audit_action: str + audit_detail: dict[str, object] + if active_tags and changes: + review = AdvisorProfileDriftReview( + drift_no=str(uuid4()), + customer_id=customer_id, + source_assessment_id=assessment_id, + candidate_profile_uuid=profile_uuid, + candidate_profile_version=version, + candidate_snapshot=snapshot, + candidate_generation_basis=basis, + changed_tags=changes, + status="pending_review", + reviewer_user_id=None, + reviewed_at=None, + review_comment=None, + created_at=now, + updated_at=now, + ) + repository.add_drift_review(review) + await session.flush() + self._add_tags( + repository, customer_id, version, candidates, active_tags, now, + status="pending_review", drift_review_id=review.id, + drift_reasons={ + str(item["tag_key"]): item["reason"] for item in changes + }, + ) + audit_action = "onboarding.profile_drift_detected" + audit_detail = { + "assessment_id": str(assessment_id), + "drift_no": review.drift_no, + "changed_tag_keys": [str(item["tag_key"]) for item in changes], + } + else: + await repository.deactivate_current_profile(customer_id, now) + if active_tags: + await repository.supersede_active_tags( + customer_id, tuple(tag.tag_key for tag in active_tags), now + ) + repository.add_profile(ProfileSnapshot( + profile_uuid=profile_uuid, + customer_id=customer_id, + version=version, + snapshot=snapshot, + generation_basis=basis, + snapshot_hash=digest(snapshot), + is_current=True, + generated_at=now, + created_at=now, + updated_at=now, + )) + self._add_tags( + repository, customer_id, version, candidates, active_tags, now, + status="active", drift_review_id=None, drift_reasons={}, + ) + self._add_profile_sync_events( + repository, profile_uuid, version, customer_id, snapshot, now + ) + audit_action = "onboarding.risk_questionnaire_completed" + audit_detail = {"assessment_id": str(assessment_id), "profile_version": version} + session.add(InteractionAudit( + actor_type="user", + actor_id=customer_id, + target_customer_id=customer_id, + portal=context.portal, + action_type=audit_action, + detail={ + "questionnaire_version": QUESTIONNAIRE_VERSION, + "trace_id": context.trace_id, + **audit_detail, + }, + created_at=now, + )) + return { + "data": { + "completed": True, + "questionnaire_version": QUESTIONNAIRE_VERSION, + "valid_until": valid_until.isoformat() + "Z", + }, + "meta": {"trace_id": context.trace_id}, + } + + return await ApiTransactionService().execute( + context, + "onboarding:risk-questionnaire", + key, + payload.model_dump(mode="json"), + operation, + ) + + @staticmethod + def score(answers: dict[str, int]) -> _ScoreResult: + total = sum( + _SCORE_RULES[question_id][option_id] + for question_id, option_id in answers.items() + ) + for maximum, risk_level, risk_profile in _RISK_BANDS: + if total <= maximum: + return _ScoreResult(total, risk_level, risk_profile) + raise ValueError("questionnaire score exceeds configured range") + + @staticmethod + def _profile_snapshot( + score: _ScoreResult, answers: dict[str, int], valid_until: datetime + ) -> dict[str, object]: + return { + "profile_type": "opening_risk_assessment", + "risk_level": score.risk_level, + "risk_profile": score.risk_profile, + "investment_horizon": _HORIZONS[answers["q9"]], + "preferred_asset_classes": _ASSET_PREFERENCES[answers["q10"]], + "source": "formal_risk_assessment", + "valid_until": valid_until.isoformat() + "Z", + } + + @staticmethod + def _profile_tag_candidates( + score: _ScoreResult, answers: dict[str, int], assessment_id: int + ) -> dict[str, dict[str, object]]: + source_reference = f"fin_risk_assessment:{assessment_id}" + risk_confidence = RiskQuestionnaireService._risk_tag_confidence(score.total) + return { + "risk_level": { + "value": score.risk_level, + "confidence": risk_confidence, + "source_type": "formal_risk_assessment", + "source_reference": source_reference, + "source_confidence": Decimal("1.0000"), + }, + "risk_profile": { + "value": score.risk_profile, + "confidence": risk_confidence, + "source_type": "formal_risk_assessment", + "source_reference": source_reference, + "source_confidence": Decimal("1.0000"), + }, + "investment_horizon": { + "value": _HORIZONS[answers["q9"]], + "confidence": Decimal("1.0000"), + "source_type": "formal_risk_assessment", + "source_reference": source_reference, + "source_confidence": Decimal("1.0000"), + }, + "preferred_asset_classes": { + "value": _ASSET_PREFERENCES[answers["q10"]], + "confidence": Decimal("1.0000"), + "source_type": "formal_risk_assessment", + "source_reference": source_reference, + "source_confidence": Decimal("1.0000"), + }, + } + + @staticmethod + def _risk_tag_confidence(total: int) -> Decimal: + lower = 13 + for upper, _risk_level, _risk_profile in _RISK_BANDS: + if total <= upper: + distance = min(total - lower, upper - total) + return min( + Decimal("0.9500"), + Decimal("0.7000") + Decimal(distance) * Decimal("0.0500"), + ) + lower = upper + 1 + raise ValueError("questionnaire score exceeds configured range") + + @staticmethod + def _drift_changes( + active_tags: list[AdvisorProfileTag], candidates: dict[str, dict[str, object]] + ) -> list[dict[str, object]]: + active_by_key = {tag.tag_key: tag for tag in active_tags} + changes: list[dict[str, object]] = [] + for tag_key, candidate in candidates.items(): + previous = active_by_key.get(tag_key) + if previous is None: + continue + reason: str | None = None + if previous.tag_value != candidate["value"]: + reason = "value_changed" + elif previous.source_type != candidate["source_type"]: + reason = "source_changed" + elif ( + Decimal(str(candidate["confidence"])) + Decimal("0.1500") + < Decimal(previous.confidence) + ): + reason = "confidence_decreased" + if reason is not None: + changes.append({ + "tag_key": tag_key, + "reason": reason, + "previous_value": previous.tag_value, + "candidate_value": candidate["value"], + "previous_confidence": str(previous.confidence), + "candidate_confidence": str(candidate["confidence"]), + "previous_source_type": previous.source_type, + "candidate_source_type": candidate["source_type"], + }) + return changes + + @staticmethod + def _add_tags( + repository: RiskQuestionnaireRepository, + customer_id: int, + profile_version: int, + candidates: dict[str, dict[str, object]], + active_tags: list[AdvisorProfileTag], + now: datetime, + *, + status: str, + drift_review_id: int | None, + drift_reasons: dict[str, object], + ) -> None: + active_by_key = {tag.tag_key: tag for tag in active_tags} + for tag_key, candidate in candidates.items(): + previous = active_by_key.get(tag_key) + repository.add_tag(AdvisorProfileTag( + tag_uuid=str(uuid4()), + customer_id=customer_id, + tag_key=tag_key, + tag_value=candidate["value"], + tag_value_hash=digest(candidate["value"]), + confidence=Decimal(str(candidate["confidence"])), + source_type=str(candidate["source_type"]), + source_reference=str(candidate["source_reference"]), + source_confidence=Decimal(str(candidate["source_confidence"])), + profile_version=profile_version, + drift_review_id=drift_review_id, + previous_tag_id=previous.id if previous is not None else None, + drift_reason=str(drift_reasons[tag_key]) if tag_key in drift_reasons else None, + status=status, + active_customer_tag=(f"{customer_id}:{tag_key}" if status == "active" else None), + created_at=now, + updated_at=now, + )) + + @staticmethod + def _add_profile_sync_events( + repository: RiskQuestionnaireRepository, + profile_uuid: str, + version: int, + customer_id: int, + snapshot: dict[str, object], + now: datetime, + ) -> None: + event_uuid = str(uuid4()) + payload = { + "customer_id": str(customer_id), + "profile_uuid": profile_uuid, + "version": version, + "profile": snapshot, + } + for target_store in ("milvus", "neo4j"): + repository.add_sync_event(MemorySyncOutbox( + event_uuid=event_uuid, + aggregate_type="profile", + aggregate_uuid=profile_uuid, + aggregate_version=version, + target_store=target_store, + operation="upsert", + payload=payload, + status="pending", + retry_count=0, + next_retry_at=None, + last_error=None, + created_at=now, + processed_at=None, + )) + + @staticmethod + async def _validate_submission_rate( + repository: RiskQuestionnaireRepository, customer_id: int, now: datetime + ) -> None: + start_of_day = now.replace(hour=0, minute=0, second=0, microsecond=0) + if await repository.assessments_since(customer_id, start_of_day) >= 2: + raise InvalidStateError("风险测评在一个自然日内不能超过两次") + if await repository.assessments_since(customer_id, now - timedelta(days=365)) >= 8: + raise InvalidStateError("风险测评在一年内不能超过八次") + + @staticmethod + def _require_customer(context: RequestContext) -> None: + if "customer" not in context.roles: + raise ForbiddenAgentError("仅客户可以填写开户风险测评问卷") + + +def _identifier() -> int: + return (uuid4().int >> 64) or 1 + + +def _next_year(value: datetime) -> datetime: + try: + return value.replace(year=value.year + 1) + except ValueError: + return value.replace(year=value.year + 1, month=2, day=28) + + +def _utc_now() -> datetime: + return datetime.now(UTC).replace(tzinfo=None) diff --git a/docs/21-投顾Agent迁移TODO.md b/docs/21-投顾Agent迁移TODO.md index 35e394b..aea405c 100644 --- a/docs/21-投顾Agent迁移TODO.md +++ b/docs/21-投顾Agent迁移TODO.md @@ -51,6 +51,19 @@ R4=7、R5=1；资产分类成功 18 个，1 个因合同证据不足跳过。每 治理导入/分类导入 `5 passed`，合计专项 `10 passed`；Ruff、MyPy、数据库结构审计和 约束审计通过。阶段五尚未完成流动性指标细化和完整推荐侧适当性联动，暂不提交阶段完成标记。 +### 阶段六：开户风险问卷 + +已完成开户问卷查询和提交接口、服务端 C1-C5 评分与风险等级生成、开户画像快照、 +问卷有效期和每日/年度提交次数限制。首次登录访问非 onboarding 接口时会被拦截， +问卷结果通过 `MemorySyncOutbox` 同步画像投影；客户响应不包含答案、总分、风险等级、 +画像或来源置信度。提交使用 `Idempotency-Key`，重复请求按请求摘要幂等处理。 + +阶段六测试结果：专项测试 `7 passed`，全量单元测试 `464 passed, 3 warnings`，Ruff +通过，MyPy（116 个源文件）通过。数据库审计工具已执行，但默认 `.env` 连接的是旧 +`jr_agent` 库，结构审计发现旧库的场外表和基线约束差异；阶段四已验证独立库 +`jr_agent_qyqy_migration` 的迁移结构，不能将旧库结果作为新底座验收结果。 +阶段六提交：待提交。 + ## 一、迁移准备 - [ ] 确认远程仓库可访问。（当前失败：连接 `47.106.207.27:3000` 被拒绝） @@ -183,27 +196,27 @@ python tools/audit_constraints.py ## 六、开户风险问卷 -- [ ] 迁移问卷查询接口。 -- [ ] 迁移问卷提交接口。 -- [ ] 迁移服务端评分规则。 -- [ ] 迁移 C1-C5 风险等级生成。 -- [ ] 迁移开户画像快照生成。 -- [ ] 迁移问卷有效期控制。 -- [ ] 迁移每日和年度提交次数限制。 -- [ ] 迁移首次登录拦截。 -- [ ] 迁移画像同步 Outbox 事件。 -- [ ] 确认客户响应不返回答案、总分、风险等级和画像。 -- [ ] 确认问卷重复提交具有幂等性。 -- [ ] 完成风险问卷提交 `advisor/risk-questionnaire`。 +- [x] 迁移问卷查询接口。 +- [x] 迁移问卷提交接口。 +- [x] 迁移服务端评分规则。 +- [x] 迁移 C1-C5 风险等级生成。 +- [x] 迁移开户画像快照生成。 +- [x] 迁移问卷有效期控制。 +- [x] 迁移每日和年度提交次数限制。 +- [x] 迁移首次登录拦截。 +- [x] 迁移画像同步 Outbox 事件。 +- [x] 确认客户响应不返回答案、总分、风险等级和画像。 +- [x] 确认问卷重复提交具有幂等性。 +- [x] 完成风险问卷提交 `advisor/risk-questionnaire`。（专项 `7 passed`；全量单元 `464 passed`） 验收： -- [ ] 新客户访问投顾业务会被拦截。 -- [ ] 新客户可以查询问卷。 -- [ ] 新客户可以成功提交问卷。 -- [ ] 问卷结果正确写入后台表。 -- [ ] 客户响应不包含内部评分信息。 -- [ ] 投顾工具可以读取必要的内部画像投影。 +- [x] 新客户访问投顾业务会被拦截。 +- [x] 新客户可以查询问卷。 +- [x] 新客户可以成功提交问卷。 +- [x] 问卷结果正确写入后台表。 +- [x] 客户响应不包含内部评分信息。 +- [x] 投顾工具可以读取必要的内部画像投影。 ## 七、投资目标和目标书 diff --git a/tests/unit/api/test_onboarding_gate.py b/tests/unit/api/test_onboarding_gate.py new file mode 100644 index 0000000..b00489d --- /dev/null +++ b/tests/unit/api/test_onboarding_gate.py @@ -0,0 +1,73 @@ +import pytest +from fastapi import Request +from fastapi.security import HTTPAuthorizationCredentials + +from app.api.dependencies.auth import build_request_context +from app.core.contracts import RequestContext +from app.core.errors import OnboardingRequiredError + + +def request(path: str) -> Request: + return Request({"type": "http", "method": "GET", "path": path, "headers": []}) + + +@pytest.mark.asyncio +async def test_customer_without_assessment_is_gated_after_authentication( + monkeypatch: pytest.MonkeyPatch, +) -> None: + authenticated = RequestContext(user_id="7", trace_id="initial") + resolved = authenticated.model_copy(update={"roles": ("customer",)}) + + class Authenticator: + def authenticate(self, _token: str) -> RequestContext: + return authenticated + + async def resolve(_self: object, _context: RequestContext) -> RequestContext: + return resolved + + async def is_required(_self: object, _context: RequestContext) -> bool: + return True + + monkeypatch.setattr("app.api.dependencies.auth._authenticator", lambda: Authenticator()) + monkeypatch.setattr("app.service.identity_service.IdentityService.resolve", resolve) + monkeypatch.setattr( + "app.service.risk_questionnaire_service.RiskQuestionnaireService.is_required", is_required + ) + with pytest.raises(OnboardingRequiredError): + await build_request_context( + request("/api/v1/agent-runs"), HTTPAuthorizationCredentials( + scheme="Bearer", credentials="token" + ) + ) + + +@pytest.mark.asyncio +async def test_questionnaire_endpoint_is_exempt_from_the_gate( + monkeypatch: pytest.MonkeyPatch, +) -> None: + authenticated = RequestContext(user_id="7", trace_id="initial") + resolved = authenticated.model_copy(update={"roles": ("customer",)}) + + class Authenticator: + def authenticate(self, _token: str) -> RequestContext: + return authenticated + + async def resolve(_self: object, _context: RequestContext) -> RequestContext: + return resolved + + async def unexpected_check(_self: object, _context: RequestContext) -> bool: + raise AssertionError("问卷入口不应经过完成状态拦截") + + monkeypatch.setattr("app.api.dependencies.auth._authenticator", lambda: Authenticator()) + monkeypatch.setattr("app.service.identity_service.IdentityService.resolve", resolve) + monkeypatch.setattr( + "app.service.risk_questionnaire_service.RiskQuestionnaireService.is_required", + unexpected_check, + ) + context = await build_request_context( + request("/api/v1/onboarding/risk-questionnaire"), HTTPAuthorizationCredentials( + scheme="Bearer", credentials="token" + ) + ) + assert context == resolved + diff --git a/tests/unit/service/test_risk_questionnaire_service.py b/tests/unit/service/test_risk_questionnaire_service.py new file mode 100644 index 0000000..7f0ba50 --- /dev/null +++ b/tests/unit/service/test_risk_questionnaire_service.py @@ -0,0 +1,187 @@ +from decimal import Decimal +from typing import Any, cast + +import pytest +from pydantic import ValidationError + +from app.core.contracts import RequestContext +from app.core.risk_questionnaire_contracts import RiskQuestionnaireSubmission +from app.model.fund import FundRiskAssessment as RiskAssessment +from app.model.memory import MemorySyncOutbox +from app.model.profile_tag import AdvisorProfileDriftReview, AdvisorProfileTag +from app.model.risk_questionnaire import ProfileSnapshot +from app.service.api_transaction_service import ApiTransactionService +from app.service.risk_questionnaire_service import RiskQuestionnaireService + + +def context() -> RequestContext: + return RequestContext( + user_id="7", trace_id="questionnaire-test", roles=("customer",) + ) + + +def lowest_risk_answers() -> dict[str, int]: + return { + "q1": 12, "q2": 4, "q3": 5, "q4": 4, "q5": 4, "q6": 5, "q7": 5, + "q8": 4, "q9": 4, "q10": 1, "q11": 5, "q12": 4, "q13": 4, + } + + +def highest_risk_answers() -> dict[str, int]: + return { + "q1": 1, "q2": 1, "q3": 4, "q4": 1, "q5": 1, "q6": 1, "q7": 1, + "q8": 1, "q9": 1, "q10": 4, "q11": 1, "q12": 1, "q13": 1, + } + + +def submission() -> RiskQuestionnaireSubmission: + return RiskQuestionnaireSubmission( + answers=lowest_risk_answers(), declaration_accepted=True + ) + + +class FakeSession: + def __init__(self) -> None: + self.items: list[object] = [] + + def add(self, item: object) -> None: + self.items.append(item) + + async def flush(self) -> None: + for item in self.items: + if isinstance(item, AdvisorProfileDriftReview) and item.id is None: + item.id = 99 + + +class FakeRepository: + def __init__(self, session: FakeSession) -> None: + self.session = session + self.deactivated = False + + async def assessments_since(self, _customer_id: int, _since: object) -> int: + return 0 + + async def next_profile_version(self, _customer_id: int) -> int: + return 1 + + async def pending_drift_review(self, _customer_id: int, *, lock: bool = False) -> None: + del lock + return None + + async def active_tags( + self, _customer_id: int, *, lock: bool = False + ) -> list[AdvisorProfileTag]: + del lock + return [] + + async def deactivate_current_profile(self, _customer_id: int, _now: object) -> None: + self.deactivated = True + + def add_assessment(self, assessment: RiskAssessment) -> None: + self.session.add(assessment) + + def add_profile(self, profile: ProfileSnapshot) -> None: + self.session.add(profile) + + def add_tag(self, tag: AdvisorProfileTag) -> None: + self.session.add(tag) + + def add_drift_review(self, review: AdvisorProfileDriftReview) -> None: + self.session.add(review) + + async def supersede_active_tags( + self, _customer_id: int, _tag_keys: tuple[str, ...], _now: object + ) -> None: + return None + + def add_sync_event(self, event: MemorySyncOutbox) -> None: + self.session.add(event) + + +def test_submission_requires_all_questions_and_declaration() -> None: + with pytest.raises(ValidationError, match="every questionnaire question"): + RiskQuestionnaireSubmission(answers={"q1": 1}, declaration_accepted=True) + with pytest.raises(ValidationError): + RiskQuestionnaireSubmission.model_validate({ + "answers": lowest_risk_answers(), "declaration_accepted": False, + }) + + +def test_score_bands_are_deterministic_and_server_only() -> None: + low = RiskQuestionnaireService.score(lowest_risk_answers()) + high = RiskQuestionnaireService.score(highest_risk_answers()) + assert (low.total, low.risk_level) == (13, "C1") + assert (high.total, high.risk_level) == (57, "C5") + + +@pytest.mark.asyncio +async def test_submission_persists_assessment_and_profile_without_exposing_them( + monkeypatch: pytest.MonkeyPatch, +) -> None: + session = FakeSession() + + async def execute( + _self: ApiTransactionService, + _context: RequestContext, + _scope: str, + _key: str | None, + _body: object, + operation: Any, + ) -> dict[str, Any]: + return cast(dict[str, Any], await operation(session)) + + monkeypatch.setattr(ApiTransactionService, "execute", execute) + monkeypatch.setattr( + "app.service.risk_questionnaire_service.RiskQuestionnaireRepository", FakeRepository + ) + response = await RiskQuestionnaireService().submit( + submission(), context(), "questionnaire-key-0001" + ) + + data = cast(dict[str, object], response["data"]) + assert set(data) == {"completed", "questionnaire_version", "valid_until"} + assert data["completed"] is True + assert not {"total_score", "risk_level", "risk_profile", "profile"}.intersection(data) + assert any(isinstance(item, RiskAssessment) for item in session.items) + assert any(isinstance(item, ProfileSnapshot) for item in session.items) + tags = [item for item in session.items if isinstance(item, AdvisorProfileTag)] + assert {tag.tag_key for tag in tags} == { + "risk_level", "risk_profile", "investment_horizon", "preferred_asset_classes", + } + assert all(tag.source_type == "formal_risk_assessment" for tag in tags) + assert all(tag.status == "active" for tag in tags) + assert len([item for item in session.items if isinstance(item, MemorySyncOutbox)]) == 2 + + +def test_tag_confidence_is_lower_near_a_risk_band_boundary() -> None: + assert RiskQuestionnaireService._risk_tag_confidence(22) == Decimal("0.7000") + assert RiskQuestionnaireService._risk_tag_confidence(17) == Decimal("0.9000") + + +def test_tag_value_change_requires_drift_review() -> None: + candidates = RiskQuestionnaireService._profile_tag_candidates( + RiskQuestionnaireService.score(highest_risk_answers()), highest_risk_answers(), 8 + ) + previous = AdvisorProfileTag( + id=1, + tag_uuid="previous-risk-level", + customer_id=7, + tag_key="risk_level", + tag_value="C1", + tag_value_hash="different-value-hash", + confidence=0.95, + source_type="formal_risk_assessment", + source_reference="fin_risk_assessment:7", + source_confidence=1, + profile_version=1, + drift_review_id=None, + previous_tag_id=None, + drift_reason=None, + status="active", + active_customer_tag="7:risk_level", + created_at=object(), + updated_at=object(), + ) + changes = RiskQuestionnaireService._drift_changes([previous], candidates) + assert changes[0]["tag_key"] == "risk_level" + assert changes[0]["reason"] == "value_changed"