from __future__ import annotations import unittest from unittest.mock import MagicMock, patch from backend.data.providers.tushare_client import TushareClient, TushareError from backend.data.providers.tushare_helpers import _moneyflow_payload from backend.data.realtime import ( WebRealtimeAggregator, _normalize_eastmoney_limit_row, _normalize_eastmoney_stock_quote, ) from backend.data.providers.tushare_daily import DailyMarketMixin class MoneyflowPayloadTests(unittest.TestCase): def test_missing_row_is_not_zero(self) -> None: payload = _moneyflow_payload(None) self.assertFalse(payload["available"]) self.assertIsNone(payload["net_million"]) self.assertIsNone(payload["large_million"]) def test_empty_row_is_not_zero(self) -> None: payload = _moneyflow_payload({}) self.assertFalse(payload["available"]) self.assertIsNone(payload["net_million"]) def test_real_zero_net_is_kept_when_source_exists(self) -> None: payload = _moneyflow_payload( { "net_mf_amount": 0, "buy_lg_amount": 1, "sell_lg_amount": 1, "buy_elg_amount": 0, "sell_elg_amount": 0, "buy_md_amount": 0, "sell_md_amount": 0, "buy_sm_amount": 0, "sell_sm_amount": 0, } ) self.assertTrue(payload["available"]) self.assertEqual(payload["net_million"], 0) class LimitOverlayTests(unittest.TestCase): def test_normalize_limit_keeps_missing_seal_as_none(self) -> None: row = DailyMarketMixin._normalize_limit( { "ts_code": "000737.SZ", "name": "北方铜业", "close": 12.3, "pct_chg": 10, "amount": 1e8, "amount_unit": "yuan", }, "涨停", ) self.assertIsNone(row["seal_amount_million"]) self.assertEqual(row["first_time"], "--") def test_overlay_fills_board_times_from_official_list(self) -> None: mixin = DailyMarketMixin() mixin._load_limit_lists = lambda trade_date: [ { "ts_code": "000737.SZ", "first_time": "09:31:02", "last_time": "10:18:11", "fd_amount": 82000000, "open_times": 1, "turnover_ratio": 18.4, } ] mixin.realtime_aggregator = None rows = mixin._overlay_board_fields( [{"ts_code": "000737.SZ", "close": 12.3, "limit_type": "U"}], "20260908", ) self.assertEqual(rows[0]["first_time"], "09:31:02") self.assertEqual(rows[0]["fd_amount"], 82000000) self.assertEqual(rows[0]["turnover_ratio"], 18.4) class ShenwanRealtimeSourceTests(unittest.TestCase): def test_transport_refuses_rt_sw_k(self) -> None: client = TushareClient(token="demo") with self.assertRaisesRegex(TushareError, "rt_sw_k is disabled"): client.query("rt_sw_k", {"ts_code": "801074.SI"}) def test_outer_realtime_uses_eastmoney_shenwan_not_rt_sw_k(self) -> None: client = TushareClient(token="demo") client.query = MagicMock(side_effect=AssertionError("should not call tushare")) client.realtime_aggregator = MagicMock() client.realtime_aggregator.eastmoney_shenwan_quote.return_value = { "code": "801074.SI", "name": "工业金属", "close": 1234.5, "pre_close": 1200, "change": 2.88, "pct_change": 2.88, "quote_date": "20260908", "quote_time": "2026-09-08T14:50:00+08:00", "source": "eastmoney_sw", } row, source, error = client._sw_outer_realtime("801074.SI", "工业金属", "20260908") self.assertEqual(source, "eastmoney_sw") self.assertEqual(error, "") self.assertEqual(row["change"], 2.88) client.query.assert_not_called() def test_outer_waiting_state_has_no_permission_error(self) -> None: client = TushareClient(token="demo") client.realtime_aggregator = None row, source, error = client._sw_outer_realtime( "801074.SI", "工业金属", "20260908", finalized=True ) self.assertEqual(row, {}) self.assertIn("尚未入库", error) self.assertNotIn("权限", error) self.assertNotIn("rt_sw_k", error) class EastmoneyHelperTests(unittest.TestCase): def test_limit_pool_row_keeps_board_clock(self) -> None: parsed = _normalize_eastmoney_limit_row( { "c": "000737", "m": 0, "n": "北方铜业", "fbt": 93102, "lbt": 101811, "zbc": 1, "lbc": 2, "hs": 18.4, "fund": 82000000, }, "U", ) self.assertEqual(parsed["ts_code"], "000737.SZ") self.assertEqual(parsed["first_time"], "09:31:02") self.assertEqual(parsed["last_time"], "10:18:11") self.assertEqual(parsed["fd_amount"], 82000000) def test_stock_quote_keeps_moneyflow_when_present(self) -> None: quote = _normalize_eastmoney_stock_quote( { "f43": 12.3, "f60": 11.18, "f46": 11.2, "f44": 12.3, "f45": 11.1, "f47": 1000, "f48": 150000000, "f58": "北方铜业", "f86": 0, "f168": 8.5, "f62": 25000000, "f78": 3000000, "f84": -1000000, }, "000737.SZ", ) self.assertEqual(quote["net_mf_amount"], 2500) payload = _moneyflow_payload(quote) self.assertTrue(payload["available"]) self.assertEqual(payload["net_million"], 25) @patch.object(WebRealtimeAggregator, "_get_json") def test_shenwan_quote_uses_eastmoney_90_prefix(self, get_json: MagicMock) -> None: get_json.return_value = { "rc": 0, "data": { "diff": [ { "f12": "801074", "f14": "工业金属", "f2": 1234.5, "f3": 2.88, "f18": 1200, "f17": 1205, "f15": 1240, "f16": 1198, "f6": 1, "f124": 1757319000, } ] }, } quote = WebRealtimeAggregator().eastmoney_shenwan_quote("801074.SI") self.assertEqual(quote["source"], "eastmoney_sw") self.assertAlmostEqual(quote["change"], 2.88) params = get_json.call_args.args[1] self.assertEqual(params["secids"], "90.801074") class ChartWindowTests(unittest.TestCase): def test_display_window_is_45_not_250(self) -> None: from backend.features.market.charts import DAILY_CHART_LIMIT self.assertEqual(DAILY_CHART_LIMIT, 45) class MemberQuoteCoverageTests(unittest.TestCase): def test_prefers_full_hub_market_over_truncated_named_quotes(self) -> None: client = TushareClient(token="demo") wanted = [f"{index:06d}.SZ" for index in range(205)] market = [ {"ts_code": code, "close": 10.0, "pre_close": 9.0} for code in wanted ] client.try_market_quotes = MagicMock(return_value=market) client.try_quotes = MagicMock(return_value=market[:60]) client.realtime_aggregator = MagicMock() rows, source = client._load_member_realtime_quotes(wanted, "20260908") self.assertEqual(len(rows), 205) self.assertEqual(source, "datahub") client.try_quotes.assert_not_called() def test_ignores_non_member_quotes_from_market_snapshot(self) -> None: client = TushareClient(token="demo") client.try_market_quotes = MagicMock( return_value=[ {"ts_code": "000737.SZ", "close": 12.3, "pre_close": 11.2}, {"ts_code": "600000.SH", "close": 10.0, "pre_close": 9.9}, ] ) client.try_quotes = MagicMock(return_value=[]) client._free_realtime_quotes = MagicMock(return_value=([], "empty")) rows, _source = client._load_member_realtime_quotes( ["000737.SZ", "000630.SZ"], "20260908" ) self.assertEqual([row["ts_code"] for row in rows], ["000737.SZ"]) def test_closed_keeps_daily_inner_when_sw_daily_missing(self) -> None: client = TushareClient(token="demo") client.resolve_trade_context = lambda _date: ("20260908", "20260907") client.sw_stock_industry = MagicMock( return_value={"l2_code": "801074.SI", "l2_name": "工业金属"} ) client._sw_sector_members = MagicMock( return_value=[ {"ts_code": "000737.SZ", "name": "北方铜业"}, {"ts_code": "000630.SZ", "name": "铜陵有色"}, ] ) client._stock_listing_reference = MagicMock(return_value={}) client._load_daily = MagicMock( return_value=[ {"ts_code": "000737.SZ", "name": "北方铜业", "pct_chg": 2, "amount": 1e8}, {"ts_code": "000630.SZ", "name": "铜陵有色", "pct_chg": 1, "amount": 1e8}, ] ) client._confirmed_suspended_members = MagicMock(return_value=[]) client.query = MagicMock(return_value=[]) client._sw_realtime_sector_snapshot = MagicMock( side_effect=AssertionError("daily inner should be kept") ) client.realtime_aggregator = MagicMock() client.realtime_aggregator.eastmoney_shenwan_quote.return_value = { "code": "801074.SI", "name": "工业金属", "change": 1.5, "pct_change": 1.5, "quote_date": "20260908", "quote_time": "2026-09-08T15:00:00+08:00", "source": "eastmoney_sw", } snapshot = client.sw_sector_snapshot( "000737.SZ", "20260908", allow_realtime_close=True ) self.assertEqual(snapshot["quote_count"], 2) self.assertEqual(snapshot["member_count"], 2) self.assertTrue(snapshot["inner_precise"]) self.assertTrue(snapshot["outer_precise"]) self.assertEqual(snapshot["inner_source"], "tushare_member_daily") self.assertEqual(snapshot["change"], 1.5) self.assertNotIn("权限", snapshot.get("outer_error") or "") self.assertNotIn("rt_sw_k", snapshot.get("outer_error") or "")