refactor(data): unify source-health tracking on one shared router (P1.3)

src/data/fetcher.py

@@ -51,6 +51,7 @@
     SSE_INDEX_PREFIXES,
     SZSE_INDEX_PREFIXES,
 )
+from src.data.source_router import SourceDomain, get_source_router
 from src.utils.config import get_data_dir, load_config
 from src.utils.logger import get_logger
 
@@ -123,6 +124,8 @@ def __init__(self, config_path: str = "stocks") -> None:
         """
         self.config: dict[str, Any] = load_config(config_path)
         self.logger = get_logger("data.fetcher")
+        # Shared health tracker (single source of truth across all data sources).
+        self._source_router = get_source_router(config_path)
 
         # Unpack frequently-used config sections
         self._daily_cfg: dict[str, Any] = self.config.get("data_collection", {}).get(
@@ -239,6 +242,22 @@ def _check_data_freshness(self, df: pd.DataFrame, symbol: str, source: str) -> b
             return False
         return True
 
+    def _record_source(self, domain: SourceDomain, ok: bool) -> None:
+        """Record a source success/failure into the shared health router.
+
+        Defensive: health bookkeeping must never break a data fetch, so any
+        error here is swallowed. Only called on a fresh success or a hard
+        failure — stale data is data lag, not a source failure, so it is not
+        recorded.
+        """
+        try:
+            if ok:
+                self._source_router.record_success(domain)
+            else:
+                self._source_router.record_failure(domain)
+        except Exception:
+            self.logger.debug("source health record skipped", exc_info=True)
+
     def fetch_daily_ohlcv(
         self,
         symbol: str,
@@ -309,10 +328,12 @@ def _track_stale(df: pd.DataFrame) -> None:
                 df = self._qmt.get_daily_ohlcv(symbol, start, end)
                 if df is not None and not df.empty:
                     if self._check_data_freshness(df, symbol, "QMT"):
+                        self._record_source(SourceDomain.QMT, True)
                         self._save_cache(df, cache_path)
                         return df
                     _track_stale(df)
             except Exception as exc:
+                self._record_source(SourceDomain.QMT, False)
                 self.logger.warning(
                     "QMT daily OHLCV failed for %s: %s, trying EastMoney",
                     symbol,
@@ -336,10 +357,12 @@ def _track_stale(df: pd.DataFrame) -> None:
             )
             df = df.rename(columns=OHLCV_COLUMN_MAP)
             if self._check_data_freshness(df, symbol, "EastMoney"):
+                self._record_source(SourceDomain.EASTMONEY_PUSH2, True)
                 self._save_cache(df, cache_path)
                 return df
             _track_stale(df)
         except DataCollectionError:
+            self._record_source(SourceDomain.EASTMONEY_PUSH2, False)
             self.logger.warning(
                 "EastMoney source failed for %s, trying Tencent fallback",
                 symbol,
@@ -363,10 +386,12 @@ def _track_stale(df: pd.DataFrame) -> None:
                 df = df.rename(columns={"amount": "volume"})
 
             if self._check_data_freshness(df, symbol, "Tencent"):
+                self._record_source(SourceDomain.TENCENT, True)
                 self._save_cache(df, cache_path)
                 return df
             _track_stale(df)
         except DataCollectionError:
+            self._record_source(SourceDomain.TENCENT, False)
             self.logger.warning(
                 "Tencent source failed for %s, trying adata fallback",
                 symbol,
@@ -376,9 +401,11 @@ def _track_stale(df: pd.DataFrame) -> None:
         try:
             df = self._fetch_daily_via_adata(symbol, start, end, adjust, cache_path)
             if self._check_data_freshness(df, symbol, "adata"):
+                self._record_source(SourceDomain.ADATA, True)
                 return df  # _fetch_daily_via_adata already saves cache
             _track_stale(df)
         except DataCollectionError:
+            self._record_source(SourceDomain.ADATA, False)
             self.logger.warning("adata source also failed for %s", symbol)
 
         # All sources returned stale data — use the freshest one but warn

src/data/news_fetcher.py

@@ -17,7 +17,11 @@
     HOT_RANK_COLUMN_MAP,
     NEWS_COLUMN_MAP,
 )
-from src.data.source_router import DataSourceRouter, SourceDomain
+from src.data.source_router import (
+    DataSourceRouter,
+    SourceDomain,
+    get_source_router,
+)
 from src.utils.config import load_config
 from src.utils.logger import get_logger
 
@@ -74,7 +78,7 @@ def __init__(
         self._max_items: int = news_cfg.get("max_items_per_stock", 20)
         self._hot_rank_limit: int = news_cfg.get("hot_rank_limit", 50)
         self._cache_ttl: float = float(news_cfg.get("cache_ttl_seconds", 300))
-        self._source_router = source_router or DataSourceRouter()
+        self._source_router = source_router or get_source_router()
         self._cache: dict[str, tuple[float, Any]] = {}
         self._last_request_ts: float = 0.0
 

src/data/realtime.py

@@ -14,7 +14,11 @@
 import pandas as pd
 import requests as _requests
 
-from src.data.source_router import DataSourceRouter, SourceDomain
+from src.data.source_router import (
+    DataSourceRouter,
+    SourceDomain,
+    get_source_router,
+)
 from src.utils.config import load_config
 from src.utils.logger import get_logger
 
@@ -80,7 +84,7 @@ def __init__(
         self._cache_ttl: float = float(rt_cfg.get("cache_ttl_seconds", 5))
         self._batch_size: int = rt_cfg.get("batch_size", 50)
         self._rate_limit: float = 1.0 / rt_cfg.get("rate_limit_per_second", 2)
-        self._source_router = source_router or DataSourceRouter(config_name)
+        self._source_router = source_router or get_source_router(config_name)
         self._cache: dict[str, tuple[float, dict[str, Any]]] = {}
         self._last_request_ts: float = 0.0
         self._xueqiu_session: _requests.Session | None = None

src/data/source_router.py

@@ -161,3 +161,33 @@ def is_source_available(self, domain: SourceDomain) -> bool:
             True if the source is not marked DOWN.
         """
         return self._sources[domain].health != SourceHealth.DOWN
+
+
+# ---------------------------------------------------------------------------
+# Process-wide shared instance — the single source of source-health truth
+# ---------------------------------------------------------------------------
+
+_DEFAULT_ROUTER: DataSourceRouter | None = None
+
+
+def get_source_router(config_name: str = "stocks") -> DataSourceRouter:
+    """Return the process-wide shared :class:`DataSourceRouter`.
+
+    The daily OHLCV fetcher, realtime quotes, and news all record into and read
+    from this one instance, so source health lives in a single place (and the
+    ``/admin`` health view reflects the whole system rather than an empty
+    per-request router). The first caller's ``config_name`` wins.
+    """
+    global _DEFAULT_ROUTER
+    if _DEFAULT_ROUTER is None:
+        _DEFAULT_ROUTER = DataSourceRouter(config_name)
+    return _DEFAULT_ROUTER
+
+
+def reset_source_router() -> None:
+    """Drop the shared router so the next ``get_source_router`` rebuilds it.
+
+    Used by tests for isolation (health state is in-memory and mutable).
+    """
+    global _DEFAULT_ROUTER
+    _DEFAULT_ROUTER = None

src/web/routes/api_v1/admin.py

@@ -203,9 +203,9 @@ async def get_data_sources(
     Provides a unified view of all data source backends with their
     current health, latency, and availability status.
     """
-    from src.data.source_router import DataSourceRouter
+    from src.data.source_router import get_source_router
 
-    router_instance = DataSourceRouter()
+    router_instance = get_source_router()
     router_status = router_instance.get_status()
     tracker_health = tracker.get_all_health()
 

tests/conftest.py

@@ -9,6 +9,25 @@
 from unittest.mock import MagicMock, patch
 
 
+# ---------------------------------------------------------------------------
+# Shared-singleton isolation
+# ---------------------------------------------------------------------------
+
+
+@pytest.fixture(autouse=True)
+def _reset_source_router():
+    """Reset the process-wide DataSourceRouter before/after each test.
+
+    Source health is in-memory and shared; resetting keeps tests that exercise
+    the fetcher cascade from leaking health state into one another.
+    """
+    from src.data.source_router import reset_source_router
+
+    reset_source_router()
+    yield
+    reset_source_router()
+
+
 # ---------------------------------------------------------------------------
 # Database isolation — prevent ALL tests from touching production databases
 # ---------------------------------------------------------------------------

tests/unit/test_source_router_shared.py

@@ -0,0 +1,61 @@
+"""Tests for the shared DataSourceRouter — single source of health truth (#49).
+
+The daily fetcher, realtime, and news previously each held their own router, so
+health was fragmented and the /admin view showed an empty per-request router.
+These tests lock in that they now share one instance and the fetcher cascade
+records into it.
+"""
+
+from __future__ import annotations
+
+from src.data.source_router import (
+    SourceDomain,
+    get_source_router,
+    reset_source_router,
+)
+
+
+def test_get_source_router_returns_singleton():
+    reset_source_router()
+    assert get_source_router() is get_source_router()
+
+
+def test_reset_rebuilds_a_fresh_router():
+    a = get_source_router()
+    reset_source_router()
+    assert get_source_router() is not a
+
+
+def test_fetcher_uses_the_shared_router():
+    reset_source_router()
+    from src.data.fetcher import StockDataFetcher
+
+    fetcher = StockDataFetcher()
+    assert fetcher._source_router is get_source_router()
+
+
+def test_fetcher_recording_propagates_to_shared_router():
+    reset_source_router()
+    from src.data.fetcher import StockDataFetcher
+
+    fetcher = StockDataFetcher()
+    router = get_source_router()
+
+    # 3 consecutive failures marks the source DOWN (unavailable)...
+    for _ in range(3):
+        fetcher._record_source(SourceDomain.TENCENT, False)
+    assert router.is_source_available(SourceDomain.TENCENT) is False
+
+    # ...and a success recovers it — visible on the same shared instance.
+    fetcher._record_source(SourceDomain.TENCENT, True)
+    assert router.is_source_available(SourceDomain.TENCENT) is True
+
+
+def test_record_source_never_raises():
+    """Health bookkeeping must never break a fetch, even on a bad domain."""
+    reset_source_router()
+    from src.data.fetcher import StockDataFetcher
+
+    fetcher = StockDataFetcher()
+    fetcher._source_router = None  # simulate a broken router
+    fetcher._record_source(SourceDomain.QMT, True)  # must not raise