feat(nas_sync): 小表全量同步管线(constituent_unified+bs_adjust_factor)

新增 export_table.py/merge_table.py/sync_tables.sh: VPS stdin喂python全量dump → scp → NAS全量替换(DROP+CREATE+INSERT,幂等可重跑)。sync_dbbardata.sh 每日increment末尾连带跑 sync_tables,保持NAS副本成份股/复权因子不陈旧。修复NAS组合回测选股恒空(NAS副本缺constituent_unified表,phase3只导dbbardata)。

scripts/nas_sync/export_table.py

@@ -0,0 +1,48 @@
+"""VPS 端：全量导出单张小表到独立 sqlite 文件（搬数据，merge 端重建 schema）。
+
+用于 constituent_unified / bs_adjust_factor 等小表全量同步
+（无 id 增量键、行数万级、变化不频繁，全量 dump+replace 最简）。
+
+- 纯 sqlite3 标准库，零第三方依赖。
+- 只读主库（mode=ro），不干扰 VPS 生产写入。
+- CREATE TABLE AS SELECT 搬数据（列名保留；约束由 NAS merge 端权威 schema 重建）。
+
+用法:
+    python export_table.py --db PATH --out out.db --table constituent_unified
+"""
+import argparse
+import os
+import sqlite3
+
+
+def main():
+    ap = argparse.ArgumentParser(description="全量导出单张小表到独立 sqlite")
+    ap.add_argument("--db", required=True, help="源 quant_trading.db 路径")
+    ap.add_argument("--out", required=True, help="输出 sqlite 路径")
+    ap.add_argument("--table", required=True, help="表名")
+    args = ap.parse_args()
+    tbl = args.table
+
+    conn = sqlite3.connect("file:%s?mode=ro" % args.db, uri=True)
+    cur = conn.cursor()
+    exists = cur.execute(
+        "SELECT 1 FROM sqlite_master WHERE type='table' AND name=?", (tbl,)
+    ).fetchone()
+    if not exists:
+        raise SystemExit("table not found in source db: %s" % tbl)
+
+    if os.path.exists(args.out):
+        os.remove(args.out)
+    cur.execute("ATTACH DATABASE ? AS inc", (args.out,))
+    cur.execute(
+        "CREATE TABLE inc.[%s] AS SELECT * FROM main.[%s]" % (tbl, tbl)
+    )
+    n = cur.execute("SELECT COUNT(*) FROM inc.[%s]" % tbl).fetchone()[0]
+    conn.commit()
+    conn.close()
+    print("EXPORTED table=%s rows=%d sizeMB=%.2f -> %s"
+          % (tbl, n, os.path.getsize(args.out) / 1048576.0, args.out))
+
+
+if __name__ == "__main__":
+    main()

scripts/nas_sync/merge_table.py

@@ -0,0 +1,79 @@
+"""NAS 端：全量替换单张小表（DROP + CREATE + INSERT），幂等可重跑。
+
+用于 constituent_unified / bs_adjust_factor 等小表。
+- 纯 sqlite3 标准库，零第三方依赖（NAS 宿主 Python 3.8 也能跑）。
+- DROP IF EXISTS + 权威 schema 重建 + INSERT：每次全量替换，重跑安全。
+- 权威 schema（含主键/约束）硬编码在此，NAS 副本对齐 VPS。
+
+用法:
+    python merge_table.py --db /volume1/.../quant_trading.db --inc /tmp/t.db --table constituent_unified
+"""
+import argparse
+import sqlite3
+
+# 权威 schema（对齐 VPS quant_trading.db）。新增表在此登记。
+SCHEMAS = {
+    "constituent_unified": (
+        "CREATE TABLE constituent_unified("
+        "index_code TEXT, code TEXT, code_name TEXT, source TEXT,"
+        "in_current INT, was_removed INT)"
+    ),
+    "bs_adjust_factor": (
+        "CREATE TABLE bs_adjust_factor("
+        "code TEXT NOT NULL, dividOperateDate TEXT NOT NULL,"
+        "foreAdjustFactor REAL, backAdjustFactor REAL, adjustFactor REAL,"
+        "PRIMARY KEY (code, dividOperateDate))"
+    ),
+}
+# 列名（与源表对齐，用于 INSERT 列顺序）。
+COLUMNS = {
+    "constituent_unified":
+        "index_code,code,code_name,source,in_current,was_removed",
+    "bs_adjust_factor":
+        "code,dividOperateDate,foreAdjustFactor,backAdjustFactor,adjustFactor",
+}
+# 额外索引（主键自带的不列）。
+INDEXES = {
+    "constituent_unified": [
+        "CREATE INDEX IF NOT EXISTS idx_cu_index ON constituent_unified(index_code)",
+        "CREATE INDEX IF NOT EXISTS idx_cu_code ON constituent_unified(code)",
+    ],
+    "bs_adjust_factor": [],
+}
+
+
+def main():
+    ap = argparse.ArgumentParser(description="全量替换单张小表进 NAS 副本")
+    ap.add_argument("--db", required=True, help="NAS quant_trading.db 路径")
+    ap.add_argument("--inc", required=True, help="增量 sqlite 路径")
+    ap.add_argument("--table", required=True, help="表名")
+    args = ap.parse_args()
+    tbl = args.table
+    if tbl not in SCHEMAS:
+        raise SystemExit("unknown table %s; add schema/columns first" % tbl)
+    cols = COLUMNS[tbl]
+
+    conn = sqlite3.connect(args.db)
+    cur = conn.cursor()
+    for p in ("PRAGMA journal_mode=WAL", "PRAGMA synchronous=NORMAL"):
+        cur.execute(p)
+    cur.execute("ATTACH DATABASE ? AS inc", (args.inc,))
+
+    inc_count = cur.execute(
+        "SELECT COUNT(*) FROM inc.[%s]" % tbl).fetchone()[0]
+    cur.execute("DROP TABLE IF EXISTS main.[%s]" % tbl)
+    cur.execute(SCHEMAS[tbl])
+    for idx in INDEXES.get(tbl, []):
+        cur.execute(idx)
+    cur.execute(
+        "INSERT INTO main.[%s](%s) SELECT %s FROM inc.[%s]" % (tbl, cols, cols, tbl)
+    )
+    conn.commit()
+    after = cur.execute(
+        "SELECT COUNT(*) FROM main.[%s]" % tbl).fetchone()[0]
+    conn.close()
+    print("REPLACED table=%s inc=%d after=%d" % (tbl, inc_count, after))
+
+
+if __name__ == "__main__":
+    main()

scripts/nas_sync/sync_dbbardata.sh

@@ -89,3 +89,9 @@ case "${1:-increment}" in
     exit 1
     ;;
 esac
+
+# 小表全量同步(constituent_unified 成份股 / bs_adjust_factor 复权因子:
+# 月度或除权日才变,万级行,全量几秒)。独立 log(sync_tables.log),
+# 失败不影响 dbbardata 主流程。每日 increment 连带跑,保持 NAS 副本不陈旧。
+bash "$(dirname "$0")/sync_tables.sh" >> "$LOG" 2>&1 \
+  || echo "[$(date '+%T')] WARN sync_tables 非致命" >> "$LOG"

scripts/nas_sync/sync_tables.sh

@@ -0,0 +1,41 @@
+#!/bin/bash
+# NAS 端：小表全量同步（constituent_unified / bs_adjust_factor 等）。
+#
+# 这些表无 id 增量键、行数万级、变化不频繁（成份股月度调整 / 除权日），
+# 全量 dump + replace 最简，每天跑也只需几秒。
+#
+# 用法: bash sync_tables.sh
+# 设计同 sync_dbbardata.sh: VPS python(stdin 喂) 导出 sqlite → scp → NAS 全量替换。
+# 幂等: merge 端 DROP+CREATE+INSERT，重跑安全。
+set -eu
+
+KEY=/var/services/homes/admin/.ssh/id_ed25519_nas
+VPS=Administrator@49.232.102.198
+VPS_DB='C:\sanguo_vnpy_v2\data\quant_trading.db'
+VPS_OUT='C:\sanguo_vnpy_v2\data\_sync_tbl.db'
+VPS_OUT_SCP='C:/sanguo_vnpy_v2/data/_sync_tbl.db'
+VPS_PY='C:\Python310\python.exe -X utf8 -'
+
+ROOT=/volume1/stock/sanguo_vnpy_v2
+EXP=$ROOT/scripts/nas_sync/export_table.py
+MERGE=$ROOT/scripts/nas_sync/merge_table.py
+DB=$ROOT/data_backup/quant_trading.db
+STAGE=$ROOT/data_backup/_staging
+LOG=$ROOT/data_backup/sync_tables.log
+
+# 要全量同步的小表清单（merge_table.py 需登记 schema）
+TABLES=(constituent_unified bs_adjust_factor)
+
+mkdir -p "$STAGE"
+echo "=== $(date '+%F %T') tables sync start ===" >> "$LOG"
+
+for T in "${TABLES[@]}"; do
+  echo "[$(date '+%T')] export $T ..." >> "$LOG"
+  ssh -i "$KEY" -o StrictHostKeyChecking=no "$VPS" \
+    "$VPS_PY --db $VPS_DB --out $VPS_OUT --table $T" < "$EXP" >> "$LOG" 2>&1
+  scp -i "$KEY" -o StrictHostKeyChecking=no "$VPS:$VPS_OUT_SCP" "$STAGE/tbl.db" >> "$LOG" 2>&1
+  python3 "$MERGE" --db "$DB" --inc "$STAGE/tbl.db" --table "$T" >> "$LOG" 2>&1
+  rm -f "$STAGE/tbl.db"
+  echo "[$(date '+%T')] $T done" >> "$LOG"
+done
+echo "=== $(date '+%F %T') TABLES SYNC DONE ===" >> "$LOG"