"""Website-facing data steward: pick source, fail over, cache, never fake zeros. The main site asks for a business/Tushare-shaped API. This module decides whether to serve a published EOD table, live free quotes, or an internal Tushare pull. """ from __future__ import annotations import hashlib import json from typing import Any from datahub.adapters.base import AdapterError from datahub.adapters.tushare import TUSHARE_FIELDS from datahub.numbers import finite_number from datahub.realtime_serve import ( RealtimeApiError, _read_cache, _write_cache, fetch_index_quotes, fetch_market_quotes, fetch_quotes, ) from datahub.serving import ApiError, envelope from datahub.timeutil import isoformat, now_shanghai, yyyymmdd API_TO_DATASET = { "trade_cal": "calendar", "stock_basic": "stocks", "daily": "daily", "daily_basic": "valuation", "index_daily": "index_daily", "moneyflow": "moneyflow", "stk_auction": "auction", "limit_list_d": "limit_events", "ths_hot": "popularity", "dc_hot": "popularity", "hm_detail": "dragon_tiger", "ths_daily": "sector_daily", "dc_index": "sector_daily", "sw_daily": "sector_daily", } DATASET_FETCHER = { "calendar": lambda api, q: api.calendar(q.get("from") or q.get("start_date") or "", q.get("to") or q.get("end_date") or ""), "stocks": lambda api, q: api.stocks(q.get("updated_since") or "", q), "daily": lambda api, q: api.daily_bars(_hub_query(q, adjust="none")), "valuation": lambda api, q: api.valuation(_hub_query(q)), "index_daily": lambda api, q: api.index_bars(_hub_query(q)), "moneyflow": lambda api, q: api.moneyflow(_hub_query(q)), "auction": lambda api, q: api.auction(_hub_query(q)), "limit_events": lambda api, q: api.limit_events(_hub_query(q)), "popularity": lambda api, q: api.popularity(_hub_query(q)), "dragon_tiger": lambda api, q: api.dragon_tiger(_hub_query(q)), "sector_daily": lambda api, q: api.sectors(_hub_query(q)), } SCALE_TO_TUSHARE = { "daily": {"vol": 100.0, "amount": 1000.0}, "index_daily": {"vol": 100.0, "amount": 1000.0}, "valuation": {"total_mv": 10000.0, "circ_mv": 10000.0}, "moneyflow": { "buy_sm_amount": 10000.0, "sell_sm_amount": 10000.0, "buy_md_amount": 10000.0, "sell_md_amount": 10000.0, "buy_lg_amount": 10000.0, "sell_lg_amount": 10000.0, "buy_elg_amount": 10000.0, "sell_elg_amount": 10000.0, "net_mf_amount": 10000.0, }, "auction": {"vol": 100.0, "float_share": 10000.0}, "limit_events": {"limit_amount": 10000.0, "float_mv": 10000.0, "total_mv": 10000.0}, "dragon_tiger": {"buy_amount": 10000.0, "sell_amount": 10000.0, "net_amount": 10000.0}, } LIVE_TTL = { "index_member_all": 6 * 3600, "stk_limit": 3600, "suspend_d": 6 * 3600, "adj_factor": 3600, "hm_list": 24 * 3600, "ths_index": 24 * 3600, "ths_member": 6 * 3600, "stk_mins": 20, "top_list": 3600, "top_inst": 3600, } BLOCKED_LIVE_APIS = {"rt_sw_k"} IFIND_APIS = { "ifind_wencai": "wencai", "ifind_snapshots": "snapshots", "ifind_history": "history", "ifind_realtime": "realtime", "ifind_intraday": "intraday", "ifind_status": "status", } def steward_query(api, body: dict[str, Any]) -> dict[str, Any]: api_name = str(body.get("api_name") or "").strip() params = body.get("params") if isinstance(body.get("params"), dict) else {} fields = str(body.get("fields") or "") if not api_name: raise ApiError("INVALID_ARGUMENT", "api_name is required") if api_name in BLOCKED_LIVE_APIS: raise ApiError("INVALID_ARGUMENT", "rt_sw_k is disabled; use published sw_daily or free Shenwan realtime") if api_name in IFIND_APIS: return _ifind_query(api, api_name, params, fields) if api_name == "rt_k": return _realtime_quotes(api, params, fields) if api_name == "rt_idx_k": return _realtime_index_quotes(api, params, fields) dataset = API_TO_DATASET.get(api_name) if dataset: published = _try_published(api, api_name, dataset, params, fields) if published is not None: return published rows = _live_tushare(api, api_name, params, fields) return envelope( _project(rows, fields), { "tier": "live", "source": "tushare", "stale": False, "staleness_seconds": 0, "row_shape": "tushare", "published_at": isoformat(now_shanghai()), }, ) def _ifind_query(api, api_name: str, params: dict[str, Any], fields: str) -> dict[str, Any]: adapter = getattr(api, "ifind", None) dataset = IFIND_APIS[api_name] if adapter is None: raise ApiError("SOURCE_UNAVAILABLE", "iFinD adapter is not attached") if dataset == "status": return envelope( [dict(adapter.status())], { "tier": "live", "source": "ifind", "stale": False, "staleness_seconds": 0, "row_shape": "ifind", "published_at": isoformat(now_shanghai()), }, ) if not adapter.configured: raise ApiError("SOURCE_UNAVAILABLE", "iFinD 尚未配置") try: rows = adapter.fetch(dataset, dict(params)) except AdapterError as exc: raise ApiError("SOURCE_UNAVAILABLE", str(exc)) from exc return envelope( _project(rows, fields), { "tier": "live", "source": "ifind", "stale": False, "staleness_seconds": 0, "row_shape": "ifind", "published_at": isoformat(now_shanghai()), }, ) def _try_published(api, api_name: str, dataset: str, params: dict[str, Any], fields: str) -> dict[str, Any] | None: fetcher = DATASET_FETCHER.get(dataset) if fetcher is None: return None query = _hub_query(params) if dataset == "popularity": query["source"] = "ths" if api_name == "ths_hot" else "dc" if dataset == "sector_daily": query["family"] = {"ths_daily": "ths", "dc_index": "dc", "sw_daily": "sw"}.get(api_name, "") if dataset == "limit_events": limit_type = str(params.get("limit_type") or "").strip().upper() if limit_type: query["limit_type"] = limit_type if dataset == "calendar" and not (query.get("from") and query.get("to")): start = str(params.get("start_date") or params.get("from") or "") end = str(params.get("end_date") or params.get("to") or start) if not start or not end: return None query = {"from": start, "to": end} try: payload = fetcher(api, query) except ApiError as exc: if exc.code in {"DATASET_NOT_PUBLISHED", "STALE_DATA", "INVALID_ARGUMENT"}: return None raise rows = list(payload.get("data") or []) if dataset == "stocks": rows = _filter_stocks(rows, params) if dataset == "calendar": rows = _filter_calendar(rows, params) native = _to_tushare_native(dataset, rows) meta = dict(payload.get("meta") or {}) meta["row_shape"] = "tushare" meta["source"] = str(meta.get("source") or "datahub") return envelope(_project(native, fields), meta) def _realtime_quotes(api, params: dict[str, Any], fields: str) -> dict[str, Any]: codes = [item.strip() for item in str(params.get("ts_code") or params.get("codes") or "").split(",") if item.strip()] try: payload = fetch_quotes(api.db, codes) if codes else fetch_market_quotes(api.db) except RealtimeApiError as exc: raise ApiError(exc.code, exc.message) from exc rows = [_quote_to_rt_k(item) for item in (payload.get("data") or []) if isinstance(item, dict)] rows = [item for item in rows if item] meta = dict(payload.get("meta") or {}) meta["row_shape"] = "tushare" return envelope(_project(rows, fields), meta) def _realtime_index_quotes(api, params: dict[str, Any], fields: str) -> dict[str, Any]: try: payload = fetch_index_quotes(api.db) except RealtimeApiError as exc: raise ApiError(exc.code, exc.message) from exc wanted = { item.strip() for item in str(params.get("ts_code") or "").split(",") if item.strip() } rows = [] for item in payload.get("data") or []: if not isinstance(item, dict): continue converted = _quote_to_rt_k(item) if not converted: continue if wanted and converted.get("ts_code") not in wanted and str(item.get("code") or "") not in { code.split(".")[0] for code in wanted }: continue rows.append(converted) meta = dict(payload.get("meta") or {}) meta["row_shape"] = "tushare" return envelope(_project(rows, fields), meta) def _live_tushare(api, api_name: str, params: dict[str, Any], fields: str) -> list[dict[str, Any]]: wanted_fields = fields or TUSHARE_FIELDS.get(api_name, "") cache_key = _live_cache_key(api_name, params, wanted_fields) ttl = LIVE_TTL.get(api_name, 1800) cached = _read_cache(api.db, cache_key) if cached is not None: data = cached.get("data") if isinstance(data, list): return [dict(item) for item in data if isinstance(item, dict)] pipeline = api.pipeline if not pipeline.breaker.allow(): recovered = _live_lkg(api.db, cache_key) if recovered is not None: return recovered raise ApiError("SOURCE_UNAVAILABLE", "Tushare circuit open") pipeline.bucket.acquire() try: rows = pipeline.adapter.query_raw(api_name, dict(params), wanted_fields) pipeline.breaker.record_success() except Exception as exc: pipeline.breaker.record_failure(str(exc)) recovered = _live_lkg(api.db, cache_key) if recovered is not None: return recovered raise ApiError("SOURCE_UNAVAILABLE", f"Tushare {api_name} unavailable: {exc}") from exc payload = envelope( rows, { "tier": "live", "source": "tushare", "stale": False, "staleness_seconds": 0, "row_shape": "tushare", "published_at": isoformat(now_shanghai()), }, ) _write_cache(api.db, cache_key, payload, ttl, "tushare") return rows def _live_lkg(db, cache_key: str) -> list[dict[str, Any]] | None: from datahub.governance.lkg import LastKnownGood item = LastKnownGood(db).load(cache_key) payload = item.get("payload") if item else None if not isinstance(payload, dict): return None data = payload.get("data") if not isinstance(data, list) or not data: return None return [dict(row) for row in data if isinstance(row, dict)] def _live_cache_key(api_name: str, params: dict[str, Any], fields: str) -> str: packed = json.dumps({"api": api_name, "params": params, "fields": fields}, sort_keys=True, ensure_ascii=False) digest = hashlib.sha1(packed.encode("utf-8")).hexdigest() return f"steward:{api_name}:{digest}" def _hub_query(params: dict[str, Any], **extra: Any) -> dict[str, str]: query = {key: str(value) for key, value in extra.items() if value not in (None, "")} date = yyyymmdd(params.get("trade_date") or params.get("date") or "") start = yyyymmdd(params.get("start_date") or params.get("from") or date) end = yyyymmdd(params.get("end_date") or params.get("to") or date) code = str(params.get("ts_code") or params.get("code") or "").strip() if code: query["code"] = code if date and not (params.get("start_date") or params.get("end_date")): query["date"] = date else: if start: query["from"] = start if end: query["to"] = end return query def _to_tushare_native(dataset: str, rows: list[dict[str, Any]]) -> list[dict[str, Any]]: scales = SCALE_TO_TUSHARE.get(dataset) or {} converted: list[dict[str, Any]] = [] for row in rows: item = dict(row) if item.get("vol") in (None, ""): item["vol"] = item.get("volume") item.pop("volume", None) for field, factor in scales.items(): if field in item and item[field] not in (None, ""): number = finite_number(item.get(field)) item[field] = number / factor if factor else number if dataset == "popularity" and item.get("ts_name") and not item.get("name"): item["name"] = item.get("ts_name") if dataset == "dragon_tiger" and item.get("ts_name") and not item.get("name"): item["name"] = item.get("ts_name") if dataset == "sector_daily" and item.get("pct_change") is not None and item.get("pct_chg") is None: item["pct_chg"] = item.get("pct_change") if dataset == "calendar": item["is_open"] = 1 if item.get("is_open") in (True, 1, "1", "Y", "y") else 0 converted.append(item) return converted def _quote_to_rt_k(row: dict[str, Any]) -> dict[str, Any] | None: ts_code = str(row.get("ts_code") or "").strip() close = finite_number(row.get("close") if row.get("close") not in (None, "") else row.get("price")) previous = finite_number( row.get("pre_close") if row.get("pre_close") not in (None, "") else row.get("previous_close") ) if not ts_code or close <= 0: return None item = { "ts_code": ts_code, "name": row.get("name") or "", "open": row.get("open"), "high": row.get("high"), "low": row.get("low"), "close": close, "pre_close": previous, "vol": row.get("vol") if row.get("vol") not in (None, "") else row.get("volume"), "amount": row.get("amount"), "pct_chg": row.get("pct_chg") if row.get("pct_chg") not in (None, "") else row.get("change"), "trade_time": row.get("quote_time") or row.get("trade_time") or "", "quote_date": row.get("quote_date") or "", "source": row.get("source") or "", "delayed": bool(row.get("delayed")), "delay_seconds": row.get("delay_seconds") or 0, "delay_notice": row.get("delay_notice") or "", } return item def _filter_stocks(rows: list[dict[str, Any]], params: dict[str, Any]) -> list[dict[str, Any]]: ts_code = str(params.get("ts_code") or "").strip().upper() status = str(params.get("list_status") or "").strip() name = str(params.get("name") or "").strip() filtered = rows if ts_code: filtered = [row for row in filtered if str(row.get("ts_code") or "").upper() == ts_code] if status: filtered = [row for row in filtered if str(row.get("list_status") or status) == status] if name: filtered = [row for row in filtered if name.casefold() in str(row.get("name") or "").casefold()] return filtered def _filter_calendar(rows: list[dict[str, Any]], params: dict[str, Any]) -> list[dict[str, Any]]: start = yyyymmdd(params.get("start_date") or params.get("from") or "") end = yyyymmdd(params.get("end_date") or params.get("to") or start) if start and end: rows = [row for row in rows if start <= yyyymmdd(row.get("cal_date")) <= end] if params.get("is_open") in (1, "1", True): rows = [row for row in rows if int(row.get("is_open") or 0) == 1] return rows def _project(rows: list[dict[str, Any]], fields: str) -> list[dict[str, Any]]: keys = [item.strip() for item in str(fields or "").split(",") if item.strip()] if not keys: return rows return [{key: row.get(key) for key in keys} for row in rows]