fix: align executable US data routes

2026-09-20 10:53:33 +08:00 · 2026-08-29 14:22:24 +08:00
parent 6ea36abc57
commit e4d092c312
3 changed files with 40 additions and 18 deletions
--- a/docs/data-source-stability.md
+++ b/docs/data-source-stability.md
@@ -38,7 +38,7 @@ Web/API 提供 `GET /api/v1/data/overview` 和等价别名 `GET /api/v1/data/cap
 - `providers`：每个 provider 的 `enabled`、`configured`、`status`、`markets`、`datasets`、精确的 `dataset_markets` 和非敏感 warning。`markets` 与 `datasets` 只是聚合索引，不能推断为笛卡尔积；例如 YFinance 可参与 A 股行情，但 A 股基本面固定走 AkShare，因此 `dataset_markets.financial.snapshot` 不包含 `cn`。Finnhub/AlphaVantage 的美股报价 handler 当前只在 Longbridge/YFinance 已返回主报价后补字段，因此不声明 `quote.realtime` 独立路由能力。
 - `datasets`：`quote.realtime`、`kline.daily`、`index.daily`、`market.overview`、`financial.snapshot`、`news.events`、`strategy.screening`、`alert.monitor`、`portfolio.account` 的 `status/source/fallback_from/stale/warnings`；其中 `quote.realtime`、`kline.daily`、`market.overview` 和 `financial.snapshot` 都聚合市场级 coverage，避免把单市场健康度误报成全局可用。A 股实时额外拆分 `cn` 股票优先级、`cn.index.exchange` 固定指数链和 `cn.index.csi` Efinance-only 链；美股实时拆分 `us` 个股动态优先级与 `us.index` 固定 YFinance-first 链。AkShare Tencent/Sina/EM、AkShare HK 双子源、Efinance 股票/指数子源级熔断及日线 breaker 会先于 provider-wide unknown 判定，使 overview 与运行时跳过行为一致。YFinance 的港股实时声明对应其实际 `.HK` 执行路径；未配置 OpenD 时 HK priority 与运行时一致地跳过 Futu；美股个股实时也按当前请求可用性跳过处于连接冷却的 Longbridge，使 YFinance 成为实际主源而不是伪 fallback。PyTDX 不声明未接入统一 route 的实时能力，YFinance 指数日线只声明实际可执行的 CN/US route。任何按运行时顺序尝试的数据集在首个优先源尚未探测且没有更具体的 open breaker 证据时保持 `unknown`，不会越过它宣称后续源已被选中。`alert.monitor` 只有在开关启用且当前 API 进程的 scheduler 已实际注册并运行 `agent_event_monitor` 时才为 `ok`；开关关闭返回 `agent_event_monitor_disabled`，scheduler 未运行则返回 `agent_event_monitor_not_running`。
 - `priorities`：`cn.realtime`、`hk.realtime`、`us.realtime`、`daily.generic`、`cn.index.daily`、`market.overview`、`screening.snapshot`、`news.events` 的当前 source order；`daily.generic` 与运行时一样先排除当前请求不可用或处于连接冷却的 fetcher，不把已跳过的数据源记录成 fallback。
- `index.daily` 额外区分 `cn.exchange`、`cn.csi` 与 `us` coverage：沪深交易所指数使用 Tencent、AkShare、TickFlow、YFinance 固定多源链，CSI 指数按运行时契约只认 AkShare，美股指数使用 YFinance → Finnhub；Tushare 不声明指数日线能力。CN/HK realtime 都只接受对应运行时实际有 handler 的 source token，Tushare realtime 能力仅声明 A 股。
+- `index.daily` 额外区分 `cn.exchange`、`cn.csi` 与 `us` coverage：沪深交易所指数使用 Tencent、AkShare、TickFlow、YFinance 固定多源链，CSI 指数按运行时契约只认 AkShare，美股指数只声明当前能够接收指数 symbol 的 YFinance；Finnhub 当前 fetcher 会拒绝指数 symbol，因此不作为伪 fallback。Tushare 不声明指数日线能力。CN/HK realtime 都只接受对应运行时实际有 handler 的 source token，Tushare realtime 能力仅声明 A 股。

 其中 `unknown` 表示尚未执行运行态探测，`unconfigured` 表示缺少必要配置，`degraded` 表示前置优先源不可用但后续源仍可消费。选股 source health 在冷启动、成功数和失败数都为 0 时保持 `unknown`，不会把尚未调用的数据源提前标成 `ok`。后续 Data Center 可以在同一结构上追加 last_success、coverage、cooldown 和运行历史，不需要各页面各自维护数据质量口径。

--- a/src/services/data_capability_service.py
+++ b/src/services/data_capability_service.py
@@ -147,7 +147,6 @@ _PROVIDER_DEFINITIONS: Sequence[_ProviderDefinition] = (
        fetcher_name="FinnhubFetcher",
        dataset_markets={
            "kline.daily": ("us",),
-            "index.daily": ("us",),
        },
    ),
    _ProviderDefinition(
@@ -568,7 +567,7 @@ class DataCapabilityService:
                market_priorities={
                    "cn.exchange": priority_map.get("cn.index.daily", {}),
                    "cn.csi": {"providers": ["akshare"], "warnings": []},
-                    "us": {"providers": ["yfinance", "finnhub"], "warnings": []},
+                    "us": {"providers": ["yfinance"], "warnings": []},
                },
                provider_map=provider_map,
                status_resolvers={
@@ -623,6 +622,7 @@ class DataCapabilityService:
            "us": {
                "providers": self._us_daily_priority(fetchers),
                "warnings": [],
+                "empty_status": "unavailable",
            },
            **{
                market: generic_market_priority(market)
@@ -831,14 +831,19 @@ class DataCapabilityService:
        return resolve

    def _us_daily_priority(self, fetchers: Sequence[Any]) -> List[str]:
-        fetcher_map = {str(getattr(fetcher, "name", "")): fetcher for fetcher in fetchers}
-        longbridge = fetcher_map.get("LongbridgeFetcher")
-        if longbridge is not None and self._fetcher_available_for_capability(
-            longbridge,
-            capability="daily_data",
-        ):
-            return ["longbridge", "finnhub", "alphavantage", "yfinance"]
-        return ["finnhub", "alphavantage", "yfinance", "longbridge"]
+        available = {
+            provider
+            for fetcher in fetchers
+            if (provider := _FETCHER_TO_PROVIDER.get(str(getattr(fetcher, "name", ""))))
+            in {"longbridge", "finnhub", "alphavantage", "yfinance"}
+            and self._fetcher_available_for_capability(fetcher, capability="daily_data")
+        }
+        preferred = (
+            ["longbridge", "finnhub", "alphavantage", "yfinance"]
+            if "longbridge" in available
+            else ["finnhub", "alphavantage", "yfinance", "longbridge"]
+        )
+        return [provider for provider in preferred if provider in available]

    @staticmethod
    def _fetcher_available_for_capability(fetcher: Any, *, capability: str) -> bool:
--- a/tests/test_data_capability_service.py
+++ b/tests/test_data_capability_service.py
@@ -219,6 +219,7 @@ def test_provider_dataset_market_matrix_matches_fundamental_runtime_routes() ->
    assert "quote.realtime" not in _provider(overview, "pytdx")["datasets"]
    assert "quote.realtime" not in _provider(overview, "finnhub")["datasets"]
    assert "quote.realtime" not in _provider(overview, "alphavantage")["datasets"]
+    assert "index.daily" not in _provider(overview, "finnhub")["datasets"]


 def test_hk_realtime_priority_skips_futu_when_opend_is_unconfigured() -> None:
@@ -488,7 +489,7 @@ def test_daily_dataset_quality_is_market_aware() -> None:
    for market in ("jp", "kr", "tw"):
        assert daily_quality["coverage"]["markets"][market]["status"] == "unavailable"
    assert "hk:request_available_priority_empty" in daily_quality["warnings"]
-    assert "us:source_status:yfinance:unavailable" in daily_quality["warnings"]
+    assert "us:request_available_priority_empty" in daily_quality["warnings"]


 def test_daily_dataset_quality_prefers_longbridge_for_us_when_available() -> None:
@@ -509,6 +510,22 @@ def test_daily_dataset_quality_prefers_longbridge_for_us_when_available() -> Non
    assert "us:finnhub" not in daily_quality["fallback_from"]


+def test_us_daily_priority_omits_unregistered_configured_sources() -> None:
+    manager = _FetcherManager([
+        _Fetcher("YfinanceFetcher", 4, available=True),
+    ])
+    service = DataCapabilityService(config=_config(), fetcher_manager=manager)
+
+    overview = service.get_overview()
+    priorities = {item["scenario"]: item for item in overview["priorities"]}
+    us_quality = _dataset(overview, "kline.daily")["coverage"]["markets"]["us"]
+
+    assert priorities["daily.generic"]["providers"] == ["yfinance"]
+    assert us_quality["status"] == "ok"
+    assert us_quality["source"] == "yfinance"
+    assert us_quality["fallback_from"] == []
+
+
 def test_daily_priority_excludes_request_unavailable_fetchers() -> None:
    service = DataCapabilityService(
        config=_config(),
@@ -551,19 +568,19 @@ def test_daily_dataset_quality_honors_market_specific_circuit_breakers() -> None
    assert daily_quality["coverage"]["markets"]["cn"]["status"] == "ok"
    assert daily_quality["coverage"]["markets"]["hk"]["status"] == "unavailable"
    assert daily_quality["coverage"]["markets"]["hk"]["source"] is None
-    assert daily_quality["coverage"]["markets"]["us"]["status"] == "degraded"
+    assert daily_quality["coverage"]["markets"]["us"]["status"] == "ok"
    assert daily_quality["coverage"]["markets"]["us"]["source"] == "yfinance"
    assert "hk:source_status:yfinance:cooldown" in daily_quality["warnings"]


-def test_index_daily_quality_includes_us_runtime_route() -> None:
+def test_index_daily_quality_uses_only_the_executable_us_runtime_route() -> None:
    service = DataCapabilityService(
        config=_config(finnhub_api_key="key"),
        fetcher_manager=_FetcherManager([
            _Fetcher("TencentFetcher", 0, available=True),
            _Fetcher("AkshareFetcher", 1, available=True),
            _Fetcher("YfinanceFetcher", 2, available=False),
-            _Fetcher("FinnhubFetcher", 3, available=False),
+            _Fetcher("FinnhubFetcher", 3, available=True),
        ]),
    )

@@ -574,10 +591,10 @@ def test_index_daily_quality_includes_us_runtime_route() -> None:
    assert index_quality["status"] == "partial"
    assert us_quality["status"] == "unavailable"
    assert us_quality["source"] is None
-    assert us_quality["fallback_from"] == ["yfinance", "finnhub"]
+    assert us_quality["fallback_from"] == ["yfinance"]
    assert "us:source_status:yfinance:unavailable" in index_quality["warnings"]
-    assert "us:source_status:finnhub:unavailable" in index_quality["warnings"]
-    assert _provider(overview, "finnhub")["dataset_markets"]["index.daily"] == ["us"]
+    assert not any("finnhub" in warning for warning in index_quality["warnings"])
+    assert "index.daily" not in _provider(overview, "finnhub")["datasets"]


 def test_market_overview_dataset_quality_is_market_aware() -> None: