From b640e7726492c96f77d250d3d784c730aeadf5f3 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=80=BB=E7=AE=A1?= Date: Wed, 9 Sep 2026 11:15:56 +0800 Subject: [PATCH] fix(HEL-356): harden realtime quote completeness Co-authored-by: multica-agent --- backend/data/providers/tushare_dashboard.py | 6 + config/architecture-inventory.json | 4 +- tests/test_realtime_dashboard.py | 4 + .../config/hub-quality.config.json | 1 + xiaobai-datahub/datahub/adapters/eastmoney.py | 27 +- xiaobai-datahub/datahub/realtime_serve.py | 308 ++++++++++++++++-- xiaobai-datahub/datahub/scheduler.py | 26 ++ xiaobai-datahub/datahub/settings.py | 4 + .../tests/test_realtime_intraday.py | 45 ++- xiaobai-datahub/tests/test_scheduler.py | 27 ++ 10 files changed, 415 insertions(+), 37 deletions(-) diff --git a/backend/data/providers/tushare_dashboard.py b/backend/data/providers/tushare_dashboard.py index 468d1e6..1099323 100644 --- a/backend/data/providers/tushare_dashboard.py +++ b/backend/data/providers/tushare_dashboard.py @@ -400,6 +400,10 @@ class DashboardMixin: market_date, _number(row.get("vol")) / 100, ) + trade_time = str(row.get("trade_time") or row.get("quote_time") or "") + trade_date = str(row.get("quote_date") or "").replace("-", "")[:8] + if not trade_date and trade_time: + trade_date = trade_time[:10].replace("-", "") return { "code": ts_code.split(".")[0], "ts_code": ts_code, @@ -425,6 +429,8 @@ class DashboardMixin: "capital_trade_date": str(capital.get("trade_date") or ""), "turnover_source": "rt_volume/latest_float_share" if float_share else "unavailable", "data_source": str(row.get("source") or "tushare"), + "trade_date": trade_date, + "trade_time": trade_time, "realtime": True, } diff --git a/config/architecture-inventory.json b/config/architecture-inventory.json index 70b24ee..9a91efa 100644 --- a/config/architecture-inventory.json +++ b/config/architecture-inventory.json @@ -498,8 +498,8 @@ }, { "path": "backend/data/providers/tushare_dashboard.py", - "bytes": 33230, - "lines": 770 + "bytes": 33560, + "lines": 776 }, { "path": "database.py", diff --git a/tests/test_realtime_dashboard.py b/tests/test_realtime_dashboard.py index 39de1a4..73d3166 100644 --- a/tests/test_realtime_dashboard.py +++ b/tests/test_realtime_dashboard.py @@ -177,6 +177,9 @@ class RealtimeDashboardTests(unittest.TestCase): def test_realtime_stock_quote_uses_cached_industry(self): self.client._load_realtime_reference("20260720", "20260717") + self.client.try_quotes = lambda codes: [ + row for row in FREE_QUOTES if row["ts_code"] in set(codes) + ] quote = self.client.realtime_stock_quote("000003.SZ") self.assertEqual(quote["name"], "丙") @@ -184,6 +187,7 @@ class RealtimeDashboardTests(unittest.TestCase): self.assertAlmostEqual(quote["change"], 6.6667) self.assertEqual(quote["amount_billion"], 3.0) self.assertAlmostEqual(quote["turnover_rate"], 0.01) + self.assertEqual(quote["trade_date"], "20260720") def test_close_dashboard_marks_official_limit_data(self): dashboard = self.client.dashboard("20260720") diff --git a/xiaobai-datahub/config/hub-quality.config.json b/xiaobai-datahub/config/hub-quality.config.json index 2f208cd..fc5f397 100644 --- a/xiaobai-datahub/config/hub-quality.config.json +++ b/xiaobai-datahub/config/hub-quality.config.json @@ -27,6 +27,7 @@ "20:00", "23:10" ], + "realtime_warmup_interval_seconds": 120, "cross_gates": [ { "left": "daily", diff --git a/xiaobai-datahub/datahub/adapters/eastmoney.py b/xiaobai-datahub/datahub/adapters/eastmoney.py index 2dde4e2..bb1bffc 100644 --- a/xiaobai-datahub/datahub/adapters/eastmoney.py +++ b/xiaobai-datahub/datahub/adapters/eastmoney.py @@ -303,10 +303,20 @@ class EastmoneyAdapter(MarketAdapter): code = str(ts_code or "").split(".")[0] if not code: raise AdapterError("Invalid Shenwan code") + rows = self.fetch_shenwan_quotes([code]) + if not rows: + raise AdapterError(f"Eastmoney Shenwan quote missing for {code}") + return rows[0] + + def fetch_shenwan_quotes(self, ts_codes: list[str]) -> list[dict[str, Any]]: + codes = [str(item or "").split(".")[0] for item in ts_codes] + codes = list(dict.fromkeys(code for code in codes if code)) + if not codes: + return [] payload = self._get_json( EASTMONEY_INDEX_URL, { - "secids": f"90.{code}", + "secids": ",".join(f"90.{code}" for code in codes), "fltt": "2", "invt": "2", "fields": "f12,f14,f2,f3,f4,f15,f16,f17,f18,f6,f8,f104,f105,f128,f136,f140,f124", @@ -314,14 +324,21 @@ class EastmoneyAdapter(MarketAdapter): referer="https://quote.eastmoney.com/", ) rows = list((payload.get("data") or {}).get("diff") or []) - row = next((item for item in rows if item), None) - if not row: - raise AdapterError(f"Eastmoney Shenwan quote missing for {code}") + result = [quote for row in rows if row and (quote := self._normalize_shenwan_quote(row))] + if not result: + raise AdapterError(f"Eastmoney Shenwan quotes missing for {','.join(codes)}") + return result + + @staticmethod + def _normalize_shenwan_quote(row: dict[str, Any]) -> dict[str, Any] | None: + code = str(row.get("f12") or "") + if not code: + return None epoch = int(finite_number(row.get("f124")) or 0) close = round4(finite_number(row.get("f2"))) previous = round4(finite_number(row.get("f18"))) if close <= 0 or previous <= 0: - raise AdapterError(f"Eastmoney Shenwan quote empty for {code}") + return None quote_time = ( datetime.fromtimestamp(epoch).astimezone().isoformat(timespec="seconds") if epoch diff --git a/xiaobai-datahub/datahub/realtime_serve.py b/xiaobai-datahub/datahub/realtime_serve.py index bd338bd..e2e8196 100644 --- a/xiaobai-datahub/datahub/realtime_serve.py +++ b/xiaobai-datahub/datahub/realtime_serve.py @@ -8,6 +8,7 @@ from __future__ import annotations import hashlib import json import time +from concurrent.futures import ThreadPoolExecutor, as_completed from datetime import datetime from typing import Any @@ -84,13 +85,13 @@ 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 = TencentAdapter().fetch_quotes(listed) + rows = _tencent_named_quotes(listed) if len(rows) < 200: raise AdapterError(f"Tencent market snapshot too small: {len(rows)}") source = "tencent:qt" except Exception as backup_exc: errors.append(f"tencent:{backup_exc}") - recovered = _load_quotes_lkg(db, cache_key) + recovered = _load_quotes_lkg(db, cache_key, yyyymmdd(now_shanghai())) if recovered is not None: return recovered raise RealtimeApiError( @@ -117,35 +118,77 @@ def fetch_quotes(db: HubDB, codes: list[str]) -> dict[str, Any]: digest = hashlib.sha1(",".join(sorted(resolved)).encode("utf-8")).hexdigest() cache_key = f"quotes:{digest}:{len(resolved)}" cached = _read_cache(db, cache_key) - if cached is not None: + if cached is not None and _quote_codes(cached.get("data") or []) >= set(resolved): return cached minimum = max(1, int(len(resolved) * 0.5)) errors: list[str] = [] - rows: list[dict[str, Any]] = [] - source = "" + by_code = _quote_map((cached or {}).get("data") or [], resolved) + sources: list[str] = [] + cached_source = str(((cached or {}).get("meta") or {}).get("source") or "") + if by_code and cached_source: + sources.append(cached_source) + + missing = [code for code in resolved if code not in by_code] try: - rows = _eastmoney_named_quotes(resolved) - if len(rows) < minimum: - raise AdapterError(f"Eastmoney named quotes too small: {len(rows)}/{len(resolved)}") - source = "eastmoney:ulist" + rows = _eastmoney_named_quotes(missing) + by_code.update(_quote_map(rows, missing)) + if rows: + sources.append("eastmoney:ulist") except Exception as exc: errors.append(f"eastmoney:{exc}") + + missing = [code for code in resolved if code not in by_code] + if missing: try: - rows = TencentAdapter().fetch_quotes(resolved) - if len(rows) < minimum: - raise AdapterError(f"Tencent named quotes too small: {len(rows)}/{len(resolved)}") - source = "tencent:qt" + rows = _tencent_named_quotes(missing) + by_code.update(_quote_map(rows, missing)) + if rows: + sources.append("tencent:qt") except Exception as backup_exc: errors.append(f"tencent:{backup_exc}") - recovered = _load_quotes_lkg(db, cache_key) - if recovered is not None: - return recovered - raise RealtimeApiError( - "SOURCE_UNAVAILABLE", - "quotes unavailable: " + "；".join(errors), - ) from backup_exc + + fresh_rows = [by_code[code] for code in resolved if code in by_code] + source = "+".join(dict.fromkeys(sources)) or "unavailable" + if fresh_rows: + _store_quote_rows_lkg(db, fresh_rows, source) + + missing = [code for code in resolved if code not in by_code] + today = yyyymmdd(now_shanghai()) + recovered_rows, stale_age = _recover_quote_rows(db, missing, today) + by_code.update(_quote_map(recovered_rows, missing)) + missing = [code for code in resolved if code not in by_code] + if missing: + group_lkg = _load_quotes_lkg(db, cache_key, today) + group_rows = _quote_map((group_lkg or {}).get("data") or [], missing) + by_code.update(group_rows) + if group_rows: + recovered_rows.extend(group_rows.values()) + stale_age = max( + stale_age, + int(((group_lkg or {}).get("meta") or {}).get("staleness_seconds") or 0), + ) + rows = [by_code[code] for code in resolved if code in by_code] + if len(rows) < minimum: + detail = "；".join(errors) or f"only {len(rows)}/{len(resolved)} quotes returned" + raise RealtimeApiError("SOURCE_UNAVAILABLE", "quotes unavailable: " + detail) + + if recovered_rows: + source = "+".join(dict.fromkeys([*sources, "same-day-lkg"])) payload = _quote_payload(rows, source) - _write_cache(db, cache_key, payload, QUOTE_TTL, source) + payload["meta"].update({ + "requested_count": len(resolved), + "returned_count": len(rows), + "complete": len(rows) == len(resolved), + "missing_codes": [code for code in resolved if code not in by_code], + }) + if recovered_rows: + payload["meta"].update({ + "stale": True, + "staleness_seconds": stale_age, + "delay_notice": f"主备免费行情暂不完整，已用当天 {stale_age} 秒前的真实快照补齐", + }) + else: + _write_cache(db, cache_key, payload, QUOTE_TTL, source) return payload @@ -163,7 +206,7 @@ def fetch_sector_quote(db: HubDB, code: str, expected_date: str = "") -> dict[st source = str(row.get("source") or "eastmoney_sw") except Exception as exc: errors.append(f"eastmoney:{exc}") - recovered = _load_quotes_lkg(db, cache_key) + recovered = _load_quotes_lkg(db, cache_key, expected_date) if recovered is not None: return recovered raise RealtimeApiError( @@ -173,7 +216,7 @@ def fetch_sector_quote(db: HubDB, code: str, expected_date: str = "") -> dict[st want = str(expected_date or "").replace("-", "")[:8] quote_date = str(row.get("quote_date") or "") if want and quote_date and quote_date != want: - recovered = _load_quotes_lkg(db, cache_key) + recovered = _load_quotes_lkg(db, cache_key, want) if recovered is not None: return recovered raise RealtimeApiError("SOURCE_UNAVAILABLE", f"sector quote date {quote_date} != {want}") @@ -222,10 +265,28 @@ def fetch_limit_pool(db: HubDB, trade_date: str = "") -> dict[str, Any]: def _eastmoney_named_quotes(codes: list[str]) -> list[dict[str, Any]]: - adapter = EastmoneyAdapter() + return _parallel_named_quotes(EastmoneyAdapter, codes) + + +def _tencent_named_quotes(codes: list[str]) -> list[dict[str, Any]]: + return _parallel_named_quotes(TencentAdapter, codes) + + +def _parallel_named_quotes(adapter_factory, codes: list[str]) -> list[dict[str, Any]]: + chunks = [codes[index:index + QUOTE_BATCH] for index in range(0, len(codes), QUOTE_BATCH)] + if not chunks: + return [] rows: list[dict[str, Any]] = [] - for index in range(0, len(codes), QUOTE_BATCH): - rows.extend(adapter.fetch_quotes(codes[index:index + QUOTE_BATCH])) + errors: list[str] = [] + with ThreadPoolExecutor(max_workers=min(8, len(chunks))) as executor: + futures = [executor.submit(adapter_factory().fetch_quotes, chunk) for chunk in chunks] + for future in as_completed(futures): + try: + rows.extend(future.result()) + except Exception as exc: + errors.append(str(exc)) + if not rows and errors: + raise AdapterError("; ".join(errors[:3])) return rows @@ -251,7 +312,7 @@ def _quote_payload( "stale": False, "staleness_seconds": 0, "published_at": isoformat(now_shanghai()), - "failover": source.startswith("tencent"), + "failover": "tencent" in source, "delay_notice": "", } if scope: @@ -259,15 +320,32 @@ def _quote_payload( return _envelope(rows, meta) -def _load_quotes_lkg(db: HubDB, cache_key: str) -> dict[str, Any] | None: +def _load_quotes_lkg( + db: HubDB, + cache_key: str, + expected_date: str = "", +) -> dict[str, Any] | None: store = LastKnownGood(db) item = store.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: + if not isinstance(data, (list, dict)) or not data: return None + day = yyyymmdd(expected_date) if expected_date else "" + if day: + payload_day = yyyymmdd((payload.get("meta") or {}).get("trade_date")) + if isinstance(data, list): + dated = [ + row for row in data + if isinstance(row, dict) and _row_quote_date(row, payload_day) == day + ] + if not dated: + return None + payload = {**payload, "data": dated} + elif _row_quote_date(data, payload_day) != day: + return None stamped = dict(payload) meta = dict(stamped.get("meta") or {}) stored = str((item or {}).get("stored_at") or "") @@ -283,6 +361,178 @@ def _load_quotes_lkg(db: HubDB, cache_key: str) -> dict[str, Any] | None: return stamped +def _quote_map(rows: list[Any], wanted: list[str]) -> dict[str, dict[str, Any]]: + allowed = set(wanted) + result: dict[str, dict[str, Any]] = {} + for raw in rows: + if not isinstance(raw, dict): + continue + code = str(raw.get("ts_code") or "").upper() + if code in allowed and float(raw.get("close") or raw.get("price") or 0) > 0: + result[code] = dict(raw) + return result + + +def _quote_codes(rows: list[Any]) -> set[str]: + return { + str(row.get("ts_code") or "").upper() + for row in rows + if isinstance(row, dict) and row.get("ts_code") + } + + +def _row_quote_date(row: dict[str, Any], fallback: str = "") -> str: + raw_day = str(row.get("quote_date") or "").strip() + if raw_day: + try: + return yyyymmdd(raw_day) + except ValueError: + pass + stamp = str(row.get("quote_time") or row.get("trade_time") or "") + raw_fallback = stamp[:10] if stamp else str(fallback or "") + if not raw_fallback: + return "" + try: + return yyyymmdd(raw_fallback) + except ValueError: + return "" + + +def _store_quote_rows_lkg(db: HubDB, rows: list[dict[str, Any]], source: str) -> None: + now = now_shanghai() + today = yyyymmdd(now) + stored = isoformat(now) + values = [] + for row in rows: + code = str(row.get("ts_code") or "").upper() + if not code or _row_quote_date(row, today) != today: + continue + payload = _envelope( + dict(row), + {"tier": "provisional", "trade_date": today, "source": source, "stale": False}, + ) + values.append((f"quote:{code}", json.dumps(payload, ensure_ascii=False), source, stored)) + if not values: + return + with db.write() as connection: + connection.executemany( + """ + 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 + """, + values, + ) + + +def _recover_quote_rows( + db: HubDB, + codes: list[str], + expected_date: str, +) -> tuple[list[dict[str, Any]], int]: + if not codes: + return [], 0 + today = yyyymmdd(expected_date) + recovered: dict[str, dict[str, Any]] = {} + oldest_age = 0 + placeholders = ",".join("?" for _ in codes) + keys = [f"quote:{code}" for code in codes] + rows = db.fetchall( + f"SELECT cache_key,payload,stored_at FROM last_known_good WHERE cache_key IN ({placeholders})", + tuple(keys), + ) + now_epoch = time.time() + for item in rows: + try: + payload = json.loads(item["payload"]) + except (json.JSONDecodeError, TypeError): + continue + quote = payload.get("data") if isinstance(payload, dict) else None + if not isinstance(quote, dict) or _row_quote_date(quote, (payload.get("meta") or {}).get("trade_date")) != today: + continue + code = str(quote.get("ts_code") or "").upper() + if code not in codes: + continue + recovered[code] = dict(quote) + try: + oldest_age = max(oldest_age, int(now_epoch - datetime.fromisoformat(item["stored_at"]).timestamp())) + except (TypeError, ValueError): + oldest_age = max(oldest_age, QUOTE_TTL) + + remaining = [code for code in codes if code not in recovered] + if remaining: + market = _load_quotes_lkg(db, "quotes:market", today) + market_rows = _quote_map((market or {}).get("data") or [], remaining) + recovered.update(market_rows) + oldest_age = max(oldest_age, int(((market or {}).get("meta") or {}).get("staleness_seconds") or 0)) + return [recovered[code] for code in codes if code in recovered], oldest_age + + +def warm_realtime(db: HubDB) -> dict[str, Any]: + """Proactively keep same-day market snapshots warm during trading hours.""" + today = yyyymmdd(now_shanghai()) + result: dict[str, Any] = {"trade_date": today, "rows": 0, "errors": []} + try: + indexes = fetch_index_quotes(db) + result["indexes"] = len(indexes.get("data") or []) + except Exception as exc: + result["errors"].append(f"indexes:{exc}") + try: + market = fetch_market_quotes(db) + result["market"] = len(market.get("data") or []) + result["rows"] += result["market"] + except Exception as exc: + result["errors"].append(f"market:{exc}") + + sector_rows: list[dict[str, Any]] = [] + try: + masters = db.fetchall( + "SELECT ts_code FROM sector_master WHERE family = 'sw' ORDER BY ts_code" + ) + codes = [str(row.get("ts_code") or "") for row in masters if row.get("ts_code")] + sector_rows = _eastmoney_sector_quotes(codes) + for row in sector_rows: + if _row_quote_date(row, today) != today: + continue + code = str(row.get("ts_code") or "").upper() + payload = _envelope( + row, + { + "tier": "provisional", + "trade_date": today, + "source": str(row.get("source") or "eastmoney_sw"), + "stale": False, + "staleness_seconds": 0, + "published_at": isoformat(now_shanghai()), + }, + ) + _write_cache(db, f"sector:{code}", payload, INDEX_TTL, "eastmoney_sw") + result["sectors"] = len(sector_rows) + result["rows"] += len(sector_rows) + except Exception as exc: + result["errors"].append(f"sectors:{exc}") + return result + + +def _eastmoney_sector_quotes(codes: list[str]) -> list[dict[str, Any]]: + chunks = [codes[index:index + QUOTE_BATCH] for index in range(0, len(codes), QUOTE_BATCH)] + if not chunks: + return [] + rows: list[dict[str, Any]] = [] + errors: list[str] = [] + with ThreadPoolExecutor(max_workers=min(8, len(chunks))) as executor: + futures = [executor.submit(EastmoneyAdapter().fetch_shenwan_quotes, chunk) for chunk in chunks] + for future in as_completed(futures): + try: + rows.extend(future.result()) + except Exception as exc: + errors.append(str(exc)) + if not rows and errors: + raise AdapterError("; ".join(errors[:3])) + return rows + + 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: diff --git a/xiaobai-datahub/datahub/scheduler.py b/xiaobai-datahub/datahub/scheduler.py index d3b3ff7..d4d1e6d 100644 --- a/xiaobai-datahub/datahub/scheduler.py +++ b/xiaobai-datahub/datahub/scheduler.py @@ -74,6 +74,7 @@ class Scheduler: "cleanup": self._cleanup, "backup": self._backup, "history_backfill": self._history_backfill, + "realtime_warmup": self._realtime_warmup, } self._stop = threading.Event() self._thread: threading.Thread | None = None @@ -146,10 +147,30 @@ class Scheduler: if job_id in {"eod_a", "eod_b"}: self._settle_eod(day) ran.extend(self._eod_retry_tick(now, day, open_day)) + ran.extend(self._realtime_warmup_tick(now, day, open_day)) ran.extend(self._revision_review_tick(now, day, open_day)) ran.extend(self._revision_catchup_tick(now, day)) return ran + def _realtime_warmup_tick(self, now: datetime, day: str, open_day: bool) -> list[str]: + if not open_day or "realtime_warmup" not in self.jobs: + return [] + current = now.time() + in_session = time(9, 15) <= current <= time(11, 35) or time(12, 55) <= current <= time(15, 5) + if not in_session: + return [] + interval = self.pipeline.settings.realtime_warmup_interval_seconds + slot = str(int(now.timestamp()) // interval) + key = ("realtime_warmup", day, slot) + if key in self._fired: + return [] + self._fired.add(key) + try: + self.run_job("realtime_warmup", day) + except Exception: + LOGGER.exception("realtime warmup failed for %s", day) + return ["realtime_warmup"] + # ------------------------------------------------------------------ # EOD retry window # ------------------------------------------------------------------ @@ -572,6 +593,11 @@ class Scheduler: def _history_backfill(self, trade_date: str) -> dict[str, Any]: return self.pipeline.backfill_history(trade_date) + def _realtime_warmup(self, trade_date: str) -> dict[str, Any]: + from datahub.realtime_serve import warm_realtime + + return warm_realtime(self.db) + def _cleanup(self, trade_date: str) -> dict[str, Any]: result = self.pipeline.cleanup() if now_shanghai().weekday() == 6: diff --git a/xiaobai-datahub/datahub/settings.py b/xiaobai-datahub/datahub/settings.py index 6c07f8e..3aa86b0 100644 --- a/xiaobai-datahub/datahub/settings.py +++ b/xiaobai-datahub/datahub/settings.py @@ -73,6 +73,10 @@ class Settings: raw = [raw] return tuple(str(item) for item in raw) + @property + def realtime_warmup_interval_seconds(self) -> int: + return max(30, int(self.quality.get("realtime_warmup_interval_seconds") or 120)) + @property def eod_retry_start(self) -> str: return str(self.quality.get("eod_retry_start") or "15:15") diff --git a/xiaobai-datahub/tests/test_realtime_intraday.py b/xiaobai-datahub/tests/test_realtime_intraday.py index b049562..a7b27cf 100644 --- a/xiaobai-datahub/tests/test_realtime_intraday.py +++ b/xiaobai-datahub/tests/test_realtime_intraday.py @@ -252,13 +252,56 @@ class MarketQuotesTests(unittest.TestCase): self.assertTrue(payload["meta"]["failover"]) self.assertFalse(payload["meta"]["stale"]) + def test_partial_sources_are_merged_instead_of_discarded(self) -> None: + with patch("datahub.realtime_serve.EastmoneyAdapter") as eastmoney, patch( + "datahub.realtime_serve.TencentAdapter" + ) as tencent: + eastmoney.return_value.fetch_quotes.return_value = [ + {"ts_code": "000001.SZ", "close": 10, "pre_close": 9, "quote_date": yyyymmdd(now_shanghai())}, + ] + tencent.return_value.fetch_quotes.return_value = [ + {"ts_code": "000002.SZ", "close": 20, "pre_close": 19, "quote_date": yyyymmdd(now_shanghai())}, + ] + payload = self.api.handle( + "/v1/quotes/latest", {"codes": ["000001.SZ,000002.SZ"]} + ) + self.assertEqual({row["ts_code"] for row in payload["data"]}, {"000001.SZ", "000002.SZ"}) + self.assertTrue(payload["meta"]["complete"]) + self.assertIn("eastmoney", payload["meta"]["source"]) + self.assertIn("tencent", payload["meta"]["source"]) + + def test_per_stock_same_day_snapshot_fills_a_different_group(self) -> None: + from datahub.realtime_serve import _store_quote_rows_lkg + + today = yyyymmdd(now_shanghai()) + _store_quote_rows_lkg( + self.db, + [ + {"ts_code": "000001.SZ", "close": 10, "pre_close": 9, "quote_date": today}, + {"ts_code": "000002.SZ", "close": 20, "pre_close": 19, "quote_date": today}, + ], + "eastmoney:ulist", + ) + with patch("datahub.realtime_serve.EastmoneyAdapter") as eastmoney, patch( + "datahub.realtime_serve.TencentAdapter" + ) as tencent: + eastmoney.return_value.fetch_quotes.side_effect = AdapterError("closed") + tencent.return_value.fetch_quotes.side_effect = AdapterError("timeout") + payload = self.api.handle( + "/v1/quotes/latest", {"codes": ["000002.SZ,000001.SZ"]} + ) + self.assertEqual(len(payload["data"]), 2) + self.assertTrue(payload["meta"]["stale"]) + self.assertEqual(payload["meta"]["source"], "same-day-lkg") + def test_both_quote_sources_return_last_known_good(self) -> None: from datahub.realtime_serve import _envelope, _write_cache cache_key = "quotes:placeholder:1" + today = yyyymmdd(now_shanghai()) payload = _envelope( [{"ts_code": "000737.SZ", "close": 12.3, "pre_close": 11.2}], - {"source": "eastmoney:ulist", "stale": False, "trade_date": "20260908"}, + {"source": "eastmoney:ulist", "stale": False, "trade_date": today}, ) _write_cache(self.db, cache_key, payload, 60, "eastmoney:ulist") self.db.execute( diff --git a/xiaobai-datahub/tests/test_scheduler.py b/xiaobai-datahub/tests/test_scheduler.py index da0de5b..7c8d9e2 100644 --- a/xiaobai-datahub/tests/test_scheduler.py +++ b/xiaobai-datahub/tests/test_scheduler.py @@ -57,6 +57,33 @@ class SchedulerTests(unittest.TestCase): self.assertEqual(ran["eod_a"], 1) tmp.cleanup() + def test_realtime_warmup_runs_once_per_interval_during_market(self) -> None: + tmp = tempfile.TemporaryDirectory() + db = HubDB(Path(tmp.name) / "hub.db") + adapter = TushareAdapter("x", transport=fake_transport) + settings = Settings( + encryption_key=SecretVault.generate_key(), + scheduler_enabled=False, + db_path=db.path, + quality={"realtime_warmup_interval_seconds": 120}, + ) + pipe = Pipeline(db, adapter, settings) + pipe.ingest_reference("20240902") + calls: list[str] = [] + sched = Scheduler(db, pipe, jobs={"realtime_warmup": lambda day: calls.append(day) or {}}) + + first = sched.tick(datetime(2024, 9, 2, 10, 0, 0, tzinfo=SHANGHAI)) + duplicate = sched.tick(datetime(2024, 9, 2, 10, 1, 0, tzinfo=SHANGHAI)) + second = sched.tick(datetime(2024, 9, 2, 10, 2, 0, tzinfo=SHANGHAI)) + lunch = sched.tick(datetime(2024, 9, 2, 12, 0, 0, tzinfo=SHANGHAI)) + + self.assertIn("realtime_warmup", first) + self.assertNotIn("realtime_warmup", duplicate) + self.assertIn("realtime_warmup", second) + self.assertNotIn("realtime_warmup", lunch) + self.assertEqual(calls, ["20240902", "20240902"]) + tmp.cleanup() + if __name__ == "__main__": unittest.main()