From 93e32f4e63615734e99164337d7a3a034a34c318 Mon Sep 17 00:00:00 2001 From: Simon Date: Wed, 9 Sep 2026 07:29:26 +0800 Subject: [PATCH] =?UTF-8?q?feat(data):=20B1-2=20=E6=8C=87=E6=95=B0?= =?UTF-8?q?=E6=88=90=E5=88=86=E5=90=8C=E6=AD=A5=EF=BC=88Provider=20+=20CLI?= =?UTF-8?q?=20sync=20index=5Fweight=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - MarketDataProvider.get_index_weight（协议）；Tushare 实现 normalize_index_weight + get_index_weight（ts_code=... 全历史成分权重）；Sina 抛 DataSourceNotSupported； FailoverProvider 代理并审计每次尝试 - CLI：sync index_weight --code 000300.SH（拉取→幂等落库 index_weight→打印最新快照； 失败走 sync_log 审计并返回非零） - tests：Tushare 映射与调用（FakePro）、Failover 主源单源语义（新浪不支持被审计）； 全量 pytest 通过 --- backend/app/cli/sync.py | 29 ++++++++++++++++ backend/app/domain/providers.py | 7 ++++ .../infrastructure/data_sources/failover.py | 7 ++++ .../app/infrastructure/data_sources/sina.py | 3 ++ .../infrastructure/data_sources/tushare.py | 28 ++++++++++++++++ backend/tests/test_failover.py | 33 +++++++++++++++++++ backend/tests/test_tushare_provider.py | 33 +++++++++++++++++++ 7 files changed, 140 insertions(+) diff --git a/backend/app/cli/sync.py b/backend/app/cli/sync.py index 191f53f..3407e16 100644 --- a/backend/app/cli/sync.py +++ b/backend/app/cli/sync.py @@ -41,6 +41,9 @@ from app.infrastructure.data_sources.errors import DataSourceError from app.infrastructure.data_sources.sina import SinaProvider from app.infrastructure.data_sources.tushare import TushareProvider from app.infrastructure.persistence.sqlalchemy.models.market import StockModel +from app.infrastructure.persistence.sqlalchemy.repositories.index_impl import ( + SqlAlchemyIndexConstituentRepository, +) from app.infrastructure.persistence.sqlalchemy.repositories.market_impl import ( SqlAlchemyAdjustFactorRepository, SqlAlchemyDailyBarRepository, @@ -277,6 +280,28 @@ def cmd_financial(args) -> int: return 0 +def cmd_index_weight(args) -> int: + """同步指数历史成分（Tushare index_weight；新浪不支持 → failover 审计留痕）。""" + + code = args.code + with _session_ctx() as session: + provider = _failover_provider(session) + try: + rows = provider.get_index_weight(code) + except DataSourceError as exc: + print(f"[index_weight] {code} 失败：{exc}") + return 1 + repo = SqlAlchemyIndexConstituentRepository(session) + touched = repo.upsert_many(rows) + latest = repo.latest_date(code) + session.commit() + print( + f"[index_weight] {code} 拉取 {len(rows)} 期成分行，落库 {touched} 条" + f"，最新快照 {latest}（as_of 查询见 Universe.index_code）" + ) + return 0 + + def cmd_verify(args) -> int: """新浪交叉验证：取新浪最新前复权收盘，与本地最新交易日对照。 @@ -399,6 +424,10 @@ def build_parser() -> argparse.ArgumentParser: ) p_fin.set_defaults(func=cmd_financial) + p_idx = sub.add_parser("index_weight", help="同步指数历史成分（如沪深300 000300.SH）") + p_idx.add_argument("--code", required=True, help="指数代码，如 000300.SH / 000905.SH") + p_idx.set_defaults(func=cmd_index_weight) + p_verify = sub.add_parser("verify", help="新浪交叉验证最新行情") p_verify.add_argument("--symbol", required=True) p_verify.set_defaults(func=cmd_verify) diff --git a/backend/app/domain/providers.py b/backend/app/domain/providers.py index 4d1ba81..6031c9d 100644 --- a/backend/app/domain/providers.py +++ b/backend/app/domain/providers.py @@ -9,6 +9,7 @@ from __future__ import annotations from datetime import date from typing import Protocol +from app.domain.entities.index import IndexWeight from app.domain.entities.market import ( AdjustFactor, DailyBar, @@ -49,3 +50,9 @@ class MarketDataProvider(Protocol): start_date/end_date 参数，按报告期过滤）；不传表示全量历史。 新浪接口不支持按窗口拉取，提供方会忽略窗口后由调用方自行过滤。 """ + + def get_index_weight(self, index_code: str) -> list[IndexWeight]: + """指数历史成分（含权重）：每期成分快照 → IndexWeight(index_code, trade_date, symbol)。 + + 供 index_weight 同步与历史成分 Universe（v3 §9）。""" + diff --git a/backend/app/infrastructure/data_sources/failover.py b/backend/app/infrastructure/data_sources/failover.py index 39d42d5..acfdda9 100644 --- a/backend/app/infrastructure/data_sources/failover.py +++ b/backend/app/infrastructure/data_sources/failover.py @@ -60,6 +60,13 @@ class FailoverProvider: fallback_call=lambda: self.fallback.get_daily(symbol, start, end), ) + def get_index_weight(self, index_code: str) -> list: + return self._with_failover( + "get_index_weight", + primary_call=lambda: self.primary.get_index_weight(index_code), + fallback_call=lambda: self.fallback.get_index_weight(index_code), + ) + def get_adjust_factor(self, symbol: str, start: date, end: date) -> list: return self._with_failover( "get_adjust_factor", diff --git a/backend/app/infrastructure/data_sources/sina.py b/backend/app/infrastructure/data_sources/sina.py index a9a03b6..943b005 100644 --- a/backend/app/infrastructure/data_sources/sina.py +++ b/backend/app/infrastructure/data_sources/sina.py @@ -225,6 +225,9 @@ class SinaProvider: def get_stock_basic(self): raise DataSourceNotSupported("新浪不提供股票基础信息列表") + def get_index_weight(self, index_code): + raise DataSourceNotSupported("新浪不提供指数成分接口") + def get_trade_cal(self, start, end): raise DataSourceNotSupported("新浪不提供交易日历") diff --git a/backend/app/infrastructure/data_sources/tushare.py b/backend/app/infrastructure/data_sources/tushare.py index fb6bc4a..59250dc 100644 --- a/backend/app/infrastructure/data_sources/tushare.py +++ b/backend/app/infrastructure/data_sources/tushare.py @@ -14,6 +14,7 @@ from datetime import date, datetime, timedelta from decimal import Decimal from typing import Any +from app.domain.entities.index import IndexWeight from app.domain.entities.market import ( AdjustFactor, DailyBar, @@ -96,6 +97,28 @@ class TushareProvider: for rec in records ] + @staticmethod + def normalize_index_weight( + records: list[dict[str, Any]], index_code_fallback: str = "" + ) -> list[IndexWeight]: + """index_weight 接口行 → IndexWeight（index_code/con_code/trade_date/weight）。""" + out: list[IndexWeight] = [] + for rec in records: + code = str(rec.get("index_code") or index_code_fallback or "") + symbol = str(rec.get("con_code") or "") + if not code or not symbol: + continue + out.append( + IndexWeight( + index_code=code, + index_name=rec.get("index_name"), + trade_date=_to_date(rec.get("trade_date")) or date.min, + symbol=symbol, + weight=_to_decimal(rec.get("weight")), + ) + ) + return out + @staticmethod def normalize_daily(records: list[dict[str, Any]]) -> list[DailyBar]: bars: list[DailyBar] = [] @@ -224,6 +247,11 @@ class TushareProvider: _RATE_LIMIT_MARKERS = ("频率超限", "每分钟", "frequenc", "too many") + def get_index_weight(self, index_code: str) -> list[IndexWeight]: + """指数成分（Tushare index_weight 全历史，ts_code 过滤）。""" + records = self._call("index_weight", ts_code=index_code) + return self.normalize_index_weight(records, index_code_fallback=index_code) + def _call(self, api: str, **kwargs) -> list[dict[str, Any]]: """带限速退避的调用：频率超限按指数退避（最长 _rate_limit_wait）等待后重试。""" last_error: Exception | None = None diff --git a/backend/tests/test_failover.py b/backend/tests/test_failover.py index ca732ea..ebbae23 100644 --- a/backend/tests/test_failover.py +++ b/backend/tests/test_failover.py @@ -91,3 +91,36 @@ class TestFailover: with pytest.raises(DataSourceError, match="无备用源"): f.get_daily("600519.SH", date(2024, 1, 1), date(2024, 1, 5)) assert len(collected) == 1 + + +class TestIndexFailover: + def test_primary_only_index(self) -> None: + """指数成分：主源成功即用（新浪不支持会自动尝试并审计失败 → 结果仍来自主源）。""" + from datetime import date + from decimal import Decimal + + from app.domain.entities.index import IndexWeight + from app.infrastructure.data_sources.failover import FailoverProvider + from app.infrastructure.data_sources.sina import SinaProvider + from app.infrastructure.data_sources.tushare import TushareProvider + + class _P(TushareProvider): + name = "tushare" + + def __init__(self): # 不触网 + pass + + def get_index_weight(self, index_code): + return [IndexWeight(index_code=index_code, index_name="沪深300", + trade_date=date(2024, 6, 28), symbol="600519.SH", + weight=Decimal("1"))] + + logs: list[SyncLog] = [] + + def audit(log: SyncLog): + logs.append(log) + + f = FailoverProvider(_P(), SinaProvider(), audit=audit) + rows = f.get_index_weight("000300.SH") + assert len(rows) == 1 and rows[0].symbol == "600519.SH" + assert len(logs) >= 1 # 每次尝试已审计 diff --git a/backend/tests/test_tushare_provider.py b/backend/tests/test_tushare_provider.py index bde9247..b6d0f6c 100644 --- a/backend/tests/test_tushare_provider.py +++ b/backend/tests/test_tushare_provider.py @@ -224,3 +224,36 @@ class TestGetFinancialWindow: assert len(fake.kwargs) == 2 assert fake.kwargs[1]["end_date"] < fake.kwargs[0]["end_date"] assert {r.report_date for r in rows} == {date.fromisoformat(e[:4] + "-" + e[4:6] + "-" + e[6:]) for e in newest + older} + + +class TestIndexWeight: + def test_normalize_mapping(self) -> None: + from app.infrastructure.data_sources.tushare import TushareProvider + + rows = TushareProvider.normalize_index_weight( + [ + {"index_code": "000300.SH", "con_code": "600519.SH", + "trade_date": "20240628", "weight": 1.53}, + {"con_code": "000001.SZ", "trade_date": "20240628", "weight": 0.9}, + ], + index_code_fallback="000300.SH", + ) + assert len(rows) == 2 + assert rows[0].index_code == "000300.SH" + assert rows[0].symbol == "600519.SH" + assert rows[0].trade_date.isoformat() == "2024-06-28" + assert float(rows[0].weight) == 1.53 + # 无 index_code 时用 fallback；con_code 缺失跳过 + assert rows[1].index_code == "000300.SH" + + def test_provider_calls_index_weight(self) -> None: + from app.infrastructure.data_sources.tushare import TushareProvider + + pro = FakePro( + payload=[{"index_code": "000300.SH", "con_code": "600519.SH", + "trade_date": "20240628", "weight": 1.0}] + ) + p = TushareProvider(token="t", pro=pro) + rows = p.get_index_weight("000300.SH") + assert pro.calls == ["index_weight"] + assert len(rows) == 1 and rows[0].symbol == "600519.SH"