sanguo/sanguo_vnpy_v2
2
0
Fork 0
Code Issues 6 Pull Requests Actions Packages Projects Releases Wiki Activity
Files
e5e96324603a0923df2e1253d4d56cbd68205294
sanguo_vnpy_v2/docs/archive/data/2026-07-22-data-arch-migration.md
T
claude_dev c3e53fbef3 docs(data): 归档数据层验证产物 + 数据层总览README 
- scripts/data_platform/_archive/legacy/: 归档20个独立探针/诊断/旧降级脚本(零引用验证)
- docs/archive/data/: 归档17个数据相关旧设计/plan/report(保留fusion spec作深读)
- docs/data-platform/README.md: 数据层单一权威记录(8节:架构/布局/源/管线/铁律/API/缺口/待办)
- 删除 _mootdx_depth_result.txt
- Phase2待办: 15m灌库链+旧回填import链(有测试/wrapper依赖,VPS schtask确认后归档)
2026-07-29 10:11:38 +08:00

6.0 KiB
Raw Blame History

数据架构方案A迁移 + schtask 改造 实施计划

For agentic workers: REQUIRED SUB-SKILL: superpowers:executing-plans。Steps use checkbox。

Goal: 落地 spec §14 方案A定稿 — DB 唯一表、每类数据唯一权威源、4 个新 schtask、迁移 5 单元,全程备份+staging+可回滚+审计。

Architecture: 以本地 DB 迁移为主(daily_baostock_full→dbbardata/parquet,无网络),schtask 改造(废弃旧 4 个新建 4 个)。每单元独立可回滚,按风险升序。

Tech Stack: Python3.10 / sqlite3(WAL) / pandas parquet / Windows schtasks / baostock+xtata+akshare

Global Constraints

  • baostock:单进程单登录,DAILY_LIMIT=48000,sleep 限速,login 探针 graceful skip,直连不走代理(unset proxy)
  • xtata:单进程 download 不并发,无限流
  • akshare:interval 4s 单线程,防东财封 IP
  • 每迁移单元前:sqlite3 .backup 全库 + rsync 到 NAS /volume1/stock/backup/ + WAL checkpoint
  • 每单元:staging 隔离 → 验证探针 → 用户确认合并 → 旧 rename _old 保留 7 天
  • 全程 nohup + 审计日志 data/migration_logs/<unit>_<ts>.log
  • 不破坏 vnpy 回测:dbbardata schema 不动(只灌数据),dbbardata 12 列保持

文件结构

  • 迁移脚本:scripts/data_platform/migrate_*.py(每单元一个)
  • 验证脚本:scripts/data_platform/verify_*.py
  • schtask wrapper:scripts/data_platform/*_wrapper.ps1
  • 审计日志:data/migration_logs/

前置 Task 0:全库备份(所有单元前必做)

Files: scripts/data_platform/backup_db.py(新建,可复用)

  • 写脚本:sqlite3 .backup → quant_trading.db.bak_<YYYYMMDD>(在线一致);WAL checkpoint;rsync 到 NAS
  • 执行
  • verify:.bak 存在 + 大小≈28GB + PRAGMA integrity_check ok

单元 1:存量垃圾清理(零风险)

Files: scripts/data_platform/cleanup_staging.py(新建)

  • 写脚本:--dry-run 先列清单 → 删 _staging_xtdata/(14万)、_xtdata.tar(1.4G);移 cta_*/dbg_*/smoke_*/trace_* → backtest_files/
  • dry-run 输出清单给用户确认
  • 执行删除/移动
  • verify:data/ 根目录无散落 json/log;backtest_files/ 收纳;du -sh data/ 体积下降
  • 回滚:staging 可由 build_daily_from_xtdata 重建(已合并到 qfq/raw)

单元 2:config 统一 VPS 路径

Files: config/data_platform.yaml(VPS 实例)

  • 核实 VPS 实际 config 路径(当前仓库版指 NAS /volume1,是容器版遗留)
  • daily_dir/raw_dir/qfq_dir/minute_15_dir → C:\sanguo_vnpy_v2\data\...
  • daily_dir 统一指 qfq(消除 daily/ vs qfq/ 分叉,daily/68文件归档)
  • NAS config 保留 + 注释"备份用"
  • verify:datareader.read_parquet_daily 抽样能读 + LocalParquetProvider 抽样
  • 回滚:yaml 改回

单元 3:成份股合并 → constituent_unified

Files: scripts/data_platform/migrate_constituent.py + verify_constituent.py Interfaces: 读 bs_index_constituent(baostock 300/500/50)+ data/index_const_hist/*_union.parquet(akshare cni 深证);写 constituent_unified(date,index_code,code,code_name,source)

  • 写迁移脚本:按指数代码去重(300/500/50=baostock;深证 399xxx=akshare cni union;新浪 300/50 作校验丢弃);schema 映射 INSERT
  • staging:先写 constituent_unified_staging
  • verify:行数 / 指数覆盖 / 抽样某指数某日成份集 vs 源一致 / 无同指数同日重复
  • 合并:rename staging → constituent_unified;bs_index_constituent → _old
  • 7 天后删 _old
  • 回滚:rename bs_index_constituent_old 回来

单元 4:daily_baostock_full 拆分(本地 DB 迁移,无网络)

Files: scripts/data_platform/migrate_daily_baostock.py + verify_daily_migration.py Interfaces: 读 daily_baostock_full(含退市);写 dbbardata('d')(OHLCV 12 列)+ data/valuation_baostock/<year>.parquet

  • 写迁移脚本:
    • OHLCV:daily_baostock_full → dbbardata INSERT OR REPLACE(interval='d',exchange SH/SZ→SSE/SZSE,datetime=date)。含退市(治偏差)。ETF 不碰(已在 dbbardata)
    • pe/pb:按年 group → valuation_baostock/<year>.parquet 宽表
  • staging:先写 dbbardata_staging_daily 表 + parquet staging 目录,不动 dbbardata
  • verify:
    • 退市股(000005 等)在 dbbardata('d') 有了(治偏差验证)
    • 在市股(600519)日线行数 / 抽样价格 vs daily_baostock_full 一致
    • pe/pb parquet 按年覆盖 + 抽样值合理
    • dbbardata 总行数变化合理(+退市日线)
  • 合并:staging → dbbardata;daily_baostock_full → _old;valuation parquet → 正式目录
  • 7 天后删 _old
  • 回滚:daily_baostock_full_old 还原 + dbbardata 从 .bak 恢复

单元 5:schtask 改造(废弃旧 4 个,新建 4 个)

Files: scripts/data_platform/bs_eod.py(日线+15min+pe/pb 拆)+ xt_eod.py(ETF+实时)+ *_wrapper.ps1

  • 写 bs_eod.py:基于 daily_update_static.py 扩展,+15min 增量,+pe/pb 拆 parquet;落 dbbardata('d'/'15m');DAILY_LIMIT=48000
  • 写 xt_eod.py:基于 daily_update_xtdata.py,universe 收窄 ETF/基金 + 个股当天实时;落 dbbardata('d')
  • verify:--limit 10 小样本跑通 + 数据到当天
  • 部署 schtask:废弃 sanguo-daily-update/sanguo-bs-daily-increment/sanguo-index-hist;新建 sanguo-bs-eod(18:05)/sanguo-xt-eod(18:40);sanguo-bs-akshare 调到 19:00;sanguo-index(月度 19:50)
  • verify:schtasks /query + 首日运行结果码 + 数据抽查到当天
  • 回滚:重新注册旧 schtask

收尾:E2E 验证

  • 回测 all_weather 一轮(读 dbbardata 日线含退市 + valuation parquet + constituent_unified)无回归
  • LocalParquetProvider 接 constituent_unified + valuation_baostock 单测
  • 更新 memory:data-fusion-design-finalized(标方案A落地)+ 新建 data-arch-migration-done

执行节奏

  • 每单元独立提交 + 用户 review staging 再合并(单元 4/5 关键)
  • 全程 VPS nohup 跑(Mac Mini 防休眠 caffeinate,长迁移)
  • 顺序:0 → 1 → 2 → 3 → 4 → 5 → 收尾(严格风险升序)
Reference in New Issue View Git Blame Copy Permalink