Simon
2da234220a
- domain：市场数据实体（Stock / 交易日历 / 日线 / 复权 / 财务含 announce_date）+ Repository 与 MarketDataProvider Protocol
- 数据源：TushareProvider（归一化、重试、鉴权错误归类）、SinaProvider（备用，明确前复权口径与能力边界）、FailoverProvider + SyncLog 审计（禁止静默切换）
- 持久化：SQLAlchemy 2.x Models + Repository 实现（按业务键幂等 upsert、as_of_date 防未来函数过滤）+ Alembic 迁移
- CLI：uv run python -m app.cli.sync {basic|calendar|daily|financial|verify}，支持 --resume 断点续传
- 真实 Tushare 验证：stock 5556 / 交易日历 366 / daily+factor 242 / 财务 55；sync_log 审计完整
- 测试：38 passed（domain / provider / failover / repository / 未来函数 / 迁移），ruff clean
78 lines
2.4 KiB
Python
78 lines
2.4 KiB
Python
"""新浪 Provider 测试：代码转换、JSONP 解析、能力边界（不触网）。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
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_adjust_factor("600519.SH", date(2024, 1, 1), date(2024, 1, 31))
|
|
with pytest.raises(DataSourceNotSupported):
|
|
provider.get_financial("600519.SH")
|