sanguo/sanguo_vnpy_v2
3
0
Fork 0
Code Issues 3 Pull Requests Actions Packages Projects Releases Wiki Activity

docs: 新增 Plan 1 数据层实施计划

Browse Source 
8 个 TDD task：脚手架+配置 / validator / DataReader(parquet+SQLite) /
DataFeed(多源fallback+BaoStock超时) / DataWriter(原子写) /
UpdateScheduler(增量+断点续传+熔断) / vnpy 4.4.0 spike + 端到端冒烟
含 self-review 与执行指引
This commit is contained in:
Claude Dev
2026-07-05 11:31:20 +08:00
parent f255e5062e
commit 9595879655
1 changed files with 854 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/superpowers/plans/2026-07-05-plan1-data-layer.md
+854
View File
@@ -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 检查点。