feat(data): UpdateScheduler 增量 + 断点续传 + 熔断

sanguo_data/scheduler.py

@@ -0,0 +1,57 @@
+# 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"

tests/data/test_scheduler.py

@@ -0,0 +1,24 @@
+# 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