diff --git a/xiaobai-datahub/datahub/admin_api.py b/xiaobai-datahub/datahub/admin_api.py index c5ab3f2..01a488c 100644 --- a/xiaobai-datahub/datahub/admin_api.py +++ b/xiaobai-datahub/datahub/admin_api.py @@ -6,9 +6,11 @@ from typing import Any from datahub.adapters import RESERVED from datahub.auth import AuthService from datahub.db import HubDB +from datahub import lineage as lineage_module from datahub.pipeline import OFFICIAL_DATASETS, STOCKS_DATASET, Pipeline from datahub.scheduler import Scheduler from datahub.serving import ApiError +from datahub import source_catalog from datahub.timeutil import isoformat, now_shanghai, session_phase, yyyymmdd @@ -105,6 +107,40 @@ class AdminAPI: raise ApiError("INVALID_ARGUMENT", f"unknown provider: {provider}") return adapter.probe() + # ------------------------------------------------------------------ + # HEL-543: read-only side-channel status/catalog/lineage. These never + # change routing, credentials, or adapters; they only read the + # provider_call_log/provider_health tables (observability.py) plus the + # static registries in source_catalog.py / lineage.py. + # ------------------------------------------------------------------ + def providers_status(self, provider: str = "", limit: int = 50) -> dict[str, Any]: + limit = max(1, min(int(limit or 50), 200)) + health_sql = "SELECT * FROM provider_health" + params: tuple[Any, ...] = () + if provider: + health_sql += " WHERE provider = ?" + params = (provider,) + health_sql += " ORDER BY provider, interface" + health = self.db.fetchall(health_sql, params) + calls_sql = "SELECT * FROM provider_call_log" + if provider: + calls_sql += " WHERE provider = ?" + calls_sql += " ORDER BY id DESC LIMIT ?" + recent = self.db.fetchall(calls_sql, (*params, limit)) + return {"health": health, "recent_calls": recent} + + def source_catalog(self) -> dict[str, Any]: + return {"items": source_catalog.snapshot(self.db, self.auth)} + + def lineage(self, trade_date: str = "") -> dict[str, Any]: + day = yyyymmdd(trade_date) if trade_date else yyyymmdd(now_shanghai()) + return {"trade_date": day, "items": lineage_module.snapshot(self.db, day)} + + def lineage_affected(self, provider: str = "", interface: str = "") -> dict[str, Any]: + if not provider: + raise ApiError("INVALID_ARGUMENT", "provider is required") + return {"provider": provider, "interface": interface, "items": lineage_module.affected(self.db, provider, interface)} + def jobs(self) -> dict[str, Any]: runs = self.db.fetchall("SELECT * FROM job_runs ORDER BY id DESC LIMIT 100") stocks_times = "/".join(self.pipeline.settings.stocks_refresh_times) or "20:00" diff --git a/xiaobai-datahub/datahub/db.py b/xiaobai-datahub/datahub/db.py index e7bdb28..c23d3f2 100644 --- a/xiaobai-datahub/datahub/db.py +++ b/xiaobai-datahub/datahub/db.py @@ -8,6 +8,7 @@ from pathlib import Path from typing import Any from datahub.datasets_ext import EXTENDED_DATASET_TABLES, EXTENDED_SCHEMA +from datahub.observability import OBS_SCHEMA from datahub.timeutil import isoformat _BASE_SCHEMA = """ @@ -296,7 +297,7 @@ CREATE INDEX IF NOT EXISTS idx_eod_bars_date ON eod_bars(trade_date, batch_id); CREATE INDEX IF NOT EXISTS idx_calendar_open ON trade_calendar(is_open, cal_date); """ -SCHEMA = _BASE_SCHEMA + EXTENDED_SCHEMA +SCHEMA = _BASE_SCHEMA + EXTENDED_SCHEMA + OBS_SCHEMA DATASET_TABLES = { "daily": ("eod_bars", "staging_bars"), diff --git a/xiaobai-datahub/datahub/httpapp.py b/xiaobai-datahub/datahub/httpapp.py index 8540b22..7685dd1 100644 --- a/xiaobai-datahub/datahub/httpapp.py +++ b/xiaobai-datahub/datahub/httpapp.py @@ -122,6 +122,26 @@ class HubRequestHandler(BaseHTTPRequestHandler): if path == "/admin/api/sources" and method == "GET": self._json(self.hub.admin.sources(), HTTPStatus.OK) return + if path == "/admin/api/providers/status" and method == "GET": + query = parse_query(urlparse(self.path).query) + provider = (query.get("provider") or [""])[0] + limit = (query.get("limit") or ["50"])[0] + self._json(self.hub.admin.providers_status(provider, int(limit or 50)), HTTPStatus.OK) + return + if path == "/admin/api/source-catalog" and method == "GET": + self._json(self.hub.admin.source_catalog(), HTTPStatus.OK) + return + if path == "/admin/api/lineage" and method == "GET": + query = parse_query(urlparse(self.path).query) + date = (query.get("date") or [""])[0] + self._json(self.hub.admin.lineage(date), HTTPStatus.OK) + return + if path == "/admin/api/lineage/affected" and method == "GET": + query = parse_query(urlparse(self.path).query) + provider = (query.get("provider") or [""])[0] + interface = (query.get("interface") or [""])[0] + self._json(self.hub.admin.lineage_affected(provider, interface), HTTPStatus.OK) + return if path.startswith("/admin/api/sources/") and path.endswith("/probe") and method == "POST": provider = path.split("/")[4] self._json(self.hub.admin.probe(provider), HTTPStatus.OK) diff --git a/xiaobai-datahub/datahub/lineage.py b/xiaobai-datahub/datahub/lineage.py new file mode 100644 index 0000000..cd1cc9e --- /dev/null +++ b/xiaobai-datahub/datahub/lineage.py @@ -0,0 +1,258 @@ +"""Read-only lineage/impact inventory (HEL-543). + +Answers, without changing any routing decision: for a given main-site data +item, which datahub dataset backs it, which provider/interface currently +serves it (primary and backup), and — when a provider/interface is +unhealthy — which datasets and, best-effort, which main-site consumers are +affected. + +Every row cites where it was verified so a reviewer does not have to trust +a paraphrase: + +- ``v1_endpoint``/``primary_source``/``backup_source`` are taken verbatim + from ``datahub/serving.py`` (the ``source=`` string literal passed to + ``_published_rows``/``_official_meta``) or from the provider/interface + pairs wired into ``datahub/realtime_serve.py`` for HEL-543. +- ``known_consumers`` lists only call sites this round actually found via + code search in the ``xiaobai-review`` website tree (cited as + ``file:line`` in the comment above each dataset). Anything not backed by + a citation is left out rather than guessed; a fuller page-by-page map is + tracked separately (HEL-549) and can extend this table later without + touching its shape. + +This module never talks to a provider and never mutates anything; it only +reads ``provider_health``/``provider_call_log`` (HEL-543) and the existing +``publications``/``batches`` tables to attach live status to each row. +""" + +from __future__ import annotations + +from typing import Any + +# Verified against backend/features/screener/data_sync.py (calendar, +# stock_basic, daily, daily_basic, index_daily-as-benchmark, stk_auction, +# moneyflow, ths_hot, dc_hot all called via `self.client.query(...)`) and +# backend/features/heaven/market_context.py (stock_basic, index_daily via +# `self._tushare_client().query(...)` / `client.query(...)`). +DATASETS: list[dict[str, Any]] = [ + { + "dataset": "calendar", + "tier": "official", + "v1_endpoint": "/v1/calendar", + "primary_source": "tushare:trade_cal", + "backup_source": None, + "known_consumers": ["智能选股(data_sync.py 交易日历解析)", "问天(交易日推算)"], + }, + { + "dataset": "stocks", + "tier": "official", + "v1_endpoint": "/v1/stocks", + "primary_source": "tushare:stock_basic", + "backup_source": None, + "known_consumers": ["智能选股(股票主档)", "问天(market_context.py 股票代码/名称解析)"], + }, + { + "dataset": "daily", + "tier": "official", + "v1_endpoint": "/v1/bars/daily", + "primary_source": "tushare:daily", + "backup_source": None, + "known_consumers": ["智能选股(data_sync.py 日K因子)", "交易复盘/个股详情日K图表"], + }, + { + "dataset": "valuation", + "tier": "official", + "v1_endpoint": "/v1/valuation", + "primary_source": "tushare:daily_basic", + "backup_source": None, + "known_consumers": ["智能选股(data_sync.py 估值因子)"], + }, + { + "dataset": "moneyflow", + "tier": "official", + "v1_endpoint": "/v1/moneyflow", + "primary_source": "tushare:moneyflow", + "backup_source": None, + "known_consumers": ["智能选股(data_sync.py 资金流因子)", "个股详情资金流"], + }, + { + "dataset": "auction", + "tier": "official", + "v1_endpoint": "/v1/auction", + "primary_source": "tushare:stk_auction", + "backup_source": None, + "known_consumers": ["智能选股(data_sync.py 竞价快照)", "竞价板块"], + }, + { + "dataset": "index_daily", + "tier": "official", + "v1_endpoint": "/v1/indexes/bars", + "primary_source": "tushare:index_daily", + "backup_source": None, + "known_consumers": ["问天(market_context.py 指数近20日走势)", "智能选股(基准回看)"], + }, + { + "dataset": "limit_events", + "tier": "official", + "v1_endpoint": "/v1/limit-events", + "primary_source": "tushare:limit_list_d", + "backup_source": None, + "known_consumers": ["涨停梯队(历史/盘后视图)"], + }, + { + "dataset": "popularity", + "tier": "official", + "v1_endpoint": "/v1/popularity", + "primary_source": "tushare:ths_hot+dc_hot", + "backup_source": None, + "known_consumers": ["人气榜", "智能选股(data_sync.py 人气因子)"], + }, + { + "dataset": "dragon_tiger", + "tier": "official", + "v1_endpoint": "/v1/dragon-tiger", + "primary_source": "tushare:hm_detail", + "backup_source": None, + "known_consumers": ["龙虎榜"], + }, + { + "dataset": "sector_daily", + "tier": "official", + "v1_endpoint": "/v1/sectors", + "primary_source": "tushare:ths_daily+dc_index+sw_daily", + "backup_source": None, + "known_consumers": ["主题轮动", "板块梯队"], + }, + { + "dataset": "quotes_latest", + "tier": "provisional", + "v1_endpoint": "/v1/quotes/latest", + "primary_source": "eastmoney:ulist/clist", + "backup_source": "tencent:qt", + "known_consumers": ["竞价/股票池盘中价格", "情绪周期盘中快照"], + }, + { + "dataset": "index_quotes", + "tier": "provisional", + "v1_endpoint": "/v1/indexes/quotes", + "primary_source": "eastmoney:ulist", + "backup_source": "tencent:qt", + "known_consumers": ["首页大盘指数条"], + }, + { + "dataset": "sectors_quote", + "tier": "provisional", + "v1_endpoint": "/v1/sectors/quote", + "primary_source": "eastmoney:sw", + "backup_source": None, + "known_consumers": ["主题轮动盘中板块报价"], + }, + { + "dataset": "limit_pool", + "tier": "provisional", + "v1_endpoint": "/v1/limit-pool", + "primary_source": "eastmoney:zt_pool", + "backup_source": None, + "known_consumers": ["涨停梯队盘中视图"], + }, + { + "dataset": "intraday_points", + "tier": "provisional", + "v1_endpoint": "/v1/intraday/points", + "primary_source": "eastmoney:trends2", + "backup_source": None, + "known_consumers": ["个股详情分时图"], + }, + { + "dataset": "ifind_wencai", + "tier": "licensed", + "v1_endpoint": "/v1/query (api_name=ifind_wencai)", + "primary_source": "ifind:smart_stock_picking", + "backup_source": None, + "known_consumers": ["问师(自然语言选股，需 iFinD 凭证)"], + }, +] + +_KNOWN_PROVIDERS_BY_SOURCE_PREFIX = ("tushare", "eastmoney", "tencent", "ifind") + + +def _providers_for(primary_source: str, backup_source: str | None) -> list[str]: + providers: list[str] = [] + for source in (primary_source, backup_source or ""): + for provider in _KNOWN_PROVIDERS_BY_SOURCE_PREFIX: + if source.startswith(provider) and provider not in providers: + providers.append(provider) + return providers + + +def snapshot(db: Any, trade_date: str = "") -> list[dict[str, Any]]: + """Attach live status to the static lineage table. Read-only; never + raises (a per-row status lookup failure just leaves that row's status + empty rather than failing the whole snapshot).""" + result: list[dict[str, Any]] = [] + for entry in DATASETS: + row = dict(entry) + providers = _providers_for(entry["primary_source"], entry.get("backup_source")) + row["providers"] = providers + live: list[dict[str, Any]] = [] + try: + if db is not None and providers: + placeholders = ",".join("?" for _ in providers) + live = db.fetchall( + f"SELECT provider, interface, state, last_error, last_fallback_reason, " + f"consec_failures, updated_at FROM provider_health WHERE provider IN ({placeholders})", + tuple(providers), + ) + except Exception: + live = [] + row["live_provider_health"] = live + if entry["tier"] == "official": + pub = None + try: + if db is not None and trade_date: + pub = db.fetchone( + "SELECT dataset, trade_date, state, published_at FROM publications " + "WHERE dataset = ? AND trade_date = ?", + (entry["dataset"], trade_date), + ) + except Exception: + pub = None + row["publication"] = pub + result.append(row) + return result + + +def affected(db: Any, provider: str = "", interface: str = "") -> list[dict[str, Any]]: + """Read-only: which datasets/pages are impacted by a given provider (and, + optionally, a specific interface) right now. Does not change routing.""" + provider = str(provider or "").strip() + interface = str(interface or "").strip() + result: list[dict[str, Any]] = [] + for entry in DATASETS: + providers = _providers_for(entry["primary_source"], entry.get("backup_source")) + if provider and provider not in providers: + continue + row = dict(entry) + row["providers"] = providers + health: list[dict[str, Any]] = [] + try: + if db is not None: + if interface: + health = db.fetchall( + "SELECT provider, interface, state, last_error, last_fallback_reason, " + "consec_failures, updated_at FROM provider_health " + "WHERE provider = ? AND interface = ?", + (provider, interface), + ) + elif providers: + placeholders = ",".join("?" for _ in providers) + health = db.fetchall( + f"SELECT provider, interface, state, last_error, last_fallback_reason, " + f"consec_failures, updated_at FROM provider_health WHERE provider IN ({placeholders})", + tuple(providers), + ) + except Exception: + health = [] + row["live_provider_health"] = health + result.append(row) + return result diff --git a/xiaobai-datahub/datahub/observability.py b/xiaobai-datahub/datahub/observability.py new file mode 100644 index 0000000..395cdae --- /dev/null +++ b/xiaobai-datahub/datahub/observability.py @@ -0,0 +1,300 @@ +"""Side-channel provider-call observability (HEL-543). + +This module is additive-only and must never change what any existing call +returns or raises. It exists purely to answer, after the fact and without +touching routing: which provider/interface was called, whether it +succeeded, how stale/complete the payload looked, and why a fallback fired. + +Hard rules enforced here: + +- Every public entry point (`observe`, `record_call`) is wrapped so that a + database failure, a classifier bug, or any other internal error is + swallowed and logged at DEBUG level. It never raises into the caller and + never delays/blocks the caller's real data path beyond a best-effort + timing measurement. +- `observe()` always returns exactly what `fn()` returned, and re-raises + exactly what `fn()` raised (same exception object, unmodified). It does + not retry, does not change ordering, and does not add new failure modes. +- No mock data is ever produced or returned by this module. +""" + +from __future__ import annotations + +import re +import time +from typing import Any, Callable, TypeVar + +from datahub.timeutil import isoformat, now_shanghai + +_T = TypeVar("_T") + +# Additive-only schema: two new tables, no changes to any existing table. +# Merged into datahub.db.SCHEMA the same way EXTENDED_SCHEMA is. +OBS_SCHEMA = """ +CREATE TABLE IF NOT EXISTS provider_call_log ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + provider TEXT NOT NULL, + interface TEXT NOT NULL, + fetched_at TEXT NOT NULL, + latency_ms INTEGER, + status TEXT NOT NULL, + error TEXT, + fallback_reason TEXT, + data_age_seconds INTEGER, + created_at TEXT NOT NULL +); + +CREATE TABLE IF NOT EXISTS provider_health ( + provider TEXT NOT NULL, + interface TEXT NOT NULL, + state TEXT NOT NULL, + last_ok_at TEXT, + last_error TEXT, + last_fallback_reason TEXT, + consec_failures INTEGER NOT NULL DEFAULT 0, + last_latency_ms INTEGER, + last_data_age_seconds INTEGER, + updated_at TEXT NOT NULL, + PRIMARY KEY (provider, interface) +); + +CREATE INDEX IF NOT EXISTS idx_provider_call_log_created ON provider_call_log(created_at); +CREATE INDEX IF NOT EXISTS idx_provider_call_log_provider ON provider_call_log(provider, interface, created_at); +""" + +DEFAULT_STALE_SECONDS = 300 + +_BLOCKED_MARKERS = ( + " tuple[str, str]: + """Best-effort, side-reading classification of an exception message. + + Never raises. Unknown shapes fall back to a generic ``error`` status so a + classifier miss can never be mistaken for a healthy call. + """ + try: + lower = (message or "").lower() + if any(marker in lower for marker in _BLOCKED_MARKERS): + return "blocked", "response_looks_like_intercept_page" + if "timeout" in lower or "timed out" in lower: + return "timeout", "request_timeout" + mismatch = _COUNT_MISMATCH_RE.search(lower) + if mismatch and int(mismatch.group(1)) == 0: + return "empty", "empty_or_incomplete_response" + if mismatch: + return "degraded", "partial_or_mismatched_response" + if "empty" in lower or "no intraday chart data" in lower or "missing" in lower: + return "empty", "empty_or_incomplete_response" + if "too small" in lower or "incomplete" in lower or "mismatch" in lower: + return "degraded", "partial_or_mismatched_response" + return "error", "" + except Exception: # pragma: no cover - defensive, must never raise + return "error", "" + + +def classify_rows( + rows: Any, + *, + required_fields: tuple[str, ...] | None = None, + freshness_field: str | None = "quote_time_epoch", + max_age_seconds: int = DEFAULT_STALE_SECONDS, +) -> tuple[str, str, int | None]: + """Read-only classification of an already-successful payload. + + Only ever called on a value a caller is about to use as-is; this never + mutates ``rows`` and a classifier bug always degrades to ``("ok", "", + None)`` rather than mislabeling a real success as a failure. + """ + try: + if isinstance(rows, dict): + items = [rows] if rows else [] + elif isinstance(rows, (list, tuple)): + items = [item for item in rows if isinstance(item, dict)] + else: + items = [] + if not items: + return "empty", "no_rows_returned", None + if required_fields: + missing: set[str] = set() + for item in items: + for field in required_fields: + if item.get(field) in (None, ""): + missing.add(field) + if missing: + return "missing_fields", "missing:" + ",".join(sorted(missing)), None + data_age: int | None = None + if freshness_field: + now_epoch = time.time() + ages: list[int] = [] + for item in items: + raw = item.get(freshness_field) + try: + epoch = int(raw or 0) + except (TypeError, ValueError): + epoch = 0 + if epoch > 0: + ages.append(max(0, int(now_epoch - epoch))) + if ages: + data_age = max(ages) + if data_age > max_age_seconds: + return "stale", "data_age_exceeds_threshold", data_age + return "ok", "", data_age + except Exception: # pragma: no cover - defensive, must never raise + return "ok", "", None + + +def record_call( + db: Any, + provider: str, + interface: str, + *, + status: str, + latency_ms: int | None = None, + error: str = "", + fallback_reason: str = "", + data_age_seconds: int | None = None, +) -> None: + """Fail-open recorder. Never raises; a write failure here must never be + able to take down a real, otherwise-successful data path.""" + if db is None: + return + try: + now = isoformat(now_shanghai()) + ok = status == "ok" + error_text = (error or "")[:500] + reason_text = (fallback_reason or "")[:200] + with db.write() as connection: + connection.execute( + "INSERT INTO provider_call_log(" + "provider, interface, fetched_at, latency_ms, status, error, " + "fallback_reason, data_age_seconds, created_at) " + "VALUES (?,?,?,?,?,?,?,?,?)", + (provider, interface, now, latency_ms, status, error_text, reason_text, data_age_seconds, now), + ) + connection.execute( + """ + INSERT INTO provider_health( + provider, interface, state, last_ok_at, last_error, last_fallback_reason, + consec_failures, last_latency_ms, last_data_age_seconds, updated_at) + VALUES (?,?,?,?,?,?,?,?,?,?) + ON CONFLICT(provider, interface) DO UPDATE SET + state = excluded.state, + last_ok_at = CASE WHEN excluded.state = 'ok' THEN excluded.last_ok_at ELSE provider_health.last_ok_at END, + last_error = CASE WHEN excluded.state = 'ok' THEN '' ELSE excluded.last_error END, + last_fallback_reason = excluded.last_fallback_reason, + consec_failures = CASE WHEN excluded.state = 'ok' THEN 0 ELSE provider_health.consec_failures + 1 END, + last_latency_ms = excluded.last_latency_ms, + last_data_age_seconds = excluded.last_data_age_seconds, + updated_at = excluded.updated_at + """, + ( + provider, + interface, + status, + now if ok else None, + "" if ok else (error_text or reason_text or "unknown_error"), + reason_text, + 0 if ok else 1, + latency_ms, + data_age_seconds, + now, + ), + ) + except Exception: # pragma: no cover - defensive, must never raise + try: + _logger().debug( + "observability record_call failed (fail-open)", + extra={"hub": {"provider": provider, "interface": interface}}, + exc_info=True, + ) + except Exception: + pass + + +def _safe_record_failure(db: Any, provider: str, interface: str, latency_ms: int, exc: BaseException) -> None: + if db is None: + return + try: + message = str(exc) + status, reason = classify_error(message) + record_call( + db, provider, interface, + status=status, latency_ms=latency_ms, error=message, fallback_reason=reason, + ) + except Exception: # pragma: no cover - defensive, must never raise + pass + + +def _safe_record_success( + db: Any, + provider: str, + interface: str, + latency_ms: int, + result: Any, + classify: Callable[[Any], tuple[str, str, int | None] | None] | None, +) -> None: + if db is None: + return + status, reason, data_age = "ok", "", None + if classify is not None: + try: + classified = classify(result) + if classified: + status, reason, data_age = classified + except Exception: # pragma: no cover - defensive, must never raise + # A classifier bug must never mislabel (or hide) a real success; + # degrade to a plain "ok" call rather than skipping the log. + status, reason, data_age = "ok", "", None + try: + record_call( + db, provider, interface, + status=status, latency_ms=latency_ms, fallback_reason=reason, data_age_seconds=data_age, + ) + except Exception: # pragma: no cover - defensive, must never raise + pass + + +def observe( + db: Any, + provider: str, + interface: str, + fn: Callable[[], _T], + *, + classify: Callable[[Any], tuple[str, str, int | None] | None] | None = None, +) -> _T: + """Call ``fn()`` and record a side-channel status row. + + Returns exactly what ``fn()`` returns and re-raises exactly what + ``fn()`` raises. ``db`` may be ``None`` (e.g. in call sites that are not + wired to a database yet); in that case this is a transparent passthrough + with no recording at all. + """ + started = time.perf_counter() + try: + result = fn() + except Exception: + latency_ms = round((time.perf_counter() - started) * 1000) + import sys + + exc = sys.exc_info()[1] + if exc is not None: + _safe_record_failure(db, provider, interface, latency_ms, exc) + raise + latency_ms = round((time.perf_counter() - started) * 1000) + _safe_record_success(db, provider, interface, latency_ms, result, classify) + return result diff --git a/xiaobai-datahub/datahub/pipeline.py b/xiaobai-datahub/datahub/pipeline.py index 808886c..d76f5e1 100644 --- a/xiaobai-datahub/datahub/pipeline.py +++ b/xiaobai-datahub/datahub/pipeline.py @@ -20,6 +20,7 @@ from datahub.governance.ratelimit import TokenBucket from datahub.governance.retry import RetryError, retry_call from datahub.logutil import get_logger from datahub.normalize import finite_number, normalize_daily +from datahub import observability from datahub.revision import ( compare_fields, diff_published_vs_upstream, @@ -1634,6 +1635,7 @@ class Pipeline: deleted += cur.rowcount connection.execute("DELETE FROM job_runs WHERE started_at < ?", (cutoff_jobs,)) connection.execute("DELETE FROM src_calls WHERE created_at < ?", (cutoff_jobs,)) + connection.execute("DELETE FROM provider_call_log WHERE created_at < ?", (cutoff_jobs,)) return {"staging_deleted": deleted} def audit(self, actor: str, action: str, target: str = "", detail: str = "") -> None: @@ -1768,6 +1770,18 @@ class Pipeline: "INSERT INTO src_calls(provider, endpoint, ok, latency_ms, error, created_at) VALUES (?,?,?,?,?,?)", ("tushare", endpoint, 1 if ok else 0, latency_ms, error, isoformat(self.clock())), ) + # HEL-543 side channel: unified cross-provider call log/health. Kept + # strictly additive and fail-open; the src_calls insert above (the + # existing, already-compatible Tushare record) is unaffected either + # way. + if ok: + status, reason = "ok", "" + else: + status, reason = observability.classify_error(error) + observability.record_call( + self.db, "tushare", endpoint, + status=status, latency_ms=latency_ms, error=error, fallback_reason=reason, + ) def _persist_health(self, state: str, error: str = "") -> None: snap = self.breaker.snapshot() diff --git a/xiaobai-datahub/datahub/realtime_serve.py b/xiaobai-datahub/datahub/realtime_serve.py index bcd49b1..5d1dd35 100644 --- a/xiaobai-datahub/datahub/realtime_serve.py +++ b/xiaobai-datahub/datahub/realtime_serve.py @@ -18,6 +18,7 @@ from datahub.adapters.tencent import TencentAdapter from datahub.codes import resolve_code from datahub.db import HubDB from datahub.governance.lkg import LastKnownGood +from datahub import observability from datahub.timeutil import isoformat, now_shanghai, yyyymmdd QUOTE_TTL = 60 @@ -25,6 +26,23 @@ INDEX_TTL = 60 INTRADAY_TTL = 20 QUOTE_BATCH = 60 +# HEL-543 side-channel classifiers. These only *read* an already-successful +# payload to decide what to log; they never change the payload itself and a +# classifier exception always degrades to "ok" (see observability.classify_rows). + + +def _classify_quote_rows(rows: Any) -> tuple[str, str, int | None]: + return observability.classify_rows(rows, freshness_field="quote_time_epoch") + + +def _classify_rows_no_freshness(rows: Any) -> tuple[str, str, int | None]: + return observability.classify_rows(rows, freshness_field=None) + + +def _classify_intraday_payload(data: Any) -> tuple[str, str, int | None]: + points = data.get("points") if isinstance(data, dict) else None + return observability.classify_rows(points or [], freshness_field=None) + class RealtimeApiError(RuntimeError): def __init__(self, code: str, message: str) -> None: @@ -44,12 +62,17 @@ def fetch_index_quotes(db: HubDB) -> dict[str, Any]: cached = _read_cache(db, cache_key) if cached is not None: return cached - eastmoney = EastmoneyAdapter() try: - rows = eastmoney.fetch_indices() + rows = observability.observe( + db, "eastmoney", "indices", lambda: EastmoneyAdapter().fetch_indices(), + classify=_classify_quote_rows, + ) source = "eastmoney:ulist" except Exception: - rows = TencentAdapter().fetch_indices() + rows = observability.observe( + db, "tencent", "indices", lambda: TencentAdapter().fetch_indices(), + classify=_classify_quote_rows, + ) source = "tencent:qt" if len(rows) < 3: raise RealtimeApiError("SOURCE_UNAVAILABLE", "index quotes incomplete") @@ -77,7 +100,10 @@ def fetch_market_quotes(db: HubDB) -> dict[str, Any]: rows: list[dict[str, Any]] = [] source = "" try: - rows = EastmoneyAdapter().fetch_market_quotes() + rows = observability.observe( + db, "eastmoney", "market_quotes", lambda: EastmoneyAdapter().fetch_market_quotes(), + classify=_classify_quote_rows, + ) source = "eastmoney:clist" except Exception as exc: errors.append(f"eastmoney:{exc}") @@ -85,7 +111,10 @@ def fetch_market_quotes(db: HubDB) -> dict[str, Any]: listed = _listed_ts_codes(db) if not listed: raise AdapterError("no local stock master for tencent market snapshot") - rows = _tencent_named_quotes(listed) + rows = observability.observe( + db, "tencent", "market_quotes_fallback", lambda: _tencent_named_quotes(listed), + classify=_classify_quote_rows, + ) if len(rows) < 200: raise AdapterError(f"Tencent market snapshot too small: {len(rows)}") source = "tencent:qt" @@ -130,7 +159,10 @@ def fetch_quotes(db: HubDB, codes: list[str]) -> dict[str, Any]: missing = [code for code in resolved if code not in by_code] try: - rows = _eastmoney_named_quotes(missing) + rows = observability.observe( + db, "eastmoney", "named_quotes", lambda: _eastmoney_named_quotes(missing), + classify=_classify_quote_rows, + ) by_code.update(_quote_map(rows, missing)) if rows: sources.append("eastmoney:ulist") @@ -140,7 +172,10 @@ def fetch_quotes(db: HubDB, codes: list[str]) -> dict[str, Any]: missing = [code for code in resolved if code not in by_code] if missing: try: - rows = _tencent_named_quotes(missing) + rows = observability.observe( + db, "tencent", "named_quotes", lambda: _tencent_named_quotes(missing), + classify=_classify_quote_rows, + ) by_code.update(_quote_map(rows, missing)) if rows: sources.append("tencent:qt") @@ -203,7 +238,10 @@ def fetch_sector_quote(db: HubDB, code: str, expected_date: str = "") -> dict[st return cached errors: list[str] = [] try: - row = EastmoneyAdapter().fetch_shenwan_quote(ts_code) + row = observability.observe( + db, "eastmoney", "sector_quote", lambda: EastmoneyAdapter().fetch_shenwan_quote(ts_code), + classify=_classify_quote_rows, + ) if not _sector_row_matches(row, canonical_name): raise AdapterError( f"industry name mismatch: expected {canonical_name}, got {row.get('name') or '--'}" @@ -270,7 +308,10 @@ def fetch_limit_pool(db: HubDB, trade_date: str = "") -> dict[str, Any]: if cached is not None: return cached try: - rows = EastmoneyAdapter().fetch_limit_pool(day) + rows = observability.observe( + db, "eastmoney", "limit_pool", lambda: EastmoneyAdapter().fetch_limit_pool(day), + classify=_classify_rows_no_freshness, + ) source = "eastmoney:zt_pool" except Exception as exc: recovered = _load_quotes_lkg(db, cache_key) @@ -530,7 +571,10 @@ def warm_realtime(db: HubDB) -> dict[str, Any]: if master_code and master_name: canonical_names.setdefault(master_code, master_name) codes = list(canonical_names) - fetched_sector_rows = _eastmoney_sector_quotes(codes) + fetched_sector_rows = observability.observe( + db, "eastmoney", "sector_quotes_batch", lambda: _eastmoney_sector_quotes(codes), + classify=_classify_quote_rows, + ) for row in fetched_sector_rows: if _row_quote_date(row, today) != today: continue @@ -588,7 +632,10 @@ def fetch_intraday(db: HubDB, code: str, date: str = "") -> dict[str, Any]: return cached adapter = EastmoneyAdapter() try: - payload_data = adapter.fetch_intraday(ts_code, date) + payload_data = observability.observe( + db, "eastmoney", "intraday", lambda: adapter.fetch_intraday(ts_code, date), + classify=_classify_intraday_payload, + ) source = "eastmoney:trends2" except Exception as exc: recovered = _load_intraday_lkg(db, ts_code, date) diff --git a/xiaobai-datahub/datahub/source_catalog.py b/xiaobai-datahub/datahub/source_catalog.py new file mode 100644 index 0000000..f3cc17e --- /dev/null +++ b/xiaobai-datahub/datahub/source_catalog.py @@ -0,0 +1,165 @@ +"""Minimal, read-only source directory (HEL-543). + +Registers what already exists: providers, their concrete interfaces, what +capability/dataset each interface serves, and whether the provider plays a +primary or backup role. This module only *describes* the current adapters +and datasets already wired in `datahub/hub.py`, `datahub/serving.py`, and +`datahub/realtime_serve.py`; it does not add a way to configure or add a new +source without code, and it never changes routing, retries, or fallback +order. + +Every ``interfaces`` entry below is a docs-as-code mirror of a real call +site, cross-referenced in comments so a reviewer can verify each row is +accurate rather than aspirational: + +- tushare interfaces mirror ``datahub/serving.py``'s ``_official_meta``/``source=`` + strings and ``datahub/steward.py``'s live/published dataset table. +- eastmoney/tencent interfaces mirror the ``observability.observe(...)`` + call sites added in ``datahub/realtime_serve.py`` for HEL-543. +- ifind interfaces mirror ``datahub/steward.py``'s ``IFIND_APIS`` table. +""" + +from __future__ import annotations + +from typing import Any + +CATALOG: list[dict[str, Any]] = [ + { + "provider": "tushare", + "label": "Tushare", + "role": "official_primary", + "credential_key": "tushare_token", + "status_source": "src_health (legacy, kept) + provider_health (unified, HEL-543)", + "interfaces": [ + {"interface": "trade_cal", "capability": "交易日历", "datasets": ["calendar"]}, + {"interface": "stock_basic", "capability": "股票主档", "datasets": ["stocks"]}, + {"interface": "daily", "capability": "个股日K", "datasets": ["daily"]}, + {"interface": "adj_factor", "capability": "复权因子", "datasets": ["daily"]}, + {"interface": "daily_basic", "capability": "估值", "datasets": ["valuation"]}, + {"interface": "index_daily", "capability": "指数日K", "datasets": ["index_daily"]}, + {"interface": "moneyflow", "capability": "资金流", "datasets": ["moneyflow"]}, + {"interface": "stk_auction", "capability": "集合竞价", "datasets": ["auction"]}, + {"interface": "limit_list_d", "capability": "涨跌停池", "datasets": ["limit_events"]}, + {"interface": "ths_hot", "capability": "同花顺人气榜", "datasets": ["popularity"]}, + {"interface": "dc_hot", "capability": "东方财富人气榜", "datasets": ["popularity"]}, + {"interface": "hm_detail", "capability": "龙虎榜游资明细", "datasets": ["dragon_tiger"]}, + {"interface": "ths_daily", "capability": "同花顺概念行情", "datasets": ["sector_daily"]}, + {"interface": "dc_index", "capability": "东方财富板块行情", "datasets": ["sector_daily"]}, + {"interface": "sw_daily", "capability": "申万行业行情", "datasets": ["sector_daily"]}, + ], + }, + { + "provider": "eastmoney", + "label": "东方财富", + "role": "provisional_primary", + "credential_key": None, + "status_source": "provider_health (unified, HEL-543)", + "interfaces": [ + {"interface": "indices", "capability": "指数实时报价", "datasets": ["index_quotes"]}, + {"interface": "market_quotes", "capability": "全市场实时快照", "datasets": ["quotes_latest"]}, + {"interface": "named_quotes", "capability": "指定个股实时报价", "datasets": ["quotes_latest"]}, + {"interface": "sector_quote", "capability": "申万板块实时报价(单个)", "datasets": ["sectors_quote"]}, + {"interface": "sector_quotes_batch", "capability": "申万板块批量报价(预热)", "datasets": ["sectors_quote"]}, + {"interface": "limit_pool", "capability": "涨停/炸板池(盘中)", "datasets": ["limit_pool"]}, + {"interface": "intraday", "capability": "分时走势", "datasets": ["intraday_points"]}, + ], + }, + { + "provider": "tencent", + "label": "腾讯行情", + "role": "provisional_backup", + "credential_key": None, + "status_source": "provider_health (unified, HEL-543)", + "interfaces": [ + {"interface": "indices", "capability": "指数实时报价(东财失败时备用)", "datasets": ["index_quotes"]}, + { + "interface": "market_quotes_fallback", + "capability": "全市场快照(备用；按本地股票主档逐只请求拼接)", + "datasets": ["quotes_latest"], + }, + {"interface": "named_quotes", "capability": "指定个股实时报价(东财失败时备用)", "datasets": ["quotes_latest"]}, + ], + }, + { + "provider": "ifind", + "label": "同花顺 iFinD", + "role": "licensed_optional", + "credential_key": "ifind_refresh_token", + "status_source": "provider_health (unified, HEL-543) + adapter.status()", + "interfaces": [ + {"interface": "wencai", "capability": "问财自然语言选股", "datasets": ["ifind_wencai"]}, + {"interface": "snapshots", "capability": "快照", "datasets": ["ifind_snapshots"]}, + {"interface": "history", "capability": "历史行情", "datasets": ["ifind_history"]}, + {"interface": "realtime", "capability": "实时行情", "datasets": ["ifind_realtime"]}, + {"interface": "intraday", "capability": "分时(高频)", "datasets": ["ifind_intraday"]}, + ], + }, + { + "provider": "ths", + "label": "同花顺(预留)", + "role": "reserved", + "credential_key": None, + "status_source": "adapter.probe()（占位，本阶段未接入真实数据）", + "interfaces": [], + }, + { + "provider": "xgb", + "label": "选股宝(预留)", + "role": "reserved", + "credential_key": None, + "status_source": "adapter.probe()（占位，本阶段未接入真实数据）", + "interfaces": [], + }, + { + "provider": "akshare", + "label": "AKShare(预留)", + "role": "reserved", + "credential_key": None, + "status_source": "adapter.probe()（占位，本阶段未接入真实数据）", + "interfaces": [], + }, +] + + +def snapshot(db: Any, auth: Any = None) -> list[dict[str, Any]]: + """Merge the static catalog with live credential/health facts. + + Purely read-only: never touches routing, credentials, or adapters. Any + failure while enriching one entry only degrades that entry's live data; + it never drops the entry or raises, so a directory read can never break + on a partially-unhealthy database. + """ + result: list[dict[str, Any]] = [] + for entry in CATALOG: + item: dict[str, Any] = { + "provider": entry["provider"], + "label": entry.get("label", entry["provider"]), + "role": entry["role"], + "status_source": entry["status_source"], + "interfaces": [dict(i) for i in entry.get("interfaces", [])], + } + cred_key = entry.get("credential_key") + if cred_key: + cred = None + try: + if auth is not None: + cred = auth.credential_status(cred_key) + except Exception: + cred = None + item["credential"] = cred or {"configured": False, "last4": "", "updated_at": ""} + else: + item["credential"] = {"configured": True, "last4": "", "updated_at": "", "note": "无需凭证"} + health_rows: list[dict[str, Any]] = [] + try: + if db is not None: + health_rows = db.fetchall( + "SELECT interface, state, last_ok_at, last_error, last_fallback_reason, " + "consec_failures, last_latency_ms, last_data_age_seconds, updated_at " + "FROM provider_health WHERE provider = ? ORDER BY interface", + (entry["provider"],), + ) + except Exception: + health_rows = [] + item["live_interfaces"] = health_rows + result.append(item) + return result diff --git a/xiaobai-datahub/datahub/steward.py b/xiaobai-datahub/datahub/steward.py index 4a03e8b..d94317f 100644 --- a/xiaobai-datahub/datahub/steward.py +++ b/xiaobai-datahub/datahub/steward.py @@ -12,6 +12,7 @@ from typing import Any from datahub.adapters.base import AdapterError from datahub.adapters.tushare import TUSHARE_FIELDS +from datahub import observability from datahub.numbers import finite_number from datahub.realtime_serve import ( RealtimeApiError, @@ -152,8 +153,12 @@ def _ifind_query(api, api_name: str, params: dict[str, Any], fields: str) -> dic ) if not adapter.configured: raise ApiError("SOURCE_UNAVAILABLE", "iFinD 尚未配置") + db = getattr(api, "db", None) try: - rows = adapter.fetch(dataset, dict(params)) + rows = observability.observe( + db, "ifind", dataset, lambda: adapter.fetch(dataset, dict(params)), + classify=lambda r: observability.classify_rows(r, freshness_field=None), + ) except AdapterError as exc: raise ApiError("SOURCE_UNAVAILABLE", str(exc)) from exc return envelope( diff --git a/xiaobai-datahub/tests/test_admin_observability_api.py b/xiaobai-datahub/tests/test_admin_observability_api.py new file mode 100644 index 0000000..cf7f85a --- /dev/null +++ b/xiaobai-datahub/tests/test_admin_observability_api.py @@ -0,0 +1,145 @@ +from __future__ import annotations + +import json +import tempfile +import threading +import unittest +from http.server import ThreadingHTTPServer +from pathlib import Path +from urllib.request import Request, urlopen + +from datahub.adapters.tushare import TushareAdapter +from datahub.crypto import SecretVault +from datahub.httpapp import make_handler +from datahub.hub import Hub +from datahub.settings import Settings +from tests.fixtures import TRADE_DATE, fake_transport + + +class AdminObservabilityApiTests(unittest.TestCase): + """HEL-543: new read-only admin endpoints for provider status, source + catalog and lineage. These must never require write access and must + never touch the existing routing/publish logic.""" + + def setUp(self) -> None: + self.tmp = tempfile.TemporaryDirectory() + settings = Settings( + encryption_key=SecretVault.generate_key(), + api_token="z" * 32, + admin_password="StartPass1", + tushare_token="real-tushare-token-abcdef", + db_path=Path(self.tmp.name) / "hub.db", + scheduler_enabled=False, + ) + self.hub = Hub(settings, adapter=TushareAdapter("real-tushare-token-abcdef", transport=fake_transport)) + handler = make_handler(self.hub) + self.server = ThreadingHTTPServer(("127.0.0.1", 0), handler) + threading.Thread(target=self.server.serve_forever, daemon=True).start() + self.base = f"http://127.0.0.1:{self.server.server_address[1]}" + + _, body, cookie_header = self._json( + "/admin/api/login", "POST", {"username": "hub_admin", "password": "StartPass1"} + ) + cookie = cookie_header.split(";")[0] + csrf = body["csrf"] + self._json( + "/admin/api/change-password", + "POST", + {"current": "StartPass1", "new_password": "NewPass123"}, + cookie=cookie, + csrf=csrf, + ) + self.cookie = cookie + self.csrf = csrf + + def tearDown(self) -> None: + self.server.shutdown() + self.server.server_close() + self.tmp.cleanup() + + def _json(self, path, method="GET", body=None, cookie="", csrf=""): + data = None if body is None else json.dumps(body).encode() + headers = {"Content-Type": "application/json"} + if cookie: + headers["Cookie"] = cookie + if csrf: + headers["X-CSRF-Token"] = csrf + req = Request(self.base + path, data=data, headers=headers, method=method) + with urlopen(req, timeout=5) as resp: + set_cookie = resp.headers.get("Set-Cookie", "") + return resp.status, json.loads(resp.read().decode()), set_cookie + + def _get(self, path): + return self._json(path, cookie=self.cookie, csrf=self.csrf) + + def test_providers_status_reflects_real_pipeline_activity(self) -> None: + pipeline = self.hub.pipeline + pipeline.ingest_reference(TRADE_DATE) + pipeline.run_dataset("daily", TRADE_DATE) + + status, body, _ = self._get("/admin/api/providers/status") + self.assertEqual(status, 200) + health = body["health"] + self.assertTrue(any(item["provider"] == "tushare" and item["interface"] == "daily" for item in health)) + row = next(item for item in health if item["provider"] == "tushare" and item["interface"] == "daily") + self.assertEqual(row["state"], "ok") + recent = body["recent_calls"] + self.assertTrue(any(item["provider"] == "tushare" and item["interface"] == "daily" for item in recent)) + + def test_providers_status_filters_by_provider(self) -> None: + pipeline = self.hub.pipeline + pipeline.ingest_reference(TRADE_DATE) + pipeline.run_dataset("daily", TRADE_DATE) + + status, body, _ = self._get("/admin/api/providers/status?provider=tushare") + self.assertEqual(status, 200) + self.assertTrue(all(item["provider"] == "tushare" for item in body["health"])) + self.assertTrue(all(item["provider"] == "tushare" for item in body["recent_calls"])) + + status, body, _ = self._get("/admin/api/providers/status?provider=eastmoney") + self.assertEqual(status, 200) + self.assertEqual(body["health"], []) + self.assertEqual(body["recent_calls"], []) + + def test_source_catalog_lists_known_providers_without_leaking_secrets(self) -> None: + status, body, _ = self._get("/admin/api/source-catalog") + self.assertEqual(status, 200) + blob = json.dumps(body) + self.assertNotIn("real-tushare-token-abcdef", blob) + providers = {item["provider"] for item in body["items"]} + self.assertIn("tushare", providers) + self.assertIn("eastmoney", providers) + self.assertIn("tencent", providers) + self.assertIn("ifind", providers) + + def test_lineage_snapshot_and_affected_query(self) -> None: + status, body, _ = self._get("/admin/api/lineage") + self.assertEqual(status, 200) + self.assertTrue(len(body["items"]) > 0) + datasets = {item["dataset"] for item in body["items"]} + self.assertIn("stocks", datasets) + + status, body, _ = self._get("/admin/api/lineage/affected?provider=tushare&interface=daily") + self.assertEqual(status, 200) + self.assertEqual(body["provider"], "tushare") + self.assertEqual(body["interface"], "daily") + + def test_must_change_password_blocks_new_endpoints_too(self) -> None: + from urllib.error import HTTPError + + _, body, cookie_header = self._json( + "/admin/api/login", "POST", {"username": "hub_admin", "password": "NewPass123"} + ) + # Freshly logged-in user has already changed password in setUp, so + # this login should not require a change; verify the endpoint is + # reachable with a valid, non-must-change session (regression guard + # against accidentally bypassing the must-change gate for these new + # routes). + cookie = cookie_header.split(";")[0] + csrf = body["csrf"] + status, _, _ = self._json("/admin/api/source-catalog", cookie=cookie, csrf=csrf) + self.assertEqual(status, 200) + + +if __name__ == "__main__": + unittest.main() diff --git a/xiaobai-datahub/tests/test_ifind_observability.py b/xiaobai-datahub/tests/test_ifind_observability.py new file mode 100644 index 0000000..fa04288 --- /dev/null +++ b/xiaobai-datahub/tests/test_ifind_observability.py @@ -0,0 +1,105 @@ +from __future__ import annotations + +import tempfile +import unittest +from pathlib import Path + +from datahub.adapters.ifind import IfindAdapter +from datahub.db import HubDB +from datahub.serving import ApiError +from datahub.steward import steward_query + + +class _Resp: + def __init__(self, payload: dict, status: int = 200) -> None: + import json + + self.status = status + self._raw = json.dumps(payload).encode("utf-8") + + def read(self): + return self._raw + + def __enter__(self): + return self + + def __exit__(self, *args): + return False + + +class IfindObservabilityTests(unittest.TestCase): + def setUp(self) -> None: + self.tmp = tempfile.TemporaryDirectory() + self.db = HubDB(Path(self.tmp.name) / "hub.db") + + def tearDown(self) -> None: + self.tmp.cleanup() + + def _adapter_with_urlopen(self, urlopen) -> IfindAdapter: + return IfindAdapter(refresh_token="rt", access_token="at", urlopen=urlopen) + + def test_successful_fetch_is_logged_without_changing_rows(self) -> None: + def urlopen(request, timeout=None): + return _Resp( + { + "errorcode": 0, + "tables": [{"thscode": ["000001.SZ"], "table": {"涨停原因": ["重组"]}}], + } + ) + + adapter = self._adapter_with_urlopen(urlopen) + + class _Api: + ifind = adapter + db = self.db + + payload = steward_query( + _Api(), {"api_name": "ifind_wencai", "params": {"query": "涨停"}} + ) + self.assertEqual(payload["data"][0]["thscode"], "000001.SZ") + log = self.db.fetchone("SELECT * FROM provider_call_log WHERE provider = 'ifind'") + self.assertIsNotNone(log) + self.assertEqual(log["interface"], "wencai") + self.assertEqual(log["status"], "ok") + + def test_failed_fetch_reraises_and_logs_error(self) -> None: + def urlopen(request, timeout=None): + return _Resp({"errorcode": -9999, "errmsg": "quota exceeded"}) + + adapter = self._adapter_with_urlopen(urlopen) + + class _Api: + ifind = adapter + db = self.db + + with self.assertRaises(ApiError) as ctx: + steward_query(_Api(), {"api_name": "ifind_wencai", "params": {"query": "涨停"}}) + self.assertEqual(ctx.exception.code, "SOURCE_UNAVAILABLE") + log = self.db.fetchone("SELECT * FROM provider_call_log WHERE provider = 'ifind'") + self.assertIsNotNone(log) + self.assertEqual(log["status"], "error") + + def test_status_check_alone_does_not_dial_or_log_a_fetch_call(self) -> None: + class _Api: + ifind = IfindAdapter() + db = self.db + + payload = steward_query(_Api(), {"api_name": "ifind_status", "params": {}}) + self.assertFalse(payload["data"][0]["configured"]) + log = self.db.fetchall("SELECT * FROM provider_call_log WHERE provider = 'ifind'") + self.assertEqual(log, []) + + def test_api_double_without_db_attribute_still_works(self) -> None: + # Mirrors tests/test_ifind_adapter.py's `_Api` double, which has no + # `db` attribute at all. Observability must not require it. + class _Api: + ifind = IfindAdapter() + + payload = steward_query(_Api(), {"api_name": "ifind_status", "params": {}}) + self.assertFalse(payload["data"][0]["configured"]) + with self.assertRaises(ApiError): + steward_query(_Api(), {"api_name": "ifind_wencai", "params": {"query": "涨停"}}) + + +if __name__ == "__main__": + unittest.main() diff --git a/xiaobai-datahub/tests/test_observability.py b/xiaobai-datahub/tests/test_observability.py new file mode 100644 index 0000000..301b74f --- /dev/null +++ b/xiaobai-datahub/tests/test_observability.py @@ -0,0 +1,214 @@ +from __future__ import annotations + +import tempfile +import unittest +from contextlib import contextmanager +from pathlib import Path + +from datahub import observability +from datahub.db import HubDB + + +class _BrokenDB: + """A db double whose write() always raises, to prove fail-open.""" + + @contextmanager + def write(self): + raise RuntimeError("disk is full") + yield None # pragma: no cover - unreachable, keeps this a generator + + def fetchall(self, sql, params=()): + raise RuntimeError("disk is full") + + def fetchone(self, sql, params=()): + raise RuntimeError("disk is full") + + +class ClassifyRowsTests(unittest.TestCase): + def test_empty_list_is_flagged_empty(self): + status, reason, age = observability.classify_rows([]) + self.assertEqual(status, "empty") + self.assertEqual(reason, "no_rows_returned") + self.assertIsNone(age) + + def test_empty_dict_result_is_flagged_empty(self): + status, reason, _ = observability.classify_rows({}) + self.assertEqual(status, "empty") + self.assertEqual(reason, "no_rows_returned") + + def test_missing_required_field_is_flagged(self): + rows = [{"ts_code": "600000.SH", "close": 10.2}, {"ts_code": "000001.SZ"}] + status, reason, _ = observability.classify_rows(rows, required_fields=("close",), freshness_field=None) + self.assertEqual(status, "missing_fields") + self.assertIn("close", reason) + + def test_fresh_rows_are_ok(self): + import time + + rows = [{"ts_code": "600000.SH", "close": 10.2, "quote_time_epoch": int(time.time())}] + status, reason, age = observability.classify_rows(rows) + self.assertEqual(status, "ok") + self.assertEqual(reason, "") + self.assertIsNotNone(age) + self.assertLess(age, 5) + + def test_stale_rows_are_flagged(self): + import time + + rows = [{"ts_code": "600000.SH", "close": 10.2, "quote_time_epoch": int(time.time()) - 3600}] + status, reason, age = observability.classify_rows(rows, max_age_seconds=300) + self.assertEqual(status, "stale") + self.assertEqual(reason, "data_age_exceeds_threshold") + self.assertGreaterEqual(age, 3600 - 5) + + def test_classifier_never_raises_on_garbage_input(self): + status, reason, age = observability.classify_rows(object()) + self.assertEqual(status, "empty") + self.assertIsNone(age) + # Malformed rows inside a list must not raise either. + status, _, _ = observability.classify_rows(["not-a-dict", 123, None]) + self.assertEqual(status, "empty") + + +class ClassifyErrorTests(unittest.TestCase): + def test_blocked_page_markers_are_detected(self): + status, reason = observability.classify_error("eastmoney request failed: Expecting value: line 1 column 1") + self.assertEqual(status, "blocked") + self.assertEqual(reason, "response_looks_like_intercept_page") + + def test_timeout_is_detected(self): + status, _ = observability.classify_error("tencent request failed: timed out") + self.assertEqual(status, "timeout") + + def test_generic_error_falls_back(self): + status, reason = observability.classify_error("connection reset by peer") + self.assertEqual(status, "error") + self.assertEqual(reason, "") + + def test_never_raises_on_none(self): + status, reason = observability.classify_error(None) + self.assertEqual(status, "error") + + +class RecordCallTests(unittest.TestCase): + def setUp(self) -> None: + self.tmp = tempfile.TemporaryDirectory() + self.db = HubDB(Path(self.tmp.name) / "hub.db") + + def tearDown(self) -> None: + self.tmp.cleanup() + + def test_record_call_writes_log_and_health(self): + observability.record_call(self.db, "eastmoney", "indices", status="ok", latency_ms=42) + log_rows = self.db.fetchall("SELECT * FROM provider_call_log") + self.assertEqual(len(log_rows), 1) + self.assertEqual(log_rows[0]["provider"], "eastmoney") + self.assertEqual(log_rows[0]["interface"], "indices") + self.assertEqual(log_rows[0]["status"], "ok") + health = self.db.fetchone( + "SELECT * FROM provider_health WHERE provider = ? AND interface = ?", + ("eastmoney", "indices"), + ) + self.assertIsNotNone(health) + self.assertEqual(health["state"], "ok") + self.assertEqual(health["consec_failures"], 0) + + def test_consecutive_failures_increment_and_reset(self): + observability.record_call(self.db, "tencent", "named_quotes", status="error", error="boom") + observability.record_call(self.db, "tencent", "named_quotes", status="error", error="boom again") + health = self.db.fetchone( + "SELECT * FROM provider_health WHERE provider = ? AND interface = ?", + ("tencent", "named_quotes"), + ) + self.assertEqual(health["consec_failures"], 2) + self.assertEqual(health["state"], "error") + observability.record_call(self.db, "tencent", "named_quotes", status="ok") + health = self.db.fetchone( + "SELECT * FROM provider_health WHERE provider = ? AND interface = ?", + ("tencent", "named_quotes"), + ) + self.assertEqual(health["consec_failures"], 0) + self.assertEqual(health["state"], "ok") + + def test_none_db_is_a_silent_noop(self): + # Must not raise even though there is nowhere to write. + observability.record_call(None, "ifind", "wencai", status="ok") + + def test_broken_db_write_does_not_raise(self): + observability.record_call(_BrokenDB(), "eastmoney", "indices", status="ok") + + +class ObserveTests(unittest.TestCase): + def setUp(self) -> None: + self.tmp = tempfile.TemporaryDirectory() + self.db = HubDB(Path(self.tmp.name) / "hub.db") + + def tearDown(self) -> None: + self.tmp.cleanup() + + def test_returns_exact_success_value_unmodified(self): + sentinel = {"ts_code": "600000.SH", "close": 10.2} + result = observability.observe(self.db, "eastmoney", "indices", lambda: sentinel) + self.assertIs(result, sentinel) + rows = self.db.fetchall("SELECT * FROM provider_call_log") + self.assertEqual(len(rows), 1) + self.assertEqual(rows[0]["status"], "ok") + + def test_reraises_exact_exception_on_failure(self): + boom = ValueError("upstream exploded") + + def fn(): + raise boom + + with self.assertRaises(ValueError) as ctx: + observability.observe(self.db, "eastmoney", "indices", fn) + self.assertIs(ctx.exception, boom) + rows = self.db.fetchall("SELECT * FROM provider_call_log") + self.assertEqual(len(rows), 1) + self.assertEqual(rows[0]["status"], "error") + self.assertIn("upstream exploded", rows[0]["error"]) + + def test_classify_downgrades_success_to_stale_without_changing_return_value(self): + sentinel = [{"ts_code": "600000.SH", "quote_time_epoch": 1}] + result = observability.observe( + self.db, "eastmoney", "indices", lambda: sentinel, + classify=lambda rows: observability.classify_rows(rows), + ) + self.assertIs(result, sentinel) + row = self.db.fetchone("SELECT * FROM provider_call_log") + self.assertEqual(row["status"], "stale") + + def test_broken_db_never_breaks_a_successful_call(self): + sentinel = {"ok": True} + result = observability.observe(_BrokenDB(), "eastmoney", "indices", lambda: sentinel) + self.assertIs(result, sentinel) + + def test_broken_db_never_masks_a_real_failure(self): + def fn(): + raise RuntimeError("real upstream failure") + + with self.assertRaises(RuntimeError) as ctx: + observability.observe(_BrokenDB(), "eastmoney", "indices", fn) + self.assertEqual(str(ctx.exception), "real upstream failure") + + def test_classifier_exception_does_not_break_the_call(self): + sentinel = {"ok": True} + + def bad_classify(_result): + raise KeyError("classifier bug") + + result = observability.observe(self.db, "eastmoney", "indices", lambda: sentinel, classify=bad_classify) + self.assertIs(result, sentinel) + row = self.db.fetchone("SELECT * FROM provider_call_log") + # A classifier bug must degrade to "ok", never silently drop the row + # nor claim the call failed when it did not. + self.assertEqual(row["status"], "ok") + + def test_none_db_is_transparent_passthrough(self): + sentinel = object() + result = observability.observe(None, "eastmoney", "indices", lambda: sentinel) + self.assertIs(result, sentinel) + + +if __name__ == "__main__": + unittest.main() diff --git a/xiaobai-datahub/tests/test_pipeline_observability.py b/xiaobai-datahub/tests/test_pipeline_observability.py new file mode 100644 index 0000000..5c797ae --- /dev/null +++ b/xiaobai-datahub/tests/test_pipeline_observability.py @@ -0,0 +1,82 @@ +from __future__ import annotations + +import unittest + +from datahub.pipeline import RetryError +from tests.fixtures import TRADE_DATE +from tests.test_pipeline import make_pipeline + + +class PipelineObservabilityTests(unittest.TestCase): + """HEL-543: Tushare calls must keep writing the existing `src_calls` + record unchanged, while also feeding the new cross-provider + `provider_call_log` / `provider_health` side channel.""" + + def test_successful_fetch_logs_to_both_src_calls_and_provider_call_log(self) -> None: + pipe, db = make_pipeline() + pipe.ingest_reference(TRADE_DATE) + result = pipe.run_dataset("daily", TRADE_DATE) + self.assertEqual(result["state"], "published") + + src_calls = db.fetchall("SELECT * FROM src_calls WHERE provider = 'tushare' AND endpoint = 'daily'") + self.assertTrue(any(row["ok"] == 1 for row in src_calls)) + + log = db.fetchall( + "SELECT * FROM provider_call_log WHERE provider = 'tushare' AND interface = 'daily'" + ) + self.assertTrue(len(log) >= 1) + self.assertEqual(log[-1]["status"], "ok") + + health = db.fetchone( + "SELECT * FROM provider_health WHERE provider = 'tushare' AND interface = 'daily'" + ) + self.assertIsNotNone(health) + self.assertEqual(health["state"], "ok") + self.assertEqual(health["consec_failures"], 0) + + def test_failed_fetch_logs_error_to_both_channels_and_still_raises(self) -> None: + pipe, db = make_pipeline() + pipe.ingest_reference(TRADE_DATE) + + def boom(dataset, params): + raise RuntimeError("tushare upstream 500") + + pipe.adapter.fetch = boom # type: ignore[assignment] + + with self.assertRaises(RetryError): + pipe.run_dataset("daily", TRADE_DATE, attempts=1) + + src_calls = db.fetchall("SELECT * FROM src_calls WHERE provider = 'tushare' AND endpoint = 'daily' AND ok = 0") + self.assertTrue(len(src_calls) >= 1) + self.assertIn("tushare upstream 500", src_calls[-1]["error"]) + + log = db.fetchall( + "SELECT * FROM provider_call_log WHERE provider = 'tushare' AND interface = 'daily' AND status != 'ok'" + ) + self.assertTrue(len(log) >= 1) + self.assertIn("tushare upstream 500", log[-1]["error"]) + + health = db.fetchone( + "SELECT * FROM provider_health WHERE provider = 'tushare' AND interface = 'daily'" + ) + self.assertIsNotNone(health) + self.assertNotEqual(health["state"], "ok") + self.assertGreaterEqual(health["consec_failures"], 1) + + def test_provider_call_log_is_purged_by_existing_cleanup_job(self) -> None: + pipe, db = make_pipeline() + pipe.ingest_reference(TRADE_DATE) + pipe.run_dataset("daily", TRADE_DATE) + self.assertTrue(db.fetchall("SELECT * FROM provider_call_log")) + + # Force everything to look ancient so cleanup() sweeps it. + db.execute("UPDATE provider_call_log SET created_at = '2000-01-01T00:00:00+08:00'") + db.execute("UPDATE src_calls SET created_at = '2000-01-01T00:00:00+08:00'") + db.execute("UPDATE job_runs SET started_at = '2000-01-01T00:00:00+08:00'") + + pipe.cleanup() + self.assertEqual(db.fetchall("SELECT * FROM provider_call_log"), []) + + +if __name__ == "__main__": + unittest.main() diff --git a/xiaobai-datahub/tests/test_realtime_observability.py b/xiaobai-datahub/tests/test_realtime_observability.py new file mode 100644 index 0000000..572bc3a --- /dev/null +++ b/xiaobai-datahub/tests/test_realtime_observability.py @@ -0,0 +1,163 @@ +from __future__ import annotations + +import tempfile +import unittest +from contextlib import contextmanager +from pathlib import Path +from unittest.mock import patch + +from datahub.adapters.base import AdapterError +from datahub.db import HubDB +from datahub.realtime_serve import fetch_index_quotes, fetch_intraday, fetch_quotes + + +class _WriteBreaksDB: + """Wraps a real HubDB but breaks only the write path, to prove the + real serving path (reads/caches) is untouched by an observability + failure while still exercising real fetch/cache code around it.""" + + def __init__(self, real: HubDB) -> None: + self._real = real + + def fetchall(self, sql, params=()): + return self._real.fetchall(sql, params) + + def fetchone(self, sql, params=()): + return self._real.fetchone(sql, params) + + def execute(self, sql, params=()): + return self._real.execute(sql, params) + + def executemany(self, sql, rows): + return self._real.executemany(sql, rows) + + @contextmanager + def write(self): + raise RuntimeError("db is not writable right now") + yield None # pragma: no cover + + +class RealtimeObservabilityTests(unittest.TestCase): + def setUp(self) -> None: + self.tmp = tempfile.TemporaryDirectory() + self.db = HubDB(Path(self.tmp.name) / "hub.db") + + def tearDown(self) -> None: + self.tmp.cleanup() + + def test_eastmoney_success_is_logged_without_changing_payload(self) -> None: + rows = [ + {"ts_code": "000001.SH", "code": "000001", "name": "上证指数", "price": 3000.0, + "previous_close": 2990.0, "quote_time_epoch": 0, "source": "eastmoney_push2"}, + {"ts_code": "399001.SZ", "code": "399001", "name": "深证成指", "price": 9000.0, + "previous_close": 8990.0, "quote_time_epoch": 0, "source": "eastmoney_push2"}, + {"ts_code": "399006.SZ", "code": "399006", "name": "创业板指", "price": 1800.0, + "previous_close": 1790.0, "quote_time_epoch": 0, "source": "eastmoney_push2"}, + ] + with patch("datahub.realtime_serve.EastmoneyAdapter") as mocked: + mocked.return_value.fetch_indices.return_value = rows + payload = fetch_index_quotes(self.db) + self.assertEqual(payload["data"], rows) + self.assertEqual(payload["meta"]["source"], "eastmoney:ulist") + log = self.db.fetchall("SELECT * FROM provider_call_log WHERE provider = 'eastmoney'") + self.assertEqual(len(log), 1) + self.assertEqual(log[0]["interface"], "indices") + self.assertEqual(log[0]["status"], "ok") + health = self.db.fetchone( + "SELECT * FROM provider_health WHERE provider = 'eastmoney' AND interface = 'indices'" + ) + self.assertEqual(health["state"], "ok") + + def test_eastmoney_failure_falls_back_to_tencent_and_logs_both(self) -> None: + tencent_rows = [ + {"ts_code": "000001.SH", "code": "000001", "name": "上证指数", "price": 3000.0, + "previous_close": 2990.0, "quote_time_epoch": 0, "source": "tencent_qt"}, + {"ts_code": "399001.SZ", "code": "399001", "name": "深证成指", "price": 9000.0, + "previous_close": 8990.0, "quote_time_epoch": 0, "source": "tencent_qt"}, + {"ts_code": "399006.SZ", "code": "399006", "name": "创业板指", "price": 1800.0, + "previous_close": 1790.0, "quote_time_epoch": 0, "source": "tencent_qt"}, + ] + with patch("datahub.realtime_serve.EastmoneyAdapter") as eastmoney, patch( + "datahub.realtime_serve.TencentAdapter" + ) as tencent: + eastmoney.return_value.fetch_indices.side_effect = AdapterError("Eastmoney returned 0/3 indices") + tencent.return_value.fetch_indices.return_value = tencent_rows + payload = fetch_index_quotes(self.db) + self.assertEqual(payload["meta"]["source"], "tencent:qt") + self.assertEqual(payload["data"], tencent_rows) + east_log = self.db.fetchone("SELECT * FROM provider_call_log WHERE provider = 'eastmoney'") + self.assertEqual(east_log["status"], "empty") + tencent_log = self.db.fetchone("SELECT * FROM provider_call_log WHERE provider = 'tencent'") + self.assertEqual(tencent_log["status"], "ok") + + def test_observability_db_failure_never_breaks_a_real_successful_fetch(self) -> None: + rows = [ + {"ts_code": "000001.SH", "price": 3000.0, "previous_close": 2990.0, "quote_time_epoch": 0}, + {"ts_code": "399001.SZ", "price": 9000.0, "previous_close": 8990.0, "quote_time_epoch": 0}, + {"ts_code": "399006.SZ", "price": 1800.0, "previous_close": 1790.0, "quote_time_epoch": 0}, + ] + broken = _WriteBreaksDB(self.db) + with patch("datahub.realtime_serve.EastmoneyAdapter") as mocked: + mocked.return_value.fetch_indices.return_value = rows + payload = fetch_index_quotes(broken) + self.assertEqual(payload["data"], rows) + self.assertEqual(payload["meta"]["source"], "eastmoney:ulist") + + def test_observability_db_failure_never_masks_a_real_source_outage(self) -> None: + broken = _WriteBreaksDB(self.db) + with patch("datahub.realtime_serve.EastmoneyAdapter") as eastmoney, patch( + "datahub.realtime_serve.TencentAdapter" + ) as tencent: + eastmoney.return_value.fetch_indices.side_effect = AdapterError("down") + tencent.return_value.fetch_indices.side_effect = AdapterError("also down") + with self.assertRaises(Exception): + fetch_index_quotes(broken) + + def test_named_quotes_records_both_providers_on_partial_merge(self) -> None: + with patch("datahub.realtime_serve.EastmoneyAdapter") as eastmoney, patch( + "datahub.realtime_serve.TencentAdapter" + ) as tencent: + eastmoney.return_value.fetch_quotes.return_value = [ + {"ts_code": "000001.SZ", "close": 10, "pre_close": 9, "quote_date": "20260907"}, + ] + tencent.return_value.fetch_quotes.return_value = [ + {"ts_code": "000002.SZ", "close": 20, "pre_close": 19, "quote_date": "20260907"}, + ] + payload = fetch_quotes(self.db, ["000001.SZ", "000002.SZ"]) + self.assertEqual(payload["meta"]["complete"], True) + east_log = self.db.fetchone( + "SELECT * FROM provider_call_log WHERE provider = 'eastmoney' AND interface = 'named_quotes'" + ) + self.assertEqual(east_log["status"], "ok") + tencent_log = self.db.fetchone( + "SELECT * FROM provider_call_log WHERE provider = 'tencent' AND interface = 'named_quotes'" + ) + self.assertEqual(tencent_log["status"], "ok") + + def test_intraday_success_is_logged_as_ok(self) -> None: + payload_data = { + "entity_type": "stock", "ts_code": "601318.SH", "trade_date": "2026-09-07", + "previous_close": 55.8, "points": [{"date": "2026-09-07", "time": "09:30", "close": 55.9}], + } + with patch("datahub.realtime_serve.EastmoneyAdapter") as mocked: + mocked.return_value.fetch_intraday.return_value = payload_data + payload = fetch_intraday(self.db, "601318.SH") + self.assertEqual(payload["data"], payload_data) + log = self.db.fetchone( + "SELECT * FROM provider_call_log WHERE provider = 'eastmoney' AND interface = 'intraday'" + ) + self.assertEqual(log["status"], "ok") + + def test_intraday_failure_is_logged_as_empty(self) -> None: + with patch("datahub.realtime_serve.EastmoneyAdapter") as mocked: + mocked.return_value.fetch_intraday.side_effect = AdapterError("No intraday chart data returned") + with self.assertRaises(Exception): + fetch_intraday(self.db, "000001.SZ") + log = self.db.fetchone( + "SELECT * FROM provider_call_log WHERE provider = 'eastmoney' AND interface = 'intraday'" + ) + self.assertEqual(log["status"], "empty") + + +if __name__ == "__main__": + unittest.main()