250 lines
9.3 KiB
Python
250 lines
9.3 KiB
Python
"""Opening risk-questionnaire application service with server-only scoring."""
|
|
|
|
from dataclasses import dataclass
|
|
from datetime import UTC, datetime, timedelta
|
|
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 ConflictAgentError, ForbiddenAgentError
|
|
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.memory import MemorySyncOutbox
|
|
from app.model.risk_questionnaire import ProfileSnapshot, RiskAssessment
|
|
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)
|
|
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)
|
|
await repository.deactivate_current_profile(customer_id, now)
|
|
repository.add_assessment(assessment)
|
|
repository.add_profile(ProfileSnapshot(
|
|
profile_uuid=profile_uuid,
|
|
customer_id=customer_id,
|
|
version=version,
|
|
snapshot=snapshot,
|
|
generation_basis={
|
|
"source": "fin_risk_assessment",
|
|
"assessment_id": str(assessment_id),
|
|
"questionnaire_version": QUESTIONNAIRE_VERSION,
|
|
},
|
|
snapshot_hash=digest(snapshot),
|
|
is_current=True,
|
|
generated_at=now,
|
|
created_at=now,
|
|
updated_at=now,
|
|
))
|
|
self._add_profile_sync_events(
|
|
repository, profile_uuid, version, customer_id, snapshot, now
|
|
)
|
|
session.add(InteractionAudit(
|
|
actor_type="user",
|
|
actor_id=customer_id,
|
|
target_customer_id=customer_id,
|
|
portal=context.portal,
|
|
action_type="onboarding.risk_questionnaire_completed",
|
|
detail={
|
|
"assessment_id": str(assessment_id),
|
|
"questionnaire_version": QUESTIONNAIRE_VERSION,
|
|
"profile_version": version,
|
|
"trace_id": context.trace_id,
|
|
},
|
|
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 _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 ConflictAgentError("风险测评在一个自然日内不能超过两次")
|
|
if await repository.assessments_since(customer_id, now - timedelta(days=365)) >= 8:
|
|
raise ConflictAgentError("风险测评在一年内不能超过八次")
|
|
|
|
@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)
|