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

fix(HEL-356): harden realtime quote completeness

Browse Source 
Co-authored-by: multica-agent <github@multica.ai>
This commit is contained in:
总管
2026-09-09 11:15:56 +08:00
co-authored by multica-agent
parent b950ea4344
commit b640e77264
10 changed files with 415 additions and 37 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/data/providers/tushare_dashboard.py
+6
View File
@@ -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,
}
config/architecture-inventory.json
+2 -2
View File
@@ -498,8 +498,8 @@
},
{
"path": "backend/data/providers/tushare_dashboard.py",
"bytes": 33230,
"lines": 770
"bytes": 33560,
"lines": 776
},
{
"path": "database.py",
tests/test_realtime_dashboard.py
+4
View File
@@ -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")
xiaobai-datahub/config/hub-quality.config.json
+1
View File
@@ -27,6 +27,7 @@
"20:00",
"23:10"
],
"realtime_warmup_interval_seconds": 120,
"cross_gates": [
{
"left": "daily",
xiaobai-datahub/datahub/adapters/eastmoney.py
+22 -5
View File
@@ -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
xiaobai-datahub/datahub/realtime_serve.py
+279 -29
View File
@@ -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:
xiaobai-datahub/datahub/scheduler.py
+26
View File
@@ -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:
xiaobai-datahub/datahub/settings.py
+4
View File
@@ -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")
xiaobai-datahub/tests/test_realtime_intraday.py
+44 -1
View File
@@ -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(
xiaobai-datahub/tests/test_scheduler.py
+27
View File
@@ -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()