from __future__ import annotations import tempfile import unittest from pathlib import Path from unittest.mock import patch from datahub.adapters.base import AdapterError from datahub.adapters.eastmoney import HIS_TRENDS_URL, TRENDS_URL, EastmoneyAdapter from datahub.db import HubDB from datahub.realtime_serve import fetch_intraday from datahub.serving import ApiError, V1API from datahub.timeutil import now_shanghai, yyyymmdd class FakeEastmoney(EastmoneyAdapter): def __init__(self) -> None: super().__init__(timeout=2) self.urls: list[str] = [] def _get_json(self, url, params, referer): self.urls.append(f"{url}|{params.get('ndays')}") if url == TRENDS_URL: return {"data": {"name": "中国平安", "code": "601318", "preClose": 56.36, "trends": []}} if url == HIS_TRENDS_URL: return { "data": { "name": "中国平安", "code": "601318", "preClose": 55.8, "trends": [ "2026-09-07 09:30,55.80,55.90,56.00,55.70,100,5580.00,55.900", "2026-09-07 15:00,56.10,56.20,56.30,56.00,200,11240.00,56.150", "2026-09-08 09:30,0,0,0,0,0,0.00,0", ], } } raise AdapterError(f"unexpected url {url}") class EastmoneyIntradayLookbackTests(unittest.TestCase): def test_empty_today_uses_latest_available_session(self): adapter = FakeEastmoney() payload = adapter.fetch_intraday("601318.SH") self.assertEqual(adapter.urls, [f"{TRENDS_URL}|1", f"{TRENDS_URL}|5", f"{HIS_TRENDS_URL}|5"]) self.assertEqual(payload["trade_date"], "2026-09-07") self.assertEqual([point["time"] for point in payload["points"]], ["09:30", "15:00"]) self.assertEqual(payload["points"][0]["close"], 55.9) def test_preferred_date_keeps_that_session(self): adapter = FakeEastmoney() payload = adapter.fetch_intraday("601318.SH", "20260907") self.assertEqual(payload["trade_date"], "2026-09-07") self.assertEqual(len(payload["points"]), 2) class IntradayLkgTests(unittest.TestCase): def setUp(self) -> None: self.tmp = tempfile.TemporaryDirectory() self.db = HubDB(Path(self.tmp.name) / "hub.db") def tearDown(self) -> None: self.tmp.cleanup() def test_source_failure_returns_last_known_good(self): from datahub.realtime_serve import _envelope, _write_cache payload = _envelope( { "entity_type": "stock", "ts_code": "601318.SH", "trade_date": "2026-09-07", "previous_close": 55.8, "points": [{"date": "2026-09-07", "time": "09:30", "close": 55.9}], }, { "tier": "provisional", "trade_date": "20260907", "source": "eastmoney:trends2", "stale": False, }, ) _write_cache(self.db, "intraday:601318.SH:today", payload, 20, "eastmoney:trends2") self.db.execute( "UPDATE rt_cache SET expires_at = ? WHERE cache_key = ?", ("2000-01-01T00:00:00+08:00", "intraday:601318.SH:today"), ) with patch("datahub.realtime_serve.EastmoneyAdapter") as mocked: mocked.return_value.fetch_intraday.side_effect = AdapterError("down") recovered = fetch_intraday(self.db, "601318.SH") self.assertTrue(recovered["meta"]["stale"]) self.assertEqual(recovered["data"]["points"][0]["close"], 55.9) def test_source_failure_without_lkg_raises(self): with patch("datahub.realtime_serve.EastmoneyAdapter") as mocked: mocked.return_value.fetch_intraday.side_effect = AdapterError("down") with self.assertRaises(Exception) as ctx: fetch_intraday(self.db, "000001.SZ") self.assertIn("intraday unavailable", str(ctx.exception)) class ServingIntradayDateTests(unittest.TestCase): def setUp(self) -> None: self.tmp = tempfile.TemporaryDirectory() self.db = HubDB(Path(self.tmp.name) / "hub.db") self.api = V1API(self.db, pipeline=None, settings=None) def tearDown(self) -> None: self.tmp.cleanup() def _assert_usable_intraday(self, payload: dict) -> None: data = payload["data"] points = [point for point in data.get("points") or [] if float(point.get("close") or 0) > 0] self.assertGreaterEqual(len(points), 1) self.assertTrue(str(data.get("trade_date") or "")) self.assertFalse((payload.get("meta") or {}).get("stale")) def test_serving_omitted_or_empty_date_uses_today_and_returns_points(self) -> None: today = yyyymmdd(now_shanghai()) omitted = self.api.handle("/v1/intraday/points", {"code": ["601318"]}) empty = self.api.handle("/v1/intraday/points", {"code": ["601318"], "date": [""]}) explicit = self.api.handle("/v1/intraday/points", {"code": ["601318"], "date": [today]}) self._assert_usable_intraday(omitted) self._assert_usable_intraday(empty) self._assert_usable_intraday(explicit) self.assertEqual(omitted["data"]["trade_date"], empty["data"]["trade_date"]) self.assertEqual(explicit["data"]["trade_date"], omitted["data"]["trade_date"]) def test_serving_normalizes_empty_date_to_today_and_keeps_history(self) -> None: today = yyyymmdd(now_shanghai()) captured: list[str] = [] def fake_fetch(db, code, date=""): captured.append(date) return { "schema_version": 1, "data": { "trade_date": f"{date[:4]}-{date[4:6]}-{date[6:8]}", "points": [{"date": f"{date[:4]}-{date[4:6]}-{date[6:8]}", "time": "09:30", "close": 55.9}], }, "meta": {"stale": False, "trade_date": date}, } with patch("datahub.realtime_serve.fetch_intraday", side_effect=fake_fetch): omitted = self.api.handle("/v1/intraday/points", {"code": ["601318"]}) empty = self.api.handle("/v1/intraday/points", {"code": ["601318"], "date": [" "]}) history = self.api.handle("/v1/intraday/points", {"code": ["601318"], "date": ["20260907"]}) self.assertEqual(captured, [today, today, "20260907"]) self.assertEqual(omitted["data"]["trade_date"], f"{today[:4]}-{today[4:6]}-{today[6:8]}") self.assertEqual(empty["data"]["trade_date"], omitted["data"]["trade_date"]) self.assertEqual(history["data"]["trade_date"], "2026-09-07") def test_serving_invalid_date_is_invalid_argument(self) -> None: with self.assertRaises(ApiError) as ctx: self.api.handle("/v1/intraday/points", {"code": ["601318"], "date": ["not-a-date"]}) self.assertEqual(ctx.exception.code, "INVALID_ARGUMENT") self.assertIn("invalid trade_date", ctx.exception.message) def test_serving_missing_code_is_invalid_argument(self) -> None: with self.assertRaises(ApiError) as ctx: self.api.handle("/v1/intraday/points", {"date": [yyyymmdd(now_shanghai())]}) self.assertEqual(ctx.exception.code, "INVALID_ARGUMENT") self.assertIn("code is required", ctx.exception.message) def test_serving_no_data_keeps_source_unavailable(self) -> None: with patch("datahub.realtime_serve.EastmoneyAdapter") as mocked: mocked.return_value.fetch_intraday.side_effect = AdapterError("No intraday chart data returned") with self.assertRaises(ApiError) as ctx: self.api.handle("/v1/intraday/points", {"code": ["000001"]}) self.assertEqual(ctx.exception.code, "SOURCE_UNAVAILABLE") self.assertIn("intraday unavailable", ctx.exception.message) class MarketQuotesTests(unittest.TestCase): def setUp(self) -> None: self.tmp = tempfile.TemporaryDirectory() self.db = HubDB(Path(self.tmp.name) / "hub.db") self.api = V1API(self.db, None, None) # type: ignore[arg-type] def tearDown(self) -> None: self.tmp.cleanup() def test_empty_codes_returns_full_market_snapshot(self) -> None: rows = [ { "ts_code": f"{600000 + index:06d}.SH", "name": f"股票{index}", "pre_close": 10.0, "close": 10.2, "open": 10.1, "high": 10.3, "low": 10.0, "vol": 1000, "amount": 2000000, } for index in range(220) ] with patch("datahub.realtime_serve.EastmoneyAdapter") as mocked: mocked.return_value.fetch_market_quotes.return_value = rows omitted = self.api.handle("/v1/quotes/latest", {}) empty = self.api.handle("/v1/quotes/latest", {"codes": [""]}) self.assertEqual(len(omitted["data"]), 220) self.assertEqual(omitted["meta"]["scope"], "market") self.assertEqual(omitted["meta"]["source"], "eastmoney:clist") self.assertEqual(len(empty["data"]), 220) def test_explicit_codes_still_use_named_quote_path(self) -> None: named = [ { "ts_code": "600000.SH", "name": "浦发银行", "price": 10.2, "previous_close": 10.0, } ] with patch("datahub.realtime_serve.EastmoneyAdapter") as mocked: mocked.return_value.fetch_quotes.return_value = named payload = self.api.handle("/v1/quotes/latest", {"codes": ["600000.SH"]}) mocked.return_value.fetch_market_quotes.assert_not_called() self.assertEqual(payload["data"][0]["ts_code"], "600000.SH") self.assertEqual(payload["meta"]["source"], "eastmoney:ulist") def test_named_quotes_page_beyond_sixty_codes(self) -> None: codes = [f"{index:06d}.SZ" for index in range(70)] def fake_fetch(chunk): return [{"ts_code": code, "close": 10, "pre_close": 9} for code in chunk] with patch("datahub.realtime_serve.EastmoneyAdapter") as mocked: mocked.return_value.fetch_quotes.side_effect = fake_fetch payload = self.api.handle("/v1/quotes/latest", {"codes": [",".join(codes)]}) self.assertEqual(mocked.return_value.fetch_quotes.call_count, 2) self.assertEqual(len(payload["data"]), 70) def test_named_quotes_fail_over_to_tencent(self) -> None: named = [ { "ts_code": "000737.SZ", "name": "北方铜业", "close": 12.3, "pre_close": 11.2, } ] with patch("datahub.realtime_serve.EastmoneyAdapter") as eastmoney, patch( "datahub.realtime_serve.TencentAdapter" ) as tencent: eastmoney.return_value.fetch_quotes.side_effect = AdapterError("HTTP 503") tencent.return_value.fetch_quotes.return_value = named payload = self.api.handle("/v1/quotes/latest", {"codes": ["000737.SZ"]}) self.assertEqual(payload["data"][0]["ts_code"], "000737.SZ") self.assertEqual(payload["meta"]["source"], "tencent:qt") 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": today}, ) _write_cache(self.db, cache_key, payload, 60, "eastmoney:ulist") self.db.execute( "UPDATE rt_cache SET expires_at = ? WHERE cache_key = ?", ("2000-01-01T00:00:00+08:00", cache_key), ) with patch("datahub.realtime_serve.EastmoneyAdapter") as eastmoney, patch( "datahub.realtime_serve.TencentAdapter" ) as tencent, patch("datahub.realtime_serve.hashlib.sha1") as sha1: eastmoney.return_value.fetch_quotes.side_effect = AdapterError("HTTP 503") tencent.return_value.fetch_quotes.side_effect = AdapterError("timeout") sha1.return_value.hexdigest.return_value = "placeholder" recovered = self.api.handle("/v1/quotes/latest", {"codes": ["000737.SZ"]}) self.assertTrue(recovered["meta"]["stale"]) self.assertIn("真实快照", recovered["meta"]["delay_notice"]) self.assertEqual(recovered["data"][0]["close"], 12.3) self.assertNotEqual(recovered["data"][0]["close"], 0) def test_market_unavailable_stays_source_error(self) -> None: with patch("datahub.realtime_serve.EastmoneyAdapter") as eastmoney, patch( "datahub.realtime_serve.TencentAdapter" ) as tencent: eastmoney.return_value.fetch_market_quotes.side_effect = AdapterError("too small") tencent.return_value.fetch_quotes.side_effect = AdapterError("empty master") with self.assertRaises(ApiError) as ctx: self.api.handle("/v1/quotes/latest", {}) self.assertEqual(ctx.exception.code, "SOURCE_UNAVAILABLE") if __name__ == "__main__": unittest.main()