from __future__ import annotations from datetime import datetime from typing import Any from backend.data.datahub.bridge import DatahubBridge from backend.data.realtime import RealtimeAggregateError class HubRealtimeProxy: """Realtime observation facade. Talks only to xiaobai-datahub.""" def __init__(self, datahub: DatahubBridge) -> None: self._datahub = datahub def health_snapshot(self, sector: str = "") -> dict[str, Any]: started = datetime.now().astimezone() indices: list[dict[str, Any]] = [] error = "" try: indices = self.tencent_indices() except RealtimeAggregateError as exc: error = str(exc) epochs = [int(item.get("quote_time_epoch") or 0) for item in indices] max_skew = 120 index_consistent = bool(epochs) and max(epochs) - min(epochs) <= max_skew ready = len(indices) == 3 and index_consistent return { "ready": ready, "isolated": True, "generated_at": started.isoformat(timespec="seconds"), "elapsed_ms": 0, "indices": indices, "index_consistent": index_consistent, "sector": None, "sources": { "datahub_indices": { "ok": ready, "error": error, "source": "datahub", } }, "observations": {}, "policy": { "integration": "datahub_exclusive", "max_index_time_skew_seconds": max_skew, "notice": "实时观察只走数据中枢，主网站不再直连东财/腾讯。", }, } def tencent_indices(self) -> list[dict[str, Any]]: rows = self._datahub.try_index_quotes() or [] result = [_as_index(item) for item in rows if _as_index(item)] wanted = {"000001", "399001", "399006"} result = [item for item in result if item.get("code") in wanted] result.sort(key=lambda item: str(item.get("code") or "")) if len(result) != 3: raise RealtimeAggregateError(f"datahub returned {len(result)}/3 indices") return result def eastmoney_indices(self) -> list[dict[str, Any]]: return self.tencent_indices() def tencent_stock_quote(self, code: str, expected_date: str = "") -> dict[str, Any]: return self._stock_quote(code, expected_date) def eastmoney_stock_quote(self, code: str, expected_date: str = "") -> dict[str, Any]: return self._stock_quote(code, expected_date) def tencent_stock_quotes( self, codes: list[str], expected_date: str = "", minimum: int | None = None, ) -> list[dict[str, Any]]: return self._stock_quotes(codes, expected_date, minimum) def eastmoney_stock_quotes( self, codes: list[str], expected_date: str = "", ) -> list[dict[str, Any]]: return self._stock_quotes(codes, expected_date, None) def eastmoney_shenwan_quote(self, ts_code: str, expected_date: str = "") -> dict[str, Any]: quote = self._datahub.try_sector_quote(ts_code, expected_date) if not quote: raise RealtimeAggregateError(f"datahub shenwan quote unavailable for {ts_code}") return quote def _stock_quote(self, code: str, expected_date: str) -> dict[str, Any]: rows = self._stock_quotes([code], expected_date, 1) if not rows: raise RealtimeAggregateError(f"datahub stock quote unavailable for {code}") return rows[0] def _stock_quotes( self, codes: list[str], expected_date: str, minimum: int | None, ) -> list[dict[str, Any]]: cleaned = [str(item or "").strip() for item in codes if str(item or "").strip()] rows = self._datahub.try_quotes(cleaned) if cleaned else (self._datahub.try_market_quotes(expected_date) or []) quotes = [_as_stock(item) for item in (rows or []) if _as_stock(item)] if expected_date: compact = str(expected_date).replace("-", "") quotes = [ item for item in quotes if not item.get("quote_date") or str(item.get("quote_date") or "").replace("-", "") == compact ] if minimum is not None and len(quotes) < minimum: raise RealtimeAggregateError(f"datahub returned {len(quotes)} quotes, need {minimum}") return quotes def _as_index(row: dict[str, Any]) -> dict[str, Any] | None: code = str(row.get("code") or str(row.get("ts_code") or "").split(".")[0] or "") price = _number(row.get("price") if row.get("price") not in (None, "") else row.get("close")) if not code or price <= 0: return None epoch = int(_number(row.get("quote_time_epoch"))) amount = _number(row.get("amount_billion")) if amount <= 0: amount = round(_number(row.get("amount")) / 100_000_000, 2) return { "code": code, "name": row.get("name") or code, "price": price, "change": _number(row.get("change") if row.get("change") not in (None, "") else row.get("pct_chg")), "change_amount": _number(row.get("change_amount")), "open": _number(row.get("open")), "high": _number(row.get("high")), "low": _number(row.get("low")), "previous_close": _number( row.get("previous_close") if row.get("previous_close") not in (None, "") else row.get("pre_close") ), "amount_billion": amount, "quote_time_epoch": epoch, "quote_time": str(row.get("quote_time") or ""), "source": str(row.get("source") or "datahub"), "cache_age_seconds": 0, } def _as_stock(row: dict[str, Any]) -> dict[str, Any] | None: close = _number(row.get("close") if row.get("close") not in (None, "") else row.get("price")) if close <= 0: return None ts_code = str(row.get("ts_code") or "") code = str(row.get("code") or ts_code.split(".")[0] or "") return { "ts_code": ts_code or code, "code": code, "name": row.get("name") or "", "close": close, "pre_close": _number( row.get("pre_close") if row.get("pre_close") not in (None, "") else row.get("previous_close") ), "open": _number(row.get("open")), "high": _number(row.get("high")), "low": _number(row.get("low")), "volume": _number(row.get("volume") if row.get("volume") not in (None, "") else row.get("vol")), "vol": _number(row.get("vol") if row.get("vol") not in (None, "") else row.get("volume")), "amount": _number(row.get("amount")), "quote_time_epoch": int(_number(row.get("quote_time_epoch"))), "quote_time": str(row.get("quote_time") or ""), "quote_date": str(row.get("quote_date") or ""), "source": str(row.get("source") or "datahub"), "delayed": bool(row.get("delayed")), } def _number(value: Any) -> float: try: return float(value or 0) except (TypeError, ValueError): return 0.0