from __future__ import annotations import unittest from backend.data.providers.ifind_client import IfindHttpClient from backend.features.market.charts import ChartDataError, EastmoneyChartClient, HIS_TRENDS_URL, MarketChartClient, TRENDS_URL from server import DashboardService class FakeChartClient(EastmoneyChartClient): def __init__(self) -> None: super().__init__(cache_ttl_seconds=20) self.requests: list[tuple[str, dict[str, str]]] = [] def _request_json(self, url, params, referer): self.requests.append((url, params)) if "trends2" in url: return { "data": { "code": params["secid"].split(".", 1)[1], "name": "测试行情", "preClose": 10.0, "trends": [ "2026-07-24 09:30,10.10,10.20,10.30,10.00,100,1020.00,10.200", "2026-07-24 09:31,10.20,10.15,10.25,10.10,80,812.00,10.178", ], } } return { "data": { "diff": [ {"f12": "BK0474", "f14": "保险Ⅱ"}, {"f12": "BK1040", "f14": "中药Ⅱ"}, ] } } class ChartDataProviderTests(unittest.TestCase): def setUp(self) -> None: EastmoneyChartClient._cache.clear() EastmoneyChartClient._board_catalog.clear() EastmoneyChartClient._board_catalog_at = 0 self.client = FakeChartClient() def test_stock_intraday_maps_market_and_parses_points(self): payload = self.client.stock_intraday("601318") self.assertEqual(self.client.requests[0][1]["secid"], "1.601318") self.assertEqual(payload["trade_date"], "2026-07-24") self.assertEqual(payload["points"][0]["time"], "09:30") self.assertEqual(payload["points"][0]["average"], 10.2) def test_short_cache_avoids_duplicate_hover_requests(self): self.client.stock_intraday("002141") self.client.stock_intraday("002141") trend_requests = [item for item in self.client.requests if "trends2" in item[0]] self.assertEqual(len(trend_requests), 1) def test_index_and_board_use_the_same_chart_shape(self): index = self.client.index_intraday("000001.SH") board = self.client.board_intraday("BK0474") self.assertEqual(index["points"][1]["close"], 10.15) self.assertEqual(board["points"][1]["volume"], 80.0) secids = [params["secid"] for url, params in self.client.requests if "trends2" in url] self.assertIn("1.000001", secids) self.assertIn("90.BK0474", secids) def test_invalid_identifier_is_rejected(self): with self.assertRaises(ChartDataError): self.client.stock_intraday("abc") class LookbackChartClient(EastmoneyChartClient): def __init__(self) -> None: super().__init__(cache_ttl_seconds=20) self.requests: list[tuple[str, dict[str, str]]] = [] def _request_json(self, url, params, referer): self.requests.append((url, params)) if url == TRENDS_URL and params.get("ndays") == "1": return {"data": {"code": "601318", "name": "中国平安", "preClose": 56.0, "trends": []}} if url == TRENDS_URL and params.get("ndays") == "5": return {"data": {"code": "601318", "name": "中国平安", "preClose": 56.0, "trends": []}} if url == HIS_TRENDS_URL: return { "data": { "code": "601318", "name": "中国平安", "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 ChartDataError("unexpected url") class ChartLookbackTests(unittest.TestCase): def setUp(self) -> None: EastmoneyChartClient._cache.clear() self.client = LookbackChartClient() def test_empty_today_falls_back_to_latest_available_session(self): payload = self.client.stock_intraday("601318") urls = [url for url, _ in self.client.requests] self.assertEqual(urls[0], TRENDS_URL) self.assertEqual(self.client.requests[0][1]["ndays"], "1") self.assertEqual(urls[1], TRENDS_URL) self.assertEqual(self.client.requests[1][1]["ndays"], "5") self.assertEqual(urls[2], HIS_TRENDS_URL) 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_delay_multiday_can_recover_without_his(self): class DelayFive(EastmoneyChartClient): def __init__(self): super().__init__(cache_ttl_seconds=20) self.requests = [] def _request_json(self, url, params, referer): self.requests.append((url, params)) if params.get("ndays") == "1": return {"data": {"code": "000001", "name": "平安银行", "preClose": 11.7, "trends": []}} return { "data": { "code": "000001", "name": "平安银行", "preClose": 11.5, "trends": [ "2026-09-07 09:30,11.50,11.60,11.70,11.40,100,1160.00,11.600", "2026-09-07 15:00,11.70,11.80,11.90,11.60,200,2360.00,11.750", ], } } EastmoneyChartClient._cache.clear() client = DelayFive() payload = client.stock_intraday("000001") self.assertEqual(payload["trade_date"], "2026-09-07") self.assertEqual(len(payload["points"]), 2) self.assertEqual([url for url, _ in client.requests], [TRENDS_URL, TRENDS_URL]) def test_sh_sz_cyb_codes_use_correct_secid(self): for code, secid in (("601318", "1.601318"), ("000001", "0.000001"), ("300750", "0.300750")): EastmoneyChartClient._cache.clear() client = LookbackChartClient() client.stock_intraday(code) self.assertEqual(client.requests[0][1]["secid"], secid) class FakeHub: def __init__(self, chart=None, error=None): self.chart = chart self.error = error self.calls: list[str] = [] def try_intraday(self, code): self.calls.append(code) if self.error: raise self.error return self.chart class DatahubChartFallbackTests(unittest.TestCase): def setUp(self) -> None: EastmoneyChartClient._cache.clear() def test_datahub_success_skips_old_channel(self): hub = FakeHub( { "entity_type": "stock", "identifier": "601318", "name": "中国平安", "code": "601318", "trade_date": "2026-09-08", "previous_close": 56.36, "points": [{"date": "2026-09-08", "time": "09:30", "close": 56.5, "average": 56.4}], "source": "datahub", } ) fallback = LookbackChartClient() client = MarketChartClient(IfindHttpClient(), fallback, hub) payload = client.stock_intraday("601318") self.assertEqual(payload["source"], "datahub") self.assertEqual(hub.calls, ["601318"]) self.assertEqual(fallback.requests, []) def test_datahub_timeout_or_empty_falls_back_to_eastmoney(self): fallback = LookbackChartClient() for hub in ( FakeHub(chart=None), FakeHub(error=RuntimeError("timeout")), FakeHub(error=RuntimeError("datahub exploded")), FakeHub(chart={"points": []}), ): EastmoneyChartClient._cache.clear() fallback.requests.clear() client = MarketChartClient(IfindHttpClient(), fallback, hub) payload = client.stock_intraday("000001") self.assertEqual(payload["trade_date"], "2026-09-07") self.assertGreaterEqual(len(payload["points"]), 1) self.assertTrue(fallback.requests) class ChartServiceStub: @staticmethod def _payload(code: str, name: str): return { "code": code, "name": name, "trade_date": "2026-07-24", "previous_close": 10, "points": [{"date": "2026-07-24", "time": "09:30", "close": 10.1}], } def stock_intraday(self, code): return self._payload(code, "测试股票") def index_intraday(self, identifier): return self._payload(identifier, "上证指数") def board_intraday(self, identifier, name=""): return self._payload("BK0474", name) class ChartDirectoryStub: @staticmethod def get_data_snapshot(kind, cache_key): if (kind, cache_key) != ("search_directory", "ths"): return None return { "schema_version": 2, "items": [ {"id": "881107.TI", "name": "保险", "type": "sector"}, {"id": "885728.TI", "name": "人工智能", "type": "theme"}, ], } class IntradayChartServiceTests(unittest.TestCase): def setUp(self): self.service = DashboardService.__new__(DashboardService) self.service.chart_data = ChartServiceStub() self.service.database = ChartDirectoryStub() def test_stock_index_sector_and_theme_share_display_only_contract(self): cases = ( ("stock", "601318"), ("index", "000001.SH"), ("sector", "881107.TI"), ("theme", "885728.TI"), ) for entity_type, identifier in cases: with self.subTest(entity_type=entity_type): payload = self.service.get_intraday_chart(entity_type, identifier) self.assertEqual(payload["entity"]["type"], entity_type) self.assertEqual(payload["meta"]["trade_date"], "2026-07-24") self.assertEqual(len(payload["points"]), 1) self.assertNotIn("source", payload["meta"]) if __name__ == "__main__": unittest.main()