diff --git a/src/data/fetcher.py b/src/data/fetcher.py index 41a5270..0910259 100644 --- a/src/data/fetcher.py +++ b/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 diff --git a/src/data/news_fetcher.py b/src/data/news_fetcher.py index b668ee6..3201190 100644 --- a/src/data/news_fetcher.py +++ b/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 diff --git a/src/data/realtime.py b/src/data/realtime.py index 10ee8a6..d5b2475 100644 --- a/src/data/realtime.py +++ b/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 diff --git a/src/data/source_router.py b/src/data/source_router.py index 40d29d4..3303593 100644 --- a/src/data/source_router.py +++ b/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 diff --git a/src/web/routes/api_v1/admin.py b/src/web/routes/api_v1/admin.py index 957daf4..da32ef1 100644 --- a/src/web/routes/api_v1/admin.py +++ b/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() diff --git a/tests/conftest.py b/tests/conftest.py index 4af57b3..2eb10b8 100644 --- a/tests/conftest.py +++ b/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 # --------------------------------------------------------------------------- diff --git a/tests/unit/test_source_router_shared.py b/tests/unit/test_source_router_shared.py new file mode 100644 index 0000000..44f0bff --- /dev/null +++ b/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