From 56254172b395326b6768f4baf588087c10759fb5 Mon Sep 17 00:00:00 2001 From: Simon Date: Sun, 6 Sep 2026 20:52:13 +0800 Subject: [PATCH] =?UTF-8?q?feat(data):=20Tushare=20=E9=99=90=E9=80=9F?= =?UTF-8?q?=E9=80=80=E9=81=BF=20+=20=E6=96=B0=E6=B5=AA=E5=85=9C=E5=BA=95?= =?UTF-8?q?=EF=BC=88=E8=B4=A2=E5=8A=A1=20getFinanceReport2022=20/=20?= =?UTF-8?q?=E6=97=A5K=20=E5=89=8D=E5=A4=8D=E6=9D=83=EF=BC=89=EF=BC=8Csourc?= =?UTF-8?q?e+adjust=20=E5=8F=A3=E5=BE=84=E6=A0=87=E8=AE=B0?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 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+ 项（新浪财务解析/格式一致/限速退避等）通过 --- backend/app/cli/sync.py | 11 +- backend/app/domain/entities/market.py | 10 +- .../app/infrastructure/data_sources/sina.py | 171 +++++++++++++++--- .../infrastructure/data_sources/tushare.py | 16 +- ...4e27a03fb_stock_daily_source_and_adjust.py | 37 ++++ .../persistence/sqlalchemy/models/market.py | 2 + backend/tests/test_sina_provider.py | 81 ++++++++- backend/tests/test_tushare_provider.py | 21 +++ 8 files changed, 314 insertions(+), 35 deletions(-) create mode 100644 backend/app/infrastructure/persistence/migrations/versions/91c4e27a03fb_stock_daily_source_and_adjust.py diff --git a/backend/app/cli/sync.py b/backend/app/cli/sync.py index 930ba98..605c171 100644 --- a/backend/app/cli/sync.py +++ b/backend/app/cli/sync.py @@ -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(): diff --git a/backend/app/domain/entities/market.py b/backend/app/domain/entities/market.py index bd58cb3..810a514 100644 --- a/backend/app/domain/entities/market.py +++ b/backend/app/domain/entities/market.py @@ -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 diff --git a/backend/app/infrastructure/data_sources/sina.py b/backend/app/infrastructure/data_sources/sina.py index 6b9be3d..7a09e0c 100644 --- a/backend/app/infrastructure/data_sources/sina.py +++ b/backend/app/infrastructure/data_sources/sina.py @@ -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("新浪不提供复权因子（日线接口为前复权口径）") diff --git a/backend/app/infrastructure/data_sources/tushare.py b/backend/app/infrastructure/data_sources/tushare.py index c51d4f1..baa2bb9 100644 --- a/backend/app/infrastructure/data_sources/tushare.py +++ b/backend/app/infrastructure/data_sources/tushare.py @@ -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 diff --git a/backend/app/infrastructure/persistence/migrations/versions/91c4e27a03fb_stock_daily_source_and_adjust.py b/backend/app/infrastructure/persistence/migrations/versions/91c4e27a03fb_stock_daily_source_and_adjust.py new file mode 100644 index 0000000..b8ca73c --- /dev/null +++ b/backend/app/infrastructure/persistence/migrations/versions/91c4e27a03fb_stock_daily_source_and_adjust.py @@ -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") diff --git a/backend/app/infrastructure/persistence/sqlalchemy/models/market.py b/backend/app/infrastructure/persistence/sqlalchemy/models/market.py index 8132eea..9fe947d 100644 --- a/backend/app/infrastructure/persistence/sqlalchemy/models/market.py +++ b/backend/app/infrastructure/persistence/sqlalchemy/models/market.py @@ -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) diff --git a/backend/tests/test_sina_provider.py b/backend/tests/test_sina_provider.py index 94f39e2..6c3775f 100644 --- a/backend/tests/test_sina_provider.py +++ b/backend/tests/test_sina_provider.py @@ -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 diff --git a/backend/tests/test_tushare_provider.py b/backend/tests/test_tushare_provider.py index 7472a1a..2215c9f 100644 --- a/backend/tests/test_tushare_provider.py +++ b/backend/tests/test_tushare_provider.py @@ -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