From 605f97e5dffd25132f6a857b7870ab0f023c8f2a Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=80=BB=E5=B7=A5?= Date: Sat, 5 Sep 2026 17:30:58 +0800 Subject: [PATCH] =?UTF-8?q?feat(HEL-463):=20=E6=8E=A5=E5=85=A5=E5=89=A9?= =?UTF-8?q?=E4=BD=99=E8=A1=8C=E6=83=85=E6=95=B0=E6=8D=AE=E5=88=B0=20datahu?= =?UTF-8?q?b?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 扩展盘后正式集(涨跌停/人气/龙虎榜/板块日线)与盘中观察 API(报价/指数/分时),网站 bridge 按开关接入并回退旧链路;问天改为按数据依赖跟随开关,不再整栈强制旧路径。 Co-authored-by: Cursor Co-authored-by: multica-agent --- .env.example | 3 + backend/data/datahub/bridge.py | 36 ++- backend/data/datahub/client.py | 21 ++ backend/data/datahub/native.py | 51 ++++ backend/data/datahub/settings.py | 14 + config/datahub.config.json | 7 + tests/test_datahub_bridge.py | 7 +- xiaobai-datahub/README.md | 5 +- xiaobai-datahub/datahub/adapters/__init__.py | 8 +- xiaobai-datahub/datahub/adapters/eastmoney.py | 251 +++++++++++++++++- xiaobai-datahub/datahub/adapters/tencent.py | 100 ++++++- xiaobai-datahub/datahub/adapters/tushare.py | 119 ++++++++- xiaobai-datahub/datahub/datasets_ext.py | 180 +++++++++++++ xiaobai-datahub/datahub/db.py | 6 +- xiaobai-datahub/datahub/normalize.py | 94 +++++++ xiaobai-datahub/datahub/pipeline.py | 81 +++++- xiaobai-datahub/datahub/realtime_serve.py | 188 +++++++++++++ xiaobai-datahub/datahub/scheduler.py | 24 +- xiaobai-datahub/datahub/serving.py | 90 ++++++- xiaobai-datahub/tests/fixtures.py | 31 ++- xiaobai-datahub/tests/test_extended_eod.py | 65 +++++ xiaobai-datahub/tests/test_layout.py | 8 +- 22 files changed, 1348 insertions(+), 41 deletions(-) create mode 100644 xiaobai-datahub/datahub/datasets_ext.py create mode 100644 xiaobai-datahub/datahub/realtime_serve.py create mode 100644 xiaobai-datahub/tests/test_extended_eod.py diff --git a/.env.example b/.env.example index c89b184..46c4c14 100644 --- a/.env.example +++ b/.env.example @@ -7,6 +7,9 @@ TUSHARE_TOKEN=your_tushare_token_here # Optional xiaobai-datahub client. All DATAHUB_READ_* / DATAHUB_SHADOW_* flags # default off in config/datahub.config.json, so the website keeps using Tushare. +# Extended datasets (HEL-463): LIMIT_EVENTS POPULARITY DRAGON_TIGER SECTOR_DAILY +# QUOTES INDEX_QUOTES INTRADAY — plus first-batch CALENDAR STOCKS DAILY INDEX_DAILY +# VALUATION MONEYFLOW AUCTION STATUS. DATAHUB_BASE_URL=http://127.0.0.1:8766 DATAHUB_TOKEN= diff --git a/backend/data/datahub/bridge.py b/backend/data/datahub/bridge.py index b4e36ac..938a92f 100644 --- a/backend/data/datahub/bridge.py +++ b/backend/data/datahub/bridge.py @@ -21,11 +21,14 @@ from backend.data.providers.tushare_client import TushareClient LOGGER = logging.getLogger("xiaobai.datahub") ShadowSink = Callable[[dict[str, Any]], None] -EMPTY_FAIL_DATASETS = {"stocks", "daily", "index_daily", "valuation", "moneyflow", "auction"} +EMPTY_FAIL_DATASETS = { + "stocks", "daily", "index_daily", "valuation", "moneyflow", "auction", + "limit_events", "sector_daily", +} def looks_like_heaven(module_name: str, filename: str = "") -> bool: - """问天调用栈识别。问天未永久冻结,只是本阶段仍走旧 Tushare 链路。""" + """问天调用栈识别(诊断用)。问天按数据集依赖接入,不再整栈强制旧链路。""" path = filename.replace("\\", "/") return module_name.startswith("backend.features.heaven") or "/features/heaven/" in path @@ -96,8 +99,8 @@ class DatahubBridge: legacy_query: Callable[..., list[dict[str, Any]]], ) -> list[dict[str, Any]]: dataset = API_TO_DATASET.get(api_name) - # 问天允许后续纳入 datahub;首批只读接入仍保持旧链路,避免误切。 - if not dataset or self.heaven_guard(): + # 问天按实际数据依赖接入:已映射到 hub 的 API 跟随开关;未映射的继续旧链路。 + if not dataset: return legacy_query(api_name, params, fields) flags = self.settings.flags(dataset) if not flags.read and not flags.shadow: @@ -108,7 +111,7 @@ class DatahubBridge: hub_error: str | None = None hub_canonical: list[dict[str, Any]] = [] try: - response = self._fetch_dataset(dataset, params or {}) + response = self._fetch_dataset(dataset, params or {}, api_name=api_name) hub_canonical = self._extract_rows(dataset, response, params or {}) hub_rows = to_native_rows(dataset, hub_canonical) hub_meta = dict(response.meta) @@ -136,7 +139,7 @@ class DatahubBridge: return project_fields(hub_rows, fields) return legacy_query(api_name, params, fields) - def _fetch_dataset(self, dataset: str, params: dict[str, Any]) -> DatahubResponse: + def _fetch_dataset(self, dataset: str, params: dict[str, Any], api_name: str = "") -> DatahubResponse: date = yyyymmdd(params.get("trade_date") or params.get("date")) start = yyyymmdd(params.get("start_date") or params.get("from") or date) end = yyyymmdd(params.get("end_date") or params.get("to") or date) @@ -153,6 +156,10 @@ class DatahubBridge: "valuation": self.client.valuation, "moneyflow": self.client.moneyflow, "auction": self.client.auction, + "limit_events": self.client.limit_events, + "popularity": self.client.popularity, + "dragon_tiger": self.client.dragon_tiger, + "sector_daily": self.client.sectors, } fetcher = fetchers[dataset] query: dict[str, Any] = {} @@ -167,6 +174,23 @@ class DatahubBridge: query["to"] = end if dataset == "daily": query["adjust"] = "none" + if dataset == "limit_events": + limit_type = str(params.get("limit_type") or "").strip().upper() + if limit_type: + query["limit_type"] = limit_type + if dataset == "popularity": + if api_name == "ths_hot": + query["source"] = "ths" + elif api_name == "dc_hot": + query["source"] = "dc" + if dataset == "sector_daily": + family = { + "ths_daily": "ths", + "dc_index": "dc", + "sw_daily": "sw", + }.get(api_name, "") + if family: + query["family"] = family return self._paginate(fetcher, query) def _paginate(self, fetcher: Callable[..., DatahubResponse], params: dict[str, Any]) -> DatahubResponse: diff --git a/backend/data/datahub/client.py b/backend/data/datahub/client.py index 7f247f2..5b8d737 100644 --- a/backend/data/datahub/client.py +++ b/backend/data/datahub/client.py @@ -60,6 +60,27 @@ class DatahubClient: def auction(self, **params: Any) -> DatahubResponse: return self.get("/v1/auction", params) + def limit_events(self, **params: Any) -> DatahubResponse: + return self.get("/v1/limit-events", params) + + def popularity(self, **params: Any) -> DatahubResponse: + return self.get("/v1/popularity", params) + + def dragon_tiger(self, **params: Any) -> DatahubResponse: + return self.get("/v1/dragon-tiger", params) + + def sectors(self, **params: Any) -> DatahubResponse: + return self.get("/v1/sectors", params) + + def quotes_latest(self, **params: Any) -> DatahubResponse: + return self.get("/v1/quotes/latest", params) + + def index_quotes(self, **params: Any) -> DatahubResponse: + return self.get("/v1/indexes/quotes", params) + + def intraday_points(self, **params: Any) -> DatahubResponse: + return self.get("/v1/intraday/points", params) + def dataset_status(self, date: str) -> DatahubResponse: return self.get("/v1/datasets/status", {"date": date}) diff --git a/backend/data/datahub/native.py b/backend/data/datahub/native.py index e68b7ef..cdc46ca 100644 --- a/backend/data/datahub/native.py +++ b/backend/data/datahub/native.py @@ -17,6 +17,13 @@ API_TO_DATASET = { "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", } SCALE_FIELDS = { @@ -35,6 +42,16 @@ SCALE_FIELDS = { "net_mf_amount": AMOUNT_WAN_YUAN, }, "auction": {"vol": VOLUME_LOT, "float_share": AMOUNT_WAN_YUAN}, + "limit_events": { + "limit_amount": AMOUNT_WAN_YUAN, + "float_mv": AMOUNT_WAN_YUAN, + "total_mv": AMOUNT_WAN_YUAN, + }, + "dragon_tiger": { + "buy_amount": AMOUNT_WAN_YUAN, + "sell_amount": AMOUNT_WAN_YUAN, + "net_amount": AMOUNT_WAN_YUAN, + }, } @@ -67,6 +84,16 @@ def to_native_row(dataset: str, row: dict[str, Any]) -> dict[str, Any]: converted[field] = _unscale(converted.get(field), factor) if dataset == "stocks": converted.pop("updated_at", None) + if dataset == "popularity": + # keep hub source; callers filter ths/dc themselves when needed + if converted.get("ts_name") and not converted.get("name"): + converted["name"] = converted.get("ts_name") + if dataset == "dragon_tiger": + if converted.get("ts_name") and not converted.get("name"): + converted["name"] = converted.get("ts_name") + if dataset == "sector_daily": + if converted.get("pct_change") is not None and converted.get("pct_chg") is None: + converted["pct_chg"] = converted.get("pct_change") return converted @@ -96,6 +123,30 @@ def row_key(dataset: str, row: dict[str, Any]) -> tuple[str, ...]: return (str(row.get("ts_code") or "").upper(),) if dataset == "status": return (str(row.get("dataset") or ""), yyyymmdd(row.get("trade_date"))) + if dataset == "limit_events": + return ( + str(row.get("ts_code") or "").upper(), + yyyymmdd(row.get("trade_date")), + str(row.get("limit_type") or ""), + ) + if dataset == "popularity": + return ( + str(row.get("ts_code") or "").upper(), + yyyymmdd(row.get("trade_date")), + str(row.get("source") or ""), + ) + if dataset == "dragon_tiger": + return ( + str(row.get("ts_code") or "").upper(), + yyyymmdd(row.get("trade_date")), + str(row.get("hm_name") or ""), + ) + if dataset == "sector_daily": + return ( + str(row.get("ts_code") or "").upper(), + yyyymmdd(row.get("trade_date")), + str(row.get("family") or ""), + ) return (str(row.get("ts_code") or "").upper(), yyyymmdd(row.get("trade_date"))) diff --git a/backend/data/datahub/settings.py b/backend/data/datahub/settings.py index 419b3c7..fedacb4 100644 --- a/backend/data/datahub/settings.py +++ b/backend/data/datahub/settings.py @@ -17,6 +17,13 @@ DATASETS = ( "valuation", "moneyflow", "auction", + "limit_events", + "popularity", + "dragon_tiger", + "sector_daily", + "quotes", + "index_quotes", + "intraday", "status", ) @@ -28,6 +35,13 @@ ENV_DATASET = { "valuation": "VALUATION", "moneyflow": "MONEYFLOW", "auction": "AUCTION", + "limit_events": "LIMIT_EVENTS", + "popularity": "POPULARITY", + "dragon_tiger": "DRAGON_TIGER", + "sector_daily": "SECTOR_DAILY", + "quotes": "QUOTES", + "index_quotes": "INDEX_QUOTES", + "intraday": "INTRADAY", "status": "STATUS", } diff --git a/config/datahub.config.json b/config/datahub.config.json index 1877fbc..31f622a 100644 --- a/config/datahub.config.json +++ b/config/datahub.config.json @@ -13,6 +13,13 @@ "valuation": { "read": false, "shadow": false }, "moneyflow": { "read": false, "shadow": false }, "auction": { "read": false, "shadow": false }, + "limit_events": { "read": false, "shadow": false }, + "popularity": { "read": false, "shadow": false }, + "dragon_tiger": { "read": false, "shadow": false }, + "sector_daily": { "read": false, "shadow": false }, + "quotes": { "read": false, "shadow": false }, + "index_quotes": { "read": false, "shadow": false }, + "intraday": { "read": false, "shadow": false }, "status": { "read": false, "shadow": false } } } diff --git a/tests/test_datahub_bridge.py b/tests/test_datahub_bridge.py index 23155df..3949769 100644 --- a/tests/test_datahub_bridge.py +++ b/tests/test_datahub_bridge.py @@ -290,8 +290,8 @@ class DatahubBridgeTests(unittest.TestCase): self.assertEqual(canonical["vol"], 100000.0) self.assertEqual(canonical["amount"], 2000000.0) - def test_heaven_keeps_legacy_on_first_batch_even_when_read_flag_is_on(self) -> None: - """问天未永久冻结;首批只读接入仍走旧链路,后续迁移可以纳入。""" + def test_heaven_can_use_hub_when_dataset_flag_is_on(self) -> None: + """问天按数据依赖接入:已映射 API 跟随开关,不再整栈强制旧链路。""" self.assertTrue(looks_like_heaven("backend.features.heaven.market_context", "backend/features/heaven/market_context.py")) self.assertFalse(looks_like_heaven("backend.features.market.service", "backend/features/market/service.py")) client = FakeClient() @@ -302,7 +302,8 @@ class DatahubBridgeTests(unittest.TestCase): ) rows = wrapped.query("daily", {"trade_date": "20240902"}, "amount") self.assertEqual(rows[0]["amount"], 2000.0) - self.assertEqual(client.paths, []) + self.assertEqual(client.paths, ["/v1/bars/daily"]) + self.assertEqual(legacy.calls, []) def test_status_flag_does_not_run_when_off_and_falls_back_when_on(self) -> None: off = DatahubBridge(flags(), FakeClient(error=DatahubError("UNAVAILABLE", "down"))) diff --git a/xiaobai-datahub/README.md b/xiaobai-datahub/README.md index 607e299..7d82fa4 100644 --- a/xiaobai-datahub/README.md +++ b/xiaobai-datahub/README.md @@ -6,11 +6,12 @@ ## 做什么 - SQLite WAL `datahub.db`,容器名 `xiaobai-datahub`,端口 `8766` -- Tushare 盘后正式数据:交易日历、股票主档、daily、daily_basic、adj_factor、index_daily、moneyflow、stk_auction +- Tushare 盘后正式数据:交易日历、股票主档、daily、daily_basic、adj_factor、index_daily、moneyflow、stk_auction、limit_list_d、ths_hot/dc_hot、hm_detail、ths_daily/dc_index/sw_daily +- 盘中观察(provisional):东财/腾讯指数报价、个股最新价、分时点(`/v1/quotes/latest` `/v1/indexes/quotes` `/v1/intraday/points`);永不写入 eod_* 正式表 - 暂存 → 校验 → 整批原子发布 → 可回滚 - `/v1` 稳定接口(`X-Datahub-Token`) - `/admin/` 最小管理后台(总览 / 数据源 / 调度 / 发布 / 数据集 / 审计) -- 东财/腾讯/同花顺/选股宝/AKShare/iFinD 适配器位已预留,本阶段不拉实时源 +- 同花顺/选股宝/AKShare/iFinD 适配器位仍预留;东财/腾讯已接入盘中观察 ## 单位口径(相对现站) diff --git a/xiaobai-datahub/datahub/adapters/__init__.py b/xiaobai-datahub/datahub/adapters/__init__.py index a5dcfb8..c9050f9 100644 --- a/xiaobai-datahub/datahub/adapters/__init__.py +++ b/xiaobai-datahub/datahub/adapters/__init__.py @@ -1,13 +1,13 @@ from datahub.adapters.akshare import ADAPTER as akshare -from datahub.adapters.eastmoney import ADAPTER as eastmoney +from datahub.adapters.eastmoney import EastmoneyAdapter from datahub.adapters.ifind import ADAPTER as ifind -from datahub.adapters.tencent import ADAPTER as tencent +from datahub.adapters.tencent import TencentAdapter from datahub.adapters.ths import ADAPTER as ths from datahub.adapters.xgb import ADAPTER as xgb RESERVED = { - "eastmoney": eastmoney, - "tencent": tencent, + "eastmoney": EastmoneyAdapter(), + "tencent": TencentAdapter(), "ths": ths, "xgb": xgb, "akshare": akshare, diff --git a/xiaobai-datahub/datahub/adapters/eastmoney.py b/xiaobai-datahub/datahub/adapters/eastmoney.py index 29d5034..32b0bc7 100644 --- a/xiaobai-datahub/datahub/adapters/eastmoney.py +++ b/xiaobai-datahub/datahub/adapters/eastmoney.py @@ -1,3 +1,250 @@ -from datahub.adapters.base import ReservedAdapter +from __future__ import annotations -ADAPTER = ReservedAdapter("eastmoney") +import json +import time +import urllib.error +import urllib.parse +import urllib.request +from datetime import datetime +from typing import Any + +from datahub.adapters.base import AdapterError, MarketAdapter +from datahub.numbers import finite_number, round4 + +EASTMONEY_INDEX_URL = "https://push2.eastmoney.com/api/qt/ulist.np/get" +EASTMONEY_CLIST_URL = "https://push2.eastmoney.com/api/qt/clist/get" +TRENDS_URL = "https://push2delay.eastmoney.com/api/qt/stock/trends2/get" +BROWSER_UA = ( + "Mozilla/5.0 (Windows NT 10.0; Win64; x64) " + "AppleWebKit/537.36 (KHTML, like Gecko) Chrome/138.0.0.0 Safari/537.36" +) +INDEX_SECIDS = { + "000001.SH": "1.000001", + "399001.SZ": "0.399001", + "399006.SZ": "0.399006", +} + + +class EastmoneyAdapter(MarketAdapter): + name = "eastmoney" + + def __init__(self, timeout: int = 8) -> None: + self.timeout = timeout + + def probe(self) -> dict[str, Any]: + started = time.perf_counter() + try: + rows = self.fetch_indices() + state = "ok" if len(rows) == 3 else "empty" + except AdapterError as exc: + return { + "provider": self.name, + "configured": True, + "state": "error", + "message": str(exc), + "latency_ms": round((time.perf_counter() - started) * 1000), + } + return { + "provider": self.name, + "configured": True, + "state": state, + "latency_ms": round((time.perf_counter() - started) * 1000), + } + + def fetch(self, dataset: str, params: dict[str, Any]) -> list[dict[str, Any]]: + if dataset in {"indexes_quotes", "index_quotes"}: + return self.fetch_indices() + if dataset in {"quotes", "quotes_latest"}: + codes = params.get("codes") or [] + if isinstance(codes, str): + codes = [item.strip() for item in codes.split(",") if item.strip()] + return self.fetch_quotes(list(codes)) + raise AdapterError(f"{self.name} unsupported dataset: {dataset}") + + def normalize(self, dataset: str, rows: list[dict[str, Any]]) -> list[dict[str, Any]]: + return list(rows) + + def fetch_indices(self) -> list[dict[str, Any]]: + payload = self._get_json( + EASTMONEY_INDEX_URL, + { + "secids": "1.000001,0.399001,0.399006", + "fltt": "2", + "invt": "2", + "fields": "f12,f14,f2,f3,f4,f15,f16,f17,f18,f6,f124", + }, + referer="https://quote.eastmoney.com/", + ) + rows = list((payload.get("data") or {}).get("diff") or []) + result = [] + for row in rows: + code = str(row.get("f12") or "") + if code not in {"000001", "399001", "399006"}: + continue + epoch = int(finite_number(row.get("f124")) or 0) + ts_code = f"{code}.SH" if code.startswith("0") and code == "000001" else f"{code}.SZ" + if code == "000001": + ts_code = "000001.SH" + result.append( + { + "ts_code": ts_code, + "code": code, + "name": row.get("f14") or code, + "price": round4(finite_number(row.get("f2"))), + "pct_chg": round4(finite_number(row.get("f3"))), + "change_amount": round4(finite_number(row.get("f4"))), + "open": round4(finite_number(row.get("f17"))), + "high": round4(finite_number(row.get("f15"))), + "low": round4(finite_number(row.get("f16"))), + "previous_close": round4(finite_number(row.get("f18"))), + "amount": round4(finite_number(row.get("f6"))), + "quote_time_epoch": epoch, + "quote_time": ( + datetime.fromtimestamp(epoch).astimezone().isoformat(timespec="seconds") + if epoch + else "" + ), + "source": "eastmoney_push2", + } + ) + if len(result) != 3: + raise AdapterError(f"Eastmoney returned {len(result)}/3 indices") + return result + + def fetch_quotes(self, codes: list[str]) -> list[dict[str, Any]]: + # Eastmoney clist does not accept arbitrary code lists well; use ulist.np for batches. + secids = [] + for code in codes: + ts = str(code or "").upper() + symbol = ts.split(".")[0] + if ts.endswith(".SH") or symbol.startswith(("5", "6", "9")): + secids.append(f"1.{symbol}") + else: + secids.append(f"0.{symbol}") + if not secids: + return [] + payload = self._get_json( + EASTMONEY_INDEX_URL, + { + "secids": ",".join(secids[:60]), + "fltt": "2", + "invt": "2", + "fields": "f12,f14,f2,f3,f4,f15,f16,f17,f18,f5,f6,f8,f124", + }, + referer="https://quote.eastmoney.com/", + ) + rows = list((payload.get("data") or {}).get("diff") or []) + result = [] + for row in rows: + symbol = str(row.get("f12") or "") + if not symbol: + continue + ts_code = f"{symbol}.SH" if symbol.startswith(("5", "6", "9")) else f"{symbol}.SZ" + epoch = int(finite_number(row.get("f124")) or 0) + result.append( + { + "ts_code": ts_code, + "name": row.get("f14") or symbol, + "price": round4(finite_number(row.get("f2"))), + "pct_chg": round4(finite_number(row.get("f3"))), + "change_amount": round4(finite_number(row.get("f4"))), + "open": round4(finite_number(row.get("f17"))), + "high": round4(finite_number(row.get("f15"))), + "low": round4(finite_number(row.get("f16"))), + "previous_close": round4(finite_number(row.get("f18"))), + "volume": round4(finite_number(row.get("f5"))), + "amount": round4(finite_number(row.get("f6"))), + "turnover_rate": round4(finite_number(row.get("f8"))), + "quote_time_epoch": epoch, + "quote_time": ( + datetime.fromtimestamp(epoch).astimezone().isoformat(timespec="seconds") + if epoch + else "" + ), + "source": "eastmoney_push2", + } + ) + return result + + def fetch_intraday(self, ts_code: str) -> dict[str, Any]: + code = str(ts_code or "").upper() + if code in INDEX_SECIDS: + secid = INDEX_SECIDS[code] + entity = "index" + identifier = code + else: + symbol = code.split(".")[0] + market = "1" if symbol.startswith(("5", "6", "9")) else "0" + secid = f"{market}.{symbol}" + entity = "stock" + identifier = symbol + payload = self._get_json( + TRENDS_URL, + { + "secid": secid, + "fields1": "f1,f2,f3,f4,f5,f6,f7,f8,f9,f10,f11,f12,f13", + "fields2": "f51,f52,f53,f54,f55,f56,f57,f58", + "iscr": "0", + "ndays": "1", + }, + referer="https://quote.eastmoney.com/", + ) + data = payload.get("data") or {} + points = [] + for raw in data.get("trends") or []: + point = _parse_trend(raw) + if point: + points.append(point) + if not points: + raise AdapterError("No intraday chart data returned") + return { + "entity_type": entity, + "identifier": identifier, + "ts_code": code if "." in code else f"{identifier}.{'SH' if identifier.startswith(('5','6','9')) else 'SZ'}", + "name": str(data.get("name") or ""), + "code": str(data.get("code") or identifier), + "trade_date": points[-1]["date"], + "previous_close": round4(finite_number(data.get("preClose"))), + "points": points, + "source": "eastmoney_trends2", + } + + def _get_json(self, url: str, params: dict[str, str], referer: str) -> dict[str, Any]: + request_url = f"{url}?{urllib.parse.urlencode(params)}" + request = urllib.request.Request( + request_url, + headers={ + "Accept": "application/json,text/plain,*/*", + "User-Agent": BROWSER_UA, + "Referer": referer, + }, + method="GET", + ) + try: + with urllib.request.urlopen(request, timeout=self.timeout) as response: + return json.loads(response.read().decode("utf-8")) + except Exception as exc: + raise AdapterError(f"eastmoney request failed: {exc}") from exc + + +def _parse_trend(raw: Any) -> dict[str, Any] | None: + text = str(raw or "") + parts = text.split(",") + if len(parts) < 8: + return None + stamp = parts[0] + try: + when = datetime.strptime(stamp, "%Y-%m-%d %H:%M") + except ValueError: + return None + return { + "time": when.strftime("%H:%M"), + "date": when.strftime("%Y-%m-%d"), + "open": round4(finite_number(parts[1])), + "close": round4(finite_number(parts[2])), + "high": round4(finite_number(parts[3])), + "low": round4(finite_number(parts[4])), + "avg_price": round4(finite_number(parts[7] if len(parts) > 7 else parts[2])), + "volume": round4(finite_number(parts[5])), + "amount": round4(finite_number(parts[6])), + } diff --git a/xiaobai-datahub/datahub/adapters/tencent.py b/xiaobai-datahub/datahub/adapters/tencent.py index 015c701..b992e99 100644 --- a/xiaobai-datahub/datahub/adapters/tencent.py +++ b/xiaobai-datahub/datahub/adapters/tencent.py @@ -1,3 +1,99 @@ -from datahub.adapters.base import ReservedAdapter +from __future__ import annotations -ADAPTER = ReservedAdapter("tencent") +import time +import urllib.error +import urllib.request +from datetime import datetime +from typing import Any + +from datahub.adapters.base import AdapterError, MarketAdapter +from datahub.numbers import finite_number, round4 + +TENCENT_INDEX_URL = "https://qt.gtimg.cn/q=sh000001,sz399001,sz399006" +BROWSER_UA = ( + "Mozilla/5.0 (Windows NT 10.0; Win64; x64) " + "AppleWebKit/537.36 (KHTML, like Gecko) Chrome/138.0.0.0 Safari/537.36" +) + + +class TencentAdapter(MarketAdapter): + name = "tencent" + + def __init__(self, timeout: int = 8) -> None: + self.timeout = timeout + + def probe(self) -> dict[str, Any]: + started = time.perf_counter() + try: + rows = self.fetch_indices() + state = "ok" if len(rows) == 3 else "empty" + except AdapterError as exc: + return { + "provider": self.name, + "configured": True, + "state": "error", + "message": str(exc), + "latency_ms": round((time.perf_counter() - started) * 1000), + } + return { + "provider": self.name, + "configured": True, + "state": state, + "latency_ms": round((time.perf_counter() - started) * 1000), + } + + def fetch(self, dataset: str, params: dict[str, Any]) -> list[dict[str, Any]]: + if dataset in {"indexes_quotes", "index_quotes"}: + return self.fetch_indices() + raise AdapterError(f"{self.name} unsupported dataset: {dataset}") + + def normalize(self, dataset: str, rows: list[dict[str, Any]]) -> list[dict[str, Any]]: + return list(rows) + + def fetch_indices(self) -> list[dict[str, Any]]: + request = urllib.request.Request( + TENCENT_INDEX_URL, + headers={"User-Agent": BROWSER_UA, "Referer": "https://gu.qq.com/"}, + method="GET", + ) + try: + with urllib.request.urlopen(request, timeout=self.timeout) as response: + raw = response.read().decode("gb18030", errors="ignore") + except Exception as exc: + raise AdapterError(f"tencent request failed: {exc}") from exc + result = [] + for line in raw.splitlines(): + if '="' not in line: + continue + fields = line.split('="', 1)[1].rsplit('";', 1)[0].split("~") + if len(fields) < 38: + continue + code = fields[2] + if code not in {"000001", "399001", "399006"}: + continue + try: + quote_time = datetime.strptime(fields[30], "%Y%m%d%H%M%S").astimezone() + except ValueError as exc: + raise AdapterError(f"Tencent invalid quote time for {code}") from exc + ts_code = "000001.SH" if code == "000001" else f"{code}.SZ" + result.append( + { + "ts_code": ts_code, + "code": code, + "name": fields[1] or code, + "price": round4(finite_number(fields[3])), + "pct_chg": round4(finite_number(fields[32])), + "change_amount": round4(finite_number(fields[31])), + "open": round4(finite_number(fields[5])), + "high": round4(finite_number(fields[33])), + "low": round4(finite_number(fields[34])), + "previous_close": round4(finite_number(fields[4])), + "amount": round4(finite_number(fields[37]) * 10000), + "quote_time_epoch": int(quote_time.timestamp()), + "quote_time": quote_time.isoformat(timespec="seconds"), + "source": "tencent_qt", + } + ) + if len(result) != 3: + raise AdapterError(f"Tencent returned {len(result)}/3 indices") + return result diff --git a/xiaobai-datahub/datahub/adapters/tushare.py b/xiaobai-datahub/datahub/adapters/tushare.py index e0c1c11..4362b1b 100644 --- a/xiaobai-datahub/datahub/adapters/tushare.py +++ b/xiaobai-datahub/datahub/adapters/tushare.py @@ -11,8 +11,12 @@ from datahub.normalize import ( normalize_auction, normalize_calendar, normalize_daily, + normalize_dragon_tiger, normalize_index_daily, + normalize_limit_event, normalize_moneyflow, + normalize_popularity, + normalize_sector_daily, normalize_stock, normalize_valuation, ) @@ -31,6 +35,21 @@ TUSHARE_FIELDS = { "buy_lg_amount,sell_lg_amount,buy_elg_amount,sell_elg_amount,net_mf_amount" ), "stk_auction": "ts_code,trade_date,vol,price,amount,pre_close,turnover_rate,volume_ratio,float_share", + "limit_list_d": ( + "trade_date,ts_code,industry,name,close,pct_chg,amount,limit_amount," + "float_mv,total_mv,turnover_ratio,fd_amount,first_time,last_time," + "open_times,up_stat,limit_times,limit_type" + ), + "ths_hot": "ts_code,ts_name,hot,rank,pct_change,current_price,concept,data_type,trade_date", + "dc_hot": "ts_code,ts_name,rank,pct_change,current_price,hot,concept,data_type,trade_date", + "hm_detail": "trade_date,ts_code,ts_name,buy_amount,sell_amount,net_amount,hm_name,hm_orgs,tag", + "hm_list": "name,desc,orgs", + "top_list": "trade_date,ts_code,name,pct_change,reason", + "top_inst": "trade_date,ts_code,exalter,buy,buy_rate,sell,sell_rate,net_buy,side,reason", + "ths_index": "ts_code,name,count,exchange,list_date,type", + "ths_daily": "ts_code,trade_date,open,high,low,close,pre_close,pct_change,vol,turnover_rate", + "dc_index": "ts_code,trade_date,name,open,high,low,close,pre_close,pct_change,vol,amount,turnover_rate", + "sw_daily": "ts_code,trade_date,name,open,high,low,close,pct_change,vol,amount", } DATASET_API = { @@ -42,12 +61,15 @@ DATASET_API = { "index_daily": "index_daily", "moneyflow": "moneyflow", "auction": "stk_auction", + "limit_events": "limit_list_d", + "popularity": "ths_hot", + "dragon_tiger": "hm_detail", + "sector_daily": "ths_daily", } -# Website actual index usage: market cards / 90-day charts (SH/SZ/CYB) plus -# screener 沪深300 benchmark (lookback up to 260 trading days). WEBSITE_INDEX_CODES = ("000001.SH", "399001.SZ", "399006.SZ", "000300.SH") DEFAULT_INDEX_CODES = WEBSITE_INDEX_CODES +LIMIT_TYPES = ("U", "D", "Z") class TushareAdapter(MarketAdapter): @@ -85,6 +107,14 @@ class TushareAdapter(MarketAdapter): } def fetch(self, dataset: str, params: dict[str, Any]) -> list[dict[str, Any]]: + if dataset == "limit_events": + return self.fetch_limit_events(str(params.get("trade_date") or "")) + if dataset == "popularity": + return self.fetch_popularity(str(params.get("trade_date") or "")) + if dataset == "dragon_tiger": + return self.fetch_dragon_tiger(str(params.get("trade_date") or "")) + if dataset == "sector_daily": + return self.fetch_sector_daily(str(params.get("trade_date") or "")) api_name = DATASET_API.get(dataset, dataset) fields = TUSHARE_FIELDS.get(api_name, "") query_params = dict(params) @@ -93,10 +123,67 @@ class TushareAdapter(MarketAdapter): if api_name == "trade_cal" and "exchange" not in query_params: query_params["exchange"] = "SSE" if api_name == "index_daily" and "ts_code" not in query_params: - # Caller typically loops codes; a missing code would pull nothing useful. query_params.setdefault("ts_code", DEFAULT_INDEX_CODES[0]) return self._query(api_name, query_params, fields) + def fetch_limit_events(self, trade_date: str) -> list[dict[str, Any]]: + rows: list[dict[str, Any]] = [] + for limit_type in LIMIT_TYPES: + part = self._query( + "limit_list_d", + {"trade_date": trade_date, "limit_type": limit_type}, + TUSHARE_FIELDS["limit_list_d"], + ) + for row in part: + row = dict(row) + row.setdefault("limit_type", limit_type) + rows.append(row) + return rows + + def fetch_popularity(self, trade_date: str) -> list[dict[str, Any]]: + rows: list[dict[str, Any]] = [] + for api_name, source in (("ths_hot", "ths"), ("dc_hot", "dc")): + for row in self._query(api_name, {"trade_date": trade_date}, TUSHARE_FIELDS[api_name]): + item = dict(row) + item["source"] = source + item.setdefault("trade_date", trade_date) + rows.append(item) + return rows + + def fetch_dragon_tiger(self, trade_date: str) -> list[dict[str, Any]]: + details = self._query("hm_detail", {"trade_date": trade_date}, TUSHARE_FIELDS["hm_detail"]) + top_rows = self._query("top_list", {"trade_date": trade_date}, TUSHARE_FIELDS["top_list"]) + context = { + str(row.get("ts_code") or ""): row + for row in top_rows + if str(row.get("ts_code") or "") + } + rows: list[dict[str, Any]] = [] + for row in details: + item = dict(row) + stock = context.get(str(item.get("ts_code") or ""), {}) + if item.get("pct_change") is None and stock.get("pct_change") is not None: + item["pct_change"] = stock.get("pct_change") + if not item.get("reason") and stock.get("reason"): + item["reason"] = stock.get("reason") + if not item.get("ts_name") and stock.get("name"): + item["ts_name"] = stock.get("name") + rows.append(item) + return rows + + def fetch_sector_daily(self, trade_date: str) -> list[dict[str, Any]]: + rows: list[dict[str, Any]] = [] + for api_name, family in (("ths_daily", "ths"), ("dc_index", "dc"), ("sw_daily", "sw")): + try: + part = self._query(api_name, {"trade_date": trade_date}, TUSHARE_FIELDS[api_name]) + except AdapterError: + part = [] + for row in part: + item = dict(row) + item["family"] = family + rows.append(item) + return rows + def fetch_index_daily(self, trade_date: str, codes: tuple[str, ...] = DEFAULT_INDEX_CODES) -> list[dict[str, Any]]: rows: list[dict[str, Any]] = [] for ts_code in codes: @@ -104,6 +191,17 @@ class TushareAdapter(MarketAdapter): return rows def normalize(self, dataset: str, rows: list[dict[str, Any]]) -> list[dict[str, Any]]: + if dataset in {"limit_events", "limit_list_d"}: + return [normalize_limit_event(row) for row in rows] + if dataset == "popularity": + return [normalize_popularity(row, source=str(row.get("source") or "")) for row in rows] + if dataset == "dragon_tiger": + return [normalize_dragon_tiger(row) for row in rows] + if dataset == "sector_daily": + return [ + normalize_sector_daily(row, family=str(row.get("family") or "ths")) + for row in rows + ] mapping = { "calendar": normalize_calendar, "trade_cal": normalize_calendar, @@ -148,12 +246,11 @@ class TushareAdapter(MarketAdapter): try: with urllib.request.urlopen(request, timeout=self.timeout) as response: result = json.loads(response.read().decode("utf-8")) - except json.JSONDecodeError: - raise AdapterError("Tushare returned invalid json") from None - except (urllib.error.URLError, TimeoutError) as exc: - raise AdapterError(f"Tushare request failed: {exc}") from exc - if result.get("code") != 0: - raise AdapterError(result.get("msg") or "Tushare returned an unknown error") + except (urllib.error.URLError, TimeoutError, json.JSONDecodeError) as exc: + raise AdapterError(f"Tushare 请求失败: {exc}") from exc + if result.get("code") not in (0, "0", None): + raise AdapterError(str(result.get("msg") or f"Tushare error {result.get('code')}")) data = result.get("data") or {} - columns = data.get("fields") or [] - return [dict(zip(columns, item)) for item in data.get("items") or []] + items = data.get("items") or [] + fields_list = data.get("fields") or (fields.split(",") if fields else []) + return [dict(zip(fields_list, item)) for item in items] diff --git a/xiaobai-datahub/datahub/datasets_ext.py b/xiaobai-datahub/datahub/datasets_ext.py new file mode 100644 index 0000000..08bced5 --- /dev/null +++ b/xiaobai-datahub/datahub/datasets_ext.py @@ -0,0 +1,180 @@ +"""Extended EOD datasets beyond the first-batch A/B release groups. + +These publish independently (soft): a failure here must not block daily/valuation +release. Scheduler runs them after the core EOD window. +""" + +from __future__ import annotations + +from typing import Any + +# Independent soft datasets (not part of A/B atomic groups). +EXTENDED_SOFT_DATASETS = { + "limit_events", + "popularity", + "dragon_tiger", + "sector_daily", +} + +EXTENDED_SCHEMA = """ +CREATE TABLE IF NOT EXISTS eod_limit_events ( + ts_code TEXT NOT NULL, trade_date TEXT NOT NULL, limit_type TEXT NOT NULL, + name TEXT, industry TEXT, close REAL, pct_chg REAL, amount REAL, + limit_amount REAL, float_mv REAL, total_mv REAL, turnover_ratio REAL, + fd_amount REAL, first_time TEXT, last_time TEXT, + open_times INTEGER, up_stat TEXT, limit_times INTEGER, + batch_id TEXT NOT NULL, + PRIMARY KEY (ts_code, trade_date, limit_type, batch_id) +) WITHOUT ROWID; + +CREATE TABLE IF NOT EXISTS staging_limit_events ( + ts_code TEXT NOT NULL, trade_date TEXT NOT NULL, limit_type TEXT NOT NULL, batch_id TEXT NOT NULL, + name TEXT, industry TEXT, close REAL, pct_chg REAL, amount REAL, + limit_amount REAL, float_mv REAL, total_mv REAL, turnover_ratio REAL, + fd_amount REAL, first_time TEXT, last_time TEXT, + open_times INTEGER, up_stat TEXT, limit_times INTEGER, + PRIMARY KEY (batch_id, ts_code, trade_date, limit_type) +); + +CREATE TABLE IF NOT EXISTS eod_popularity ( + ts_code TEXT NOT NULL, trade_date TEXT NOT NULL, source TEXT NOT NULL, + ts_name TEXT, rank INTEGER, pct_change REAL, current_price REAL, + hot REAL, concept TEXT, data_type TEXT, + batch_id TEXT NOT NULL, + PRIMARY KEY (ts_code, trade_date, source, batch_id) +) WITHOUT ROWID; + +CREATE TABLE IF NOT EXISTS staging_popularity ( + ts_code TEXT NOT NULL, trade_date TEXT NOT NULL, source TEXT NOT NULL, batch_id TEXT NOT NULL, + ts_name TEXT, rank INTEGER, pct_change REAL, current_price REAL, + hot REAL, concept TEXT, data_type TEXT, + PRIMARY KEY (batch_id, ts_code, trade_date, source) +); + +CREATE TABLE IF NOT EXISTS eod_dragon_tiger ( + ts_code TEXT NOT NULL, trade_date TEXT NOT NULL, hm_name TEXT NOT NULL, + ts_name TEXT, buy_amount REAL, sell_amount REAL, net_amount REAL, + hm_orgs TEXT, tag TEXT, pct_change REAL, reason TEXT, + batch_id TEXT NOT NULL, + PRIMARY KEY (ts_code, trade_date, hm_name, batch_id) +) WITHOUT ROWID; + +CREATE TABLE IF NOT EXISTS staging_dragon_tiger ( + ts_code TEXT NOT NULL, trade_date TEXT NOT NULL, hm_name TEXT NOT NULL, batch_id TEXT NOT NULL, + ts_name TEXT, buy_amount REAL, sell_amount REAL, net_amount REAL, + hm_orgs TEXT, tag TEXT, pct_change REAL, reason TEXT, + PRIMARY KEY (batch_id, ts_code, trade_date, hm_name) +); + +CREATE TABLE IF NOT EXISTS eod_sector_daily ( + ts_code TEXT NOT NULL, trade_date TEXT NOT NULL, family TEXT NOT NULL, + name TEXT, open REAL, high REAL, low REAL, close REAL, pre_close REAL, + pct_change REAL, vol REAL, turnover_rate REAL, amount REAL, + batch_id TEXT NOT NULL, + PRIMARY KEY (ts_code, trade_date, family, batch_id) +) WITHOUT ROWID; + +CREATE TABLE IF NOT EXISTS staging_sector_daily ( + ts_code TEXT NOT NULL, trade_date TEXT NOT NULL, family TEXT NOT NULL, batch_id TEXT NOT NULL, + name TEXT, open REAL, high REAL, low REAL, close REAL, pre_close REAL, + pct_change REAL, vol REAL, turnover_rate REAL, amount REAL, + PRIMARY KEY (batch_id, ts_code, trade_date, family) +); + +CREATE TABLE IF NOT EXISTS sector_master ( + ts_code TEXT PRIMARY KEY, + name TEXT, + family TEXT NOT NULL, + exchange TEXT, + list_date TEXT, + member_count INTEGER, + type TEXT, + updated_at TEXT NOT NULL +); + +CREATE INDEX IF NOT EXISTS idx_eod_limit_date ON eod_limit_events(trade_date, batch_id); +CREATE INDEX IF NOT EXISTS idx_eod_pop_date ON eod_popularity(trade_date, batch_id); +CREATE INDEX IF NOT EXISTS idx_eod_lhb_date ON eod_dragon_tiger(trade_date, batch_id); +CREATE INDEX IF NOT EXISTS idx_eod_sector_date ON eod_sector_daily(trade_date, family, batch_id); +""" + +EXTENDED_DATASET_TABLES = { + "limit_events": ("eod_limit_events", "staging_limit_events"), + "popularity": ("eod_popularity", "staging_popularity"), + "dragon_tiger": ("eod_dragon_tiger", "staging_dragon_tiger"), + "sector_daily": ("eod_sector_daily", "staging_sector_daily"), +} + +EXTENDED_STAGING_INSERT: dict[str, tuple[str, Any]] = { + "limit_events": ( + "INSERT INTO staging_limit_events(" + "ts_code,trade_date,limit_type,batch_id,name,industry,close,pct_chg,amount," + "limit_amount,float_mv,total_mv,turnover_ratio,fd_amount,first_time,last_time," + "open_times,up_stat,limit_times) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)", + lambda r, b: ( + r["ts_code"], r["trade_date"], r["limit_type"], b, + r.get("name"), r.get("industry"), r.get("close"), r.get("pct_chg"), r.get("amount"), + r.get("limit_amount"), r.get("float_mv"), r.get("total_mv"), r.get("turnover_ratio"), + r.get("fd_amount"), r.get("first_time"), r.get("last_time"), + r.get("open_times"), r.get("up_stat"), r.get("limit_times"), + ), + ), + "popularity": ( + "INSERT INTO staging_popularity(" + "ts_code,trade_date,source,batch_id,ts_name,rank,pct_change,current_price,hot,concept,data_type) " + "VALUES (?,?,?,?,?,?,?,?,?,?,?)", + lambda r, b: ( + r["ts_code"], r["trade_date"], r["source"], b, + r.get("ts_name"), r.get("rank"), r.get("pct_change"), r.get("current_price"), + r.get("hot"), r.get("concept"), r.get("data_type"), + ), + ), + "dragon_tiger": ( + "INSERT INTO staging_dragon_tiger(" + "ts_code,trade_date,hm_name,batch_id,ts_name,buy_amount,sell_amount,net_amount," + "hm_orgs,tag,pct_change,reason) VALUES (?,?,?,?,?,?,?,?,?,?,?,?)", + lambda r, b: ( + r["ts_code"], r["trade_date"], r["hm_name"], b, + r.get("ts_name"), r.get("buy_amount"), r.get("sell_amount"), r.get("net_amount"), + r.get("hm_orgs"), r.get("tag"), r.get("pct_change"), r.get("reason"), + ), + ), + "sector_daily": ( + "INSERT INTO staging_sector_daily(" + "ts_code,trade_date,family,batch_id,name,open,high,low,close,pre_close," + "pct_change,vol,turnover_rate,amount) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?)", + lambda r, b: ( + r["ts_code"], r["trade_date"], r["family"], b, + r.get("name"), r.get("open"), r.get("high"), r.get("low"), r.get("close"), + r.get("pre_close"), r.get("pct_change"), r.get("vol"), r.get("turnover_rate"), + r.get("amount"), + ), + ), +} + +EXTENDED_EOD_COPY = { + "limit_events": ( + "INSERT OR REPLACE INTO eod_limit_events " + "SELECT ts_code,trade_date,limit_type,name,industry,close,pct_chg,amount," + "limit_amount,float_mv,total_mv,turnover_ratio,fd_amount,first_time,last_time," + "open_times,up_stat,limit_times,batch_id " + "FROM staging_limit_events WHERE batch_id = ?" + ), + "popularity": ( + "INSERT OR REPLACE INTO eod_popularity " + "SELECT ts_code,trade_date,source,ts_name,rank,pct_change,current_price,hot,concept,data_type,batch_id " + "FROM staging_popularity WHERE batch_id = ?" + ), + "dragon_tiger": ( + "INSERT OR REPLACE INTO eod_dragon_tiger " + "SELECT ts_code,trade_date,hm_name,ts_name,buy_amount,sell_amount,net_amount," + "hm_orgs,tag,pct_change,reason,batch_id " + "FROM staging_dragon_tiger WHERE batch_id = ?" + ), + "sector_daily": ( + "INSERT OR REPLACE INTO eod_sector_daily " + "SELECT ts_code,trade_date,family,name,open,high,low,close,pre_close," + "pct_change,vol,turnover_rate,amount,batch_id " + "FROM staging_sector_daily WHERE batch_id = ?" + ), +} diff --git a/xiaobai-datahub/datahub/db.py b/xiaobai-datahub/datahub/db.py index bcea373..65cb88a 100644 --- a/xiaobai-datahub/datahub/db.py +++ b/xiaobai-datahub/datahub/db.py @@ -7,9 +7,10 @@ from contextlib import contextmanager from pathlib import Path from typing import Any +from datahub.datasets_ext import EXTENDED_DATASET_TABLES, EXTENDED_SCHEMA from datahub.timeutil import isoformat -SCHEMA = """ +_BASE_SCHEMA = """ CREATE TABLE IF NOT EXISTS schema_migrations ( version INTEGER PRIMARY KEY, applied_at TEXT NOT NULL @@ -282,6 +283,8 @@ 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 + DATASET_TABLES = { "daily": ("eod_bars", "staging_bars"), "valuation": ("eod_valuation", "staging_valuation"), @@ -289,6 +292,7 @@ DATASET_TABLES = { "auction": ("eod_auction", "staging_auction"), "index_daily": ("eod_index_bars", "staging_index_bars"), "stocks": ("eod_stocks", "staging_stocks"), + **EXTENDED_DATASET_TABLES, } diff --git a/xiaobai-datahub/datahub/normalize.py b/xiaobai-datahub/datahub/normalize.py index bf4c5f4..29eb3f8 100644 --- a/xiaobai-datahub/datahub/normalize.py +++ b/xiaobai-datahub/datahub/normalize.py @@ -156,6 +156,95 @@ def normalize_stock(row: dict[str, Any]) -> dict[str, Any]: } +def normalize_limit_event(row: dict[str, Any]) -> dict[str, Any]: + """limit_list_d. float_mv/total_mv/limit_amount are 万元 → yuan; amount/fd_amount already yuan.""" + return { + "ts_code": _code(row.get("ts_code")), + "trade_date": _date(row.get("trade_date")), + "limit_type": str(row.get("limit_type") or "").strip().upper() or "U", + "name": str(row.get("name") or "").strip() or None, + "industry": str(row.get("industry") or "").strip() or None, + "close": round4(finite_number(row.get("close"))), + "pct_chg": round4(finite_number(row.get("pct_chg"))), + "amount": round4(finite_number(row.get("amount"))), + "limit_amount": round4(_scale(row.get("limit_amount"), AMOUNT_WAN_YUAN)), + "float_mv": round4(_scale(row.get("float_mv"), AMOUNT_WAN_YUAN)), + "total_mv": round4(_scale(row.get("total_mv"), AMOUNT_WAN_YUAN)), + "turnover_ratio": round4(finite_number(row.get("turnover_ratio"))), + "fd_amount": round4(finite_number(row.get("fd_amount"))), + "first_time": str(row.get("first_time") or "").strip() or None, + "last_time": str(row.get("last_time") or "").strip() or None, + "open_times": _optional_int(row.get("open_times")), + "up_stat": str(row.get("up_stat") or "").strip() or None, + "limit_times": _optional_int(row.get("limit_times")), + } + + +def normalize_popularity(row: dict[str, Any], source: str = "") -> dict[str, Any]: + src = str(source or row.get("source") or "").strip().lower() or "ths" + return { + "ts_code": _code(row.get("ts_code")), + "trade_date": _date(row.get("trade_date")), + "source": src, + "ts_name": str(row.get("ts_name") or row.get("name") or "").strip() or None, + "rank": _optional_int(row.get("rank")), + "pct_change": round4( + finite_number(row.get("pct_change") if row.get("pct_change") is not None else row.get("pct_chg")) + ), + "current_price": round4(finite_number(row.get("current_price") or row.get("price"))), + "hot": round4(finite_number(row.get("hot"))), + "concept": str(row.get("concept") or "").strip() or None, + "data_type": str(row.get("data_type") or "").strip() or None, + } + + +def normalize_dragon_tiger(row: dict[str, Any]) -> dict[str, Any]: + """hm_detail amounts are 万元 → yuan.""" + return { + "ts_code": _code(row.get("ts_code")), + "trade_date": _date(row.get("trade_date")), + "hm_name": str(row.get("hm_name") or "未命名游资").strip() or "未命名游资", + "ts_name": str(row.get("ts_name") or row.get("name") or "").strip() or None, + "buy_amount": round4(_scale(row.get("buy_amount"), AMOUNT_WAN_YUAN)), + "sell_amount": round4(_scale(row.get("sell_amount"), AMOUNT_WAN_YUAN)), + "net_amount": round4(_scale(row.get("net_amount"), AMOUNT_WAN_YUAN)), + "hm_orgs": str(row.get("hm_orgs") or "").strip() or None, + "tag": str(row.get("tag") or "").strip() or None, + "pct_change": round4(finite_number(row.get("pct_change"))), + "reason": str(row.get("reason") or "").strip() or None, + } + + +def normalize_sector_daily(row: dict[str, Any], family: str = "ths") -> dict[str, Any]: + fam = str(family or row.get("family") or "ths").strip().lower() + return { + "ts_code": _code(row.get("ts_code")), + "trade_date": _date(row.get("trade_date")), + "family": fam, + "name": str(row.get("name") or "").strip() or None, + "open": round4(finite_number(row.get("open"))), + "high": round4(finite_number(row.get("high"))), + "low": round4(finite_number(row.get("low"))), + "close": round4(finite_number(row.get("close"))), + "pre_close": round4(finite_number(row.get("pre_close"))), + "pct_change": round4( + finite_number(row.get("pct_change") if row.get("pct_change") is not None else row.get("pct_chg")) + ), + "vol": round4(finite_number(row.get("vol"))), + "turnover_rate": round4(finite_number(row.get("turnover_rate"))), + "amount": round4(finite_number(row.get("amount"))), + } + + +def _optional_int(value: Any) -> int | None: + if value in (None, ""): + return None + try: + return int(float(value)) + except (TypeError, ValueError): + return None + + def apply_qfq(price: float | None, factor: float | None, latest_factor: float | None) -> float | None: if price is None: return None @@ -184,6 +273,11 @@ NORMALIZERS = { "calendar": normalize_calendar, "stock_basic": normalize_stock, "stocks": normalize_stock, + "limit_events": normalize_limit_event, + "limit_list_d": normalize_limit_event, + "popularity": normalize_popularity, + "dragon_tiger": normalize_dragon_tiger, + "sector_daily": normalize_sector_daily, } diff --git a/xiaobai-datahub/datahub/pipeline.py b/xiaobai-datahub/datahub/pipeline.py index c6c5d8c..ec63558 100644 --- a/xiaobai-datahub/datahub/pipeline.py +++ b/xiaobai-datahub/datahub/pipeline.py @@ -9,6 +9,11 @@ from typing import Any from datahub.adapters.base import AdapterError from datahub.adapters.tushare import DEFAULT_INDEX_CODES, WEBSITE_INDEX_CODES, TushareAdapter +from datahub.datasets_ext import ( + EXTENDED_EOD_COPY, + EXTENDED_SOFT_DATASETS, + EXTENDED_STAGING_INSERT, +) from datahub.db import DATASET_TABLES, HubDB from datahub.governance.circuit import CircuitBreaker from datahub.governance.ratelimit import TokenBucket @@ -21,12 +26,16 @@ from datahub.timeutil import add_days, isoformat, now_shanghai, yyyymmdd LOGGER = get_logger() HARD_DATASETS = {"daily", "valuation", "index_daily"} -SOFT_DATASETS = {"moneyflow", "auction"} -OFFICIAL_DATASETS = HARD_DATASETS | SOFT_DATASETS +SOFT_DATASETS = {"moneyflow", "auction"} | EXTENDED_SOFT_DATASETS +OFFICIAL_DATASETS = HARD_DATASETS | {"moneyflow", "auction"} # A/B retry scope unchanged STOCKS_DATASET = "stocks" STOCK_SNAPSHOT_FIELDS = ("ts_code", "symbol", "name", "area", "industry", "market", "list_status", "list_date") EOD_A_DATASETS = ("daily", "valuation", "moneyflow", "auction") EOD_B_DATASETS = ("index_daily",) +EOD_C_DATASETS = ("limit_events",) +EOD_D_DATASETS = ("dragon_tiger",) +EOD_E_DATASETS = ("sector_daily",) +EOD_F_DATASETS = ("popularity",) EMPTY_BATCH_ERROR = "empty official batch: 0 valid rows" STAGING_INSERT = { @@ -80,6 +89,7 @@ STAGING_INSERT = { r.get("close"), r.get("pct_chg"), r.get("volume"), r.get("amount"), ), ), + **EXTENDED_STAGING_INSERT, } EOD_COPY = { @@ -114,6 +124,7 @@ EOD_COPY = { "SELECT ts_code,trade_date,open,high,low,close,pct_chg,volume,amount,batch_id " "FROM staging_index_bars WHERE batch_id = ?" ), + **EXTENDED_EOD_COPY, } @@ -672,19 +683,69 @@ class Pipeline: def run_eod_batch_b(self, trade_date: str, force: bool = False) -> dict[str, Any]: return self.run_release_group(EOD_B_DATASETS, trade_date, force=force) + def run_extended_soft(self, datasets: tuple[str, ...], trade_date: str, force: bool = False) -> dict[str, Any]: + """Publish extended soft datasets independently (not A/B atomic).""" + results: dict[str, Any] = {} + day = yyyymmdd(trade_date) + for dataset in datasets: + if not force and self.active_batch(dataset, day): + results[dataset] = { + "dataset": dataset, + "trade_date": day, + "state": "skipped", + "reason": "already_published", + } + continue + try: + rows = self._fetch_dataset(dataset, day) + if not rows and dataset in {"popularity", "dragon_tiger"}: + results[dataset] = { + "dataset": dataset, + "trade_date": day, + "state": "skipped", + "reason": "upstream_empty", + "rows": 0, + } + continue + results[dataset] = self.run_dataset(dataset, day, prepared_rows=rows) + except Exception as exc: + results[dataset] = { + "dataset": dataset, + "trade_date": day, + "state": "failed", + "error": str(exc), + } + LOGGER.exception("extended soft publish failed dataset=%s date=%s", dataset, day) + return results + + def run_eod_batch_c(self, trade_date: str, force: bool = False) -> dict[str, Any]: + return self.run_extended_soft(EOD_C_DATASETS, trade_date, force=force) + + def run_eod_batch_d(self, trade_date: str, force: bool = False) -> dict[str, Any]: + return self.run_extended_soft(EOD_D_DATASETS, trade_date, force=force) + + def run_eod_batch_e(self, trade_date: str, force: bool = False) -> dict[str, Any]: + return self.run_extended_soft(EOD_E_DATASETS, trade_date, force=force) + + def run_eod_batch_f(self, trade_date: str, force: bool = False) -> dict[str, Any]: + return self.run_extended_soft(EOD_F_DATASETS, trade_date, force=force) + def force_republish_boundary(self, dataset: str, trade_date: str) -> dict[str, Any]: """Force-republish the full A/B consistency boundary that owns ``dataset``. CLI ``eod-refresh --force`` and admin manual backfill must not publish a single official member alone — that would mix old and new batches inside the same trade date. Naming any A-group member (or stocks) rebuilds the - whole A group; naming ``index_daily`` rebuilds B. + whole A group; naming ``index_daily`` rebuilds B. Extended soft datasets + republish independently. """ name = str(dataset or "").strip() if name in EOD_A_DATASETS or name == STOCKS_DATASET: return self.run_eod_batch_a(trade_date, force=True) if name in EOD_B_DATASETS: return self.run_eod_batch_b(trade_date, force=True) + if name in EXTENDED_SOFT_DATASETS: + return self.run_extended_soft((name,), trade_date, force=True) raise ValueError(f"dataset is not part of an EOD release boundary: {dataset}") def run_release_group( @@ -1022,7 +1083,16 @@ class Pipeline: ) listed_n = int((listed or {}).get("n") or 0) row_n = len(rows) - keys = [(row.get("ts_code"), row.get("trade_date")) for row in rows] + if dataset == "limit_events": + keys = [(row.get("ts_code"), row.get("trade_date"), row.get("limit_type")) for row in rows] + elif dataset == "popularity": + keys = [(row.get("ts_code"), row.get("trade_date"), row.get("source")) for row in rows] + elif dataset == "dragon_tiger": + keys = [(row.get("ts_code"), row.get("trade_date"), row.get("hm_name")) for row in rows] + elif dataset == "sector_daily": + keys = [(row.get("ts_code"), row.get("trade_date"), row.get("family")) for row in rows] + else: + keys = [(row.get("ts_code"), row.get("trade_date")) for row in rows] dup = row_n - len(set(keys)) if dup: errors.append(f"duplicate keys: {dup}") @@ -1043,7 +1113,8 @@ class Pipeline: errors.append(EMPTY_BATCH_ERROR) field_report = self._field_gate(dataset, trade_date, rows, errors) if dataset in SOFT_DATASETS: - hard_fail = bool(dup or bad_date or empty) + allow_empty = dataset in {"popularity", "dragon_tiger", "moneyflow", "auction"} + hard_fail = bool(dup or bad_date or (empty and not allow_empty)) else: hard_fail = bool(errors) and (dataset in HARD_DATASETS or dataset == STOCKS_DATASET) report = { diff --git a/xiaobai-datahub/datahub/realtime_serve.py b/xiaobai-datahub/datahub/realtime_serve.py new file mode 100644 index 0000000..ed883f1 --- /dev/null +++ b/xiaobai-datahub/datahub/realtime_serve.py @@ -0,0 +1,188 @@ +"""Provisional (盘中观察) serving: quotes, index quotes, intraday points. + +Free sources only. Never writes official eod_* tables. Uses rt_cache + LKG. +""" + +from __future__ import annotations + +import json +import time +from datetime import datetime +from typing import Any + +from datahub.adapters.eastmoney import EastmoneyAdapter +from datahub.adapters.tencent import TencentAdapter +from datahub.codes import resolve_code +from datahub.db import HubDB +from datahub.timeutil import isoformat, now_shanghai, yyyymmdd + +QUOTE_TTL = 60 +INDEX_TTL = 60 +INTRADAY_TTL = 20 + + +class RealtimeApiError(RuntimeError): + def __init__(self, code: str, message: str) -> None: + super().__init__(message) + self.code = code + self.message = message + + +def _envelope(data: Any, meta: dict[str, Any]) -> dict[str, Any]: + from datahub import SCHEMA_VERSION + + return {"schema_version": SCHEMA_VERSION, "data": data, "meta": meta} + + +def fetch_index_quotes(db: HubDB) -> dict[str, Any]: + cache_key = "indexes:quotes" + cached = _read_cache(db, cache_key) + if cached is not None: + return cached + eastmoney = EastmoneyAdapter() + try: + rows = eastmoney.fetch_indices() + source = "eastmoney:ulist" + except Exception: + rows = TencentAdapter().fetch_indices() + source = "tencent:qt" + if len(rows) < 3: + raise RealtimeApiError("SOURCE_UNAVAILABLE", "index quotes incomplete") + payload = _envelope( + rows, + { + "tier": "provisional", + "trade_date": yyyymmdd(now_shanghai()), + "source": source, + "stale": False, + "staleness_seconds": 0, + "published_at": isoformat(now_shanghai()), + }, + ) + _write_cache(db, cache_key, payload, INDEX_TTL, source) + return payload + + +def fetch_quotes(db: HubDB, codes: list[str]) -> dict[str, Any]: + if not codes: + raise RealtimeApiError("INVALID_ARGUMENT", "codes is required") + resolved: list[str] = [] + for code in codes[:60]: + item = resolve_code(db, code) or _guess_ts_code(code) + if item: + resolved.append(item) + if not resolved: + raise RealtimeApiError("INVALID_ARGUMENT", "no resolvable codes") + cache_key = "quotes:" + ",".join(sorted(resolved)) + cached = _read_cache(db, cache_key) + if cached is not None: + return cached + adapter = EastmoneyAdapter() + try: + rows = adapter.fetch_quotes(resolved) + source = "eastmoney:clist" + except Exception as exc: + raise RealtimeApiError("SOURCE_UNAVAILABLE", f"quotes unavailable: {exc}") from exc + payload = _envelope( + rows, + { + "tier": "provisional", + "trade_date": yyyymmdd(now_shanghai()), + "source": source, + "stale": False, + "staleness_seconds": 0, + "published_at": isoformat(now_shanghai()), + }, + ) + _write_cache(db, cache_key, payload, QUOTE_TTL, source) + return payload + + +def fetch_intraday(db: HubDB, code: str, date: str = "") -> dict[str, Any]: + ts_code = resolve_code(db, code) or _guess_ts_code(code) + if not ts_code: + raise RealtimeApiError("INVALID_ARGUMENT", f"ambiguous code: {code}") + cache_key = f"intraday:{ts_code}:{date or 'today'}" + cached = _read_cache(db, cache_key) + if cached is not None: + return cached + adapter = EastmoneyAdapter() + try: + payload_data = adapter.fetch_intraday(ts_code) + source = "eastmoney:trends2" + except Exception as exc: + raise RealtimeApiError("SOURCE_UNAVAILABLE", f"intraday unavailable: {exc}") from exc + payload = _envelope( + payload_data, + { + "tier": "provisional", + "trade_date": yyyymmdd(payload_data.get("trade_date") or date or now_shanghai()), + "source": source, + "stale": False, + "staleness_seconds": 0, + "published_at": isoformat(now_shanghai()), + }, + ) + _write_cache(db, cache_key, payload, INTRADAY_TTL, source) + return payload + + +def _guess_ts_code(code: str) -> str | None: + raw = str(code or "").strip().upper() + if "." in raw: + return raw + if len(raw) == 6 and raw.isdigit(): + if raw.startswith(("5", "6", "9")): + return f"{raw}.SH" + return f"{raw}.SZ" + return None + + +def _read_cache(db: HubDB, cache_key: str) -> dict[str, Any] | None: + row = db.fetchone("SELECT * FROM rt_cache WHERE cache_key = ?", (cache_key,)) + if not row: + return None + expires = str(row.get("expires_at") or "") + now = isoformat(now_shanghai()) + if expires and expires < now: + return None + try: + payload = json.loads(row["payload"]) + except json.JSONDecodeError: + return None + if isinstance(payload, dict) and isinstance(payload.get("meta"), dict): + stored = str(row.get("stored_at") or "") + try: + age = max(0, int(time.time() - datetime.fromisoformat(stored).timestamp())) + except Exception: + age = 0 + payload["meta"]["staleness_seconds"] = age + payload["meta"]["stale"] = age > QUOTE_TTL + return payload + + +def _write_cache(db: HubDB, cache_key: str, payload: dict[str, Any], ttl: int, source: str) -> None: + from datetime import timedelta + + now = now_shanghai() + stored = isoformat(now) + expires = isoformat(now + timedelta(seconds=ttl)) + db.execute( + """ + INSERT INTO rt_cache(cache_key, payload, source, stored_at, expires_at) + VALUES (?,?,?,?,?) + ON CONFLICT(cache_key) DO UPDATE SET + payload=excluded.payload, source=excluded.source, + stored_at=excluded.stored_at, expires_at=excluded.expires_at + """, + (cache_key, json.dumps(payload, ensure_ascii=False), source, stored, expires), + ) + db.execute( + """ + INSERT INTO last_known_good(cache_key, payload, source, stored_at) + VALUES (?,?,?,?) + ON CONFLICT(cache_key) DO UPDATE SET + payload=excluded.payload, source=excluded.source, stored_at=excluded.stored_at + """, + (cache_key, json.dumps(payload, ensure_ascii=False), source, stored), + ) diff --git a/xiaobai-datahub/datahub/scheduler.py b/xiaobai-datahub/datahub/scheduler.py index 51e2cb3..d8530db 100644 --- a/xiaobai-datahub/datahub/scheduler.py +++ b/xiaobai-datahub/datahub/scheduler.py @@ -48,6 +48,10 @@ class Scheduler: "precheck": self._precheck, "eod_a": self._eod_a, "eod_b": self._eod_b, + "eod_c": self._eod_c, + "eod_d": self._eod_d, + "eod_e": self._eod_e, + "eod_f": self._eod_f, "eod_retry": self._eod_retry, "stocks_refresh": self._stocks_refresh, "cleanup": self._cleanup, @@ -87,6 +91,10 @@ class Scheduler: ("precheck", time(8, 45)), ("eod_a", time(15, 5)), ("eod_b", time(15, 10)), + ("eod_c", time(16, 40)), + ("eod_d", time(16, 45)), + ("eod_e", time(18, 5)), + ("eod_f", time(22, 40)), ("cleanup", time(0, 30)), ("backup", time(0, 40)), ] @@ -99,7 +107,7 @@ class Scheduler: key = (job_id, day, at.strftime("%H%M")) if key in self._fired: continue - if job_id in {"eod_a", "eod_b", "stocks_refresh"} and not open_day: + if job_id in {"eod_a", "eod_b", "eod_c", "eod_d", "eod_e", "eod_f", "stocks_refresh"} and not open_day: self._fired.add(key) continue self._fired.add(key) @@ -110,7 +118,7 @@ class Scheduler: try: self.run_job(job_id, day) except Exception: - if job_id not in {"eod_a", "eod_b", "stocks_refresh"}: + if job_id not in {"eod_a", "eod_b", "eod_c", "eod_d", "eod_e", "eod_f", "stocks_refresh"}: raise # Keep the tick alive; evening retries take over. LOGGER.exception("scheduled job %s failed for %s", job_id, day) @@ -317,6 +325,18 @@ class Scheduler: def _eod_b(self, trade_date: str) -> dict[str, Any]: return self.pipeline.run_eod_batch_b(trade_date) + def _eod_c(self, trade_date: str) -> dict[str, Any]: + return self.pipeline.run_eod_batch_c(trade_date) + + def _eod_d(self, trade_date: str) -> dict[str, Any]: + return self.pipeline.run_eod_batch_d(trade_date) + + def _eod_e(self, trade_date: str) -> dict[str, Any]: + return self.pipeline.run_eod_batch_e(trade_date) + + def _eod_f(self, trade_date: str) -> dict[str, Any]: + return self.pipeline.run_eod_batch_f(trade_date) + def _eod_retry(self, trade_date: str) -> dict[str, Any]: return self.pipeline.run_eod_missing(trade_date) diff --git a/xiaobai-datahub/datahub/serving.py b/xiaobai-datahub/datahub/serving.py index ff8eb71..cd36cd0 100644 --- a/xiaobai-datahub/datahub/serving.py +++ b/xiaobai-datahub/datahub/serving.py @@ -73,6 +73,20 @@ class V1API: return self.moneyflow(q) if path == "/v1/auction": return self.auction(q) + if path == "/v1/limit-events": + return self.limit_events(q) + if path == "/v1/popularity": + return self.popularity(q) + if path == "/v1/dragon-tiger": + return self.dragon_tiger(q) + if path == "/v1/sectors": + return self.sectors(q) + if path == "/v1/quotes/latest": + return self.quotes_latest(q) + if path == "/v1/indexes/quotes": + return self.index_quotes(q) + if path == "/v1/intraday/points": + return self.intraday_points(q) if path == "/v1/datasets/status": return self.dataset_status(q.get("date") or "") if path == "/v1/batches": @@ -197,9 +211,75 @@ class V1API: def auction(self, q: dict[str, str]) -> dict[str, Any]: return self._published_rows(dataset="auction", table="eod_auction", q=q, source="tushare:stk_auction") + def limit_events(self, q: dict[str, str]) -> dict[str, Any]: + return self._published_rows( + dataset="limit_events", + table="eod_limit_events", + q=q, + source="tushare:limit_list_d", + extra_filters={"limit_type": q.get("limit_type") or ""}, + ) + + def popularity(self, q: dict[str, str]) -> dict[str, Any]: + return self._published_rows( + dataset="popularity", + table="eod_popularity", + q=q, + source="tushare:ths_hot+dc_hot", + extra_filters={"source": q.get("source") or ""}, + ) + + def dragon_tiger(self, q: dict[str, str]) -> dict[str, Any]: + return self._published_rows( + dataset="dragon_tiger", + table="eod_dragon_tiger", + q=q, + source="tushare:hm_detail", + ) + + def sectors(self, q: dict[str, str]) -> dict[str, Any]: + return self._published_rows( + dataset="sector_daily", + table="eod_sector_daily", + q=q, + source="tushare:ths_daily+dc_index+sw_daily", + extra_filters={"family": q.get("family") or ""}, + ) + + def quotes_latest(self, q: dict[str, str]) -> dict[str, Any]: + from datahub.realtime_serve import RealtimeApiError, fetch_quotes + + codes = [item.strip() for item in str(q.get("codes") or "").split(",") if item.strip()] + try: + return fetch_quotes(self.db, codes) + except RealtimeApiError as exc: + raise ApiError(exc.code, exc.message) from exc + + def index_quotes(self, q: dict[str, str]) -> dict[str, Any]: + from datahub.realtime_serve import RealtimeApiError, fetch_index_quotes + + try: + return fetch_index_quotes(self.db) + except RealtimeApiError as exc: + raise ApiError(exc.code, exc.message) from exc + + def intraday_points(self, q: dict[str, str]) -> dict[str, Any]: + from datahub.realtime_serve import RealtimeApiError, fetch_intraday + + code = str(q.get("code") or "").strip() + if not code: + raise ApiError("INVALID_ARGUMENT", "code is required") + try: + return fetch_intraday(self.db, code, yyyymmdd(q.get("date") or "")) + except RealtimeApiError as exc: + raise ApiError(exc.code, exc.message) from exc + def dataset_status(self, date: str) -> dict[str, Any]: trade_date = yyyymmdd(date or now_shanghai()) - datasets = ("daily", "valuation", "moneyflow", "auction", "index_daily", "stocks") + datasets = ( + "daily", "valuation", "moneyflow", "auction", "index_daily", "stocks", + "limit_events", "popularity", "dragon_tiger", "sector_daily", + ) items = [] for dataset in datasets: pub = self.db.fetchone( @@ -244,6 +324,7 @@ class V1API: source: str, adjust: str = "none", default_code: str = "", + extra_filters: dict[str, str] | None = None, ) -> dict[str, Any]: trade_date = q.get("date") or q.get("trade_date") or "" code = q.get("code") or default_code @@ -264,6 +345,7 @@ class V1API: if resolved is None: raise ApiError("INVALID_ARGUMENT", f"ambiguous code: {code}") ts_code = resolved + filters = {key: value for key, value in (extra_filters or {}).items() if value} # For a range, use per-date published batch. Single-date is the common path. if start == end: pub = self.db.fetchone( @@ -282,6 +364,9 @@ class V1API: if ts_code: sql += " AND ts_code = ?" params.append(ts_code) + for key, value in filters.items(): + sql += f" AND {key} = ?" + params.append(value) sql += " ORDER BY ts_code LIMIT ? OFFSET ?" params.extend([limit, offset]) rows = [dict(row) for row in self.db.fetchall(sql, tuple(params))] @@ -317,6 +402,9 @@ class V1API: if ts_code: sql += " AND ts_code = ?" params.append(ts_code) + for key, value in filters.items(): + sql += f" AND {key} = ?" + params.append(value) sql += " ORDER BY ts_code" rows.extend(self.db.fetchall(sql, tuple(params))) sliced = rows[offset: offset + limit] diff --git a/xiaobai-datahub/tests/fixtures.py b/xiaobai-datahub/tests/fixtures.py index 78db5b4..a9f735d 100644 --- a/xiaobai-datahub/tests/fixtures.py +++ b/xiaobai-datahub/tests/fixtures.py @@ -45,6 +45,30 @@ RAW = { {"ts_code": "600000.SH", "trade_date": "20240902", "vol": 100, "price": 10.15, "amount": 1500000, "pre_close": 10.00, "turnover_rate": 0.1, "volume_ratio": 1.2, "float_share": 2000}, {"ts_code": "000001.SZ", "trade_date": "20240902", "vol": 80, "price": 11.05, "amount": 1200000, "pre_close": 11.10, "turnover_rate": 0.2, "volume_ratio": 0.9, "float_share": 1800}, ], + "limit_list_d": [ + {"trade_date": "20240902", "ts_code": "600000.SH", "industry": "银行", "name": "浦发银行", "close": 10.2, "pct_chg": 9.95, "amount": 1e8, "limit_amount": 5000, "float_mv": 800, "total_mv": 1000, "turnover_ratio": 5.0, "fd_amount": 2e7, "first_time": "09:30:01", "last_time": "14:55:00", "open_times": 0, "up_stat": "1/1", "limit_times": 1, "limit_type": "U"}, + ], + "ths_hot": [ + {"ts_code": "600000.SH", "ts_name": "浦发银行", "hot": 90.0, "rank": 1, "pct_change": 1.2, "current_price": 10.2, "concept": "银行", "data_type": "热股", "trade_date": "20240902"}, + ], + "dc_hot": [ + {"ts_code": "600000.SH", "ts_name": "浦发银行", "rank": 2, "pct_change": 1.2, "current_price": 10.2, "hot": 80.0, "concept": "银行", "data_type": "A股市场", "trade_date": "20240902"}, + ], + "hm_detail": [ + {"trade_date": "20240902", "ts_code": "600000.SH", "ts_name": "浦发银行", "buy_amount": 1000, "sell_amount": 200, "net_amount": 800, "hm_name": "测试游资", "hm_orgs": "某某营业部", "tag": "超买"}, + ], + "top_list": [ + {"trade_date": "20240902", "ts_code": "600000.SH", "name": "浦发银行", "pct_change": 9.95, "reason": "涨幅偏离值达7%"}, + ], + "ths_daily": [ + {"ts_code": "885811.TI", "trade_date": "20240902", "open": 1000, "high": 1010, "low": 990, "close": 1005, "pre_close": 995, "pct_change": 1.0, "vol": 100, "turnover_rate": 1.2}, + ], + "dc_index": [ + {"ts_code": "BK0475", "trade_date": "20240902", "name": "银行", "open": 100, "high": 101, "low": 99, "close": 100.5, "pre_close": 99.5, "pct_change": 1.0, "vol": 10, "amount": 1e8, "turnover_rate": 0.5}, + ], + "sw_daily": [ + {"ts_code": "801780.SI", "trade_date": "20240902", "name": "银行", "open": 2000, "high": 2010, "low": 1990, "close": 2005, "pct_change": 0.8, "vol": 50, "amount": 2e8}, + ], } @@ -66,4 +90,9 @@ def fake_transport(api_name: str, params: dict, fields: str): start = str(params.get("start_date") or "") end = str(params.get("end_date") or "99999999") return [row for row in RAW["trade_cal"] if start <= row["cal_date"] <= end] - return list(RAW.get(api_name) or []) + rows = list(RAW.get(api_name) or []) + if api_name == "limit_list_d": + limit_type = str(params.get("limit_type") or "") + if limit_type: + rows = [row for row in rows if str(row.get("limit_type") or "") == limit_type] + return rows diff --git a/xiaobai-datahub/tests/test_extended_eod.py b/xiaobai-datahub/tests/test_extended_eod.py new file mode 100644 index 0000000..e282616 --- /dev/null +++ b/xiaobai-datahub/tests/test_extended_eod.py @@ -0,0 +1,65 @@ +from __future__ import annotations + +import tempfile +import unittest +from pathlib import Path + +from datahub.adapters.tushare import TushareAdapter +from datahub.crypto import SecretVault +from datahub.hub import Hub +from datahub.settings import Settings +from tests.fixtures import TRADE_DATE, fake_transport + + +class ExtendedEodTests(unittest.TestCase): + def setUp(self) -> None: + self.tmp = tempfile.TemporaryDirectory() + key = SecretVault.generate_key() + settings = Settings( + host="127.0.0.1", + port=0, + encryption_key=key, + api_token="k" * 32, + admin_password="StartPass1", + tushare_token="tushare-secret", + db_path=Path(self.tmp.name) / "hub.db", + backup_dir=Path(self.tmp.name) / "backups", + scheduler_enabled=False, + quality={"daily_row_ratio": 0.5, "null_rate_max": 0.5, "list_limit_default": 5000, "list_limit_max": 5000}, + ) + adapter = TushareAdapter("tushare-secret", transport=fake_transport) + self.hub = Hub(settings, adapter=adapter) + self.hub.pipeline.ingest_reference(TRADE_DATE) + for dataset in ("daily", "valuation", "moneyflow", "auction", "index_daily"): + self.hub.pipeline.run_dataset(dataset, TRADE_DATE) + + def tearDown(self) -> None: + self.hub.stop() + self.tmp.cleanup() + + def test_extended_soft_datasets_publish_and_serve(self) -> None: + results = self.hub.pipeline.run_extended_soft( + ("limit_events", "popularity", "dragon_tiger", "sector_daily"), + TRADE_DATE, + ) + for name in ("limit_events", "popularity", "dragon_tiger", "sector_daily"): + self.assertEqual(results[name]["state"], "published", results[name]) + api = self.hub.api + limits = api.handle("/v1/limit-events", {"date": [TRADE_DATE]}) + self.assertGreaterEqual(len(limits["data"]), 1) + self.assertEqual(limits["meta"]["tier"], "official") + pop = api.handle("/v1/popularity", {"date": [TRADE_DATE], "source": ["ths"]}) + self.assertEqual(pop["data"][0]["source"], "ths") + lhb = api.handle("/v1/dragon-tiger", {"date": [TRADE_DATE]}) + self.assertEqual(lhb["data"][0]["hm_name"], "测试游资") + # hub stores 万元→元 + self.assertEqual(lhb["data"][0]["buy_amount"], 10_000_000.0) + sectors = api.handle("/v1/sectors", {"date": [TRADE_DATE], "family": ["ths"]}) + self.assertEqual(sectors["data"][0]["family"], "ths") + status = api.handle("/v1/datasets/status", {"date": [TRADE_DATE]}) + names = {item["dataset"] for item in status["data"]} + self.assertTrue({"limit_events", "popularity", "dragon_tiger", "sector_daily"} <= names) + + +if __name__ == "__main__": + unittest.main() diff --git a/xiaobai-datahub/tests/test_layout.py b/xiaobai-datahub/tests/test_layout.py index 7f70bf1..2469db9 100644 --- a/xiaobai-datahub/tests/test_layout.py +++ b/xiaobai-datahub/tests/test_layout.py @@ -19,11 +19,17 @@ class LayoutTests(unittest.TestCase): def test_reserved_adapters_present(self) -> None: from datahub.adapters import RESERVED - for name in ("eastmoney", "tencent", "ths", "xgb", "akshare", "ifind"): + for name in ("ths", "xgb", "akshare", "ifind"): self.assertIn(name, RESERVED) probe = RESERVED[name].probe() self.assertEqual(probe["state"], "reserved") self.assertFalse(probe["configured"]) + for name in ("eastmoney", "tencent"): + self.assertIn(name, RESERVED) + probe = RESERVED[name].probe() + # Live free adapters: probe may be ok/error/empty depending on network. + self.assertIn(probe["state"], {"ok", "empty", "error"}) + self.assertTrue(probe["configured"]) if __name__ == "__main__":