diff --git a/data/feed/index_feed.py b/data/feed/index_feed.py index 820bdb6..3314b43 100644 --- a/data/feed/index_feed.py +++ b/data/feed/index_feed.py @@ -14,13 +14,10 @@ from __future__ import annotations -import json -from pathlib import Path - import pandas as pd +from data.feed.secret import read_token from data.feed.throttle import request_with_retry -from data.feed.tushare_feed import _lookup_dotted # canonical constituents columns CONSTITUENT_COLUMNS: tuple[str, ...] = ("date", "symbol", "weight") @@ -49,26 +46,12 @@ def __init__( self._cache = cache # -- secret handling (token never logged) ------------------------------- # - def _read_token(self) -> str: - path = Path(self._secret_file) - if not path.exists(): - raise ValueError( - f"Secret config file not found: {self._secret_file}. " - f"Set data.external_secret_file to your .config.json path." - ) - try: - data = json.loads(path.read_text(encoding="utf-8")) - except json.JSONDecodeError as exc: - raise ValueError( - f"Secret config file is not valid JSON: {self._secret_file} ({exc.msg})." - ) from None - return _lookup_dotted(data, self._token_key) - def _client(self): if self._pro is None: import tushare as ts - self._pro = ts.pro_api(self._read_token()) # token handed straight in + # token via the shared external-config reader, handed straight in + self._pro = ts.pro_api(read_token(self._secret_file, self._token_key)) return self._pro # tushare index_weight caps a single response at ~6000 rows; a ~300-name diff --git a/data/feed/tushare_feed.py b/data/feed/tushare_feed.py index 7ed1ffe..dd9c29b 100644 --- a/data/feed/tushare_feed.py +++ b/data/feed/tushare_feed.py @@ -17,13 +17,11 @@ from __future__ import annotations -import json -from pathlib import Path - import pandas as pd from data.clean.schema import CORE_COLUMNS, normalize_panel from data.feed.base import DataFeed +from data.feed.secret import read_token from data.feed.throttle import request_with_retry # tushare raw column -> canonical column. @@ -34,27 +32,6 @@ } -def _lookup_dotted(data: dict, dotted_key: str) -> str: - """Resolve a dotted key (e.g. 'tushare.token') in a nested dict. - - Raises a readable ValueError if any segment is missing — the message names the - missing key path but NEVER echoes any value (so no secret can leak). - """ - node: object = data - for part in dotted_key.split("."): - if not isinstance(node, dict) or part not in node: - raise ValueError( - f"Secret config is missing key '{dotted_key}'. " - f"Expected a nested entry reachable via that dotted path." - ) - node = node[part] - if not isinstance(node, str) or not node: - raise ValueError( - f"Secret config key '{dotted_key}' must map to a non-empty string token." - ) - return node - - class TushareFeed(DataFeed): """A DataFeed backed by the tushare pro API. @@ -85,20 +62,8 @@ def __init__( # -- secret handling ---------------------------------------------------- # def _read_token(self) -> str: - """Read the token from the external json config (dotted-key lookup).""" - path = Path(self._secret_file) - if not path.exists(): - raise ValueError( - f"Secret config file not found: {self._secret_file}. " - f"Set data.external_secret_file to your .config.json path." - ) - try: - data = json.loads(path.read_text(encoding="utf-8")) - except json.JSONDecodeError as exc: - raise ValueError( - f"Secret config file is not valid JSON: {self._secret_file} ({exc.msg})." - ) from None - return _lookup_dotted(data, self._token_key) + """Read the token via the shared external-config reader (dotted-key lookup).""" + return read_token(self._secret_file, self._token_key) def _client(self): """Build (once) and return the tushare pro client. Lazy + no logging.""" diff --git a/docs/data/data_layer_contracts.md b/docs/data/data_layer_contracts.md new file mode 100644 index 0000000..cc2d814 --- /dev/null +++ b/docs/data/data_layer_contracts.md @@ -0,0 +1,56 @@ +# 数据层契约（Data Layer Contracts） + +> 目的：把 **`cache`（缓存）** 与 **`store`（面板存储）** 的边界写成可提交的契约， +> 让后续改动有明确不变量可守。本文件只描述 **当前已实现** 的行为，不预告未实现的阶段。 + +本文件对应 **D1**（边界文档化 + 低风险 token 解析去重）。**D2–D6 尚未实现**，本文件不声称其存在。 + +--- + +## 1. `cache` vs `store` —— 两层职责互不替代 + +| 层 | 路径 | 是什么 | 不是什么 | +|---|---|---|---| +| **缓存（cache）** | `data/cache/` | **可复用的 endpoint 级 raw 缓存**，是行情/成分/可交易性/财务等原始事实的 source of truth（SoT） | 不是派生数据（不存 qfq、不存因子、不存任何 flag 派生值） | +| **面板存储（store）** | `data/store/panel_store.py`（`PanelStore`） | **单次 run 的 canonical 面板 artifact**（per-run），供该次回测/分析消费 | **不是 raw data lake，不是缓存的 SoT**；`artifacts/data/*.parquet` 绝不当缓存权威 | + +要点： + +- **缓存是跨 run 复用的原始数据底座**；`PanelStore` 是某一次 run 内组装好的面板快照。 + 两者方向不同：缓存向上游（API）负责，`PanelStore` 向下游（本次回测）负责。 +- **缓存只存 raw**。`data/cache/` 下的持久化行（`parquet_store.py` / `intraday_parquet_store.py`） + 一律是未派生的原始 endpoint 事实；覆盖账本（`coverage.py` / `intraday_coverage.py`）只记 + endpoint 元数据，**不含 token、不含 secret**。 +- 日频缓存（`tushare_cache.py`）与分钟缓存（`intraday_cache.py`）各自独立的 ledger / store，互不串用。 + +## 2. 缓存只存 raw —— 所有派生/对齐/校验留在下游 + +以下逻辑 **不属于缓存层**，必须留在缓存的下游（feed/clean/universe/factors/alpha/portfolio/runtime）： + +- **前复权 `front_adjust`**：缓存存未复权 OHLCV + 原始 `adj_factor`；qfq 在内存按现有公式/时机计算。 +- **PIT 指数成分**：`index_weight` raw 快照入缓存；as-of（latest snapshot ≤ d）在 feed/universe 下游判定。 +- **PIT 申万行业**：`index_member_all` 的 in/out 区间 raw 入缓存；按 trade_date as-of 取行业在下游。 +- **财务 `ann_date` 披露日对齐**：财务字段 raw 入缓存；逐字段 `ann_date ≤ trade_date` 的 as-of 在下游。 +- **raw 涨跌停可行性检查**：`stk_limit` raw 价入缓存；限价闸门用 raw（绝不碰 qfq）在执行层判定。 +- **factors / alpha / portfolio / runtime 的全部数学**：均在缓存下游，缓存不参与。 + +换言之：**缓存换不换、暖不暖，下游数学逐字节不变**。这是缓存层 opt-in、默认 disabled 的前提。 + +## 3. `data-update` 只暖缓存，不跑研究 + +- `data-update` CLI（`qt/data_updater.py::run_data_update`）**只做 endpoint 级增量暖缓存**。 +- 它 **不跑 factor / alpha / portfolio / backtest，不写 `PanelStore`**。 +- 真实回测仍各自走 read-through，按需补自己的缺口；`data-update` 只是把常用 endpoint 提前填好。 + +## 4. 不属于 D1 的范围（后续阶段，未实现） + +以下明确 **不在 D1**，本文件 **不声称已实现**： + +- 数据质量校验（data-quality validator）； +- 并发 / 线程池 / 异步抓取（concurrency）； +- `TushareCache` 内部拆分、endpoint schema registry、`CoverageLedger` 存储格式变更、 + `PanelStore` 的 append/partition 特性。 + +D1 仅做两件低风险事：**把以上边界写成本契约文档**，以及 **把 `TushareFeed` / `IndexConstituentsFeed` +里重复的 token 解析收敛到共享的 `data/feed/secret.py::read_token`**（其余 feed 早已使用该共享读取器）。 +数据行为零改动。 diff --git a/tests/test_index_feed.py b/tests/test_index_feed.py index d9f723e..3e38c72 100644 --- a/tests/test_index_feed.py +++ b/tests/test_index_feed.py @@ -1,11 +1,31 @@ -"""IndexConstituentsFeed mapping tests — no network, fake SDK.""" +"""IndexConstituentsFeed mapping tests — no network, fake SDK. + +The token path uses a FAKE tmp config json with a FAKE token; the real +/home/shaofl/.../.config.json is NEVER read and no network call is made. +""" from __future__ import annotations +import io +import json +import logging +from contextlib import redirect_stdout + import pandas as pd +import pytest from data.feed.index_feed import IndexConstituentsFeed +FAKE_TOKEN = "FAKE_TUSHARE_TOKEN_do_not_leak_0123456789abcdef" + + +def _write_fake_config(tmp_path, token: str = FAKE_TOKEN): + """Write a fake nested config json mimicking the real .config.json layout.""" + cfg = {"tushare": {"token": token}, "other": {"unused": 1}} + path = tmp_path / "fake_config.json" + path.write_text(json.dumps(cfg), encoding="utf-8") + return path + class _FakePro: """Stand-in for the tushare pro client returning an index_weight frame.""" @@ -82,3 +102,62 @@ def index_weight(self, **_kw): out = feed.get_constituents("000300.SH", "2024-01-01", "2024-02-29") assert list(out.columns) == ["date", "symbol", "weight"] assert len(out) == 0 + + +# --------------------------------------------------------------------------- # +# Token sourcing — real _client() path through the shared read_token (D1) +# --------------------------------------------------------------------------- # +def test_index_feed_sources_token_through_read_token(tmp_path, monkeypatch, caplog): + """The REAL _client() reads the token via secret.read_token and hands it to + tushare.pro_api — sourced from the external fake config, never leaked.""" + cfg_path = _write_fake_config(tmp_path) + + captured: dict[str, object] = {} + + class _FakeProToken: + def index_weight(self, index_code, start_date, end_date): # noqa: ARG002 + return pd.DataFrame( + { + "index_code": ["000300.SH"], + "con_code": ["000001.SZ"], + "trade_date": [start_date], + "weight": [1.0], + } + ) + + def fake_pro_api(token=None): + captured["token"] = token + return _FakeProToken() + + import tushare as ts + + monkeypatch.setattr(ts, "pro_api", fake_pro_api) + + feed = IndexConstituentsFeed(secret_file=str(cfg_path), token_key="tushare.token") + stdout = io.StringIO() + with caplog.at_level(logging.DEBUG), redirect_stdout(stdout): + out = feed.get_constituents("000300.SH", "2024-01-01", "2024-01-31") + + # Token came from the external fake config via the shared reader. + assert captured["token"] == FAKE_TOKEN + assert list(out.columns) == ["date", "symbol", "weight"] + # Never leaked to stdout or logging. + assert FAKE_TOKEN not in stdout.getvalue() + assert FAKE_TOKEN not in caplog.text + + +def test_index_feed_missing_token_key_raises_readable_error(tmp_path, monkeypatch): + """A missing dotted token_key raises a readable error naming the key path, + and never echoes the real token value.""" + cfg_path = _write_fake_config(tmp_path) + feed = IndexConstituentsFeed(secret_file=str(cfg_path), token_key="tushare.nope") + + import tushare as ts + + monkeypatch.setattr(ts, "pro_api", lambda token=None: object()) + + with pytest.raises(ValueError) as exc: + feed.get_constituents("000300.SH", "2024-01-01", "2024-01-31") + msg = str(exc.value) + assert "tushare.nope" in msg + assert FAKE_TOKEN not in msg