diff --git a/app/infrastructure/neo4j_graph_driver.py b/app/infrastructure/neo4j_graph_driver.py new file mode 100644 index 0000000..dfe03f5 --- /dev/null +++ b/app/infrastructure/neo4j_graph_driver.py @@ -0,0 +1,27 @@ +"""Bounded Neo4j adapter used only behind RelationshipService.""" + +import asyncio +from collections.abc import Sequence +from typing import Any + +from neo4j import AsyncGraphDatabase + +from app.core.config import Settings + + +class Neo4jGraphDriver: + def __init__(self, settings: Settings, *, timeout_seconds: float = 2.0) -> None: + self.settings = settings + self.timeout_seconds = timeout_seconds + self._driver: Any = None + + async def execute_query(self, query: str, **parameters: Any) -> Sequence[Any]: + if self._driver is None: + self._driver = AsyncGraphDatabase.driver( + self.settings.neo4j_uri, + auth=(self.settings.neo4j_username, self.settings.neo4j_password), + ) + async with asyncio.timeout(self.timeout_seconds): + async with self._driver.session(database=self.settings.neo4j_database) as session: + result = await session.run(query, parameters) + return [record.data() async for record in result] diff --git a/app/service/portfolio_analysis_service.py b/app/service/portfolio_analysis_service.py index 5541aad..8c87887 100644 --- a/app/service/portfolio_analysis_service.py +++ b/app/service/portfolio_analysis_service.py @@ -6,20 +6,31 @@ from datetime import date from decimal import Decimal from typing import Any +from app.core.config import get_settings from app.core.contracts import RequestContext from app.core.portfolio_analysis_contracts import PortfolioAnalysisQuery from app.infrastructure.db import SessionFactory +from app.infrastructure.neo4j_graph_driver import Neo4jGraphDriver from app.model.advisor_product import AdvisorProductIndustryExposure, AdvisorProductMetricSnapshot from app.model.fund import FundHolding, FundProduct from app.repository.portfolio_analysis_repository import PortfolioAnalysisRepository from app.service.authorization_service import AuthorizationService +from app.service.relationship_service import RelationshipService HUNDRED = Decimal("100") class PortfolioAnalysisService: - def __init__(self, *, session_factory: Callable[[], Any] = SessionFactory) -> None: + def __init__( + self, + *, + session_factory: Callable[[], Any] = SessionFactory, + graph_service: RelationshipService | None = None, + ) -> None: self.session_factory = session_factory + self.graph_service = graph_service or RelationshipService( + Neo4jGraphDriver(get_settings()) + ) async def analyze_for_agent( self, _arguments: PortfolioAnalysisQuery, context: RequestContext @@ -32,12 +43,29 @@ class PortfolioAnalysisService: exposures = await repository.latest_industry_exposures(ids, date.today()) metrics = await repository.latest_metrics(ids, date.today()) result = self._analyze(positions, exposures, metrics) - result["graph_context"] = { - "degraded": True, - "reason": "neo4j_not_configured", - } + result["graph_context"] = await self._graph_context(int(context.user_id)) return result + async def _graph_context(self, customer_id: int) -> dict[str, object]: + try: + context = await self.graph_service.portfolio_industry_context(customer_id) + except Exception as exc: + return {"degraded": True, "reason": f"neo4j_unavailable:{type(exc).__name__}"} + if bool(context.get("degraded")): + return {"degraded": True, "reason": str(context.get("reason", "neo4j_unavailable"))} + raw = context.get("overlaps") + if not isinstance(raw, list): + return {"degraded": True, "reason": "neo4j_invalid_result"} + overlaps = [ + {"industry_name": item["industry_name"], "product_count": item["product_count"]} + for item in raw + if isinstance(item, dict) + and isinstance(item.get("industry_name"), str) + and isinstance(item.get("product_count"), int) + and item["product_count"] >= 2 + ] + return {"degraded": False, "overlaps": overlaps[:5]} + @classmethod def _analyze( cls, diff --git a/app/service/relationship_service.py b/app/service/relationship_service.py index 90962ec..2d1c841 100644 --- a/app/service/relationship_service.py +++ b/app/service/relationship_service.py @@ -49,3 +49,20 @@ class RelationshipService: return await self.driver.execute_query(query, customer_id=customer_id, limit=safe_limit) except Exception as exc: return [GraphDegradedResult(f"neo4j_unavailable:{type(exc).__name__}")] + + async def portfolio_industry_context(self, customer_id: int) -> dict[str, object]: + """Return only relationship enrichment; no position value comes from Neo4j.""" + query = ( + "MATCH (c:Customer {customer_id: $customer_id})-[:HOLDS]->" + "(h:Holding)-[:EXPOSED_TO_INDUSTRY]->(i:Industry) " + "RETURN i.industry_name AS industry_name, count(DISTINCT h) AS product_count " + "ORDER BY product_count DESC LIMIT $limit" + ) + try: + rows = await self.driver.execute_query(query, customer_id=customer_id, limit=20) + except Exception as exc: + return {"degraded": True, "reason": f"neo4j_unavailable:{type(exc).__name__}"} + if any(isinstance(row, GraphDegradedResult) for row in rows): + degraded = next(row for row in rows if isinstance(row, GraphDegradedResult)) + return {"degraded": True, "reason": degraded.reason} + return {"degraded": False, "overlaps": list(rows)} diff --git a/docs/21-投顾Agent迁移TODO.md b/docs/21-投顾Agent迁移TODO.md index 892b30f..0159420 100644 --- a/docs/21-投顾Agent迁移TODO.md +++ b/docs/21-投顾Agent迁移TODO.md @@ -80,15 +80,16 @@ Agent 暴露客户最新的 `confirmed` 目标,未确认目标不会进入后 阶段七测试结果:专项测试 `5 passed`,全量单元测试 `468 passed, 3 warnings`,Ruff 通过,MyPy(122 个源文件)通过。阶段七提交:待提交。 -### 阶段八:持仓分析(核心完成,图谱增强待补) +### 阶段八:持仓分析 已基于现行 `fin_holding`、`fin_product` 和投顾行业/历史指标表完成只读持仓分析: 支持持仓查询、市值汇总、单产品集中度、行业穿透、产品/行业 HHI、行业与历史指标 覆盖率,并在市值或行业数据不足时返回明确的降级状态和警告。MySQL 是数值分析的 -唯一事实来源,输出只生成分析结论,不生成交易指令;图谱上下文当前明确标记为 -`neo4j_not_configured`,真实 Neo4j 关系增强留在后续阶段。 +唯一事实来源,输出只生成分析结论,不生成交易指令;Neo4j 仅通过 +`RelationshipService` 的固定关系模板提供行业重叠辅助上下文,连接失败、超时或结果 +异常时明确降级,不影响 MySQL 数值分析。 -阶段八测试结果:专项测试 `4 passed`,全量单元测试 `471 passed, 3 warnings`,Ruff +阶段八测试结果:图谱与持仓专项测试 `8 passed`,全量单元测试 `475 passed, 3 warnings`,Ruff 通过,MyPy(127 个源文件)通过。阶段八核心提交:待提交。 ## 一、迁移准备 @@ -274,13 +275,13 @@ python tools/audit_constraints.py - [x] 迁移 HHI 指标计算。 - [x] 迁移行业穿透计算。 - [x] 迁移行情和行业数据覆盖率计算。 -- [ ] 迁移 Neo4j 图谱增强。 +- [x] 迁移 Neo4j 图谱增强。(固定关系模板、参数化查询、2 秒超时) - [x] 确认 MySQL 是数值分析权威来源。 -- [x] 确认 Neo4j 只做关系增强。(真实驱动接入待补) -- [x] 实现图谱不可用时的降级。(当前返回 `neo4j_not_configured`) +- [x] 确认 Neo4j 只做关系增强。 +- [x] 实现图谱不可用时的降级。(连接失败、超时或结果异常均标记 degraded) - [x] 实现关键数据不足时的降级。 - [x] 确认分析结果不生成交易指令。 -- [x] 完成持仓分析提交 `advisor/portfolio-analysis`。(专项 `4 passed`;全量单元 `471 passed`;图谱增强待补) +- [x] 完成持仓分析提交 `advisor/portfolio-analysis`。(专项 `8 passed`;全量单元 `475 passed`) 验收: diff --git a/tests/unit/service/test_dependency_degradation.py b/tests/unit/service/test_dependency_degradation.py index 6ec3647..d71e29e 100644 --- a/tests/unit/service/test_dependency_degradation.py +++ b/tests/unit/service/test_dependency_degradation.py @@ -24,6 +24,37 @@ async def test_neo4j_rejects_unapproved_relationship() -> None: await RelationshipService(FailingGraph()).neighbors(1, "DELETE_ALL") +@pytest.mark.asyncio +async def test_portfolio_graph_context_uses_fixed_query_and_filters_noise() -> None: + class Graph: + async def execute_query(self, query: str, **parameters: object) -> list[object]: + assert "HOLDS" in query and "EXPOSED_TO_INDUSTRY" in query + assert parameters == {"customer_id": 7, "limit": 20} + return [ + {"industry_name": "科技", "product_count": 2}, + {"industry_name": "单产品行业", "product_count": 1}, + ] + + result = await RelationshipService(Graph()).portfolio_industry_context(7) + assert result == { + "degraded": False, + "overlaps": [ + {"industry_name": "科技", "product_count": 2}, + {"industry_name": "单产品行业", "product_count": 1}, + ], + } + + +@pytest.mark.asyncio +async def test_portfolio_graph_context_failure_is_degraded() -> None: + from app.service.portfolio_analysis_service import PortfolioAnalysisService + + service = PortfolioAnalysisService(graph_service=RelationshipService(FailingGraph())) + result = await service._graph_context(7) + assert result["degraded"] is True + assert str(result["reason"]).startswith("neo4j_unavailable") + + @pytest.mark.asyncio async def test_model_router_excludes_unhealthy_endpoint() -> None: from app.model.configuration import ModelEndpointConfig