From 23765858982c8a4d1e53293ee97e58f5b90289f1 Mon Sep 17 00:00:00 2001 From: Windows Date: Fri, 11 Sep 2026 19:47:14 +0800 Subject: [PATCH] feat: add dual-source quote health orchestration --- app/model/advisor_product.py | 50 ++++ app/service/market_quote_sync_service.py | 213 ++++++++++++++++++ docs/21-投顾Agent迁移TODO.md | 7 + .../service/test_market_quote_sync_service.py | 25 ++ 4 files changed, 295 insertions(+) create mode 100644 app/service/market_quote_sync_service.py create mode 100644 tests/unit/service/test_market_quote_sync_service.py diff --git a/app/model/advisor_product.py b/app/model/advisor_product.py index b61dcf2..9d225e2 100644 --- a/app/model/advisor_product.py +++ b/app/model/advisor_product.py @@ -232,3 +232,53 @@ class AdvisorProductMarketQuoteSnapshot(Base): quote_status: Mapped[str] = mapped_column(String(16), nullable=False) source: Mapped[str] = mapped_column(String(64), nullable=False) created_at: Mapped[datetime] = mapped_column(DateTime, nullable=False) + + +class AdvisorMarketQuoteSourceRun(Base): + __tablename__ = "advisor_market_quote_source_run" + + id: Mapped[int] = mapped_column(BigInteger, primary_key=True) + sync_run_no: Mapped[str] = mapped_column(String(36), nullable=False) + source: Mapped[str] = mapped_column(String(64), nullable=False) + priority_order: Mapped[int] = mapped_column(BigInteger, nullable=False) + status: Mapped[str] = mapped_column(String(16), nullable=False) + requested_count: Mapped[int] = mapped_column(BigInteger, nullable=False) + quote_count: Mapped[int] = mapped_column(BigInteger, nullable=False) + error_type: Mapped[str | None] = mapped_column(String(128)) + started_at: Mapped[datetime] = mapped_column(DateTime, nullable=False) + completed_at: Mapped[datetime] = mapped_column(DateTime, nullable=False) + created_at: Mapped[datetime] = mapped_column(DateTime, nullable=False) + + +class AdvisorMarketQuoteSourceHealth(Base): + __tablename__ = "advisor_market_quote_source_health" + + id: Mapped[int] = mapped_column(BigInteger, primary_key=True) + source: Mapped[str] = mapped_column(String(64), nullable=False, unique=True) + status: Mapped[str] = mapped_column(String(16), nullable=False) + consecutive_failure_count: Mapped[int] = mapped_column(BigInteger, nullable=False) + last_success_at: Mapped[datetime | None] = mapped_column(DateTime) + last_failure_at: Mapped[datetime | None] = mapped_column(DateTime) + last_error_type: Mapped[str | None] = mapped_column(String(128)) + last_requested_count: Mapped[int] = mapped_column(BigInteger, nullable=False) + last_quote_count: Mapped[int] = mapped_column(BigInteger, nullable=False) + created_at: Mapped[datetime] = mapped_column(DateTime, nullable=False) + updated_at: Mapped[datetime] = mapped_column(DateTime, nullable=False) + + +class AdvisorMarketQuoteAlert(Base): + __tablename__ = "advisor_market_quote_alert" + + id: Mapped[int] = mapped_column(BigInteger, primary_key=True) + alert_no: Mapped[str] = mapped_column(String(36), nullable=False, unique=True) + source: Mapped[str] = mapped_column(String(64), nullable=False) + alert_type: Mapped[str] = mapped_column(String(64), nullable=False) + severity: Mapped[str] = mapped_column(String(16), nullable=False) + status: Mapped[str] = mapped_column(String(16), nullable=False) + occurrence_count: Mapped[int] = mapped_column(BigInteger, nullable=False) + detail: Mapped[dict[str, Any]] = mapped_column(JSON, nullable=False) + first_observed_at: Mapped[datetime] = mapped_column(DateTime, nullable=False) + last_observed_at: Mapped[datetime] = mapped_column(DateTime, nullable=False) + resolved_at: Mapped[datetime | None] = mapped_column(DateTime) + created_at: Mapped[datetime] = mapped_column(DateTime, nullable=False) + updated_at: Mapped[datetime] = mapped_column(DateTime, nullable=False) diff --git a/app/service/market_quote_sync_service.py b/app/service/market_quote_sync_service.py new file mode 100644 index 0000000..5926504 --- /dev/null +++ b/app/service/market_quote_sync_service.py @@ -0,0 +1,213 @@ +"""Dual-source exchange quote orchestration and health recording.""" + +import asyncio +from collections.abc import Callable +from datetime import UTC, datetime +from decimal import ROUND_HALF_UP, Decimal, InvalidOperation +from typing import Any +from uuid import uuid4 + +from sqlalchemy import select + +from app.infrastructure.db import SessionFactory +from app.model.advisor_product import ( + AdvisorMarketQuoteAlert, + AdvisorMarketQuoteSourceHealth, + AdvisorMarketQuoteSourceRun, + AdvisorProductMarketQuoteSnapshot, +) +from app.model.fund import FundProduct + +QuoteLoader = Callable[[list[str]], dict[str, dict[str, str | None]]] + + +class MarketQuoteSyncService: + SOURCES = ("eastmoney_exchange", "tencent_exchange") + + def __init__( + self, + *, + session_factory: Callable[[], Any] = SessionFactory, + eastmoney_loader: QuoteLoader | None = None, + tencent_loader: QuoteLoader | None = None, + ) -> None: + self.session_factory = session_factory + self.loaders = ( + eastmoney_loader or self._eastmoney_loader, + tencent_loader or self._tencent_loader, + ) + + async def sync(self, *, product_codes: tuple[str, ...] | None = None) -> dict[str, object]: + async with self.session_factory() as session: + rows = list(await session.execute( + select(FundProduct.id, FundProduct.product_code).where( + FundProduct.fund_manager == "南方基金", + FundProduct.exchange_code.in_(("SSE", "SZSE")), + FundProduct.status == "上市", + ).order_by(FundProduct.id) + )) + if product_codes is not None: + allowed = set(product_codes) + rows = [row for row in rows if row.product_code in allowed] + product_ids = {str(code): int(product_id) for product_id, code in rows} + codes = list(product_ids) + sync_run_no = str(uuid4()) + selected: dict[str, tuple[str, dict[str, str | None]]] = {} + source_results: list[dict[str, object]] = [] + for priority, (source, loader) in enumerate( + zip(self.SOURCES, self.loaders, strict=True), 1 + ): + started = datetime.now(UTC).replace(tzinfo=None) + try: + quotes = await asyncio.to_thread(loader, codes) + error_type = None + except Exception as exc: + quotes = {} + error_type = type(exc).__name__ + usable = {code: value for code, value in quotes.items() if code in product_ids} + for code, quote in usable.items(): + selected.setdefault(code, (source, quote)) + status = ( + "failed" if not usable + else "succeeded" if len(usable) == len(codes) + else "degraded" + ) + completed = datetime.now(UTC).replace(tzinfo=None) + source_results.append({ + "source": source, "priority": priority, "status": status, + "requested_count": len(codes), "quote_count": len(usable), + "error_type": error_type, + "started_at": started, "completed_at": completed, + }) + if len(selected) == len(codes): + break + + now = datetime.now(UTC).replace(tzinfo=None) + async with self.session_factory() as session, session.begin(): + for item in source_results: + await self._record_source_run(session, sync_run_no, item, now) + await self._record_health(session, item, now) + for item in source_results: + await self._record_alert(session, item, now) + for code, (source, quote) in selected.items(): + snapshot = self._snapshot(product_ids[code], source, quote, now) + if snapshot is not None: + session.add(snapshot) + return { + "sync_run_no": sync_run_no, + "requested_count": len(codes), + "quote_count": len(selected), + "sources": source_results, + "degraded": len(selected) < len(codes), + } + + @staticmethod + async def _record_source_run( + session: Any, run_no: str, item: dict[str, object], now: datetime + ) -> None: + session.add(AdvisorMarketQuoteSourceRun( + sync_run_no=run_no, source=item["source"], priority_order=item["priority"], + status=item["status"], requested_count=item["requested_count"], + quote_count=item["quote_count"], error_type=item["error_type"], + started_at=item["started_at"], completed_at=item["completed_at"], created_at=now, + )) + + @staticmethod + async def _record_health(session: Any, item: dict[str, object], now: datetime) -> None: + source = str(item["source"]) + health = await session.scalar(select(AdvisorMarketQuoteSourceHealth).where( + AdvisorMarketQuoteSourceHealth.source == source + ).with_for_update()) + source_status = str(item["status"]) + failed = source_status == "failed" + if health is None: + health = AdvisorMarketQuoteSourceHealth( + source=source, status="failed" if failed else source_status, + consecutive_failure_count=1 if failed else 0, + last_success_at=None if failed else now, + last_failure_at=now if failed else None, + last_error_type=item["error_type"], + last_requested_count=item["requested_count"], + last_quote_count=item["quote_count"], created_at=now, updated_at=now, + ) + session.add(health) + return + health.status = "failed" if failed else source_status + health.consecutive_failure_count = health.consecutive_failure_count + 1 if failed else 0 + health.last_success_at = None if failed else now + health.last_failure_at = now if failed else health.last_failure_at + health.last_error_type = item["error_type"] + health.last_requested_count = item["requested_count"] + health.last_quote_count = item["quote_count"] + health.updated_at = now + + @staticmethod + async def _record_alert(session: Any, item: dict[str, object], now: datetime) -> None: + source = str(item["source"]) + alert = await session.scalar(select(AdvisorMarketQuoteAlert).where( + AdvisorMarketQuoteAlert.source == source, + AdvisorMarketQuoteAlert.alert_type == "source_unavailable", + AdvisorMarketQuoteAlert.status == "open", + ).with_for_update()) + if item["status"] == "failed": + if alert is None: + session.add(AdvisorMarketQuoteAlert( + alert_no=str(uuid4()), source=source, alert_type="source_unavailable", + severity="high", status="open", occurrence_count=1, + detail={"error_type": item["error_type"], "quote_count": item["quote_count"]}, + first_observed_at=now, last_observed_at=now, + created_at=now, updated_at=now, + )) + else: + alert.occurrence_count += 1 + alert.last_observed_at = now + alert.updated_at = now + return + if alert is not None: + alert.status = "resolved" + alert.resolved_at = now + alert.last_observed_at = now + alert.updated_at = now + + @staticmethod + def _snapshot( + product_id: int, source: str, quote: dict[str, str | None], now: datetime + ) -> AdvisorProductMarketQuoteSnapshot | None: + try: + last_price = Decimal(str(quote.get("last_price"))).quantize( + Decimal("0.000001"), rounding=ROUND_HALF_UP + ) + except (InvalidOperation, TypeError): + return None + if last_price <= 0: + return None + def decimal(key: str, places: str) -> Decimal | None: + value = quote.get(key) + if value in (None, ""): + return None + try: + return Decimal(str(value)).quantize( + Decimal(places), rounding=ROUND_HALF_UP + ) + except InvalidOperation: + return None + return AdvisorProductMarketQuoteSnapshot( + product_id=product_id, observed_at=now, last_price=last_price, + previous_close=decimal("previous_close", "0.000001"), + change_pct=decimal("change_pct", "0.0001"), + volume=decimal("volume", "0.0001"), + turnover_amount=decimal("turnover_amount", "0.01"), + quote_status="active", source=source, created_at=now, + ) + + @staticmethod + def _eastmoney_loader(codes: list[str]) -> dict[str, dict[str, str | None]]: + from hq import fetch_southern_exchange_quotes_eastmoney + + return fetch_southern_exchange_quotes_eastmoney(codes) + + @staticmethod + def _tencent_loader(codes: list[str]) -> dict[str, dict[str, str | None]]: + from hq import fetch_southern_exchange_quotes_tencent + + return fetch_southern_exchange_quotes_tencent(codes) diff --git a/docs/21-投顾Agent迁移TODO.md b/docs/21-投顾Agent迁移TODO.md index e1be95f..cc2fcb1 100644 --- a/docs/21-投顾Agent迁移TODO.md +++ b/docs/21-投顾Agent迁移TODO.md @@ -160,6 +160,13 @@ Redis 不可用时实测按设计降级放行；生产装配模式因本机 Milv 3 warnings`，契约测试 `8 passed`，独立迁移库集成测试 `28 passed，1 skipped`，Ruff 和 MyPy 通过。 真实验收：`159511` 与 `510500` 对比返回 `ready`、2 个产品和差异字段。实现提交：`b6429e0`。 +阶段十六完成双源行情编排：新增 `MarketQuoteSyncService` 及三个行情健康/来源运行/告警表的 ORM +映射，东方财富主源失败或部分返回时按优先级切换腾讯备用源；记录来源状态、连续失败次数、开放告警 +及恢复关闭，成功行情按字段精度量化后写入场内行情快照。真实独立库同步验证两源均失败时返回 +`degraded=true`、不写入伪行情并进入失败告警链路；本次无可用行情，推荐和动态配置继续失败关闭。 +专项测试 `6 passed`，全量单元测试 `499 passed，3 warnings`，契约测试 `8 passed`，Ruff 和 MyPy +通过。实现提交：待提交。 + ## 一、迁移准备 - [ ] 确认远程仓库可访问。（当前失败：连接 `47.106.207.27:3000` 被拒绝） diff --git a/tests/unit/service/test_market_quote_sync_service.py b/tests/unit/service/test_market_quote_sync_service.py new file mode 100644 index 0000000..88c2a84 --- /dev/null +++ b/tests/unit/service/test_market_quote_sync_service.py @@ -0,0 +1,25 @@ +from datetime import datetime +from decimal import Decimal + +from app.service.market_quote_sync_service import MarketQuoteSyncService + + +def test_snapshot_parses_real_time_quote_fields() -> None: + snapshot = MarketQuoteSyncService._snapshot(7, "eastmoney_exchange", { + "last_price": "1.234", "previous_close": "1.2", "change_pct": "2.83", + "volume": "1000", "turnover_amount": "123456.78", + }, datetime(2026, 9, 10)) + + assert snapshot is not None + assert snapshot.last_price == Decimal("1.234") + assert snapshot.turnover_amount == Decimal("123456.78") + assert snapshot.source == "eastmoney_exchange" + + +def test_snapshot_rejects_missing_or_non_positive_price() -> None: + assert MarketQuoteSyncService._snapshot( + 7, "eastmoney_exchange", {"last_price": None}, datetime.now() + ) is None + assert MarketQuoteSyncService._snapshot( + 7, "eastmoney_exchange", {"last_price": "0"}, datetime.now() + ) is None