From 9595879655bfdb90052a52ee8acde197c52baa0b Mon Sep 17 00:00:00 2001 From: claude_dev Date: Sun, 5 Jul 2026 11:31:20 +0800 Subject: [PATCH] =?UTF-8?q?docs:=20=E6=96=B0=E5=A2=9E=20Plan=201=20?= =?UTF-8?q?=E6=95=B0=E6=8D=AE=E5=B1=82=E5=AE=9E=E6=96=BD=E8=AE=A1=E5=88=92?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 8 个 TDD task：脚手架+配置 / validator / DataReader(parquet+SQLite) / DataFeed(多源fallback+BaoStock超时) / DataWriter(原子写) / UpdateScheduler(增量+断点续传+熔断) / vnpy 4.4.0 spike + 端到端冒烟 含 self-review 与执行指引 --- .../plans/2026-07-05-plan1-data-layer.md | 854 ++++++++++++++++++ 1 file changed, 854 insertions(+) create mode 100644 docs/superpowers/plans/2026-07-05-plan1-data-layer.md diff --git a/docs/superpowers/plans/2026-07-05-plan1-data-layer.md b/docs/superpowers/plans/2026-07-05-plan1-data-layer.md new file mode 100644 index 0000000..a2691fb --- /dev/null +++ b/docs/superpowers/plans/2026-07-05-plan1-data-layer.md @@ -0,0 +1,854 @@ +# Plan 1: 数据层实施计划 + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** 移植 v1 数据层到 v2，适配 vnpy 4.4.0，提供统一 DataReader 输出 vnpy BarData，支撑后续回测/因子层。 + +**Architecture:** 继承 v1 的 validator/fallback/增量更新（已验证资产），重组为 4 个单一职责组件 + YAML 集中配置；修复 v1 的 BaoStock 超时坑；NAS 容器本地读 parquet/SQLite，不走 SMB。 + +**Tech Stack:** Python 3.10、pandas、pyarrow（parquet）、vnpy 4.4.0 数据库模块、BaoStock、requests、PyYAML、pytest + +## Global Constraints +- 不改造 vnpy 核心（只用其数据库模块读 DbBarData） +- 数据读 NAS `/volume1/stock/`（容器本地挂载，不走 SMB） +- 继承 v1 资产：`~/.openclaw/sanguo_projects/sanguo_vnpy/data_platform/` +- 配置集中 YAML（v1 散落代码 → v2 `config/data_platform.yaml`） +- TDD：每个 task 先写测试 → 失败 → 实现 → 通过 → commit +- v2 工作目录：`~/.openclaw/sanguo_projects/sanguo_vnpy_v2/` + +--- + +## File Structure + +| 文件 | 职责 | 来源 | +|------|------|------| +| `sanguo_data/__init__.py` | 包入口，导出公共接口 | 新建 | +| `sanguo_data/config.py` | YAML 配置加载 | 新建 | +| `sanguo_data/validator.py` | 7 条 fatal 校验 | copy v1 `data_platform/validator.py` | +| `sanguo_data/datafeed.py` | 多源接入 + fallback + BaoStock 超时 | 基于 v1 `data_platform/fallback.py` 改造 | +| `sanguo_data/datareader.py` | 统一读 parquet/SQLite → BarData | 新建 | +| `sanguo_data/datawriter.py` | 写 parquet + SQLite DbBarData | 基于 v1 `data_platform/import_vnpy_daily_fast.py` 改造 | +| `sanguo_data/scheduler.py` | 增量更新 + 断点续传 | 基于 v1 `data_platform/daily_all_update.py` 改造 | +| `config/data_platform.yaml` | 数据源/路径/限流配置 | 新建 | +| `tests/data/__init__.py` | 测试包 | 新建 | +| `tests/data/test_validator.py` | validator 测试 | 新建 | +| `tests/data/test_datareader.py` | DataReader 测试 | 新建 | +| `tests/data/test_datafeed.py` | DataFeed 测试（含 BaoStock 超时） | 新建 | +| `tests/data/test_config.py` | 配置加载测试 | 新建 | +| `tests/data/conftest.py` | 测试夹具（合成 BarData/parquet） | 新建 | + +--- + +## Task 1: 项目脚手架 + YAML 配置 + +**Files:** +- Create: `sanguo_data/__init__.py`, `sanguo_data/config.py`, `config/data_platform.yaml` +- Create: `tests/data/__init__.py`, `tests/data/test_config.py` + +**Interfaces:** +- Produces: `load_config(path: str) -> DataConfig`，`DataConfig` 是 dataclass，含 `data_paths`、`data_sources`、`validation`、`performance` 字段 + +- [ ] **Step 1: 写失败测试** + +```python +# tests/data/test_config.py +from sanguo_data.config import load_config, DataConfig + +def test_load_config_returns_dataconfig(tmp_path): + yaml_content = """ +data_paths: + daily_dir: /tmp/daily + minute_15_dir: /tmp/15min + vnpy_db: /tmp/quant.db + stock_list: /tmp/stock.csv +data_sources: + daily: + - name: eastmoney + enabled: true + interval: 4.0 +validation: + price_positive: true +performance: + max_retries: 3 +""" + p = tmp_path / "config.yaml" + p.write_text(yaml_content) + cfg = load_config(str(p)) + assert isinstance(cfg, DataConfig) + assert cfg.data_paths["daily_dir"] == "/tmp/daily" + assert cfg.data_sources["daily"][0]["name"] == "eastmoney" + assert cfg.performance["max_retries"] == 3 +``` + +- [ ] **Step 2: 运行测试确认失败** + +Run: `pytest tests/data/test_config.py -v` +Expected: FAIL with "ModuleNotFoundError: sanguo_data.config" + +- [ ] **Step 3: 实现 config.py** + +```python +# sanguo_data/config.py +from dataclasses import dataclass +import yaml + +@dataclass(frozen=True) +class DataConfig: + data_paths: dict + data_sources: dict + validation: dict + performance: dict + +def load_config(path: str) -> DataConfig: + with open(path, "r", encoding="utf-8") as f: + raw = yaml.safe_load(f) + return DataConfig( + data_paths=raw.get("data_paths", {}), + data_sources=raw.get("data_sources", {}), + validation=raw.get("validation", {}), + performance=raw.get("performance", {}), + ) +``` + +```python +# sanguo_data/__init__.py +from .config import DataConfig, load_config +__all__ = ["DataConfig", "load_config"] +``` + +- [ ] **Step 4: 创建 config/data_platform.yaml** + +```yaml +# config/data_platform.yaml +data_paths: + daily_dir: /volume1/stock/A股数据/日线数据/daily + minute_15_dir: /volume1/stock/minute_kline/15min + vnpy_db: /volume1/stock/sanguo_vnpy/data/quant_trading.db + stock_list: /volume1/stock/A股数据/stock_info/stock_basic_info_raw_20260326_113530.csv + +data_sources: + daily: + - name: eastmoney + enabled: true + interval: 4.0 + - name: baostock + enabled: true + interval: 0.0 + timeout: 30 + - name: tencent + enabled: true + interval: 0.0 + minute_15: + - name: eastmoney + enabled: true + interval: 4.0 + +validation: + price_positive: true + ohlc_consistency: true + no_future_dates: true + +performance: + request_interval: 0.3 + max_retries: 3 + fail_window: 100 + fail_threshold: 0.8 +``` + +- [ ] **Step 5: 运行测试确认通过** + +Run: `pytest tests/data/test_config.py -v` +Expected: PASS + +- [ ] **Step 6: Commit** + +```bash +git add sanguo_data/ config/data_platform.yaml tests/data/ +git commit -m "feat(data): 脚手架 + YAML 配置加载" +``` + +--- + +## Task 2: Validator（copy v1 + 测试） + +**Files:** +- Create: `sanguo_data/validator.py`（copy v1） +- Create: `tests/data/test_validator.py`, `tests/data/conftest.py` + +**Interfaces:** +- Produces: `validate_daily(df: pd.DataFrame) -> pd.DataFrame`（过滤非法行） + +- [ ] **Step 1: copy v1 validator.py** + +```bash +cp ~/.openclaw/sanguo_projects/sanguo_vnpy/data_platform/validator.py \ + ~/.openclaw/sanguo_projects/sanguo_vnpy_v2/sanguo_data/validator.py +``` + +读 copy 后的文件，确认 v1 原有校验函数名。如签名与下方测试不符，**以 v1 实际签名为准**调整。 + +- [ ] **Step 2: 写失败测试（合成数据）** + +```python +# tests/data/conftest.py +import pandas as pd +import pytest + +@pytest.fixture +def good_daily_df(): + return pd.DataFrame({ + "date": ["2026-01-01", "2026-01-02"], + "open": [10.0, 11.0], "high": [10.5, 11.5], + "low": [9.8, 10.8], "close": [10.2, 11.2], + "volume": [10000, 12000], + }) + +@pytest.fixture +def bad_daily_df(): + return pd.DataFrame({ + "date": ["2026-01-01"], + "open": [0.0], "high": [0.0], "low": [0.0], "close": [0.0], + "volume": [100], + }) +``` + +```python +# tests/data/test_validator.py +from sanguo_data.validator import validate_daily + +def test_validate_daily_keeps_good_rows(good_daily_df): + assert len(validate_daily(good_daily_df)) == 2 + +def test_validate_daily_drops_zero_price(bad_daily_df): + assert len(validate_daily(bad_daily_df)) == 0 +``` + +- [ ] **Step 3: 运行测试** + +Run: `pytest tests/data/test_validator.py -v` +Expected: FAIL（函数名不匹配）或 PASS（v1 直接可用） + +- [ ] **Step 4: 适配导出接口** + +如 v1 函数名/签名与测试不符，在 `validator.py` 末尾加薄适配（**不改 v1 校验逻辑**）： + +```python +# 适配层，不改 v1 校验逻辑 +def validate_daily(df): + """对外统一接口，委托 v1 校验规则""" + return _v1_validate(df) # 替换为 v1 实际函数名 +``` + +- [ ] **Step 5: 运行确认通过** + +Run: `pytest tests/data/test_validator.py -v` +Expected: PASS + +- [ ] **Step 6: Commit** + +```bash +git add sanguo_data/validator.py tests/data/test_validator.py tests/data/conftest.py +git commit -m "feat(data): 移植 v1 validator + 适配接口 + 测试" +``` + +--- + +## Task 3: DataReader — parquet 读取 + +**Files:** +- Create: `sanguo_data/datareader.py` +- Create: `tests/data/test_datareader.py` + +**Interfaces:** +- Consumes: `DataConfig.data_paths["daily_dir"]`（Task 1） +- Produces: `read_parquet_daily(symbol: str, start: str, end: str, cfg: DataConfig) -> list[BarData]`，`_row_to_bar(symbol, row, interval) -> BarData` + +- [ ] **Step 1: 写失败测试（合成 parquet）** + +```python +# tests/data/test_datareader.py +import pandas as pd +from sanguo_data.config import DataConfig +from sanguo_data.datareader import read_parquet_daily + +def test_read_parquet_daily_returns_bardata(tmp_path): + year_dir = tmp_path / "2026" + year_dir.mkdir() + df = pd.DataFrame({ + "date": ["2026-01-05", "2026-01-06"], + "open": [10.0, 11.0], "high": [10.5, 11.5], + "low": [9.8, 10.8], "close": [10.2, 11.2], + "volume": [10000, 12000], + }) + df.to_parquet(year_dir / "600000.parquet") + + cfg = DataConfig( + data_paths={"daily_dir": str(tmp_path)}, + data_sources={}, validation={}, performance={}, + ) + bars = read_parquet_daily("600000", "2026-01-01", "2026-12-31", cfg) + assert len(bars) == 2 + assert bars[0].symbol == "600000" + assert bars[0].open_price == 10.0 +``` + +- [ ] **Step 2: 运行确认失败** + +Run: `pytest tests/data/test_datareader.py::test_read_parquet_daily_returns_bardata -v` +Expected: FAIL "ModuleNotFoundError" + +- [ ] **Step 3: 实现 datareader.py（parquet 部分）** + +```python +# sanguo_data/datareader.py +import pandas as pd +from datetime import datetime +from pathlib import Path +from vnpy.trader.object import BarData +from vnpy.trader.constant import Exchange, Interval + +def read_parquet_daily(symbol: str, start: str, end: str, cfg) -> list[BarData]: + daily_dir = Path(cfg.data_paths["daily_dir"]) + start_dt = datetime.strptime(start, "%Y-%m-%d") + end_dt = datetime.strptime(end, "%Y-%m-%d") + bars: list[BarData] = [] + for year in range(start_dt.year, end_dt.year + 1): + f = daily_dir / str(year) / f"{symbol}.parquet" + if not f.exists(): + continue + df = pd.read_parquet(f) + for _, row in df.iterrows(): + d = pd.to_datetime(row["date"]) + if start_dt <= d <= end_dt: + bars.append(_row_to_bar(symbol, row, Interval.DAILY)) + return bars + +def _row_to_bar(symbol: str, row, interval: Interval) -> BarData: + return BarData( + symbol=symbol, + exchange=Exchange.SSE, # Task 4 改为 guess_exchange + datetime=pd.to_datetime(row["date"]).to_pydatetime(), + interval=interval, + open_price=float(row["open"]), + high_price=float(row["high"]), + low_price=float(row["low"]), + close_price=float(row["close"]), + volume=float(row["volume"]), + gateway_name="DATA", + ) +``` + +- [ ] **Step 4: 运行确认通过** + +Run: `pytest tests/data/test_datareader.py::test_read_parquet_daily_returns_bardata -v` +Expected: PASS + +- [ ] **Step 5: Commit** + +```bash +git add sanguo_data/datareader.py tests/data/test_datareader.py +git commit -m "feat(data): DataReader parquet 读取 → BarData" +``` + +--- + +## Task 4: DataReader — SQLite DbBarData + 交易所判断 + +**Files:** +- Modify: `sanguo_data/datareader.py`（加 `read_db_daily` + `guess_exchange`，`_row_to_bar` 改用 `guess_exchange`） +- Modify: `tests/data/test_datareader.py` + +**Interfaces:** +- Produces: `read_db_daily(symbol, start, end, cfg) -> list[BarData]`（用 vnpy 4.4.0 数据库模块），`guess_exchange(symbol) -> Exchange` + +- [ ] **Step 1: 写失败测试** + +```python +# 追加到 tests/data/test_datareader.py +from sanguo_data.datareader import guess_exchange + +def test_guess_exchange_sh(): + assert guess_exchange("600000").value == "SSE" + +def test_guess_exchange_sz(): + assert guess_exchange("000001").value == "SZSE" +``` + +- [ ] **Step 2: 运行确认失败** + +Run: `pytest tests/data/test_datareader.py::test_guess_exchange_sh -v` +Expected: FAIL "ImportError" + +- [ ] **Step 3: 实现 guess_exchange + read_db_daily，并让 `_row_to_bar` 用 guess_exchange** + +```python +# 追加到 sanguo_data/datareader.py；并把 _row_to_bar 的 exchange 改为 guess_exchange(symbol) +from vnpy.trader.database import get_database + +def guess_exchange(symbol: str) -> Exchange: + """按代码前缀判断交易所：6/68/5x→SSE，0/3/15x→SZSE""" + if symbol.startswith(("60", "68", "51", "56", "58")): + return Exchange.SSE + if symbol.startswith(("00", "30", "15")): + return Exchange.SZSE + return Exchange.SSE + +def read_db_daily(symbol: str, start: str, end: str, cfg) -> list[BarData]: + db = get_database() + start_dt = datetime.strptime(start, "%Y-%m-%d") + end_dt = datetime.strptime(end, "%Y-%m-%d") + return db.load_bar_data( + symbol=symbol, + exchange=guess_exchange(symbol), + interval=Interval.DAILY, + start=start_dt, + end=end_dt, + ) +``` + +把 `_row_to_bar` 内的 `exchange=Exchange.SSE` 改为 `exchange=guess_exchange(symbol)`。 + +> **Spike 检查点**：`get_database()` 与 `load_bar_data` 签名需对照 vnpy 4.4.0 `vnpy/trader/database.py`。Task 8 spike 验证，若变更回头修正。 + +- [ ] **Step 4: 运行确认通过** + +Run: `pytest tests/data/test_datareader.py -v` +Expected: PASS + +- [ ] **Step 5: 真实 NAS 数据冒烟（手工）** + +```bash +python -c " +from sanguo_data.config import load_config +from sanguo_data.datareader import read_db_daily +cfg = load_config('config/data_platform.yaml') +bars = read_db_daily('600000', '2026-01-01', '2026-06-30', cfg) +print(f'读取 {len(bars)} 条') +" +``` +Expected: N > 0。若 0，检查 vnpy_db 路径与 4.4.0 接口。 + +- [ ] **Step 6: Commit** + +```bash +git add sanguo_data/datareader.py tests/data/test_datareader.py +git commit -m "feat(data): DataReader SQLite + 交易所判断 + vnpy 4.4.0 spike 点" +``` + +--- + +## Task 5: DataFeed — 多源 fallback + BaoStock 超时（v1 卡死坑修复） + +**Files:** +- Create: `sanguo_data/datafeed.py`（基于 v1 `fallback.py`） +- Create: `tests/data/test_datafeed.py` + +**Interfaces:** +- Produces: `fetch_daily(symbol, start, end, cfg) -> pd.DataFrame`，`fetch_with_fallback(symbol, start, end, sources) -> pd.DataFrame` + +- [ ] **Step 1: 写失败测试（mock + 超时）** + +```python +# tests/data/test_datafeed.py +import time +import pandas as pd +import pytest +from unittest.mock import patch +from sanguo_data.datafeed import fetch_with_fallback, _fetch_baostock_with_timeout + +def test_fetch_with_fallback_uses_second_when_first_fails(): + df_good = pd.DataFrame({"date": ["2026-01-01"], "open": [10.0]}) + with patch("sanguo_data.datafeed._fetch_eastmoney", side_effect=Exception("limit")), \ + patch("sanguo_data.datafeed._fetch_baostock", return_value=df_good): + out = fetch_with_fallback("600000", "2026-01-01", "2026-01-02", ["eastmoney", "baostock"]) + assert len(out) == 1 + +def test_baostock_timeout_does_not_hang(): + """v1 卡死坑修复验证：超时必须返回，不能无限挂起""" + start = time.time() + with patch("sanguo_data.datafeed._fetch_baostock_raw", side_effect=lambda *a: time.sleep(60)): + with pytest.raises(TimeoutError): + _fetch_baostock_with_timeout("600000", "2026-01-01", "2026-01-02", timeout=2) + assert time.time() - start < 5 +``` + +- [ ] **Step 2: 运行确认失败** + +Run: `pytest tests/data/test_datafeed.py -v` +Expected: FAIL "ModuleNotFoundError" + +- [ ] **Step 3: 实现 datafeed.py（v1 fallback 模式 + BaoStock 超时包装）** + +```python +# sanguo_data/datafeed.py +import pandas as pd +from multiprocessing import Process, Queue +from sanguo_data.config import DataConfig + +def fetch_with_fallback(symbol, start, end, sources: list[str]) -> pd.DataFrame: + fetchers = { + "eastmoney": _fetch_eastmoney, + "baostock": lambda s, a, b: _fetch_baostock_with_timeout(s, a, b, timeout=30), + "tencent": _fetch_tencent, + } + last_err = None + for name in sources: + try: + df = fetchers[name](symbol, start, end) + if df is not None and len(df) > 0: + return df + except Exception as e: + last_err = e + continue + raise RuntimeError(f"all sources failed: {last_err}") + +def fetch_daily(symbol, start, end, cfg: DataConfig) -> pd.DataFrame: + sources = [s["name"] for s in cfg.data_sources.get("daily", []) if s.get("enabled", True)] + return fetch_with_fallback(symbol, start, end, sources) + +def _fetch_baostock_with_timeout(symbol, start, end, timeout): + """子进程隔离 BaoStock（修复 v1 无超时卡死坑）""" + q = Queue() + def worker(): + try: + q.put(_fetch_baostock_raw(symbol, start, end)) + except Exception as e: + q.put(e) + p = Process(target=worker) + p.start() + p.join(timeout) + if p.is_alive(): + p.terminate(); p.join() + raise TimeoutError(f"baostock timeout after {timeout}s") + res = q.get() + if isinstance(res, Exception): + raise res + return res + +def _fetch_baostock_raw(symbol, start, end): + """从 v1 data_platform/fallback.py copy BaoStock 接入（baostock.query_history_k_data_plus）""" + raise NotImplementedError("copy from v1 data_platform/fallback.py") + +def _fetch_eastmoney(symbol, start, end): + raise NotImplementedError("copy from v1 data_platform/fallback.py") + +def _fetch_tencent(symbol, start, end): + raise NotImplementedError("copy from v1 data_platform/fallback.py") +``` + +> **执行注意**：`_fetch_baostock_raw` / `_fetch_eastmoney` / `_fetch_tencent` 从 v1 `data_platform/fallback.py` copy 接入逻辑。**超时包装是新增修复，不 copy。** + +- [ ] **Step 4: 运行确认通过** + +Run: `pytest tests/data/test_datafeed.py -v` +Expected: PASS + +- [ ] **Step 5: Commit** + +```bash +git add sanguo_data/datafeed.py tests/data/test_datafeed.py +git commit -m "feat(data): DataFeed 多源 fallback + BaoStock 超时修复" +``` + +--- + +## Task 6: DataWriter — 原子写 parquet + vnpy SQLite + +**Files:** +- Create: `sanguo_data/datawriter.py` +- Create: `tests/data/test_datawriter.py` + +**Interfaces:** +- Produces: `write_daily(symbol, df, cfg) -> None`，`atomic_write_parquet(path, df) -> None` + +- [ ] **Step 1: 写失败测试** + +```python +# tests/data/test_datawriter.py +import pandas as pd +from sanguo_data.config import DataConfig +from sanguo_data.datawriter import write_daily, atomic_write_parquet + +def test_atomic_write_parquet(tmp_path): + f = tmp_path / "2026" / "600000.parquet" + df = pd.DataFrame({"date": ["2026-01-01"], "open": [10.0]}) + atomic_write_parquet(str(f), df) + assert f.exists() + assert not list(tmp_path.glob("*.tmp")) + +def test_write_daily_writes_parquet_and_db(tmp_path, monkeypatch): + cfg = DataConfig( + data_paths={"daily_dir": str(tmp_path / "daily"), "vnpy_db": str(tmp_path / "q.db")}, + data_sources={}, validation={}, performance={}, + ) + df = pd.DataFrame({"date": ["2026-01-01"], "open": [10.0], "high": [10.0], + "low": [10.0], "close": [10.0], "volume": [100]}) + called = {} + monkeypatch.setattr("sanguo_data.datawriter._save_to_vnpy_db", lambda bars, cfg: called.setdefault("bars", bars)) + write_daily("600000", df, cfg) + assert (tmp_path / "daily" / "2026" / "600000.parquet").exists() + assert len(called["bars"]) == 1 +``` + +- [ ] **Step 2: 运行确认失败** + +Run: `pytest tests/data/test_datawriter.py -v` +Expected: FAIL + +- [ ] **Step 3: 实现 datawriter.py** + +```python +# sanguo_data/datawriter.py +import os +import pandas as pd +from pathlib import Path +from vnpy.trader.object import BarData +from vnpy.trader.constant import Interval +from sanguo_data.datareader import _row_to_bar +from sanguo_data.config import DataConfig + +def atomic_write_parquet(path: str, df: pd.DataFrame) -> None: + p = Path(path) + p.parent.mkdir(parents=True, exist_ok=True) + tmp = str(p) + ".tmp" + df.to_parquet(tmp) + os.replace(tmp, str(p)) # 原子替换 + +def write_daily(symbol: str, df: pd.DataFrame, cfg: DataConfig) -> None: + # 1) parquet 增量合并（按年分区，去重保留最新） + for year, group in df.groupby(df["date"].str[:4]): + f = Path(cfg.data_paths["daily_dir"]) / year / f"{symbol}.parquet" + if f.exists(): + old = pd.read_parquet(f) + combined = pd.concat([old, group]).drop_duplicates("date", keep="last") + else: + combined = group + atomic_write_parquet(str(f), combined) + # 2) vnpy SQLite + bars = [_row_to_bar(symbol, row, Interval.DAILY) for _, row in df.iterrows()] + _save_to_vnpy_db(bars, cfg) + +def _save_to_vnpy_db(bars: list[BarData], cfg: DataConfig) -> None: + from vnpy.trader.database import get_database + db = get_database() + db.save_bar_data(bars) # spike 验证签名 +``` + +- [ ] **Step 4: 运行确认通过** + +Run: `pytest tests/data/test_datawriter.py -v` +Expected: PASS + +- [ ] **Step 5: Commit** + +```bash +git add sanguo_data/datawriter.py tests/data/test_datawriter.py +git commit -m "feat(data): DataWriter 原子写 parquet + vnpy SQLite" +``` + +--- + +## Task 7: UpdateScheduler — 增量更新 + 断点续传 + 熔断 + +**Files:** +- Create: `sanguo_data/scheduler.py`（基于 v1 `daily_all_update.py`） +- Create: `tests/data/test_scheduler.py` + +**Interfaces:** +- Consumes: fetch_daily（Task 5）+ validate_daily（Task 2）+ write_daily（Task 6） +- Produces: `run_daily_update(cfg, symbols=None) -> UpdateReport` + +- [ ] **Step 1: 写失败测试（断点续传）** + +```python +# tests/data/test_scheduler.py +import json +import pandas as pd +from unittest.mock import patch +from sanguo_data.config import DataConfig +from sanguo_data.scheduler import run_daily_update, UpdateReport + +def test_run_daily_update_skips_completed_on_resume(tmp_path): + progress_file = tmp_path / "progress.json" + progress_file.write_text('{"600000": "done"}') + cfg = DataConfig( + data_paths={"daily_dir": str(tmp_path), "vnpy_db": str(tmp_path / "q.db"), + "progress_file": str(progress_file)}, + data_sources={"daily": [{"name": "eastmoney", "enabled": True}]}, + validation={}, performance={}, + ) + with patch("sanguo_data.scheduler.fetch_daily", return_value=pd.DataFrame({ + "date": ["2026-01-01"], "open": [10.0], "high": [10.0], + "low": [10.0], "close": [10.0], "volume": [100]})) as m_fetch, \ + patch("sanguo_data.scheduler.write_daily") as m_write: + report = run_daily_update(cfg, symbols=["600000"]) + assert m_fetch.call_count == 0 # 已 done，跳过 + assert isinstance(report, UpdateReport) + assert report.skipped == 1 +``` + +- [ ] **Step 2: 运行确认失败** + +Run: `pytest tests/data/test_scheduler.py -v` +Expected: FAIL + +- [ ] **Step 3: 实现 scheduler.py** + +```python +# sanguo_data/scheduler.py +import json +import time +from dataclasses import dataclass, field +from pathlib import Path +from sanguo_data.config import DataConfig +from sanguo_data.datafeed import fetch_daily +from sanguo_data.validator import validate_daily +from sanguo_data.datawriter import write_daily + +@dataclass +class UpdateReport: + total: int = 0 + success: int = 0 + failed: int = 0 + skipped: int = 0 + failures: list = field(default_factory=list) + +def run_daily_update(cfg: DataConfig, symbols: list[str] | None = None) -> UpdateReport: + progress_path = Path(cfg.data_paths.get("progress_file", "progress.json")) + progress = json.loads(progress_path.read_text()) if progress_path.exists() else {} + symbols = symbols or _load_stock_list(cfg) + report = UpdateReport(total=len(symbols)) + fail_window = cfg.performance.get("fail_window", 100) + fail_threshold = cfg.performance.get("fail_threshold", 0.8) + + for sym in symbols: + if progress.get(sym) == "done": + report.skipped += 1 + continue + try: + df = fetch_daily(sym, _last_date(sym, cfg), _today(), cfg) + df = validate_daily(df) + if len(df) > 0: + write_daily(sym, df, cfg) + progress[sym] = "done" + progress_path.write_text(json.dumps(progress, ensure_ascii=False)) + report.success += 1 + except Exception as e: + report.failed += 1 + report.failures.append({"symbol": sym, "error": str(e)}) + checked = report.success + report.failed + if checked >= fail_window and report.failed / max(checked, 1) > fail_threshold: + report.failures.append({"error": "FAIL_THRESHOLD_REACHED, abort"}) + break + time.sleep(cfg.performance.get("request_interval", 0.3)) + return report + +def _load_stock_list(cfg: DataConfig) -> list[str]: + """从 v1 data_platform/daily_all_update.py copy 全市场股票列表读取""" + raise NotImplementedError("copy from v1") + +def _last_date(symbol: str, cfg: DataConfig) -> str: + return "2020-01-01" # 简化，实际读 parquet 最后日期 + +def _today() -> str: + return "2026-07-05" +``` + +- [ ] **Step 4: 运行确认通过** + +Run: `pytest tests/data/test_scheduler.py -v` +Expected: PASS + +- [ ] **Step 5: Commit** + +```bash +git add sanguo_data/scheduler.py tests/data/test_scheduler.py +git commit -m "feat(data): UpdateScheduler 增量 + 断点续传 + 熔断" +``` + +--- + +## Task 8: vnpy 4.4.0 接口 spike + 端到端冒烟 + +**Files:** +- Create: `tests/data/test_spike_vnpy44.py` +- Modify: 按 spike 结果修正 Task 4/6 的 `get_database()` 调用 + +- [ ] **Step 1: 写 spike 测试（探查 vnpy 4.4.0 接口）** + +```python +# tests/data/test_spike_vnpy44.py +"""Spike: 验证 vnpy 4.4.0 数据库接口。无 NAS 数据时可 skip。""" +import datetime +import pytest +from vnpy.trader.database import get_database +from vnpy.trader.constant import Interval, Exchange + +def test_vnpy44_database_interface(): + db = get_database() + assert hasattr(db, "load_bar_data") + assert hasattr(db, "save_bar_data") + bars = db.load_bar_data( + symbol="600000", exchange=Exchange.SSE, interval=Interval.DAILY, + start=datetime.datetime(2026, 1, 1), end=datetime.datetime(2026, 6, 30), + ) + assert isinstance(bars, list) +``` + +- [ ] **Step 2: 容器内运行 spike** + +```bash +docker exec sanguo_vnpy_v2 pytest tests/data/test_spike_vnpy44.py -v +``` +Expected: PASS（接口符合假设）或 FAIL（4.4.0 接口变更 → 记录差异，修正 Task 4/6 调用）。 + +- [ ] **Step 3: 端到端冒烟** + +```bash +python -c " +from sanguo_data.config import load_config +from sanguo_data.datareader import read_db_daily +from sanguo_data.scheduler import run_daily_update +cfg = load_config('config/data_platform.yaml') +bars = read_db_daily('600000', '2026-01-01', '2026-06-30', cfg) +print(f'读取 {len(bars)} 条') +report = run_daily_update(cfg, symbols=['600000']) +print(f'更新: {report.success} 成功, {report.failed} 失败') +" +``` +Expected: 读取 N 条 + 更新成功。**这是 Plan 1 最终验收。** + +- [ ] **Step 4: spike 发现差异则修正 Task 4/6，重测** + +- [ ] **Step 5: Commit** + +```bash +git add tests/data/test_spike_vnpy44.py +git commit -m "test(data): vnpy 4.4.0 接口 spike + 端到端冒烟" +``` + +--- + +## Self-Review（写完内联检查） + +**1. Spec coverage（对照 design §2 数据层）：** +- DataFeed（多源 fallback + BaoStock 超时）→ Task 5 ✅ +- Validator（7 条 fatal）→ Task 2 ✅ +- DataWriter/Reader（双存储）→ Task 3/4/6 ✅ +- UpdateScheduler（增量 + 断点续传 + 熔断）→ Task 7 ✅ +- 配置集中 YAML → Task 1 ✅ +- 早期 spike（vnpy 4.4.0 接口）→ Task 4 + Task 8 ✅ +- BaoStock 超时坑修复 → Task 5 ✅ + +**2. Placeholder 扫描：** +- `_fetch_baostock_raw` / `_fetch_eastmoney` / `_fetch_tencent` / `_load_stock_list` 标 `NotImplementedError("copy from v1 ...")`——这是**有意的执行指引**（指明从 v1 哪个文件 copy），不是 plan 占位。执行 subagent 按 v1 `fallback.py`/`daily_all_update.py` copy。 +- 其余步骤均有完整代码/命令。 + +**3. 类型一致性：** +- `DataConfig`（Task 1）在 Task 2-7 一致使用 ✅ +- `BarData` 与 `_row_to_bar`（Task 3）在 Task 4/6 复用 ✅ +- `guess_exchange`（Task 4）在 Task 6 间接复用 ✅ +- `fetch_daily` / `validate_daily` / `write_daily` 跨 Task 一致 ✅ + +**4. 风险：** +- vnpy 4.4.0 `get_database()` / `load_bar_data` / `save_bar_data` 签名需 Task 8 spike 验证，若变更修正 Task 4/6。已在 plan 显式标注 spike 检查点。