qlib/backend/tests/test_sina_provider.py

"""新浪 Provider 测试：代码转换、JSONP 解析、能力边界（不触网）。"""

from __future__ import annotations

import json
from datetime import date

import pytest
from app.infrastructure.data_sources.errors import (
    DataSourceError,
    DataSourceNotSupported,
)
from app.infrastructure.data_sources.sina import SinaProvider, _extract_jsonp, _to_sina_symbol

_KLINE_OK = (
    'var data=[{"day":"2024-08-30","open":"1700.0","high":"1720.0","low":"1690.0",'
    '"close":"1710.0","volume":"20000"},{"day":"2024-08-31","open":"1710.0",'
    '"high":"1725.0","low":"1705.0","close":"1720.0","volume":"18000"}]'
)


class _FakeResp:
    def __enter__(self):
        return self

    def __exit__(self, *exc):
        return False

    def read(self):
        return _KLINE_OK.encode("utf-8")


class _BoomResp(_FakeResp):
    def read(self):
        raise OSError("socket timeout")


class TestSymbolMap:
    def test_mapping(self) -> None:
        assert _to_sina_symbol("600519.SH") == "sh600519"
        assert _to_sina_symbol("000001.SZ") == "sz000001"
        assert _to_sina_symbol("830001.BJ") == "bj830001"


class TestJsonp:
    def test_extract(self) -> None:
        rows = _extract_jsonp(_KLINE_OK)
        assert len(rows) == 2
        assert rows[0]["close"] == "1710.0"

    def test_bad_payload_raises(self) -> None:
        with pytest.raises(DataSourceError, match="无法解析"):
            _extract_jsonp("not jsonp")


class TestGetDaily:
    def test_ok_with_date_filter(self) -> None:
        provider = SinaProvider(urlopen=lambda _url, **kw: _FakeResp())
        bars = provider.get_daily("600519.SH", date(2024, 8, 30), date(2024, 8, 30))
        assert len(bars) == 1
        assert bars[0].close is not None
        assert bars[0].symbol == "600519.SH"

    def test_network_error_wrapped(self) -> None:
        provider = SinaProvider(urlopen=lambda _url, **kw: _BoomResp())
        with pytest.raises(DataSourceError, match="sina 请求失败"):
            provider.get_daily("600519.SH", date(2024, 8, 1), date(2024, 8, 31))


class TestCapabilities:
    def test_not_supported(self) -> None:
        provider = SinaProvider()
        with pytest.raises(DataSourceNotSupported):
            provider.get_stock_basic()
        with pytest.raises(DataSourceNotSupported):
            provider.get_trade_cal(date(2024, 1, 1), date(2024, 1, 31))
        with pytest.raises(DataSourceNotSupported):
            provider.get_adjust_factor("600519.SH", date(2024, 1, 1), date(2024, 1, 31))


_FIN_PAYLOAD = json.dumps(
    {
        "result": {
            "data": {
                "report_date": [{"date_value": "2024-06-30"}, {"date_value": "2024-03-31"}],
                "report_list": {
                    "2024-06-30": {
                        "publish_date": "2024-08-31",
                        "data": [
                            {"item_title": "基本每股收益", "item_value": "1.23"},
                            {"item_title": "净资产收益率(ROE)", "item_value": "15.5"},
                            {"item_title": "净利润", "item_value": "1230000000"},
                        ],
                    },
                    "2024-03-31": {"publish_date": "", "data": []},
                },
            }
        }
    },
    ensure_ascii=False,
)


class _FinResp:
    def __enter__(self):
        return self

    def __exit__(self, *exc):
        return False

    def read(self):
        return _FIN_PAYLOAD.encode("utf-8")


class TestFinancial:
    def test_parse_schema_parity_with_tushare(self) -> None:
        """财务 schema 与 Tushare 对齐：report_date / announce_date / 指标。"""
        provider = SinaProvider(urlopen=lambda _req, **kw: _FinResp())
        rows = provider.get_financial("600519.SH")
        assert len(rows) == 1  # 无 publish_date 的行被跳过（防未来函数）
        row = rows[0]
        assert row.symbol == "600519.SH"
        assert row.report_date == date(2024, 6, 30)
        assert row.announce_date == date(2024, 8, 31)
        assert float(row.eps) == 1.23
        assert float(row.roe) == 15.5
        assert float(row.net_profit) == 1230000000
        assert row.source == "sina"  # 兜底行必须标记来源（与 Tushare 首选行区分）
        # FinancialIndicator 校验能通过 → 格式与 Tushare 一致
        assert row.model_dump()["eps"] is not None

    def test_window_filter(self) -> None:
        payload = json.dumps(
            {
                "result": {
                    "data": {
                        "report_date": [
                            {"date_value": "2024-06-30"},
                            {"date_value": "2024-03-31"},
                        ],
                        "report_list": {
                            "2024-06-30": {
                                "publish_date": "2024-08-31",
                                "data": [{"item_title": "基本每股收益", "item_value": "1.23"}],
                            },
                            "2024-03-31": {
                                "publish_date": "2024-04-27",
                                "data": [{"item_title": "基本每股收益", "item_value": "0.5"}],
                            },
                        },
                    }
                }
            },
            ensure_ascii=False,
        )

        class _TwoResp:
            def __enter__(self):
                return self

            def __exit__(self, *exc):
                return False

            def read(self):
                return payload.encode("utf-8")

        provider = SinaProvider(urlopen=lambda _req, **kw: _TwoResp())
        rows = provider.get_financial("600519.SH", date(2024, 6, 30), date(2024, 6, 30))
        assert [r.report_date for r in rows] == [date(2024, 6, 30)]

    def test_bad_structure_raises(self) -> None:
        class _BadResp(_FinResp):
            def read(self):
                return b'{"result": {}}'

        provider = SinaProvider(urlopen=lambda _req, **kw: _BadResp())
        try:
            provider.get_financial("600519.SH")
        except DataSourceError:
            return
        raise AssertionError("应抛 DataSourceError")


class TestFormatParity:
    """两源格式一致：统一 DailyBar schema，新浪兜底行带 source/adjust 标记。"""

    def test_sina_bar_marks_source_and_adjust(self) -> None:
        provider = SinaProvider(urlopen=lambda _url, **kw: _FakeResp())
        bars = provider.get_daily("600519.SH", date(2024, 8, 30), date(2024, 8, 30))
        assert len(bars) == 1
        bar = bars[0]
        assert bar.source == "sina"
        assert bar.adjust == "qfq"  # 新浪为前复权，与 Tushare 不复权主口径区分
        # 与 Tushare 同 schema：必备字段齐全
        assert bar.symbol == "600519.SH"
        assert bar.close is not None


class TestJsonpParens:
    def test_real_world_paren_wrapped(self) -> None:
        payload = 'var data=([{"day":"2024-08-30","close":"1710.0"}])'
        rows = _extract_jsonp(payload)
        assert rows[0]["close"] == "1710.0"

    def test_with_leading_comment_and_trailing_semicolon(self) -> None:
        payload = (
            "/*<script>location.href='//sina.com';</script>*/\n"
            'var data=([{"day":"2024-08-30","close":"1710.0"}]);'
        )
        rows = _extract_jsonp(payload)
        assert rows[0]["close"] == "1710.0"