From c6eae7eecf5525a1cc6156c88c560c2d5ef0c146 Mon Sep 17 00:00:00 2001
From: claude_dev <claude-dev@sanguo.local>
Date: Sun, 5 Jul 2026 16:14:23 +0800
Subject: [PATCH] =?UTF-8?q?feat(data):=20UpdateScheduler=20=E5=A2=9E?=
 =?UTF-8?q?=E9=87=8F=20+=20=E6=96=AD=E7=82=B9=E7=BB=AD=E4=BC=A0=20+=20?=
 =?UTF-8?q?=E7=86=94=E6=96=AD?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 sanguo_data/scheduler.py     | 57 ++++++++++++++++++++++++++++++++++++
 tests/data/test_scheduler.py | 24 +++++++++++++++
 2 files changed, 81 insertions(+)
 create mode 100644 sanguo_data/scheduler.py
 create mode 100644 tests/data/test_scheduler.py

diff --git a/sanguo_data/scheduler.py b/sanguo_data/scheduler.py
new file mode 100644
index 0000000..73cabc2
--- /dev/null
+++ b/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"
diff --git a/tests/data/test_scheduler.py b/tests/data/test_scheduler.py
new file mode 100644
index 0000000..69dd367
--- /dev/null
+++ b/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