tools(ops): 存量max_pool 30→0迁移脚本(默认dry-run,--apply才动手,幂等)——策略session 08-24移交项#1准备件 [nas]

真实盘点(只读): VPS prod=live_accounts全6行30+paper_accounts.strategies 6处JSON30(=影子实例);NAS=paper_accounts 1处JSON30,live_accounts空。
⚠️刻意排除backtest_stats.params里的3处30=历史回测运行记录,改写=伪造历史(JSON白名单只放paper_accounts活动配置面)。
⚠️执行铁律:只与数据session SQL治本(80fc517已落)+f214e2f同车,部署成功后、引擎重启前跑——池放大×慢SQL=更糟;迁移后重启引擎吃新值。
冒烟三态过:dry-run盘点/apply迁移(30→0两行,显式50保留,JSON只动max_pool键)/二次apply零行幂等+历史表原样。

scripts/ops/migrate_max_pool_30_to_0.py

@@ -0,0 +1,113 @@
+#!/usr/bin/env python3
+"""一次性迁移: 存量实例 max_pool=30 → 0(0=不限) — 2026-08-24 选股池失真处置。
+
+背景: SANGUO_LIVE_MAX_POOL 默认 30(MVP 验证遗留)把全部实盘/影子/paper 实例的
+选股池截断成「代码序前 30 只」(f214e2f 已改全链默认 30→0)。代码侧默认已翻转,
+但存量 DB 行存的是显式 30 → 需本迁移激活。
+
+⚠️ 部署铁律(策略 session 移交 #1): 必须与数据 session 的
+get_security_info_batch SQL 治本(冷跑 101s → 亚秒)同车执行——
+池放大 × 慢 SQL = 更糟。
+
+覆盖面:
+  1. live_accounts.max_pool = 30 → 0        (VPS prod: 全部 6 实例)
+  2. paper_accounts.strategies JSON 里的 "max_pool": 30 → 0
+     (VPS=6 影子实例,NAS=1 paper 实例)
+  ⚠️ 刻意排除 backtest_stats.params 里的 30——那是历史回测运行记录,
+  当时就是用 30 跑的,改写=伪造历史(2026-08-24 实盘点: VPS 命中 9 处中
+  3 处属 backtest_stats,均已保留)。JSON 白名单只放活动配置面 paper_accounts。
+
+用法:
+  python migrate_max_pool_30_to_0.py <db_path>              # dry-run 只盘点
+  python migrate_max_pool_30_to_0.py <db_path> --apply      # 实际执行
+幂等: WHERE max_pool=30 天然幂等; JSON 侧 30→0 后不再命中。
+"""
+import re
+import sqlite3
+import sys
+
+# JSON 迁移白名单: 只放「活动实例配置」面。backtest_stats 等历史记录刻意不在内(见 docstring)。
+JSON_TABLES = ("paper_accounts",)
+
+
+def scan(db_path: str) -> dict:
+    """盘点: live_accounts.max_pool 分布 + params JSON 里的 max_pool=30。"""
+    con = sqlite3.connect(f"file:{db_path}?mode=ro", uri=True, timeout=30)
+    con.row_factory = sqlite3.Row
+    out = {"live_accounts_dist": [], "json_hits": []}
+    try:
+        dist = con.execute(
+            "SELECT max_pool, COUNT(*) n FROM live_accounts GROUP BY max_pool"
+        ).fetchall()
+        out["live_accounts_dist"] = [(r["max_pool"], r["n"]) for r in dist]
+        pat = re.compile(r'"max_pool"\s*:\s*30\b')
+        for t in JSON_TABLES:
+            cols = [r[1] for r in con.execute(f"PRAGMA table_info({t})").fetchall()]
+            for c in cols:
+                try:
+                    rows = con.execute(
+                        f"SELECT rowid, {c} FROM {t} WHERE {c} LIKE '%max_pool%'"
+                    ).fetchall()
+                except sqlite3.OperationalError:
+                    continue
+                for r in rows:
+                    v = r[1]
+                    if isinstance(v, str) and pat.search(v):
+                        out["json_hits"].append((t, c, r[0]))
+    finally:
+        con.close()
+    return out
+
+
+def apply(db_path: str) -> None:
+    con = sqlite3.connect(db_path, timeout=30)
+    try:
+        cur = con.execute("UPDATE live_accounts SET max_pool=0 WHERE max_pool=30")
+        n_acct = cur.rowcount
+        n_json = 0
+        # JSON 侧: 同 scan 的白名单命中面, 逐条 30→0(正则替换保其余结构不动)
+        pat = re.compile(r'("max_pool"\s*:\s*)30\b')
+        for t in JSON_TABLES:
+            cols = [r[1] for r in con.execute(f"PRAGMA table_info({t})").fetchall()]
+            for c in cols:
+                try:
+                    rows = con.execute(
+                        f"SELECT rowid, {c} FROM {t} WHERE {c} LIKE '%max_pool%'"
+                    ).fetchall()
+                except sqlite3.OperationalError:
+                    continue
+                for rowid, v in rows:
+                    if isinstance(v, str) and pat.search(v):
+                        nv = pat.sub(r"\g<1>0", v)
+                        con.execute(
+                            f"UPDATE {t} SET {c}=? WHERE rowid=?", (nv, rowid))
+                        n_json += 1
+        con.commit()
+        print(f"✅ 已迁移: live_accounts {n_acct} 行 30→0; params JSON {n_json} 处 30→0")
+    finally:
+        con.close()
+
+
+def main() -> None:
+    if len(sys.argv) < 2 or sys.argv[1] in ("-h", "--help"):
+        print(__doc__)
+        sys.exit(0)
+    db = sys.argv[1]
+    do_apply = "--apply" in sys.argv[2:]
+    s = scan(db)
+    print(f"DB: {db}")
+    print(f"live_accounts max_pool 分布: {s['live_accounts_dist']}")
+    print(f"params JSON 含 max_pool:30 的行: {len(s['json_hits'])} {s['json_hits'][:10]}")
+    if not do_apply:
+        print("(dry-run,加 --apply 执行迁移)")
+        return
+    if not s["live_accounts_dist"] and not s["json_hits"]:
+        print("无可迁移项(可能已迁移过)")
+        return
+    apply(db)
+    s2 = scan(db)
+    print(f"迁移后分布: {s2['live_accounts_dist']}; JSON 残留: {len(s2['json_hits'])}")
+
+
+if __name__ == "__main__":
+    main()