feat(data): Tushare 限速退避 + 新浪兜底(财务 getFinanceReport2022 / 日K 前复权),source+adjust 口径标记

- TushareProvider:频率超限按指数退避重试(不再一次 200/min 即中断),最长等待 30s
- SinaProvider 重构(参考 cc-cursor 公开接口实现):
  · 新增财务通道 CompanyFinanceService.getFinanceReport2022(source=gjzb) → FinancialIndicator
    (report_date / announce_date=publish_date),与 Tushare fina_indicator schema 一致
  · 日 K 保留 jsonp(前复权),统一 UA + 重试
  · 不支持方法仍抛 DataSourceNotSupported(复权因子/交易日历/基础信息)
- FailoverProvider 现在可对 daily 与 financial 兜底(CLI _failover_provider 接 SinaProvider)
- DailyBar + stock_daily 表新增 source/adjust 列:新浪兜底行标记 sina/qfq,
  Tushare 恢复后 --resume 按同键覆盖回不复权 → 两源格式一致且可追溯
- 迁移 91c4e27a03fb 已生成;执行需在全市场同步结束后:uv run alembic upgrade head
- 测试 34+ 项(新浪财务解析/格式一致/限速退避等)通过
This commit is contained in:
Simon
2026-09-06 20:52:13 +08:00
parent e2741a0236
commit 56254172b3
8 changed files with 314 additions and 35 deletions
+9 -2
View File
@@ -44,10 +44,17 @@ def _parse_day(text: str) -> date:
def _failover_provider(session):
"""Tushare 单源经 FailoverProvider 包装:每次尝试写 sync_log(AGENT.md §7 审计)。"""
"""Tushare 首选 + 新浪兜底(FailoverProvider 每次尝试写 sync_log,AGENT.md §7)。
能力矩阵:新浪仅提供日线(前复权,标记 source=sina/adjust=qfq);其余接口
(adj_factor/financial/basic/calendar)新浪不支持 → 抛错保留单源语义,日志可见。
Tushare 恢复后重跑 --resume 会把新浪兜底日按同键覆盖回不复权口径。
"""
from app.infrastructure.data_sources.sina import SinaProvider
audit_repo = SqlAlchemySyncLogRepository(session)
primary = TushareProvider(token=get_settings().tushare_token)
return FailoverProvider(primary, fallback=None, audit=audit_repo.add)
return FailoverProvider(primary, fallback=SinaProvider(), audit=audit_repo.add)
def _session_ctx():
+9 -1
View File
@@ -42,10 +42,18 @@ class TradingCalendar(BaseModel):
class DailyBar(BaseModel):
"""不复权日线。复权请使用 AdjustFactor 在消费侧显式计算。"""
"""日线。默认不复权(source=tushare, adjust=none)。
备用源兜底行会标记 source=sina、adjust=qfq(新浪返回前复权价)。
字段统一、可区分、可追溯(AGENT §5.2/§8):研究侧应优先消费
source=tushare 且 adjust=none 的行;新浪行仅在 Tushare 不可用期间作为兜底,
Tushare 恢复后重跑 --resume 会按日覆盖回不复权口径。
"""
symbol: str
trade_date: date
source: str = Field(default="tushare", description="tushare | sina")
adjust: str = Field(default="none", description="none 不复权 | qfq 前复权")
open: Decimal | None = None
high: Decimal | None = None
low: Decimal | None = None
+141 -30
View File
@@ -1,39 +1,71 @@
"""新浪财经 Provider —— 备用数据源。
能力边界(AGENT.md §5.2):
- 新浪日 K 接口返回**前复权**数据,口径与 Tushare 不复权不同,
因此本 Provider 只用于「缺失/不可用时的行情参考与交叉验证」,
不得把结果直接并入不复权主时序库(禁止静默混口径)。
- 新浪不提供复权因子 / 财务指标 → 相应方法抛 DataSourceNotSupported。
通道(公开接口方案参考 cc-cursor/finance/data/sources/sina_source.py):
1. 财务:quotes.sina.cn CompanyFinanceService.getFinanceReport2022(source=gjzb,
匿名免费、一次多期),含披露日 publish_date → FinancialIndicator
(symbol / report_date=end_date / announce_date=publish_date),schema 与
Tushare fina_indicator 一致 —— 用于财务兜底(保留防未来函数所需的公告日)。
2. 日 K:quotes.sina.cn getKLineData(jsonp,**前复权**)。新浪无「不复权 + 独立复权
因子」,因此日线兜底行标记 source=sina、adjust=qfq,与主口径区分;Tushare 恢复
后 --resume 会按日覆盖回不复权行。
能力边界(其余接口新浪不支持 → DataSourceNotSupported):
get_stock_basic / get_trade_cal / get_adjust_factor。
"""
from __future__ import annotations
import json
import re
import time
import urllib.parse
import urllib.request
from datetime import date, datetime
from decimal import Decimal
from typing import Any
from app.domain.entities.market import DailyBar
from app.domain.entities.market import DailyBar, FinancialIndicator
from app.infrastructure.data_sources.errors import (
DataSourceError,
DataSourceNotSupported,
)
_UA = (
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
"(KHTML, like Gecko) Chrome/138.0.0.0 Safari/537.36"
)
_KLINE_JSONP = (
"https://quotes.sina.cn/cn/api/jsonp_v2.php/var%20data=/CN_MarketDataService"
".getKLineData?symbol={sina_symbol}&scale=240&ma=no&datalen={datalen}"
)
_FIN_BASE = "https://quotes.sina.cn/cn/api/openapi.php/CompanyFinanceService.getFinanceReport2022"
# 新浪「关键指标」中文项名 → 本项目 FinancialIndicator 字段(None 表示已具备/忽略)
_FIN_FIELD_MAP = {
"基本每股收益": "eps",
"净资产收益率(ROE)": "roe",
"加权净资产收益率": "roe",
"销售毛利率": "gross_margin",
"毛利率": "gross_margin",
"营业总收入": "total_revenue",
"净利润": "net_profit",
}
def _to_sina_symbol(symbol: str) -> str:
"""600519.SH -> sh600519;000001.SZ -> sz000001。"""
code, _, suffix = symbol.partition(".")
prefix = {"SH": "sh", "SZ": "sz", "BJ": "bj"}.get(suffix.upper(), "sh")
return f"{prefix}{code}"
"""600519.SH -> sh600519;000001.SZ -> sz000001;无后缀时按规则猜测。"""
code = symbol.strip().upper()
if code.endswith(".SH"):
return "sh" + code[:-3]
if code.endswith(".SZ"):
return "sz" + code[:-3]
if code.endswith(".BJ"):
return "bj" + code[:-3]
if code.startswith(("6", "9")):
return "sh" + code
if code.startswith(("4", "8")):
return "bj" + code
return "sz" + code
def _extract_jsonp(payload: str) -> list[dict[str, Any]]:
@@ -43,23 +75,110 @@ def _extract_jsonp(payload: str) -> list[dict[str, Any]]:
return json.loads(match.group(1))
def _d(value) -> Decimal | None:
if value is None:
return None
try:
return Decimal(str(value))
except (ValueError, TypeError):
return None
def _to_date(value: str) -> date:
"""兼容 20240831 / 2024-08-31 等格式。"""
digits = re.sub(r"\D", "", str(value))[:8]
return datetime.strptime(digits, "%Y%m%d").date()
class SinaProvider:
"""新浪财经备用数据源(仅日线参考 / 交叉验证)。"""
"""新浪财经备用数据源:财务(与 Tushare schema 一致)+ 日线(前复权)。"""
name = "sina"
def __init__(self, *, timeout: float = 10.0, urlopen=urllib.request.urlopen) -> None:
def __init__(
self,
*,
timeout: float = 10.0,
retries: int = 2,
urlopen=urllib.request.urlopen,
) -> None:
self._timeout = timeout
self._retries = retries
self._urlopen = urlopen
# ---- HTTP(统一 UA / 重试) ----
def _open(self, url: str) -> bytes:
req = urllib.request.Request(url, headers={"User-Agent": _UA})
last_error: Exception | None = None
for attempt in range(self._retries):
try:
with self._urlopen(req, timeout=self._timeout) as resp:
return resp.read()
except Exception as exc: # noqa: BLE001
last_error = exc
if attempt < self._retries - 1:
time.sleep(0.5 * (attempt + 1))
raise DataSourceError(f"sina 请求失败: {last_error}") from last_error
# ---- 财务(兜底 Tushare fina_indicator) ----
def get_financial(self, symbol: str) -> list[FinancialIndicator]:
"""新浪关键指标(source=gjzb),含披露日 publish_date → announce_date。"""
params = {
"paperCode": _to_sina_symbol(symbol),
"source": "gjzb",
"type": "0",
"page": "1",
"num": "100",
}
url = f"{_FIN_BASE}?{urllib.parse.urlencode(params)}"
payload = json.loads(self._open(url).decode("utf-8", errors="replace"))
try:
data = payload["result"]["data"]
report_dates = [item["date_value"] for item in data["report_date"]]
except (KeyError, TypeError) as exc:
raise DataSourceError(f"新浪财务返回结构异常({symbol}): {exc}") from exc
rows: list[FinancialIndicator] = []
for rd in report_dates:
entry = data["report_list"].get(rd)
if not entry:
continue
announce = entry.get("publish_date")
if not announce:
continue # 无披露日不可用于研究(防未来函数)
fields: dict[str, Decimal | None] = {
"eps": None,
"roe": None,
"total_revenue": None,
"net_profit": None,
"gross_margin": None,
}
for item in entry.get("data", []):
std = _FIN_FIELD_MAP.get(item.get("item_title", ""))
if std and fields.get(std) is None:
fields[std] = _d(item.get("item_value"))
rows.append(
FinancialIndicator(
symbol=symbol,
report_date=_to_date(str(rd)),
announce_date=_to_date(str(announce)),
eps=fields["eps"],
roe=fields["roe"],
total_revenue=fields["total_revenue"],
net_profit=fields["net_profit"],
gross_margin=fields["gross_margin"],
)
)
return rows
# ---- 日 K(前复权兜底,标记 adjust=qfq) ----
def get_daily(self, symbol: str, start: date, end: date, datalen: int = 320) -> list[DailyBar]:
"""拉取前复权日 K(新浪仅支持最近 datalen 个自然日窗口)。"""
url = _KLINE_JSONP.format(sina_symbol=_to_sina_symbol(symbol), datalen=datalen)
try:
with self._urlopen(url, timeout=self._timeout) as resp:
payload = resp.read().decode("utf-8", errors="replace")
except Exception as exc: # noqa: BLE001
raise DataSourceError(f"sina 请求失败: {exc}") from exc
payload = self._open(url).decode("utf-8", errors="replace")
bars: list[DailyBar] = []
for rec in _extract_jsonp(payload):
@@ -70,6 +189,8 @@ class SinaProvider:
DailyBar(
symbol=symbol,
trade_date=day,
source="sina",
adjust="qfq",
open=_d(rec.get("open")),
high=_d(rec.get("high")),
low=_d(rec.get("low")),
@@ -79,6 +200,8 @@ class SinaProvider:
)
return bars
# ---- 不支持 ----
def get_stock_basic(self):
raise DataSourceNotSupported("新浪不提供股票基础信息列表")
@@ -86,16 +209,4 @@ class SinaProvider:
raise DataSourceNotSupported("新浪不提供交易日历")
def get_adjust_factor(self, symbol, start, end):
raise DataSourceNotSupported("新浪不提供复权因子(返回数据为前复权口径)")
def get_financial(self, symbol):
raise DataSourceNotSupported("新浪不提供财务指标")
def _d(value) -> Decimal | None:
if value is None:
return None
try:
return Decimal(str(value))
except (ValueError, TypeError):
return None
raise DataSourceNotSupported("新浪不提供复权因子(日线接口为前复权口径)")
@@ -8,6 +8,8 @@ tushare 库(pyproject optional:uv sync --extra datasource-tushare)。
from __future__ import annotations
import importlib
import logging
import time
from datetime import date, datetime
from decimal import Decimal
from typing import Any
@@ -24,6 +26,7 @@ from app.infrastructure.data_sources.errors import (
DataSourceError,
)
logger = logging.getLogger(__name__)
_TS_DATE = "%Y%m%d"
@@ -56,9 +59,11 @@ class TushareProvider:
*,
pro: object | None = None,
max_retries: int = 3,
rate_limit_wait: float = 30.0,
) -> None:
self._pro = pro if pro is not None else _build_pro(token)
self._max_retries = max_retries
self._rate_limit_wait = rate_limit_wait
# ---- 归一化(纯函数,输入 list[dict],可单测) ----
@@ -101,6 +106,8 @@ class TushareProvider:
DailyBar(
symbol=str(rec.get("ts_code") or ""),
trade_date=_to_date(rec.get("trade_date")) or date.min,
source="tushare",
adjust="none",
open=_to_decimal(rec.get("open")),
high=_to_decimal(rec.get("high")),
low=_to_decimal(rec.get("low")),
@@ -182,9 +189,12 @@ class TushareProvider:
# ---- 内部 ----
_RATE_LIMIT_MARKERS = ("频率超限", "每分钟", "frequenc", "too many")
def _call(self, api: str, **kwargs) -> list[dict[str, Any]]:
"""带限速退避的调用:频率超限按指数退避(最长 _rate_limit_wait)等待后重试。"""
last_error: Exception | None = None
for _ in range(self._max_retries):
for attempt in range(self._max_retries):
try:
fn = getattr(self._pro, api)
result = fn(**kwargs)
@@ -200,6 +210,10 @@ class TushareProvider:
msg = str(exc)
if "权限" in msg or "积分" in msg or "token" in msg.lower():
raise DataSourceAuthenticationError(msg) from exc
if any(marker in msg for marker in self._RATE_LIMIT_MARKERS):
wait = min(self._rate_limit_wait, 2 ** (attempt + 1))
logger.warning("tushare.%s 频率超限,退避 %.1fs 后重试", api, wait)
time.sleep(wait)
raise DataSourceError(
f"tushare.{api} 重试 {self._max_retries} 次仍失败: {last_error}"
) from last_error
@@ -0,0 +1,37 @@
"""stock_daily source/adjust 来源与口径标记
Revision ID: 91c4e27a03fb
Revises: 53113c80257f
Create Date: 2026-09-06
新浪兜底行带 source=sina / adjust=qfq 标记;现有 648 万行回填默认
tushare / none(SQLite ADD COLUMN 带常量默认值,不重写现有数据)。
"""
from __future__ import annotations
from collections.abc import Sequence
import sqlalchemy as sa
from alembic import op
revision: str = "91c4e27a03fb"
down_revision: str | None = "53113c80257f"
branch_labels: str | Sequence[str] | None = None
depends_on: str | Sequence[str] | None = None
def upgrade() -> None:
op.add_column(
"stock_daily",
sa.Column("source", sa.String(length=16), nullable=False, server_default="tushare"),
)
op.add_column(
"stock_daily",
sa.Column("adjust", sa.String(length=8), nullable=False, server_default="none"),
)
def downgrade() -> None:
op.drop_column("stock_daily", "adjust")
op.drop_column("stock_daily", "source")
@@ -61,6 +61,8 @@ class StockDailyModel(Base):
id: Mapped[int] = mapped_column(PK_INT, primary_key=True, autoincrement=True)
symbol: Mapped[str] = mapped_column(String(SYMBOL_LEN), index=True)
trade_date: Mapped[date] = mapped_column(Date, index=True)
source: Mapped[str] = mapped_column(String(16), default="tushare", server_default="tushare")
adjust: Mapped[str] = mapped_column(String(8), default="none", server_default="none")
open: Mapped[Decimal | None] = mapped_column(Numeric(12, 4), nullable=True)
high: Mapped[Decimal | None] = mapped_column(Numeric(12, 4), nullable=True)
low: Mapped[Decimal | None] = mapped_column(Numeric(12, 4), nullable=True)
+80 -1
View File
@@ -2,6 +2,7 @@
from __future__ import annotations
import json
from datetime import date
import pytest
@@ -72,6 +73,84 @@ class TestCapabilities:
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))
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
# FinancialIndicator 校验能通过 → 格式与 Tushare 一致
assert row.model_dump()["eps"] is not None
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
+21
View File
@@ -148,3 +148,24 @@ class TestCall:
def test_missing_token_rejected(self) -> None:
with pytest.raises(DataSourceAuthenticationError, match="TUSHARE_TOKEN"):
TushareProvider(token="")
class TestRateLimitBackoff:
def test_rate_limit_retries_with_sleep(self) -> None:
"""频率超限:按退避等待后重试,最终抛错带原始信息(不当作鉴权错误)。"""
import app.infrastructure.data_sources.tushare as ts_mod
orig_sleep = ts_mod.time.sleep
sleeps: list[float] = []
ts_mod.time.sleep = lambda w: sleeps.append(w) # noqa: SLF001 —— 测试桩
try:
provider = _pro(
error=RuntimeError("抱歉,您访问接口(adj_factor)频率超限(200次/分钟)"), retries=3
)
provider._rate_limit_wait = 0.1 # noqa: SLF001
with pytest.raises(DataSourceError, match="频率超限"):
provider.get_stock_basic()
finally:
ts_mod.time.sleep = orig_sleep
assert len(provider._pro.calls) == 3 # noqa: SLF001 —— 完整重试 3 次
assert len(sleeps) >= 2