claude_dev
b2aab85105
新增 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)。
80 lines
2.9 KiB
Python
80 lines
2.9 KiB
Python
"""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()
|