From b6429e0e9c98fdccf3a33aeea09ed311414db3d2 Mon Sep 17 00:00:00 2001 From: Windows Date: Fri, 11 Sep 2026 19:20:32 +0800 Subject: [PATCH] feat: add advisory product comparison tool --- app/core/product_comparison_contracts.py | 13 ++ app/service/agent/bootstrap.py | 10 ++ app/service/agent/implementations/advisor.py | 33 ++++- app/service/product_comparison_service.py | 118 ++++++++++++++++++ docs/21-投顾Agent迁移TODO.md | 9 +- .../unit/service/test_advisor_base_adapter.py | 1 + .../test_product_comparison_service.py | 43 +++++++ tools/seed_test_rbac.py | 3 +- 8 files changed, 227 insertions(+), 3 deletions(-) create mode 100644 app/core/product_comparison_contracts.py create mode 100644 app/service/product_comparison_service.py create mode 100644 tests/unit/service/test_product_comparison_service.py diff --git a/app/core/product_comparison_contracts.py b/app/core/product_comparison_contracts.py new file mode 100644 index 0000000..8837e4f --- /dev/null +++ b/app/core/product_comparison_contracts.py @@ -0,0 +1,13 @@ +"""Read-only contract for comparing exchange-traded products.""" + +from typing import Annotated + +from pydantic import BaseModel, ConfigDict, Field, StringConstraints + +ProductCode = Annotated[str, StringConstraints(pattern=r"^\d{6}$")] + + +class ProductComparisonQuery(BaseModel): + model_config = ConfigDict(extra="forbid", frozen=True) + + product_codes: Annotated[tuple[ProductCode, ...], Field(min_length=2, max_length=4)] diff --git a/app/service/agent/bootstrap.py b/app/service/agent/bootstrap.py index d55a9a5..2c15056 100644 --- a/app/service/agent/bootstrap.py +++ b/app/service/agent/bootstrap.py @@ -10,6 +10,7 @@ from app.core.errors import RecoverableAgentError from app.core.fund_contracts import FundQuoteQuery from app.core.investment_goal_contracts import InvestmentGoalQuery from app.core.portfolio_analysis_contracts import PortfolioAnalysisQuery +from app.core.product_comparison_contracts import ProductComparisonQuery from app.core.product_recommendation_contracts import ProductRecommendationQuery from app.infrastructure.fund_quote_cache import FundQuoteCache from app.infrastructure.memory_cache import MemoryCacheAdapter @@ -31,6 +32,7 @@ from app.service.model_gateway import ( ModelGenerationService, ) from app.service.portfolio_analysis_service import portfolio_analysis_tool +from app.service.product_comparison_service import product_comparison_tool from app.service.product_recommendation_service import product_recommendation_tool from app.service.runtime_config_service import load_active_intent_configs from app.service.suitability_service import SuitabilityToolInput, suitability_tool_handler @@ -183,6 +185,14 @@ def get_agent_factory() -> AgentFactory: allowed_roles=("customer", "advisor", "operator", "admin"), timeout_seconds=15, )) + registry.register(ToolDefinition( + name="compare_products", + input_model=ProductComparisonQuery, + handler=cast(Any, product_comparison_tool), + required_permission="product-comparison:read:self", + allowed_roles=("customer", "advisor", "operator", "admin"), + timeout_seconds=10, + )) model_service = get_model_service() endpoint_resolver = DatabaseModelEndpointResolver() factory = AgentFactory( diff --git a/app/service/agent/implementations/advisor.py b/app/service/agent/implementations/advisor.py index 8a20c7a..5a285bd 100644 --- a/app/service/agent/implementations/advisor.py +++ b/app/service/agent/implementations/advisor.py @@ -1,5 +1,6 @@ """投顾 Agent 的公共底座实现。""" +import re from typing import Any from app.core.contracts import AgentDefinition, AgentRequest, CoreResult, RequestContext @@ -21,6 +22,7 @@ class AdvisorAgent(FundQueryDemoAgent): "analyze_portfolio", "generate_asset_allocation", "recommend_products", + "compare_products", ), supported_intents=( "fund_quote", @@ -77,7 +79,16 @@ class AdvisorAgent(FundQueryDemoAgent): self._classified_intent is not None and self._classified_intent.intent == "comparison" ): - return CoreResult(text="对比分析需要明确两个或多个场内基金产品，请提供基金代码或名称。") + codes = tuple(dict.fromkeys(re.findall(r"(? str: + if not isinstance(result, dict): + return "基金对比分析暂不可用，请稍后重试。" + if result.get("status") == "evidence_required": + return "部分基金缺少当前有效的权威证据，暂不能完成可靠对比。" + products = result.get("products") + if result.get("status") != "ready" or not isinstance(products, list): + return "对比分析需要至少两个有效的场内基金产品。" + names = [ + f"{item.get('product_code')} {item.get('product_name')}" + for item in products if isinstance(item, dict) + ] + common = result.get("common_industries") + common_text = "、".join(str(item) for item in common) if isinstance(common, list) else "无" + return ( + "对比分析：" + "；".join(names) + + f"。共同行业暴露：{common_text}。仅供分析参考，不构成交易指令。" + ) diff --git a/app/service/product_comparison_service.py b/app/service/product_comparison_service.py new file mode 100644 index 0000000..4ce9f13 --- /dev/null +++ b/app/service/product_comparison_service.py @@ -0,0 +1,118 @@ +"""Evidence-based comparison for listed Southern Fund products.""" + +from collections.abc import Callable +from datetime import UTC, date, datetime +from typing import Any + +from app.core.contracts import RequestContext +from app.core.product_comparison_contracts import ProductComparisonQuery +from app.infrastructure.db import SessionFactory +from app.repository.advisor_product_repository import AdvisorProductRepository +from app.repository.portfolio_analysis_repository import PortfolioAnalysisRepository +from app.service.authorization_service import AuthorizationService +from app.service.product_governance_monitor_service import SALES_INSTITUTION + + +class ProductComparisonService: + def __init__(self, *, session_factory: Callable[[], Any] = SessionFactory) -> None: + self.session_factory = session_factory + + async def compare( + self, payload: ProductComparisonQuery, context: RequestContext + ) -> dict[str, object]: + await AuthorizationService.require(context, "product-comparison:read:self") + codes = tuple(dict.fromkeys(payload.product_codes)) + if len(codes) < 2: + return {"status": "products_required", "message": "至少需要两个不同的基金代码。"} + now = datetime.now(UTC).replace(tzinfo=None) + async with self.session_factory() as session: + candidates = await AdvisorProductRepository(session).authoritative_tradable_products( + now, sales_institution=SALES_INSTITUTION, fund_manager="南方基金", limit=50 + ) + selected = [item for item in candidates if item.product.product_code in codes] + by_code = {item.product.product_code: item for item in selected} + missing = [code for code in codes if code not in by_code] + ids = tuple(item.product.id for item in selected) + exposures = await PortfolioAnalysisRepository(session).latest_industry_exposures( + ids, date.today() + ) + if missing: + return { + "status": "evidence_required", + "missing_product_codes": missing, + "message": "部分产品缺少当前有效的权威适当性或合同证据，无法完成对比。", + } + rows = [ + self._product_view(by_code[code], exposures.get(by_code[code].product.id, [])) + for code in codes + ] + industry_sets: list[set[str]] = [] + for row in rows: + raw_exposure = row.get("industry_exposure") + industry_sets.append({ + str(item["industry_name"]) + for item in raw_exposure + if isinstance(item, dict) and "industry_name" in item + } if isinstance(raw_exposure, list) else set()) + common_industries = sorted(set.intersection(*industry_sets)) if industry_sets else [] + return { + "status": "ready", + "products": rows, + "common_industries": common_industries, + "differences": self._differences(rows), + "analysis_only": True, + "disclaimer": "对比结果仅供分析参考，不构成交易指令。", + } + + @staticmethod + def _product_view(candidate: Any, exposures: list[Any]) -> dict[str, object]: + product = candidate.product + contract = candidate.contract + liquidity = candidate.liquidity + return { + "product_code": product.product_code, + "product_name": product.product_name, + "product_category": product.product_category, + "suitability_risk_level": candidate.suitability.risk_level, + "fund_type": contract.fund_type, + "investment_scope": contract.investment_scope, + "performance_benchmark": contract.performance_benchmark, + "management_fee_rate_pct": str(contract.management_fee_rate_pct) + if contract.management_fee_rate_pct is not None else None, + "custodian_fee_rate_pct": str(contract.custodian_fee_rate_pct) + if contract.custodian_fee_rate_pct is not None else None, + "asset_scale_billion": str(candidate.asset_scale_billion) + if candidate.asset_scale_billion is not None else None, + "liquidity": { + "status": liquidity.status if liquidity else "unknown", + "average_daily_turnover_amount": str(liquidity.average_daily_turnover_amount) + if liquidity and liquidity.average_daily_turnover_amount is not None else None, + }, + "industry_exposure": [ + {"industry_name": item.industry_name, + "exposure_weight_pct": str(item.exposure_weight_pct)} + for item in exposures + ], + "evidence": { + "suitability_source_url": candidate.suitability.source_url, + "contract_source_url": contract.source_url, + }, + } + + @staticmethod + def _differences(rows: list[dict[str, object]]) -> dict[str, list[object]]: + fields = ( + "product_category", "suitability_risk_level", "fund_type", + "performance_benchmark", "management_fee_rate_pct", "custodian_fee_rate_pct", + ) + return { + field: [row[field] for row in rows] + for field in fields + if len({str(row[field]) for row in rows}) > 1 + } + + +async def product_comparison_tool( + arguments: ProductComparisonQuery, context: RequestContext +) -> dict[str, object]: + return await ProductComparisonService().compare(arguments, context) diff --git a/docs/21-投顾Agent迁移TODO.md b/docs/21-投顾Agent迁移TODO.md index 5a04cf6..e8adcf8 100644 --- a/docs/21-投顾Agent迁移TODO.md +++ b/docs/21-投顾Agent迁移TODO.md @@ -153,6 +153,13 @@ Redis 不可用时实测按设计降级放行；生产装配模式因本机 Milv 端点批量请求出现 `RemoteProtocolError`，因此质量 `rejected=19`，产品推荐和动态配置真实验收仍待 行情源恢复后复验。实现提交：`930dd59`。 +阶段十五完成对比分析工具：新增 `ProductComparisonQuery` 和只读 `ProductComparisonService`，通过 +公共 `ToolExecutor` 注册 `compare_products`，读取已审核的场内产品证据、合同字段、流动性和行业暴露， +输出共同行业、差异字段和证据来源；缺少权威证据时返回 `evidence_required`，不生成交易指令。客户侧 +对比意图已可从消息提取 2 至 4 个基金代码并返回摘要。专项测试 `4 passed`，全量单元测试 `497 passed， +3 warnings`，契约测试 `8 passed`，独立迁移库集成测试 `28 passed，1 skipped`，Ruff 和 MyPy 通过。 +真实验收：`159511` 与 `510500` 对比返回 `ready`、2 个产品和差异字段。实现提交：待提交。 + ## 一、迁移准备 - [ ] 确认远程仓库可访问。（当前失败：连接 `47.106.207.27:3000` 被拒绝） @@ -415,7 +422,7 @@ python tools/audit_constraints.py - [x] 迁移产品推荐意图。 - [x] 迁移持仓分析意图。 - [x] 迁移资产配置意图。 -- [x] 迁移对比分析意图。（已完成安全路由和参数提示；计算工具待补） +- [x] 迁移对比分析意图。（`compare_products` 只读工具已接入，支持 2 至 4 个场内基金代码） - [x] 迁移投资目标意图。 - [x] 迁移会话实体抽取。（`GoalConversationService`，只抽取明确目标字段） - [x] 迁移投资目标缺口识别。（按当前会话所有用户轮次合并） diff --git a/tests/unit/service/test_advisor_base_adapter.py b/tests/unit/service/test_advisor_base_adapter.py index 54e7622..2c0d8d0 100644 --- a/tests/unit/service/test_advisor_base_adapter.py +++ b/tests/unit/service/test_advisor_base_adapter.py @@ -24,6 +24,7 @@ def test_advisor_is_registered_through_the_new_base_factory() -> None: "analyze_portfolio", "generate_asset_allocation", "recommend_products", + "compare_products", ) assert definition.supported_intents == ( "fund_quote", diff --git a/tests/unit/service/test_product_comparison_service.py b/tests/unit/service/test_product_comparison_service.py new file mode 100644 index 0000000..c4266fc --- /dev/null +++ b/tests/unit/service/test_product_comparison_service.py @@ -0,0 +1,43 @@ +import pytest +from pydantic import ValidationError + +from app.core.product_comparison_contracts import ProductComparisonQuery +from app.service.agent.implementations.advisor import AdvisorAgent +from app.service.product_comparison_service import ProductComparisonService + + +def test_comparison_query_requires_two_to_four_six_digit_codes() -> None: + assert ProductComparisonQuery(product_codes=["159511", "510500"]).product_codes == ( + "159511", "510500" + ) + with pytest.raises(ValidationError): + ProductComparisonQuery(product_codes=["159511"]) + with pytest.raises(ValidationError): + ProductComparisonQuery(product_codes=["15951x", "510500"]) + + +def test_comparison_differences_only_contains_changed_fields() -> None: + rows = [ + {"product_category": "ETF", "suitability_risk_level": "R2", "fund_type": "股票型", + "performance_benchmark": "沪深300", "management_fee_rate_pct": "0.50", + "custodian_fee_rate_pct": "0.10"}, + {"product_category": "LOF", "suitability_risk_level": "R2", "fund_type": "股票型", + "performance_benchmark": "沪深300", "management_fee_rate_pct": "0.60", + "custodian_fee_rate_pct": "0.10"}, + ] + + result = ProductComparisonService._differences(rows) + + assert set(result) == {"product_category", "management_fee_rate_pct"} + + +def test_advisor_comparison_description_is_analysis_only() -> None: + text = AdvisorAgent._describe_comparison({ + "status": "ready", + "products": [{"product_code": "159511", "product_name": "南方测试ETF"}], + "common_industries": ["科技"], + }) + + assert "159511 南方测试ETF" in text + assert "科技" in text + assert "不构成交易指令" in text diff --git a/tools/seed_test_rbac.py b/tools/seed_test_rbac.py index 5485337..e398dcb 100644 --- a/tools/seed_test_rbac.py +++ b/tools/seed_test_rbac.py @@ -58,12 +58,13 @@ PERMISSIONS: tuple[tuple[int, str, str, str, str], ...] = ( (9029, "product-recommendation:review", "product-recommendation", "review", "all"), (9030, "product-recommendation:publish", "product-recommendation", "publish", "all"), (9031, "asset-allocation:backtest", "asset-allocation", "backtest", "all"), + (9032, "product-comparison:read:self", "product-comparison", "read", "self"), ) # 客户：业务侧自助能力（自己的会话、反馈、转人工、自己的记忆画像）。 CUSTOMER_PERMISSIONS = ( 9001, 9002, 9003, 9004, 9005, 9006, 9007, 9008, 9009, 9011, - 9018, 9019, 9020, 9021, 9022, 9023, 9024, + 9018, 9019, 9020, 9021, 9022, 9023, 9024, 9032, ) # 风控专员：业务侧只读 + 跨客户记忆 + 审计只读，不含配置写权限。 RISK_PERMISSIONS = (9001, 9002, 9003, 9010, 9011, 9012)