feat: add dual-source quote health orchestration

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)

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)

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` 被拒绝）

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