group_fqcd_jr/app/service/agent/implementations/advisor.py

"""投顾 Agent 的公共底座实现。"""

from typing import Any

from app.core.contracts import AgentDefinition, AgentRequest, CoreResult, RequestContext
from app.service.agent.implementations.fund_query_demo import FundQueryDemoAgent
from app.service.goal_conversation_service import GoalConversationService


class AdvisorAgent(FundQueryDemoAgent):
    """通过公共 BaseAgent 链路运行的最小投顾 Agent。"""

    definition = AgentDefinition(
        agent_type="advisor",
        version="0.1.0",
        allowed_roles=("customer", "advisor", "operator", "admin"),
        allowed_portals=("api",),
        allowed_tools=(
            "query_fund_quote",
            "query_investment_goal",
            "analyze_portfolio",
            "generate_asset_allocation",
            "recommend_products",
        ),
        supported_intents=(
            "fund_quote",
            "investment_goal",
            "portfolio_analysis",
            "asset_allocation",
            "product_recommend",
            "comparison",
        ),
    )

    async def handle(self, request: AgentRequest, context: RequestContext) -> CoreResult:
        if (
            self._classified_intent is not None
            and self._classified_intent.intent == "investment_goal"
        ):
            if GoalConversationService.should_collect(request.message):
                goal_result = await GoalConversationService().process(
                    session_id=request.session_id, customer_id=int(context.user_id),
                    trace_id=context.trace_id, message=request.message,
                )
                return CoreResult(text=GoalConversationService.customer_prompt(goal_result))
            output = await self.call_tool(
                "query_investment_goal", {}, intent="investment_goal", context=context
            )
            if not isinstance(output, dict):
                return CoreResult(text="当前没有已确认的投资目标，暂不能用于配置或产品推荐。")
            return CoreResult(text=self._describe_goal(output))
        if (
            self._classified_intent is not None
            and self._classified_intent.intent == "portfolio_analysis"
        ):
            output = await self.call_tool(
                "analyze_portfolio", {}, intent="portfolio_analysis", context=context
            )
            return CoreResult(text=self._describe_portfolio(output))
        if (
            self._classified_intent is not None
            and self._classified_intent.intent == "asset_allocation"
        ):
            output = await self.call_tool(
                "generate_asset_allocation", {}, intent="asset_allocation", context=context
            )
            return CoreResult(text=self._describe_allocation(output))
        if (
            self._classified_intent is not None
            and self._classified_intent.intent == "product_recommend"
        ):
            output = await self.call_tool(
                "recommend_products", {"limit": 3}, intent="product_recommend", context=context
            )
            return CoreResult(text=self._describe_recommendation(output))
        if (
            self._classified_intent is not None
            and self._classified_intent.intent == "comparison"
        ):
            return CoreResult(text="对比分析需要明确两个或多个场内基金产品，请提供基金代码或名称。")
        return await super().handle(request, context)

    @staticmethod
    def _describe_goal(goal: dict[str, Any]) -> str:
        return (
            f"当前投资目标：年化收益目标 {goal['annualized_return_lower_pct']}%-"
            f"{goal['annualized_return_upper_pct']}%，最大回撤 {goal['max_drawdown_pct']}%，"
            f"流动性要求 {goal['liquidity_requirement']}，投资期限 "
            f"{goal['investment_horizon_months']} 个月，业绩比较基准 {goal['benchmark_name']}。"
            "以上为目标采集结果，不构成收益承诺或交易指令。"
        )

    @staticmethod
    def _describe_portfolio(result: object) -> str:
        if not isinstance(result, dict):
            return "持仓分析暂不可用，请稍后重试。"
        status = result.get("status")
        if status == "no_positions":
            return "当前没有可分析的场内基金持仓。"
        if status == "valuation_required":
            return "当前持仓缺少可用市值，暂不能计算集中度。"
        summary = result.get("summary")
        if not isinstance(summary, dict):
            return "持仓分析数据不完整，请稍后重试。"
        concentration = result.get("product_concentration")
        hhi = concentration.get("hhi") if isinstance(concentration, dict) else None
        return (
            f"持仓分析完成：共 {summary.get('position_count')} 个产品，"
            f"总市值 {summary.get('total_market_value')}，产品集中度 HHI 为 {hhi}。"
            "分析结果仅供参考，不生成交易指令。"
        )

    @staticmethod
    def _describe_allocation(result: object) -> str:
        if not isinstance(result, dict):
            return "资产配置分析暂不可用，请稍后重试。"
        status = result.get("status")
        if status == "profile_required":
            return "当前缺少有效风险画像，暂不能生成资产配置。"
        if status == "investment_goal_required":
            return "当前没有已确认的投资目标，暂不能生成资产配置。"
        if status != "ready":
            return "资产配置分析数据不完整，请稍后重试。"
        allocation = result.get("allocation")
        if not isinstance(allocation, list) or not allocation:
            return "当前没有足够的场内基金数据生成资产配置。"
        parts = [
            f"{item.get('label', item.get('asset_class'))} {item.get('target_pct')}%"
            for item in allocation
            if isinstance(item, dict)
        ]
        optimization = result.get("optimization")
        dynamic = isinstance(optimization, dict) and bool(optimization.get("dynamic"))
        mode = "动态历史因子优化" if dynamic else "静态配置（历史数据覆盖不足）"
        return (
            f"资产配置分析完成（{mode}）："
            + "，".join(parts)
            + "。该结果综合考虑收益目标、最大回撤、流动性和投资期限，"
            "仅供分析参考，不构成交易指令。"
        )

    @staticmethod
    def _describe_recommendation(result: object) -> str:
        if not isinstance(result, dict):
            return "产品推荐暂不可用，请稍后重试。"
        if result.get("status") == "profile_required":
            return "当前缺少有效风险画像，暂不能推荐产品。"
        if result.get("status") == "investment_goal_required":
            return "当前没有已确认的投资目标，暂不能推荐产品。"
        products = result.get("products")
        if not isinstance(products, list) or not products:
            return "当前没有通过适当性和证据校验的场内基金产品。"
        names = [
            f"{item.get('product_code')} {item.get('product_name')}"
            for item in products
            if isinstance(item, dict)
        ]
        return "推荐分析结果：" + "、".join(names) + "。方案须经审核发布，不构成交易指令。"