48 lines
2.1 KiB
Python
48 lines
2.1 KiB
Python
"""Read-only access to the internal profile projection used by advisory services."""
|
|
|
|
from datetime import UTC, datetime
|
|
from typing import Any
|
|
|
|
from app.core.advisor_allocation_contracts import CustomerProfileQuery
|
|
from app.core.contracts import RequestContext
|
|
from app.infrastructure.db import SessionFactory
|
|
from app.repository.risk_questionnaire_repository import RiskQuestionnaireRepository
|
|
from app.service.authorization_service import AuthorizationService
|
|
|
|
_RISK_LEVELS = frozenset({"C1", "C2", "C3", "C4", "C5"})
|
|
|
|
|
|
class CustomerProfileService:
|
|
"""Returns the minimum internal projection needed by governed advisor tools."""
|
|
|
|
async def current_for_agent(self, context: RequestContext) -> dict[str, object] | None:
|
|
await AuthorizationService.require(context, "customer-profile:read:self")
|
|
customer_id = int(context.user_id)
|
|
now = datetime.now(UTC).replace(tzinfo=None)
|
|
async with SessionFactory() as session:
|
|
repository = RiskQuestionnaireRepository(session)
|
|
assessment = await repository.latest_assessment(customer_id)
|
|
if assessment is None or assessment.valid_until <= now:
|
|
return None
|
|
profile = await repository.current_profile(customer_id)
|
|
if profile is None:
|
|
return None
|
|
snapshot: dict[str, Any] = profile.snapshot
|
|
risk_level = snapshot.get("risk_level")
|
|
if risk_level not in _RISK_LEVELS:
|
|
return None
|
|
preferred = snapshot.get("preferred_asset_classes")
|
|
return {
|
|
"risk_level": risk_level,
|
|
"investment_horizon": snapshot.get("investment_horizon"),
|
|
"preferred_asset_classes": list(preferred) if isinstance(preferred, list) else [],
|
|
"valid_until": assessment.valid_until.isoformat() + "Z",
|
|
}
|
|
|
|
|
|
async def customer_profile_query_tool(
|
|
_arguments: CustomerProfileQuery, context: RequestContext
|
|
) -> dict[str, object] | None:
|
|
"""Tool entry point. This projection must never be directly echoed to customers."""
|
|
return await CustomerProfileService().current_for_agent(context)
|