leefer/xiaobai-review
1
1
Fork 0
Code Issues Pull Requests 1 Actions Packages Projects Releases Wiki Activity

HEL-543: add data hub observability side-channel (provider status, source catalog, lineage)

Browse Source 
- New provider_call_log/provider_health tables (additive-only schema),
  wired via a fail-open observability.observe()/record_call() helper.
- Tushare pipeline keeps its existing src_calls record unchanged and now
  also feeds the unified provider_health/provider_call_log side channel.
- Eastmoney/Tencent realtime_serve.py call sites and the iFinD steward
  call site are wrapped with observability.observe() at the call site
  only; no adapter internals, routing, fallback order, or return values
  are touched.
- New read-only admin API endpoints: /admin/api/providers/status,
  /admin/api/source-catalog, /admin/api/lineage,
  /admin/api/lineage/affected.
- New static, read-only source_catalog.py and lineage.py registries
  documenting existing providers/interfaces/datasets and known
  main-site consumers (cited against backend/features/screener and
  backend/features/heaven call sites).
- provider_call_log is purged by the existing pipeline.cleanup() job
  alongside src_calls/job_runs.
- 47 new unit/integration tests covering classification, fail-open
  behavior under DB/log failures, unchanged payloads/exceptions on
  success and failure paths, and the new HTTP endpoints. Full suite:
  173 tests, all green.

Co-authored-by: Cursor <cursoragent@cursor.com>
Co-authored-by: multica-agent <github@multica.ai>
This commit is contained in:
总工
2026-09-14 18:34:59 +08:00
co-authored by Cursor multica-agent
parent 4a90c32fcc
commit ad70d8fccc
14 changed files with 1568 additions and 13 deletions
Download Patch File Download Diff File Expand all files Collapse all files
xiaobai-datahub/datahub/admin_api.py
+36
View File
@@ -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"
xiaobai-datahub/datahub/db.py
+2 -1
View File
@@ -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"),
xiaobai-datahub/datahub/httpapp.py
+20
View File
@@ -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)
xiaobai-datahub/datahub/lineage.py
+258
View File
@@ -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
xiaobai-datahub/datahub/observability.py
+300
View File
@@ -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 = (
"<!doctype", "<html", "expecting value", "verify you are human",
"unusual traffic", "captcha", "安全验证", "访问异常", "请完成验证",
"拦截", "禁止访问", "forbidden",
)
# Matches provider messages like "Eastmoney returned 0/3 indices" or
# "Tencent returned 2/3 indices" (see adapters/eastmoney.py, adapters/tencent.py).
_COUNT_MISMATCH_RE = re.compile(r"returned (\d+)\s*/\s*(\d+)")
def _logger():
from datahub.logutil import get_logger
return get_logger()
def classify_error(message: str) -> 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
xiaobai-datahub/datahub/pipeline.py
+14
View File
@@ -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()
xiaobai-datahub/datahub/realtime_serve.py
+58 -11
View File
@@ -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)
xiaobai-datahub/datahub/source_catalog.py
+165
View File
@@ -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
xiaobai-datahub/datahub/steward.py
+6 -1
View File
@@ -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(