xiaobai-review/tests/test_hel494_regressions.py

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")