refactor(data): 5m回灌改每日分片——用户拍板2026-08-20'每天固定下一些',废弃--full一趟灌满——①每日一跑=回灌片(若有)+每日增量(回灌期也跑,近端7天始终新鲜)②片=半年窗口newest-first(近端先到先可用,回测最常用的近期数据第一天就有),state(bs_5m_state.json next_end)逐片向2020-01-03推进;整片扫完才推进,断点/达限次日重拉同片(OR REPLACE幂等)③全程~14片≈两周补完,每片~5.5k query/~3200万行/3-4h;片+增量每日~11k query,NAS IP独立48k配额内④DAILY_LIMIT守卫改跨窗口累计计数(swept);删FULL_TIMEOUT(半年片5.8k行/股,60s超时足够);+3测试(首片=最近半年/逐片向2020走+完结盖章/损坏state重头),main测试改双sweep断言(片+增量4调用/完结后单sweep2调用);data_platform 158绿 [nas]

scripts/data_platform/bs_5m_eod.py

@@ -17,18 +17,21 @@
 - 库: BS_5M_DB 缺省 NAS 副本主库(VPS 同步目标, 也是 NAS 回测 provider 读的库,
   portfolio_worker.py 传的就是它); ensure_schema 已有唯一索引(NAS 侧名 uq_dbbardata)
   则不重复建
-- 模式: 缺省每日增量(LOOKBACK 7 天, DSM 任务计划 ~19:05); --full 一次性回灌
-  2020-01-03+(baostock 分钟数据固定起点非滚动; 全区间每股恰好 1 次调用, 一趟
-  ~5.5k query / ~3.6 亿行, 数小时级; per-stock 短事务, 中断重跑幂等)
-- 配额: NAS 出口 IP 独立核算 48k/天(与 VPS 各自计数互不相干); 本脚本一趟远低
-  于限, DAILY_LIMIT 守卫保留
-- 单实例锁 bs_5m.lock(pid 活性检测): 防 DSM 定时与 --full 长跑撞车
+- 模式(2026-08-20 用户拍板"每天固定下一些", 拒绝一趟灌满): 每日一跑(DSM ~19:05)=
+  ①回灌未完成→下一片半年窗口(newest-first, 近端先到先可用; state 推进, 片没扫完
+  不推进次日重拉同片幂等) ②每日增量(LOOKBACK 7 天, 回灌期也跑, 近端始终新鲜)。
+  全程 2020-01-03(baostock 分钟固定起点)~今 ≈ 14 片 ≈ 两周补完, 每片 ~5.5k query /
+  ~3200 万行 / 3-4h; per-stock 短事务, 中断重跑幂等
+- 配额: NAS 出口 IP 独立核算 48k/天(与 VPS 各自计数互不相干); 每日 片+增量
+  ~11k query, DAILY_LIMIT 守卫保留
+- 单实例锁 bs_5m.lock(pid 活性检测): 防手动补跑与 DSM 定时撞车
   (同 IP 双 baostock 连接红线)
 
 退出码: 0=完成/让路; 1=致命; 2=登录失败; 3=query 超限 graceful stop
 """
 import argparse
 import datetime as dt
+import json
 import logging
 import os
 import sqlite3
@@ -47,7 +50,7 @@ _DEFAULT_DB = "/volume1/stock/sanguo_vnpy_v2/data_backup/quant_trading.db"  # NA
 DB = Path(os.environ.get("BS_5M_DB", _DEFAULT_DB))
 LOOKBACK = int(os.environ.get("LOOKBACK_DAYS", "7"))
 FULL_START = "2020-01-03"  # baostock 分钟数据固定起点(官网"近5年"口径 2020-01-03)
-FULL_TIMEOUT = 300          # 全区间单股 ~6.4 万行, 默认 60s 超时不够
+CHUNK_DAYS = 183            # 回灌片宽(半年); 每片 ~5.5k query / ~3200 万行 / 3-4h
 M5_FIELDS = "date,time,code,open,high,low,close,volume,amount"
 
 logging.basicConfig(level=logging.INFO,
@@ -56,13 +59,40 @@ logging.basicConfig(level=logging.INFO,
 log = logging.getLogger(__name__)
 
 
-def calc_window(today, full):
-    """--full: 2020-01-03 固定起点全区间; 缺省: LOOKBACK 天增量窗口。"""
-    end = today.strftime("%Y-%m-%d")
-    if full:
-        return FULL_START, end
-    start = (today - dt.timedelta(days=LOOKBACK)).strftime("%Y-%m-%d")
-    return start, end
+def _state_path():
+    return DB.parent / "bs_5m_state.json"
+
+
+def load_chunk_state():
+    """{"next_end": "YYYY-MM-DD" | None}; 缺省 next_end=今天(首片=最近半年)。"""
+    if _state_path().exists():
+        try:
+            return json.loads(_state_path().read_text(encoding="utf-8"))
+        except Exception:
+            log.warning("state 文件损坏, 回灌从最近半年重头(OR REPLACE 幂等无伤)")
+    return {"next_end": dt.date.today().isoformat()}
+
+
+def save_chunk_state(state):
+    DB.parent.mkdir(parents=True, exist_ok=True)
+    _state_path().write_text(json.dumps(state, ensure_ascii=False, indent=2),
+                             encoding="utf-8")
+
+
+def next_chunk(today):
+    """下一个待回灌窗口 (start, end), newest-first 逐片向 2020 走; None=回灌完成。
+
+    片宽 CHUNK_DAYS, 尾片与 FULL_START 对齐(前段有少量重叠, OR REPLACE 去重)。
+    """
+    state = load_chunk_state()
+    if state.get("next_end") is None:
+        return None
+    end = min(dt.date.fromisoformat(state["next_end"]), today)
+    if end <= dt.date.fromisoformat(FULL_START):
+        return None
+    start = max(dt.date.fromisoformat(FULL_START),
+                end - dt.timedelta(days=CHUNK_DAYS))
+    return start.isoformat(), end.isoformat()
 
 
 def _has_unique_index(conn):
@@ -166,8 +196,6 @@ def _release_lock():
 def main():
     ap = argparse.ArgumentParser()
     ap.add_argument("--limit", type=int, default=0, help="限股数(冒烟用)")
-    ap.add_argument("--full", action="store_true",
-                    help="一次性回灌 2020-01-03+(缺省每日增量)")
     args = ap.parse_args()
 
     if not _acquire_lock():
@@ -179,11 +207,53 @@ def main():
         _release_lock()
 
 
+def _sweep(conn, stocks, start, end, swept, label):
+    """扫全 A 一个窗口: per-stock 短事务 + 周期 relogin + 间隔。
+
+    swept 为跨窗口累计股数(=query 数, 每股恰 1 次调用), 由调用方持有并回填;
+    返 (stats, limit_reached)。
+    """
+    stats = {"ok": 0, "empty": 0, "failed": 0, "db_rows": 0}
+    limit_reached = False
+    t0 = time.time()
+    for i, (code, prefix) in enumerate(stocks):
+        if swept[0] >= DAILY_LIMIT:
+            log.warning("[%s] query %d 达防线 %d, graceful stop",
+                        label, swept[0], DAILY_LIMIT)
+            limit_reached = True
+            break
+        swept[0] += 1
+        try:
+            n = _process_one(conn, code, prefix, start, end, 60)
+            stats["db_rows"] += n
+            if n:
+                stats["ok"] += 1
+            else:
+                stats["empty"] += 1
+        except Exception as e:
+            stats["failed"] += 1
+            if stats["failed"] <= 5 or stats["failed"] % 100 == 0:
+                log.warning("[%s] %s err: %s", label, code, e)
+            if not relogin():
+                log.error("[%s] %s relogin 失败, 跳过", label, code)
+        if (i + 1) % RELOGIN_EVERY == 0:
+            log.info("[%s] 进度 %d/%d ok=%d empty=%d failed=%d rows=%d (%.0fs)",
+                     label, i + 1, len(stocks), stats["ok"], stats["empty"],
+                     stats["failed"], stats["db_rows"], time.time() - t0)
+            if not relogin():
+                log.warning("[%s] 周期 relogin 失败, 继续跑", label)
+        if i < len(stocks) - 1:
+            time.sleep(BS_INTERVAL)
+    return stats, limit_reached
+
+
 def _run(args):
     today = dt.date.today()
-    start, end = calc_window(today, args.full)
-    log.info("bs_5m start window=%s~%s mode=%s db=%s", start, end,
-             "full" if args.full else "daily", DB)
+    chunk = next_chunk(today)
+    inc_start = (today - dt.timedelta(days=LOOKBACK)).strftime("%Y-%m-%d")
+    inc_end = today.strftime("%Y-%m-%d")
+    log.info("bs_5m start db=%s chunk=%s inc=%s~%s", DB,
+             "%s~%s" % chunk if chunk else "无(回灌完成)", inc_start, inc_end)
 
     if not login_with_retry():
         log.error("[SKIP] 登录失败, exit 2")
@@ -202,37 +272,28 @@ def _run(args):
     conn.execute("PRAGMA journal_mode = WAL")
     ensure_schema(conn)
 
-    timeout = FULL_TIMEOUT if args.full else 60
-    stats = {"ok": 0, "empty": 0, "failed": 0, "db_rows": 0}
+    swept = [0]  # 跨窗口累计 query 数(每股恰 1 次调用)
     limit_reached = False
     t0 = time.time()
     try:
-        for i, (code, prefix) in enumerate(stocks):
-            if i >= DAILY_LIMIT:  # 每股恰好 1 query, i 即当日 query 计数
-                log.warning("query %d 达防线 %d, graceful stop", i, DAILY_LIMIT)
-                limit_reached = True
-                break
-            try:
-                n = _process_one(conn, code, prefix, start, end, timeout)
-                stats["db_rows"] += n
-                if n:
-                    stats["ok"] += 1
-                else:
-                    stats["empty"] += 1
-            except Exception as e:
-                stats["failed"] += 1
-                if stats["failed"] <= 5 or stats["failed"] % 100 == 0:
-                    log.warning("%s err: %s", code, e)
-                if not relogin():
-                    log.error("%s relogin 失败, 跳过", code)
-            if (i + 1) % RELOGIN_EVERY == 0:
-                log.info("进度 %d/%d ok=%d empty=%d failed=%d rows=%d (%.0fs)",
-                         i + 1, len(stocks), stats["ok"], stats["empty"],
-                         stats["failed"], stats["db_rows"], time.time() - t0)
-                if not relogin():
-                    log.warning("周期 relogin 失败, 继续跑")
-            if i < len(stocks) - 1:
-                time.sleep(BS_INTERVAL)
+        if chunk:
+            cs, lr = _sweep(conn, stocks, chunk[0], chunk[1], swept,
+                            "chunk %s~%s" % chunk)
+            limit_reached = limit_reached or lr
+            log.info("[CHUNK] %s~%s ok=%d empty=%d failed=%d rows=%d",
+                     chunk[0], chunk[1], cs["ok"], cs["empty"], cs["failed"],
+                     cs["db_rows"])
+            if not lr and swept[0] < DAILY_LIMIT:
+                # 整片扫完才推进 state; 没扫完次日重拉同片(OR REPLACE 幂等续跑)
+                save_chunk_state({"next_end": chunk[0]})
+        # 每日增量照跑(回灌期也跑, 近端 7 天始终新鲜)
+        istats, lr = _sweep(conn, stocks, inc_start, inc_end, swept, "inc")
+        limit_reached = limit_reached or lr
+        log.info("[INC] ok=%d empty=%d failed=%d rows=%d",
+                 istats["ok"], istats["empty"], istats["failed"],
+                 istats["db_rows"])
+        if next_chunk(today) is None and not limit_reached:
+            save_chunk_state({"next_end": None})  # 回灌完结盖章(幂等)
     finally:
         conn.close()
         try:
@@ -241,9 +302,7 @@ def _run(args):
         except Exception:
             pass
 
-    log.info("[DONE] mode=%s ok=%d empty=%d failed=%d db_rows=%d 耗时%.0fs",
-             "full" if args.full else "daily", stats["ok"], stats["empty"],
-             stats["failed"], stats["db_rows"], time.time() - t0)
+    log.info("[DONE] query=%d 耗时%.0fs", swept[0], time.time() - t0)
     sys.exit(3 if limit_reached else 0)