diff --git a/docs/superpowers/plans/2026-07-22-data-arch-migration.md b/docs/superpowers/plans/2026-07-22-data-arch-migration.md new file mode 100644 index 0000000..8d479ac --- /dev/null +++ b/docs/superpowers/plans/2026-07-22-data-arch-migration.md @@ -0,0 +1,106 @@ +# 数据架构方案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/_.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_`(在线一致);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/.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/.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 → 收尾(严格风险升序) diff --git a/docs/superpowers/specs/2026-07-21-data-source-fusion-design.md b/docs/superpowers/specs/2026-07-21-data-source-fusion-design.md index bfd7252..29a152f 100644 --- a/docs/superpowers/specs/2026-07-21-data-source-fusion-design.md +++ b/docs/superpowers/specs/2026-07-21-data-source-fusion-design.md @@ -230,6 +230,68 @@ get_delisted_kline(...) # 退市股 K 线 --- +## 14. 方案 A 定稿(2026-07-22 讨论收敛 — 本节为最新,覆盖前文相关决策) + +### 14.1 背景 +P0 补全 + schtask 部署完成后,审计(`audit_data_layout.py` + `dbbardata_probe.py` 实证)发现「同类多源、多份落地、DB 多表」混乱。讨论收敛为方案 A:**每类数据唯一权威源(允许互补 fallback,禁止并行)、DB 表唯一、估值/三表/事件不进 DB、DB 只放高频随机读取的 K 线+成份股+复权**。 + +### 14.2 权威源最终分工 +| 数据类 | 权威源 | 备注 | +|---|---|---| +| 个股日线 OHLCV(含退市) | **baostock** | raw 真实价 → dbbardata('d') | +| ETF/基金日线 | **xtata** | baostock 盲区(只取 type=1) | +| 个股当天实时(盘后 baostock 未更窗口/盘中) | **xtata** | 拼 baostock 历史,同为 raw | +| 15min | **baostock** | 历史+增量同源 | +| 5min/1min | baostock(暂停) | 战略占位,size 扩容续 | +| 估值 pe/pb/ps/pcf/turn/pctChg/isST | **baostock** | 日线18字段拆出 | +| 基本面三表 | **akshare** | | +| 成份股 300/500/50 | **baostock**(历史) | | +| 成份股 深证 399xxx | **akshare cni**(历史 union) | | +| 成份股 中证1000/2000 | akshare csindex(当前快照) | 历史不可补=永久 gap | +| 复权因子 | **baostock** | 全链路复权基准 | +| 事件类(龙虎榜/大宗/北向/两融/解禁/预告) | **akshare** | | + +三大源:xtata(ETF+实时) / baostock(个股日线+估值+15min+复权+300.500.50) / akshare(三表+事件+深证中证成份股)。同类不并行,按标的/字段互补。 + +### 14.3 DB 边界(只放) +- `dbbardata`:日线('d')+15min('15m')+5min('5m'占位)— **唯一行情表** +- `constituent_unified`:成份股合并唯一表(date,index_code,code,code_name,source) +- `bs_adjust_factor`:复权 +- `backtest_stats`:回测(已有) +- **不进 DB**:基本面三表、估值 pe/pb、事件类、回测明细(独立 db) +- **废弃** `daily_baostock_full`(OHLCV→dbbardata,pe/pb→parquet)、`bs_index_constituent`(合入 constituent_unified) + +### 14.4 pe/pb 落地:parquet 按年宽表 +`data/valuation_baostock/.parquet`,列 `date,symbol,pe_ttm,pb_mrq,ps_ttm,pcf_ncf_ttm,turn,pct_chg,is_st`。选股排序(全市场某日)列存宽表秒级。baostock 日线增量同源拆出。 + +### 14.5 增量 schtask(4 个) +| schtask | 时间 | 源 | 内容 | 落点 | 预算 | +|---|---|---|---|---|---| +| sanguo-bs-eod | 18:05 | baostock | 个股日线+15min 增量+拆 pe/pb | dbbardata('d'/'15m')+valuation_baostock/ | ~11000/48000 | +| sanguo-xt-eod | 18:40 | xtata | ETF/基金日线+个股当天实时补 | dbbardata('d') | 无限流 | +| sanguo-bs-akshare | 19:00 | akshare | 三表+事件类 | parquet | 限频单线程 | +| sanguo-index | 月度 19:50 | baostock+akshare | 成份股合并 | constituent_unified | 小 | + +baostock 单进程单登录,`DAILY_LIMIT=48000`,sleep 限速,login 探针 graceful skip,直连不走代理。5min/1min 不设 schtask。 + +### 14.6 dbbardata 补退市(本次审计发现,顺带治回测幸存者偏差) +实证(`dbbardata_probe`):退市股(000005/000023/600811)在 dbbardata **只有 15m,无日线** → 回测读 dbbardata 日线天然幸存者偏差。迁移:`daily_baostock_full` OHLCV 全量(含退市)→ dbbardata('d') INSERT OR REPLACE(本地 DB 迁移,无网络)。ETF 已在 dbbardata(不碰)。 + +### 14.7 复权统一 +全链路 raw 真实价存储(dbbardata 存 raw),前复权由消费端按 `bs_adjust_factor`(baostock)统一算。xtata 当天实时同为 raw,可拼接。禁止混 xtata dividend_type=front。 + +### 14.8 数据迁移(5 单元,风险升序,备份+staging+可回滚+审计日志) +1. 存量垃圾清理(`_staging_xtdata` 14万文件 / `_xtdata.tar` 1.4G / 归拢 cta_* dbg_* → backtest_files/) +2. config 统一 VPS 路径(NAS 仅备份) +3. 成份股合并(bs_index_constituent + index_const_hist union → constituent_unified,按指数代码去重,新浪300/50丢弃) +4. daily_baostock_full 拆分(OHLCV→dbbardata('d') 含退市;pe/pb→valuation_baostock/.parquet;旧表 rename _old) +5. dbbardata 个股日线切 baostock 源(单元4已覆盖:daily_baostock_full 含全量,在市股 REPLACE 覆盖 xtata,数值一致) + +每单元:全库 `sqlite3 .backup` + rsync NAS → 脚本写 staging(新表/新目录)→ 验证探针(行数/distinct symbol/抽样价格/成功率)→ 用户确认合并 → 旧数据 rename _old 保留 7 天。全程 nohup + 审计日志 `data/migration_logs/`。 + +### 14.9 config 清理 +VPS config 的 daily_dir/raw_dir/qfq_dir/minute_15_dir 改 `C:\sanguo_vnpy_v2\data\...`(当前 yaml 指 NAS /volume1,容器版遗留)。NAS config 保留作备份。read_parquet_daily 的 daily_dir 统一指向 qfq(消除 daily/ vs qfq/ 分叉)。 + ## 参考(调查来源) - xtdata 官方:https://dict.thinktrader.net/nativeApi/xtdata.html diff --git a/scripts/data_platform/audit_data_layout.py b/scripts/data_platform/audit_data_layout.py new file mode 100644 index 0000000..110d455 --- /dev/null +++ b/scripts/data_platform/audit_data_layout.py @@ -0,0 +1,87 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +"""audit_data_layout.py — 只读盘点 VPS 本地数据布局(DB表行数/maxdate + data/目录)。 + +用法: + python audit_data_layout.py [BASE_DIR] +默认 BASE=C:\\sanguo_vnpy_v2 (VPS)。Mac 调试传本地路径。 + +只 SELECT / 遍历目录,不写任何东西。 +""" +import sqlite3 +import sys +from pathlib import Path + +BASE = Path(sys.argv[1]) if len(sys.argv) > 1 else Path(r"C:\sanguo_vnpy_v2") +DB = BASE / "data" / "quant_trading.db" +DATA = BASE / "data" + + +def line(s=""): + print(s) + + +# ======================== DB 表盘点 ======================== +line("=" * 72) +line(f"DB: {DB} exists={DB.exists()}") +line("=" * 72) +if DB.exists(): + c = sqlite3.connect(str(DB), timeout=60) + tables = [r[0] for r in c.execute( + "SELECT name FROM sqlite_master WHERE type='table' ORDER BY name")] + SKIP_COUNT = {"dbbardata", "daily_baostock_full"} # 千万/亿级 COUNT 慢, 跳过 + for t in tables: + if t in SKIP_COUNT: + line(f" {t:42s} (skip COUNT, see max below)") + continue + try: + n = c.execute(f"SELECT COUNT(*) FROM '{t}'").fetchone()[0] + except Exception as e: + n = f"ERR {e}" + line(f" {t:42s} {n}") + + # 关键表 schema + 时间范围 (大表 dbbardata/daily_baostock_full 跳过慢查询) + for t in ("bs_index_constituent", "bs_adjust_factor"): + try: + cols = [r[1] for r in c.execute(f"PRAGMA table_info('{t}')")] + if not cols: + continue + line("") + line(f"[{t}] cols({len(cols)}): {cols}") + if "symbol" in cols: + nd = c.execute(f"SELECT COUNT(DISTINCT symbol) FROM '{t}'").fetchone()[0] + line(f" distinct symbol = {nd}") + for dc in ("date", "datetime", "trade_date"): + if dc in cols: + mn = c.execute(f"SELECT MIN({dc}) FROM '{t}'").fetchone()[0] + mx = c.execute(f"SELECT MAX({dc}) FROM '{t}'").fetchone()[0] + line(f" {dc}: {mn} ~ {mx}") + break + except Exception as e: + line(f"[{t}] ERR {e}") + c.close() + +# ======================== data/ 目录盘点 ======================== +line("") +line("=" * 72) +line(f"DATA DIR: {DATA} exists={DATA.exists()}") +line("=" * 72) +if DATA.exists(): + for sub in sorted(DATA.iterdir()): + name = sub.name + if name.startswith("."): + continue + try: + if sub.is_dir(): + files = list(sub.rglob("*")) + n_files = sum(1 for f in files if f.is_file()) + n_pq = sum(1 for f in files if f.suffix == ".parquet") + n_xls = sum(1 for f in files if f.suffix in (".xls", ".xlsx")) + line(f" [DIR] {name:34s} files={n_files:6d} parquet={n_pq:6d} xls={n_xls}") + elif sub.is_file(): + sz = sub.stat().st_size + unit, val = ("GB", sz / 1024 ** 3) if sz > 1024 ** 3 else ( + "MB", sz / 1024 ** 2) + line(f" [FILE] {name:32s} {val:8.1f} {unit}") + except Exception as e: + line(f" {name} ERR {e}") diff --git a/scripts/data_platform/backup_db.py b/scripts/data_platform/backup_db.py new file mode 100644 index 0000000..8fde531 --- /dev/null +++ b/scripts/data_platform/backup_db.py @@ -0,0 +1,42 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +"""backup_db.py — VPS 本地全库备份(回滚保险)。 + +sqlite3 .backup 在线一致 + WAL checkpoint(TRUNCATE) + integrity_check。 +所有迁移单元的前置。BAK 同机存放,可立即恢复(满足"备份能回滚搞不坏")。 + +注:VPS(公网)↔ NAS(内网)不直连,异地冗余需 Mac 中转(28GB 大,可选)。 +此处只做 VPS 本地 .bak(核心回滚保障)。 +""" +import sqlite3 +import datetime +from pathlib import Path + +DB = Path(r"C:\sanguo_vnpy_v2\data\quant_trading.db") +ts = datetime.date.today().strftime("%Y%m%d") +BAK = DB.parent / f"quant_trading.db.bak_{ts}" + +print(f"DB : {DB} ({DB.stat().st_size / 1024**3:.1f} GB)") +print(f"BAK: {BAK}") + +src = sqlite3.connect(str(DB), timeout=120) +src.execute("PRAGMA busy_timeout = 120000") +print("WAL checkpoint(TRUNCATE)...") +print(" ", src.execute("PRAGMA wal_checkpoint(TRUNCATE)").fetchone()) + +if BAK.exists(): + print(f"BAK 已存在, 覆盖: {BAK}") + +print("sqlite3 .backup (在线一致, 可能需几分钟)...") +with sqlite3.connect(str(BAK)) as dst: + src.backup(dst) +src.close() + +print(f"BAK 写入: {BAK.stat().st_size / 1024**3:.1f} GB") + +# integrity_check on backup +chk = sqlite3.connect(str(BAK)) +result = chk.execute("PRAGMA integrity_check").fetchone()[0] +chk.close() +print(f"integrity_check: {result}") +print("BACKUP DONE" if result == "ok" else "BACKUP WARNING: integrity_check 非 ok") diff --git a/scripts/data_platform/bs_eod.py b/scripts/data_platform/bs_eod.py new file mode 100644 index 0000000..d61d347 --- /dev/null +++ b/scripts/data_platform/bs_eod.py @@ -0,0 +1,264 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +"""bs_eod.py — sanguo-bs-eod (方案A schtask 18:05): baostock 个股 EOD 增量。 + +每日收盘后跑(VPS, baostock 日终更新就绪): +- 个股日线(含退市, LOOKBACK 7 天) -> dbbardata('d') INSERT OR REPLACE (治幸存者偏差) +- 个股 15min(LOOKBACK 7) -> dbbardata('15m') +- 日线 pe/pb/turn/pctChg/isST -> data/valuation_baostock/.parquet 追加 +- DAILY_LIMIT=48000 单进程单登录, sleep 0.3s, login 探针 graceful skip + +预算: 5537股 × (1日线+1 15min) ≈ 11000 query/天 = 48000 的 23%, 安全。 +退出码: 0=完成; 1=致命; 2=黑名单 graceful skip; 3=query 超限 graceful stop +""" +import argparse +import datetime as dt +import logging +import os +import socket +import sys +import time +from pathlib import Path + +for _k in ("http_proxy", "https_proxy", "HTTP_PROXY", "HTTPS_PROXY", "all_proxy", "ALL_PROXY"): + os.environ.pop(_k, None) +socket.setdefaulttimeout(30) +try: + sys.stdout.reconfigure(line_buffering=True) +except (AttributeError, ValueError): + pass + +import baostock as bs +import pandas as pd + +BASE = Path(r"C:\sanguo_vnpy_v2") +DB = BASE / "data" / "quant_trading.db" +VAL_DIR = BASE / "data" / "valuation_baostock" +LOOKBACK = int(os.environ.get("LOOKBACK_DAYS", "7")) +DAILY_LIMIT = int(os.environ.get("BS_DAILY_LIMIT", "48000")) +BS_INTERVAL = 0.3 +QUERY_COUNT = 0 +EXC_MAP = {"sh": "SSE", "sz": "SZSE"} + +DAILY_FIELDS = ("date,code,open,high,low,close,volume,amount,turn," + "pctChg,peTTM,psTTM,pcfNcfTTM,pbMRQ,isST") +M15_FIELDS = "date,time,code,open,high,low,close,volume,amount" + +logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s", + handlers=[logging.StreamHandler(sys.stdout)]) +log = logging.getLogger(__name__) + + +def login_once(): + try: + lg = bs.login() + if lg.error_code == "0": + return True + log.error("login fail: %s %s", lg.error_code, lg.error_msg) + return False + except Exception as e: + log.error("login exc: %s", e) + return False + + +def relogin(): + try: + bs.logout() + except Exception: + pass + if login_once(): + return True + time.sleep(2) + try: + bs.logout() + except Exception: + pass + return login_once() + + +def fetch_all_stocks(): + """query_stock_basic() 无参 -> 全 A 含退市 (type=1)。返回 [(code, 'sh'/'sz')]。""" + global QUERY_COUNT + QUERY_COUNT += 1 + rs = bs.query_stock_basic() + if rs.error_code != "0": + raise RuntimeError(f"query_stock_basic: {rs.error_code} {rs.error_msg}") + fields = list(rs.fields) + idx = {n: i for i, n in enumerate(fields)} + out = [] + while rs.next(): + r = rs.get_row_data() + if r[idx["type"]] != "1": + continue + bc = r[idx["code"]] + if "." not in bc: + continue + prefix, num = bc.split(".", 1) + if prefix in ("sh", "sz") and len(num) == 6 and num.isdigit(): + out.append((num, prefix)) + return out + + +def fetch_k(bs_code, fields, freq, start, end): + global QUERY_COUNT + QUERY_COUNT += 1 + rs = bs.query_history_k_data_plus(bs_code, fields, start_date=start, + end_date=end, frequency=freq, adjustflag="3") + if rs.error_code != "0": + raise RuntimeError(f"{bs_code}: {rs.error_code} {rs.error_msg}") + rows = [] + while rs.next(): + rows.append(rs.get_row_data()) + return rows + + +def upsert_daily(conn, code, prefix, rows): + """日线 rows -> dbbardata('d') + valuation_baostock 当年 parquet 追加。""" + if not rows: + return 0 + df = pd.DataFrame(rows, columns=DAILY_FIELDS.split(",")) + for c in ["open", "high", "low", "close", "volume", "amount", + "turn", "pctChg", "peTTM", "psTTM", "pcfNcfTTM", "pbMRQ"]: + df[c] = pd.to_numeric(df[c], errors="coerce") + exc = EXC_MAP[prefix] + # OHLCV -> dbbardata('d') + db = pd.DataFrame({ + "symbol": code, "exchange": exc, + "datetime": df["date"].astype(str) + " 00:00:00", + "interval": "d", "volume": df["volume"], "turnover": df["amount"], + "open_interest": 0.0, + "open_price": df["open"], "high_price": df["high"], + "low_price": df["low"], "close_price": df["close"], + }) + conn.executemany( + "INSERT OR REPLACE INTO dbbardata " + "(symbol,exchange,datetime,interval,volume,turnover,open_interest," + "open_price,high_price,low_price,close_price) VALUES (?,?,?,?,?,?,?,?,?,?,?)", + db.itertuples(index=False, name=None)) + # pe/pb -> parquet 追加 (isST->int, 修 pyarrow ArrowTypeError) + vdf = df[["date", "peTTM", "psTTM", "pcfNcfTTM", "pbMRQ", "turn", "pctChg", "isST"]].copy() + vdf["isST"] = pd.to_numeric(vdf["isST"], errors="coerce").fillna(0).astype(int) + vdf.insert(0, "symbol", code) + vdf.insert(1, "exchange", exc) + yr = dt.date.today().year + p = VAL_DIR / f"{yr}.parquet" + if p.exists(): + try: + old = pd.read_parquet(p) + vdf = pd.concat([old, vdf]).drop_duplicates(["symbol", "date"], keep="last") + except Exception: + pass + vdf.sort_values(["symbol", "date"]).to_parquet(p, index=False) + return len(db) + + +def upsert_15m(conn, code, prefix, rows): + if not rows: + return 0 + df = pd.DataFrame(rows, columns=M15_FIELDS.split(",")) + for c in ["open", "high", "low", "close", "volume", "amount"]: + df[c] = pd.to_numeric(df[c], errors="coerce") + exc = EXC_MAP[prefix] + dt_col = (df["date"].astype(str) + " " + df["time"].astype(str).str.slice(0, 6) + ).apply(lambda s: f"{s[0:4]}-{s[4:6]}-{s[6:8]} {s[8:10]}:{s[10:12]}:00") + db = pd.DataFrame({ + "symbol": code, "exchange": exc, "datetime": dt_col, + "interval": "15m", "volume": df["volume"], "turnover": df["amount"], + "open_interest": 0.0, + "open_price": df["open"], "high_price": df["high"], + "low_price": df["low"], "close_price": df["close"], + }) + conn.executemany( + "INSERT OR REPLACE INTO dbbardata " + "(symbol,exchange,datetime,interval,volume,turnover,open_interest," + "open_price,high_price,low_price,close_price) VALUES (?,?,?,?,?,?,?,?,?,?,?)", + db.itertuples(index=False, name=None)) + return len(db) + + +def main(): + global QUERY_COUNT + ap = argparse.ArgumentParser() + ap.add_argument("--limit", type=int, default=0) + ap.add_argument("--no-15m", action="store_true") + args = ap.parse_args() + + today = dt.date.today() + end = today.strftime("%Y-%m-%d") + start = (today - dt.timedelta(days=LOOKBACK)).strftime("%Y-%m-%d") + log.info("bs_eod start window=%s~%s LOOKBACK=%d limit=%s", start, end, LOOKBACK, args.limit or "无") + + if not login_once(): + log.error("[SKIP] baostock 黑名单/冷却, graceful exit 2") + sys.exit(2) + + try: + stocks = fetch_all_stocks() + except Exception as e: + log.error("[FATAL] fetch_all: %s", e) + sys.exit(1) + log.info("全 A 含退市: %d 只", len(stocks)) + if args.limit: + stocks = stocks[:args.limit] + + import sqlite3 + VAL_DIR.mkdir(parents=True, exist_ok=True) + conn = sqlite3.connect(str(DB), timeout=60) + conn.execute("PRAGMA busy_timeout = 60000") + conn.execute("PRAGMA journal_mode = WAL") + + stats = {"ok": 0, "empty": 0, "failed": 0, "db_rows": 0} + limit_reached = False + t0 = time.time() + conn.execute("BEGIN") + try: + for i, (code, prefix) in enumerate(stocks): + if QUERY_COUNT >= DAILY_LIMIT: + log.warning("query %d 达防线 %d, graceful stop", QUERY_COUNT, DAILY_LIMIT) + limit_reached = True + break + bs_code = f"{prefix}.{code}" + try: + d_rows = fetch_k(bs_code, DAILY_FIELDS, "d", start, end) + n1 = upsert_daily(conn, code, prefix, d_rows) + n2 = 0 + if not args.no_15m: + m_rows = fetch_k(bs_code, M15_FIELDS, "15", start, end) + n2 = upsert_15m(conn, code, prefix, m_rows) + stats["db_rows"] += n1 + n2 + if n1 + n2: + stats["ok"] += 1 + else: + stats["empty"] += 1 + except Exception as e: + stats["failed"] += 1 + if stats["failed"] <= 5: + log.warning("%s err: %s", code, e) + if not relogin(): + log.error("%s relogin 失败, 跳过", code) + if (i + 1) % 500 == 0: + log.info("进度 %d/%d ok=%d empty=%d failed=%d q=%d (%.0fs)", + i + 1, len(stocks), stats["ok"], stats["empty"], + stats["failed"], QUERY_COUNT, time.time() - t0) + if i < len(stocks) - 1: + time.sleep(BS_INTERVAL) + conn.execute("COMMIT") + except Exception as e: + conn.execute("ROLLBACK") + log.error("[FATAL] rollback: %s", e) + sys.exit(1) + finally: + conn.close() + try: + bs.logout() + except Exception: + pass + + log.info("[DONE] ok=%d empty=%d failed=%d db_rows=%d query=%d 耗时%.0fs", + stats["ok"], stats["empty"], stats["failed"], stats["db_rows"], + QUERY_COUNT, time.time() - t0) + sys.exit(3 if limit_reached else 0) + + +if __name__ == "__main__": + main() diff --git a/scripts/data_platform/bs_eod_wrapper.ps1 b/scripts/data_platform/bs_eod_wrapper.ps1 new file mode 100644 index 0000000..394ef84 --- /dev/null +++ b/scripts/data_platform/bs_eod_wrapper.ps1 @@ -0,0 +1,12 @@ +# bs_eod_wrapper.ps1 — sanguo-bs-eod schtask wrapper (18:05 baostock 个股 EOD 增量) +# 方案A: 个股日线+15min -> dbbardata('d'/'15m') + pe/pb -> valuation_baostock +$env:http_proxy = '' +$env:https_proxy = '' +$env:all_proxy = '' +Set-Location C:\sanguo_vnpy_v2 +$ts = Get-Date -Format 'yyyyMMdd_HHmmss' +$logDir = 'C:\sanguo_vnpy_v2\data\migration_logs' +if (-not (Test-Path $logDir)) { New-Item -ItemType Directory -Path $logDir -Force | Out-Null } +$log = Join-Path $logDir "bs_eod_$ts.txt" +C:\Python310\python.exe -X utf8 C:\sanguo_vnpy_v2\scripts\data_platform\bs_eod.py *>> $log +exit $LASTEXITCODE diff --git a/scripts/data_platform/check_schtask_runs.ps1 b/scripts/data_platform/check_schtask_runs.ps1 new file mode 100644 index 0000000..a2d2e95 --- /dev/null +++ b/scripts/data_platform/check_schtask_runs.ps1 @@ -0,0 +1,14 @@ +# check_schtask_runs.ps1 — 查 4 个数据 schtask 上次运行+结果 + daily-update 数据抽查 +$ErrorActionPreference = 'SilentlyContinue' +Write-Output "=== NOW: $(Get-Date -Format 'yyyy-MM-dd HH:mm:ss') ===" +foreach($tn in @('sanguo-daily-update','sanguo-bs-akshare','sanguo-bs-daily-increment','sanguo-index-hist')){ + $i = Get-ScheduledTaskInfo -TaskName $tn + if($i){ + $code = ('0x{0:X}' -f $i.LastTaskResult) + Write-Output ("{0,-28} last={1} result={2} next={3}" -f $tn, $i.LastRunTime, $code, $i.NextRunTime) + } else { Write-Output "$tn NOT FOUND" } +} +Write-Output "" +Write-Output "=== daily-update 数据抽查(data/qfq/2026 max date, 应=2026-07-22) ===" +$py = 'C:\Python310\python.exe' +& $py -X utf8 -c "import pandas as pd,glob,os; [print(os.path.basename(f),'maxdate=',pd.read_parquet(f)['date'].max()) for f in [r'C:\sanguo_vnpy_v2\data\qfq\2026\sh600519_daily.parquet', r'C:\sanguo_vnpy_v2\data\qfq\2026\sh510300_daily.parquet'] if os.path.exists(f)]" diff --git a/scripts/data_platform/cleanup_staging.py b/scripts/data_platform/cleanup_staging.py new file mode 100644 index 0000000..2f78f79 --- /dev/null +++ b/scripts/data_platform/cleanup_staging.py @@ -0,0 +1,98 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +"""cleanup_staging.py — 单元1: 清理存量垃圾(零风险)。 + +删: +- data/_staging_xtdata/ (xtata build staging, 已合并到 qfq/raw) +- data/_xtdata.tar (一次性迁移包) +移: +- data/cta_* dbg_* smoke_* trace_* (json/log 回测产物) → data/backtest_files/ + +用法: + python cleanup_staging.py --dry-run # 只列清单+大小 + python cleanup_staging.py --execute # 实际删/移 +""" +import argparse +import shutil +from pathlib import Path + +BASE = Path(r"C:\sanguo_vnpy_v2") +DATA = BASE / "data" + +DELETE_DIRS = [DATA / "_staging_xtdata"] +DELETE_FILES = [DATA / "_xtdata.tar"] +MOVE_GLOBS = ["cta_*", "dbg_*", "smoke_*", "trace_*"] +MOVE_TO = DATA / "backtest_files" + + +def size_str(n): + for u in ["B", "KB", "MB", "GB"]: + if n < 1024: + return f"{n:.1f}{u}" + n /= 1024 + return f"{n:.1f}TB" + + +def dir_size(p): + total = 0 + count = 0 + for f in p.rglob("*"): + if f.is_file(): + try: + total += f.stat().st_size + count += 1 + except OSError: + pass + return total, count + + +def main(): + ap = argparse.ArgumentParser() + ap.add_argument("--dry-run", action="store_true") + ap.add_argument("--execute", action="store_true") + a = ap.parse_args() + if not a.dry_run and not a.execute: + ap.error("需指定 --dry-run 或 --execute") + + log = [] + for d in DELETE_DIRS: + if d.exists(): + sz, n = dir_size(d) + log.append(f"[DEL DIR] {d} files={n} size={size_str(sz)}") + if a.execute: + shutil.rmtree(d, ignore_errors=True) + log.append(f" -> deleted") + else: + log.append(f"[DEL DIR] {d} (不存在)") + + for f in DELETE_FILES: + if f.exists(): + sz = f.stat().st_size + log.append(f"[DEL FILE] {f} size={size_str(sz)}") + if a.execute: + f.unlink(missing_ok=True) + log.append(f" -> deleted") + else: + log.append(f"[DEL FILE] {f} (不存在)") + + MOVE_TO.mkdir(parents=True, exist_ok=True) + moved = 0 + for pat in MOVE_GLOBS: + for f in DATA.glob(pat): + if f.is_file(): + moved += 1 + if moved <= 20: + log.append(f"[MOVE] {f.name}") + if a.execute: + try: + shutil.move(str(f), str(MOVE_TO / f.name)) + except OSError as e: + log.append(f" move err {f.name}: {e}") + log.append(f"[MOVE] 总计 {moved} 个回测产物 -> backtest_files/") + + print("\n".join(log)) + print(f"\nMODE: {'DRY-RUN (未实际操作)' if a.dry_run else 'EXECUTED'}") + + +if __name__ == "__main__": + main() diff --git a/scripts/data_platform/dbbardata_probe.py b/scripts/data_platform/dbbardata_probe.py new file mode 100644 index 0000000..93ee685 --- /dev/null +++ b/scripts/data_platform/dbbardata_probe.py @@ -0,0 +1,39 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +"""dbbardata_probe.py — 只读探测 dbbardata 现状(schema + 抽样各 interval 覆盖)。 + +走 symbol 索引,避免全表 COUNT(亿级慢)。回答: + - dbbardata 有哪些 interval(d/15m/5m) + - 个股日线(interval='d' 类)含不含退市股(000005/000023/600811/600074) + - 在市股(600519/000001)日线日期范围 + - ETF(510300/159915)有没有(判断 dbbardata 是否覆盖 ETF) +用于决定迁移单元5(dbbardata 个股日线切 baostock 源)的工作量。 +""" +import sqlite3 +from pathlib import Path + +DB = Path(r"C:\sanguo_vnpy_v2\data\quant_trading.db") +c = sqlite3.connect(str(DB), timeout=60) +c.execute("PRAGMA busy_timeout = 60000") + +cols = [r[1] for r in c.execute("PRAGMA table_info(dbbardata)")] +print(f"DB: {DB}") +print(f"dbbardata cols({len(cols)}): {cols}") + +print("\n=== 抽样: 各 symbol 的 interval 覆盖 (走索引, 快) ===") +samples = { + "退市": ["000005", "000023", "600811", "600074"], + "在市个股": ["600519", "000001"], + "ETF/基金": ["510300", "159915", "510050"], +} +for label, syms in samples.items(): + print(f"\n[{label}]") + for sym in syms: + rows = c.execute( + "SELECT interval, COUNT(*), MIN(datetime), MAX(datetime) " + "FROM dbbardata WHERE symbol=? GROUP BY interval", + (sym,), + ).fetchall() + print(f" {sym}: {rows}") + +c.close() diff --git a/scripts/data_platform/diag_daily_update.ps1 b/scripts/data_platform/diag_daily_update.ps1 new file mode 100644 index 0000000..8f0f522 --- /dev/null +++ b/scripts/data_platform/diag_daily_update.ps1 @@ -0,0 +1,25 @@ +# diag_daily_update.ps1 — 诊断 sanguo-daily-update 为何跑 2.5h +Write-Output "=== NOW: $(Get-Date) ===" +Write-Output "=== python 进程 (CPU秒/内存MB/启动时间) ===" +Get-Process python -ErrorAction SilentlyContinue | + Select-Object Id, @{n='CPU_s'; e={[math]::Round($_.CPU, 1)}}, + @{n='WS_MB'; e={[math]::Round($_.WS / 1MB, 0)}}, StartTime | + Format-Table -Auto +Write-Output "=== schtask sanguo-daily-update (状态+执行的命令) ===" +$i = Get-ScheduledTaskInfo -TaskName sanguo-daily-update +Write-Output ("last={0} result=0x{1:X} next={2}" -f $i.LastRunTime, $i.LastTaskResult, $i.NextRunTime) +Write-Output "--- Task Action (实际执行) ---" +(Get-ScheduledTask -TaskName sanguo-daily-update).Actions | + Select-Object Execute, Arguments | Format-List +Write-Output "=== daily_update.log tail 40 ===" +Get-Content C:\sanguo_vnpy_v2\data\daily_update.log -Tail 40 -ErrorAction SilentlyContinue +Write-Output "=== qfq/ 各年: 文件数 + 最新写入时间 ===" +Get-ChildItem C:\sanguo_vnpy_v2\data\qfq -Directory -ErrorAction SilentlyContinue | ForEach-Object { + $files = Get-ChildItem $_.FullName -File -ErrorAction SilentlyContinue + $mx = ($files | Sort-Object LastWriteTime -Descending | Select-Object -First 1).LastWriteTime + Write-Output (" {0}: files={1} lastwrite={2}" -f $_.Name, $files.Count, $mx) +} +Write-Output "=== qfq/2026 最新写入 top 5 (看在不在写) ===" +Get-ChildItem C:\sanguo_vnpy_v2\data\qfq\2026 -ErrorAction SilentlyContinue | + Sort-Object LastWriteTime -Descending | Select-Object -First 5 Name, LastWriteTime | + Format-Table -Auto diff --git a/scripts/data_platform/fix_vps_config.py b/scripts/data_platform/fix_vps_config.py new file mode 100644 index 0000000..43fe437 --- /dev/null +++ b/scripts/data_platform/fix_vps_config.py @@ -0,0 +1,54 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +"""fix_vps_config.py — 单元2: VPS config data_paths 统一到 C:/sanguo_vnpy_v2/data (正斜杠)。 + +- 备份原 config -> data_platform.yaml.bak_nas (NAS 路径版) +- data_paths 改 VPS 本地 + daily_dir 统一 qfq (消除 daily/ vs qfq/ 分叉) +- 保留其他段 (yaml safe_dump, 丢注释但语义不变) +- 改完立即调 read_parquet_daily 验证 datareader 能读本地 + +注: rsync 同步仓库 config 会覆盖 VPS config (NAS 版)。部署 rsync 须 --exclude config, +或 VPS 用独立 config 文件。此处先改 VPS config 让回测读本地, 部署排除留作 TODO。 +""" +import shutil +import sys +from pathlib import Path + +import yaml + +CFG = Path(r"C:\sanguo_vnpy_v2\config\data_platform.yaml") +bak = CFG.with_suffix(".yaml.bak_nas") +if not bak.exists(): + shutil.copy2(CFG, bak) + print(f"backup -> {bak.name}") +else: + print(f"backup 已存在: {bak.name}") + +d = yaml.safe_load(CFG.read_text(encoding="utf-8")) +print("OLD data_paths:", d.get("data_paths")) + +d["data_paths"] = { + "daily_dir": "C:/sanguo_vnpy_v2/data/qfq", + "raw_dir": "C:/sanguo_vnpy_v2/data/raw", + "qfq_dir": "C:/sanguo_vnpy_v2/data/qfq", + "minute_15_dir": "C:/sanguo_vnpy_v2/data/minute_15", + "minute_15_qfq_dir": "C:/sanguo_vnpy_v2/data/minute_15", + "minute_15_raw_dir": "C:/sanguo_vnpy_v2/data/minute_15", + "vnpy_db": "C:/sanguo_vnpy_v2/data/quant_trading.db", + "stock_list": "C:/sanguo_vnpy_v2/data/stock_basic_info.csv", +} +CFG.write_text(yaml.safe_dump(d, allow_unicode=True, sort_keys=False), encoding="utf-8") +print("NEW data_paths:", d["data_paths"]) + +# 验证 datareader 读本地 +sys.path.insert(0, r"C:\sanguo_vnpy_v2") +from sanguo_data.config import load_config +from sanguo_data.datareader import read_parquet_daily + +cfg = load_config() +print("cfg daily_dir:", cfg.data_paths["daily_dir"]) +print("cfg vnpy_db:", cfg.data_paths["vnpy_db"]) +bars = read_parquet_daily("600519", "2026-07-01", "2026-07-22", cfg) +print("600519 bars:", len(bars), + "last:", bars[-1].datetime.date() if bars else "NONE") +print("CONFIG FIX DONE" if bars else "CONFIG FIX WARN: 0 bars") diff --git a/scripts/data_platform/kill_backup_only.ps1 b/scripts/data_platform/kill_backup_only.ps1 new file mode 100644 index 0000000..34dcff5 --- /dev/null +++ b/scripts/data_platform/kill_backup_only.ps1 @@ -0,0 +1,9 @@ +# kill_backup_only.ps1 — 杀卡在 integrity_check 的 backup_db.py python(产物 .bak 已完整验证) +$bakProc = Get-CimInstance Win32_Process -Filter "Name='python.exe'" | + Where-Object { $_.CommandLine -match 'backup_db|audit_data_layout|diag_daily' } +if ($bakProc) { + Write-Output ("killed backup python pid={0}" -f $bakProc.ProcessId) + Stop-Process -Id $bakProc.ProcessId -Force -ErrorAction SilentlyContinue +} else { + Write-Output "backup python 不在(已结束/被杀)" +} diff --git a/scripts/data_platform/kill_backup_verify.ps1 b/scripts/data_platform/kill_backup_verify.ps1 new file mode 100644 index 0000000..81728e2 --- /dev/null +++ b/scripts/data_platform/kill_backup_verify.ps1 @@ -0,0 +1,23 @@ +# kill_backup_verify.ps1 — 杀卡在 integrity_check 的 backup python, 快速验证 .bak 可读 +$bakProc = Get-CimInstance Win32_Process -Filter "Name='python.exe'" | + Where-Object { $_.CommandLine -like '*backup_db*' } +if ($bakProc) { + Write-Output ("kill backup python pid={0}" -f $bakProc.ProcessId) + Stop-Process -Id $bakProc.ProcessId -Force -ErrorAction SilentlyContinue + Start-Sleep 3 +} else { Write-Output "backup python 已不在" } + +$code = @" +import sqlite3, os +BAK = r'C:\sanguo_vnpy_v2\data\quant_trading.db.bak_20260722' +print('bak size=%.2f GB' % (os.path.getsize(BAK)/1024**3)) +c = sqlite3.connect(BAK, timeout=30) +tabs = [r[0] for r in c.execute("SELECT name FROM sqlite_master WHERE type='table' ORDER BY name")] +print('tables:', tabs) +print('backtest_stats:', c.execute('SELECT COUNT(*) FROM backtest_stats').fetchone()[0]) +print('bs_adjust_factor:', c.execute('SELECT COUNT(*) FROM bs_adjust_factor').fetchone()[0]) +print('bs_index_constituent:', c.execute('SELECT COUNT(*) FROM bs_index_constituent').fetchone()[0]) +c.close() +print('BAK READABLE OK') +"@ +$code | & 'C:\Python310\python.exe' -X utf8 - diff --git a/scripts/data_platform/merge_constituent.py b/scripts/data_platform/merge_constituent.py new file mode 100644 index 0000000..467b6bc --- /dev/null +++ b/scripts/data_platform/merge_constituent.py @@ -0,0 +1,32 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +"""merge_constituent.py — 单元3 合并: staging -> constituent_unified, bs_index_constituent -> _old。 + +前提: migrate_constituent.py 已建 constituent_unified_staging (验证通过)。 +- constituent_unified_staging RENAME constituent_unified (正式表) +- bs_index_constituent RENAME bs_index_constituent_old (保留 988 时点精度, 不删) +- 建索引 (index_code, code) 加速 get_index_stocks 查询 +回滚: rename 反向 (constituent_unified->staging, _old->bs_index_constituent) +""" +import sqlite3 + +DB = r"C:\sanguo_vnpy_v2\data\quant_trading.db" +c = sqlite3.connect(DB, timeout=60) +c.execute("PRAGMA busy_timeout = 60000") + +c.execute("ALTER TABLE constituent_unified_staging RENAME TO constituent_unified") +print("renamed constituent_unified_staging -> constituent_unified") + +c.execute("ALTER TABLE bs_index_constituent RENAME TO bs_index_constituent_old") +print("renamed bs_index_constituent -> bs_index_constituent_old (时点精度保留)") + +c.execute("CREATE INDEX IF NOT EXISTS idx_constituent_unified " + "ON constituent_unified(index_code, code)") +c.commit() + +print("constituent_unified rows:", c.execute( + "SELECT COUNT(*) FROM constituent_unified").fetchone()[0]) +print("bs_index_constituent_old rows:", c.execute( + "SELECT COUNT(*) FROM bs_index_constituent_old").fetchone()[0]) +c.close() +print("MERGE DONE") diff --git a/scripts/data_platform/merge_daily_baostock.py b/scripts/data_platform/merge_daily_baostock.py new file mode 100644 index 0000000..9c45134 --- /dev/null +++ b/scripts/data_platform/merge_daily_baostock.py @@ -0,0 +1,75 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +"""merge_daily_baostock.py — 单元4 合并: staging->dbbardata + daily_baostock_full->_old + 补 parquet。 + +前置: migrate_daily_baostock.py 已灌 dbbardata_staging_daily (1826万, verify OK); + dbbardata 有 UNIQUE(symbol,exchange,interval,datetime) (probe 确认)。 +1. INSERT OR REPLACE staging -> dbbardata('d'): 个股(含退市) REPLACE 在市/新增退市 (治偏差) +2. daily_baostock_full -> daily_baostock_full_old (保留, pe/pb 源) +3. pe/pb -> data/valuation_baostock/.parquet (从 _old, isST->int 修 ArrowTypeError) +回滚: rename daily_baostock_full_old->daily_baostock_full; dbbardata 从 .bak 恢复 +""" +import sqlite3 +from pathlib import Path + +import pandas as pd + +DB = Path(r"C:\sanguo_vnpy_v2\data\quant_trading.db") +VAL_DIR = Path(r"C:\sanguo_vnpy_v2\data\valuation_baostock") + + +def log(m): + print(m, flush=True) + + +VAL_DIR.mkdir(parents=True, exist_ok=True) +c = sqlite3.connect(str(DB), timeout=600) +c.execute("PRAGMA busy_timeout = 600000") +c.execute("PRAGMA synchronous = NORMAL") + +log("1. INSERT OR REPLACE staging(1826万) -> dbbardata('d') 含退市治偏差 ...") +# dbbardata 全列 NOT NULL; 跳过停牌(OHLC NULL), volume/turnover COALESCE 0 +c.execute( + "INSERT OR REPLACE INTO dbbardata " + "(symbol,exchange,datetime,interval,volume,turnover,open_interest," + "open_price,high_price,low_price,close_price) " + "SELECT symbol, exchange, datetime, interval, " + "COALESCE(volume, 0), COALESCE(turnover, 0), 0, " + "open_price, high_price, low_price, close_price " + "FROM dbbardata_staging_daily " + "WHERE open_price IS NOT NULL AND high_price IS NOT NULL " + "AND low_price IS NOT NULL AND close_price IS NOT NULL") +c.commit() +log(" INSERT OR REPLACE done") + +log("2. daily_baostock_full -> daily_baostock_full_old") +c.execute("ALTER TABLE daily_baostock_full RENAME TO daily_baostock_full_old") +c.commit() + +log("3. pe/pb -> valuation_baostock/.parquet (isST->int, 流式避 OOM)") +val_by_year = {} +n = 0 +for chunk in pd.read_sql( + "SELECT symbol,exchange,date,peTTM,psTTM,pcfNcfTTM,pbMRQ,turn,pctChg,isST " + "FROM daily_baostock_full_old", c, chunksize=200000): + n += 1 + chunk["isST"] = pd.to_numeric(chunk["isST"], errors="coerce").fillna(0).astype(int) + chunk["_y"] = pd.to_datetime(chunk["date"]).dt.year + for yr, sub in chunk.groupby("_y"): + sub = sub.drop(columns=["_y"]) + val_by_year.setdefault(int(yr), []).append(sub) + if n % 10 == 0: + log(f" parquet chunk#{n}") +for yr in sorted(val_by_year): + df_y = pd.concat(val_by_year[yr], ignore_index=True).sort_values(["symbol", "date"]) + df_y.to_parquet(VAL_DIR / f"{yr}.parquet", index=False) + log(f" {yr}: {len(df_y)} rows") + +log("4. verify dbbardata('d'):") +for sym, label in [("000005", "退市"), ("000023", "退市"), ("600519", "在市"), ("510300", "ETF")]: + r = c.execute( + "SELECT COUNT(*), MAX(datetime) FROM dbbardata " + "WHERE symbol=? AND interval='d'", (sym,)).fetchone() + log(f" {sym}({label}): {r}") +c.close() +log("MERGE DONE") diff --git a/scripts/data_platform/migrate_constituent.py b/scripts/data_platform/migrate_constituent.py new file mode 100644 index 0000000..87693b0 --- /dev/null +++ b/scripts/data_platform/migrate_constituent.py @@ -0,0 +1,122 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +"""migrate_constituent.py — 单元3: 合并成份股 -> constituent_unified_staging (全集型)。 + +设计 (spec §14, 治幸存者偏差选股池): +- baostock (bs_index_constituent 988 时点) -> 聚合成全集 (hs300/zz500/sz50 -> 000300/000905/000016), + in_current=最后时点成份, was_removed=历史入选过但已踢出 +- 深证/国证 _union.parquet (399001/399006/399005/399330) -> 直入 (akshare cni) +- 中证1000/2000 _snapshot.parquet (000852/932000) -> 当前 (akshare csindex) +- 新浪 _sina.parquet -> 丢弃 (baostock 300/500/50 已权威) +- code 统一 6 位无前缀 (sh.600000 / 600519.SH -> 600519) + +输出 constituent_unified_staging(index_code, code, code_name, source, in_current, was_removed)。 +验证 OK 后手动 rename: staging->constituent_unified, bs_index_constituent->_old。 +""" +import glob +import os +import re +import sqlite3 + +import pandas as pd + +DB = r"C:\sanguo_vnpy_v2\data\quant_trading.db" +HIST = r"C:\sanguo_vnpy_v2\data\index_const_hist" +STAGING = "constituent_unified_staging" +BS_MAP = {"hs300": "000300", "zz500": "000905", "sz50": "000016"} + + +def norm_code(code): + s = re.sub(r"^(sh|sz|SH|SZ)\.?", "", str(code)) + s = re.sub(r"\.(SH|SZ|sh|sz)$", "", s) + return s.zfill(6) if s.isdigit() and len(s) <= 6 else s + + +def main(): + c = sqlite3.connect(DB, timeout=60) + c.execute("PRAGMA busy_timeout = 60000") + + # 1. baostock -> 全集 + df_bs = pd.read_sql( + "SELECT updateDate, index_code, code, code_name FROM bs_index_constituent", c) + df_bs["index_code"] = df_bs["index_code"].map(BS_MAP) + df_bs["code"] = df_bs["code"].apply(norm_code) + last_sets = {} + for idx, grp in df_bs.groupby("index_code"): + last_d = grp["updateDate"].max() + last_sets[idx] = set(grp[grp["updateDate"] == last_d]["code"]) + pool = (df_bs.groupby(["index_code", "code"])["code_name"] + .first().reset_index()) + pool["in_current"] = pool.apply( + lambda r: r["code"] in last_sets.get(r["index_code"], set()), axis=1) + pool["was_removed"] = ~pool["in_current"] + pool["source"] = "baostock" + print(f"[baostock] pool rows={len(pool)} (300/500/50 全集)") + + # 2. 深证 union + deep = [] + for f in sorted(glob.glob(os.path.join(HIST, "*_union.parquet"))): + d = pd.read_parquet(f)[["index_code", "code", "code_name", + "in_current", "was_removed"]] + d["source"] = "akshare_cni" + deep.append(d) + df_deep = pd.concat(deep, ignore_index=True) if deep else pd.DataFrame( + columns=["index_code", "code", "code_name", "in_current", "was_removed", "source"]) + df_deep["code"] = df_deep["code"].apply(norm_code) + print(f"[深证 union] rows={len(df_deep)}") + + # 3. 中证 snapshot + snap = [] + for f in [os.path.join(HIST, "000852_snapshot.parquet"), + os.path.join(HIST, "932000_snapshot.parquet")]: + if os.path.exists(f): + d = pd.read_parquet(f)[["index_code", "code", "code_name"]] + d["in_current"] = True + d["was_removed"] = False + d["source"] = "akshare_csindex" + snap.append(d) + df_snap = pd.concat(snap, ignore_index=True) if snap else pd.DataFrame( + columns=["index_code", "code", "code_name", "in_current", "was_removed", "source"]) + df_snap["code"] = df_snap["code"].apply(norm_code) + print(f"[中证 snapshot] rows={len(df_snap)}") + + # 合并 + 去重 (同 index+code+source) + all_df = pd.concat([pool, df_deep, df_snap], ignore_index=True) + all_df = all_df.drop_duplicates(["index_code", "code", "source"]) + print(f"\n[TOTAL] constituent_unified: {len(all_df)} rows, " + f"{all_df['index_code'].nunique()} indices") + + print("\n各指数分布:") + print(all_df.groupby("index_code").agg( + n=("code", "count"), src=("source", "first"), + in_cur=("in_current", "sum"), removed=("was_removed", "sum"))) + + # 写 staging + c.execute(f"DROP TABLE IF EXISTS {STAGING}") + c.execute(f"""CREATE TABLE {STAGING} ( + index_code TEXT, code TEXT, code_name TEXT, source TEXT, + in_current INTEGER, was_removed INTEGER)""") + work = all_df[["index_code", "code", "code_name", "source", + "in_current", "was_removed"]].copy() + work["in_current"] = work["in_current"].astype(int) + work["was_removed"] = work["was_removed"].astype(int) + c.executemany(f"INSERT INTO {STAGING} VALUES (?,?,?,?,?,?)", + work.itertuples(index=False, name=None)) + c.commit() + n = c.execute(f"SELECT COUNT(*) FROM {STAGING}").fetchone()[0] + # 抽样验证 + print(f"\n[staging] {STAGING}: {n} rows") + print("sample 300:", c.execute( + "SELECT COUNT(*), SUM(in_current), SUM(was_removed) FROM " + f"{STAGING} WHERE index_code='000300'").fetchone()) + print("sample 399001:", c.execute( + "SELECT COUNT(*), SUM(in_current), SUM(was_removed) FROM " + f"{STAGING} WHERE index_code='399001'").fetchone()) + print("sample 000852:", c.execute( + "SELECT COUNT(*) FROM " f"{STAGING} WHERE index_code='000852'").fetchone()) + c.close() + print("\nMIGRATE STAGING DONE (未 rename, 验证 OK 后单独合并)") + + +if __name__ == "__main__": + main() diff --git a/scripts/data_platform/migrate_daily_baostock.py b/scripts/data_platform/migrate_daily_baostock.py new file mode 100644 index 0000000..48be76a --- /dev/null +++ b/scripts/data_platform/migrate_daily_baostock.py @@ -0,0 +1,99 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +"""migrate_daily_baostock.py — 单元4: daily_baostock_full 拆分 -> staging (本地DB, 无网络)。 + +- OHLCV -> dbbardata_staging_daily (interval='d', 含退市, 治回测幸存者偏差) + exchange SH/SZ -> SSE/SZSE; date -> 'YYYY-MM-DD 00:00:00'; amount->turnover +- pe/pb/turn/pctChg/isST -> data/valuation_baostock/.parquet (按年宽表) +- 全量读 + groupby year (避免 17 次全表扫; 内存峰值~5GB, VPS 16GB OK) +- staging 验证后单独合并 (INSERT OR REPLACE dbbardata + daily_baostock_full->_old) +""" +import sqlite3 +from pathlib import Path + +import pandas as pd + +DB = Path(r"C:\sanguo_vnpy_v2\data\quant_trading.db") +VAL_DIR = Path(r"C:\sanguo_vnpy_v2\data\valuation_baostock") +EXC_MAP = {"SH": "SSE", "SZ": "SZSE"} + + +def log(m): + print(m, flush=True) + + +c = sqlite3.connect(str(DB), timeout=120) +c.execute("PRAGMA busy_timeout = 120000") +c.execute("PRAGMA synchronous = NORMAL") + +cols = [r[1] for r in c.execute("PRAGMA table_info(daily_baostock_full)")] +log(f"daily_baostock_full cols({len(cols)}): {cols}") +total = c.execute("SELECT COUNT(*) FROM daily_baostock_full").fetchone()[0] +mn, mx = c.execute("SELECT MIN(date), MAX(date) FROM daily_baostock_full").fetchone() +log(f" rows={total} date {mn}~{mx}") + +# staging 表 +c.execute("DROP TABLE IF EXISTS dbbardata_staging_daily") +c.execute("""CREATE TABLE dbbardata_staging_daily ( + symbol TEXT, exchange TEXT, datetime TEXT, interval TEXT, + volume REAL, turnover REAL, open_interest REAL, + open_price REAL, high_price REAL, low_price REAL, close_price REAL)""") +c.execute("CREATE INDEX idx_staging_daily_sym ON dbbardata_staging_daily(symbol, datetime)") +c.commit() + +VAL_DIR.mkdir(parents=True, exist_ok=True) + +log("流式读 daily_baostock_full (chunksize=200000) -> staging + pe/pb 累积 (避 OOM) ...") +val_by_year = {} +db_total = val_total = 0 +n_chunk = 0 +for chunk in pd.read_sql("SELECT * FROM daily_baostock_full", c, chunksize=200000): + n_chunk += 1 + odb = pd.DataFrame({ + "symbol": chunk["symbol"].values, + "exchange": chunk["exchange"].map(EXC_MAP).values, + "datetime": (chunk["date"].astype(str) + " 00:00:00").values, + "interval": "d", + "volume": chunk["volume"].values, + "turnover": chunk["amount"].values, + "open_interest": 0.0, + "open_price": chunk["open"].values, + "high_price": chunk["high"].values, + "low_price": chunk["low"].values, + "close_price": chunk["close"].values, + }) + c.executemany( + "INSERT INTO dbbardata_staging_daily VALUES (?,?,?,?,?,?,?,?,?,?,?)", + odb.itertuples(index=False, name=None)) + c.commit() + db_total += len(odb) + chunk["_y"] = pd.to_datetime(chunk["date"]).dt.year + for yr, sub in chunk.groupby("_y"): + vdf = sub[["symbol", "exchange", "date", "peTTM", "psTTM", "pcfNcfTTM", + "pbMRQ", "turn", "pctChg", "isST"]] + val_by_year.setdefault(int(yr), []).append(vdf) + val_total += len(vdf) + if n_chunk % 10 == 0: + log(f" chunk#{n_chunk} db累计={db_total} val累计={val_total}") +log("写 valuation_baostock/.parquet ...") +for yr in sorted(val_by_year): + df_y = pd.concat(val_by_year[yr], ignore_index=True).sort_values(["symbol", "date"]) + df_y.to_parquet(VAL_DIR / f"{yr}.parquet", index=False) + log(f" {yr}: {len(df_y)} rows") + +# 退市/在市抽样验证 +log("\n[verify staging]") +for sym, label in [("000005", "退市"), ("000023", "退市"), + ("600811", "退市"), ("600519", "在市"), ("000001", "在市")]: + r = c.execute( + "SELECT COUNT(*), MIN(datetime), MAX(datetime) " + "FROM dbbardata_staging_daily WHERE symbol=?", (sym,)).fetchone() + log(f" {sym}({label}): {r}") +log(f" staging distinct symbol: " + f"{c.execute('SELECT COUNT(DISTINCT symbol) FROM dbbardata_staging_daily').fetchone()[0]}") + +c.close() +log(f"\nstaging dbbardata_staging_daily: {db_total} rows") +log(f"valuation_baostock: {val_total} rows, " + f"{len(list(VAL_DIR.glob('*.parquet')))} 年 parquet") +log("MIGRATE STAGING DONE (未合并主表, 验证 OK 后单独 merge)") diff --git a/scripts/data_platform/parse_csindex_announce.py b/scripts/data_platform/parse_csindex_announce.py new file mode 100644 index 0000000..31e322c --- /dev/null +++ b/scripts/data_platform/parse_csindex_announce.py @@ -0,0 +1,580 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +"""csindex 公告解析 - 重建中证1000/2000 历史成份股曾经入选集 (治幸存者偏差) + +数据源: + 1. 列表 API: POST https://www.csindex.com.cn/csindex-home/announcement/queryAnnouncementByVo + payload: {lang, classlist, indexlist, page:{desc,key,page,rows}, related_topics, typelist} + 2. 详情 API: GET https://www.csindex.com.cn/csindex-home/announcement/queryAnnouncementById?id={nid} + 返 {content (HTML), enclosureList:[{fileUrl, fileName}]} + 3. 附件: PDF (定期调整 多指数合并) 或 xlsx/xls (临时调整 单/多指数) + +实证覆盖 (2026-07-22 实地验证): + CSI 1000 (000852) 2014-09 发布: + - 2018-07-11 起 csindex 有调整公告 (28 份: 7 定期 + 21 临时) + - 2014-2018 期间的 June/Dec 定期调整不在 csindex (gap: ~7 round) + CSI 2000 (932000) 2023-08-10 发布: + - csindex 上无任何 CSI 2000 样本调整公告 (定期调整 PDF 中均无 CSI 2000 section) + - 仅 launch xlsx 有初始 2000 只样本 (2023-08-10) + - 历史/最新调整需查指数详情页 "拟生效样本" 或 wind/choice (不在本脚本范围) + +输出: + data/index_const_hist/000852_announce_union.parquet - CSI 1000 曾经入选集 + data/index_const_hist/932000_announce_union.parquet - CSI 2000 (initial only) + schema: updateDate / index_code / code / code_name / adjust_type / notice_id / source + +约束: + - 直连不走代理 (unset) + - 串行 + sleep(1.0~1.5s) + - UA header +""" +import argparse +import json +import logging +import os +import re +import socket +import sys +import time +import urllib.error +import urllib.parse +import urllib.request +from pathlib import Path +from typing import Dict, List, Optional, Tuple + +# ======================== 硬约束 ======================== +for _k in ("http_proxy", "https_proxy", "HTTP_PROXY", "HTTPS_PROXY", "all_proxy", "ALL_PROXY"): + os.environ.pop(_k, None) +socket.setdefaulttimeout(30) + +try: + sys.stdout.reconfigure(line_buffering=True) +except (AttributeError, ValueError): + pass + +import pandas as pd # noqa: E402 +import openpyxl # noqa: E402 +import pdfplumber # noqa: E402 + +# ======================== 配置 ======================== +BASE = "https://www.csindex.com.cn/csindex-home" +LIST_URL = f"{BASE}/announcement/queryAnnouncementByVo" +DETAIL_URL = f"{BASE}/announcement/queryAnnouncementById" +UA = "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0 Safari/537.36" + +DEFAULT_OUT_DIR = "/Users/chufeng/.openclaw/sanguo_projects/sanguo_vnpy_v2/data/index_const_hist" +OUT_DIR = Path(os.environ.get("INDEX_HIST_OUT_DIR", DEFAULT_OUT_DIR)) +CACHE_DIR = Path(os.environ.get("CSINDEX_CACHE", "/tmp/csindex_raw")) +DETAIL_CACHE_DIR = CACHE_DIR / "detail" + +# CSI 1000 相关调整公告 id (实证已发现, 脚本也支持自动发现) +CSI1000_REGULAR_IDS = [14796, 15044, 15267, 15471, 15690, 3006000, 3006137] +CSI1000_TEMP_IDS = [ + 12446, 12965, 13070, 13212, 13281, 13334, 13765, 14092, + 14842, 15019, 15232, 15342, 15357, 15390, 15575, 15613, 15648, + 1006042, 3006027, 3006041, 3006120, +] +# CSI 2000 launch xlsx (initial 2000 stocks) +CSI2000_LAUNCH_ID = 14883 + +# ======================== logging ======================== +logging.basicConfig( + level=logging.INFO, + format="%(asctime)s [%(levelname)s] %(message)s", + datefmt="%H:%M:%S", +) +log = logging.getLogger("csindex") + + +# ======================== HTTP ======================== +def _post_json(url: str, payload: dict, timeout: int = 30) -> dict: + data = json.dumps(payload).encode("utf-8") + req = urllib.request.Request( + url, data=data, method="POST", + headers={"Content-Type": "application/json", "User-Agent": UA}, + ) + with urllib.request.urlopen(req, timeout=timeout) as r: + return json.loads(r.read()) + + +def _get_json(url: str, timeout: int = 30) -> dict: + req = urllib.request.Request(url, headers={"User-Agent": UA}) + with urllib.request.urlopen(req, timeout=timeout) as r: + return json.loads(r.read()) + + +def _download(url: str, target: Path, timeout: int = 60) -> int: + """下载含中文名 URL (自动 quote, 防双重编码)""" + p = urllib.parse.urlsplit(url) + path_q = p.path if "%" in p.path else urllib.request.quote(p.path) + quoted = urllib.parse.urlunsplit((p.scheme, p.netloc, path_q, p.query, p.fragment)) + req = urllib.request.Request(quoted, headers={"User-Agent": UA}) + with urllib.request.urlopen(req, timeout=timeout) as r: + data = r.read() + target.parent.mkdir(parents=True, exist_ok=True) + with open(target, "wb") as f: + f.write(data) + return len(data) + + +# ======================== 1. 列表 ======================== +def fetch_all_notices(cache_path: Path, force: bool = False) -> List[dict]: + """拉全量公告列表 (cached)""" + if cache_path.exists() and not force: + log.info(f"使用缓存列表: {cache_path}") + with open(cache_path, encoding="utf-8") as f: + return json.load(f) + + log.info(f"分页拉取全量公告: {LIST_URL}") + all_items = [] + page = 1 + while True: + payload = { + "lang": "cn", "classlist": [], "indexlist": [], + "page": {"desc": "", "key": "", "page": page, "rows": 100}, + "related_topics": [], "typelist": [], + } + try: + d = _post_json(LIST_URL, payload) + except Exception as e: + log.error(f"page {page} err: {e}; retry once after 5s") + time.sleep(5) + d = _post_json(LIST_URL, payload) + if d.get("code") != "200": + log.error(f"page {page} API err: {d}") + break + items = d.get("data") or [] + all_items.extend(items) + total = d.get("total") or 0 + log.info(f" page {page}: +{len(items)} (cum={len(all_items)}/{total})") + if not items or len(all_items) >= total: + break + page += 1 + time.sleep(1.0) + + cache_path.parent.mkdir(parents=True, exist_ok=True) + with open(cache_path, "w", encoding="utf-8") as f: + json.dump(all_items, f, ensure_ascii=False) + log.info(f"缓存列表: {cache_path} (total={len(all_items)})") + return all_items + + +def filter_csi1000_notices(notices: List[dict]) -> List[dict]: + """筛 CSI 1000 调整公告 (theme=指数调样 + title 含 中证1000)""" + out = [] + seen_ids = set() + for x in notices: + title = x.get("title") or "" + if x.get("theme") != "指数调样": + continue + if "中证1000" not in title: + continue + if "调整" not in title: + continue + # 排除 "不实施" 通知 + if "不实施" in title: + continue + if x["id"] in seen_ids: + continue + seen_ids.add(x["id"]) + out.append(x) + out.sort(key=lambda x: x.get("publishDate", "")) + return out + + +# ======================== 2. 详情 + 下载 ======================== +def fetch_detail(nid: int, cache_dir: Path) -> dict: + """详情 (cached by id)""" + cache_file = cache_dir / f"detail_{nid}.json" + if cache_file.exists(): + with open(cache_file, encoding="utf-8") as f: + return json.load(f) + try: + d = _get_json(f"{DETAIL_URL}?id={nid}") + except Exception as e: + log.error(f"detail {nid} err: {e}") + return {} + cache_file.parent.mkdir(parents=True, exist_ok=True) + with open(cache_file, "w", encoding="utf-8") as f: + json.dump(d, f, ensure_ascii=False) + time.sleep(1.0) + return d + + +def extract_file_urls(detail_data: dict) -> List[Tuple[str, str]]: + """从 enclosureList + content 内嵌 href 抽 (fileName, fileUrl)""" + urls = [] + for e in (detail_data.get("enclosureList") or []): + fu = e.get("fileUrl") + fn = e.get("fileName") or "file" + if fu: + urls.append((fn, fu)) + content = detail_data.get("content") or "" + for m in re.finditer(r'href="([^"]+)"', content): + fu = m.group(1) + if re.search(r"\.(xlsx|pdf|xls|csv)(\?|$)", fu, re.I): + urls.append(("inline", fu)) + # 去重保序 + seen, out = set(), [] + for fn, fu in urls: + if fu not in seen: + seen.add(fu) + out.append((fn, fu)) + return out + + +def download_notice_files(nid: int, publish_date: str, urls: List[Tuple[str, str]], cache_dir: Path) -> List[Path]: + """下载该公告的所有附件,返本地路径列表""" + out = [] + for fn, fu in urls: + ext = fu.rsplit(".", 1)[-1].lower().split("?")[0] + local = cache_dir / f"{nid}_{publish_date}.{ext}" + if not local.exists(): + try: + sz = _download(fu, local) + log.info(f" dl [{sz}B] {fn} -> {local.name}") + except Exception as ex: + log.error(f" dl ERR {fu}: {ex}") + continue + else: + log.info(f" cache {local.name}") + out.append(local) + return out + + +# ======================== 3. 解析 ======================== +def _norm_code(c) -> Optional[str]: + """规整为 6 位字符串代码 (前补 0)""" + if c is None: + return None + s = str(c).strip() + # 排除空/NA + if not s or s.lower() in ("nan", "none", ""): + return None + # 纯数字 (可能 int 转 str 失去前导 0) + if s.isdigit(): + s = s.zfill(6) + if len(s) > 6: + return None # 异常长 + return s + # H开头 (港股代码, CSI 1000 不含, 跳过) + if s.startswith("H") or s.startswith("688") and not s.isdigit(): + return None + # 6 位字母数字 (如 SH/SZ 前缀) + m = re.search(r"(\d{6})", s) + return m.group(1) if m else None + + +def parse_xlsx_adjustments(path: Path, target_index_code: str = "000852") -> Tuple[List[dict], List[dict]]: + """解析 xlsx/xls -> (add_rows, remove_rows) + 每个 row = {code, code_name} + Sheet 名: 调入/调出 或 换入/换出 (反向) + """ + add_rows, remove_rows = [], [] + + if path.suffix.lower() == ".xls": + xl = pd.ExcelFile(path) + df_map = {sn: xl.parse(sn, dtype=str) for sn in xl.sheet_names} + else: + wb = openpyxl.load_workbook(path, data_only=True) + df_map = {} + for sn in wb.sheetnames: + ws = wb[sn] + rows = list(ws.iter_rows(values_only=True)) + if not rows: + continue + df_map[sn] = pd.DataFrame(rows[1:], columns=rows[0]) + + for sn, df in df_map.items(): + if df.empty or len(df.columns) < 4: + continue + # 判定方向 + sn_norm = sn.strip() + if sn_norm in ("调入", "换入", "新增"): + direction = "add" + elif sn_norm in ("调出", "换出", "删除"): + direction = "remove" + else: + continue + # 筛 target index + col0 = df.iloc[:, 0].astype(str).str.strip() + mask = col0 == target_index_code + sub = df[mask] + for _, row in sub.iterrows(): + code = _norm_code(row.iloc[2]) + name = str(row.iloc[3]).strip() if row.iloc[3] is not None else "" + if not code: + continue + r = {"code": code, "code_name": name} + if direction == "add": + add_rows.append(r) + else: + remove_rows.append(r) + return add_rows, remove_rows + + +def parse_pdf_adjustments(path: Path, target_section: str = "中证1000") -> Tuple[List[dict], List[dict]]: + """解析 PDF 的指定指数 section + target_section: '中证1000' or '中证2000' + 返 (add_rows, remove_rows) + """ + with pdfplumber.open(path) as pdf: + full_text = "\n".join((p.extract_text() or "") for p in pdf.pages) + + # 定位所有 section header + header_re = re.compile( + r"(沪深300|中证500|中证1000|中证2000|中证A\d+|上证\d+|科创50|北证\d+)\s*指数样本调整名单[:：]?" + ) + headers = list(header_re.finditer(full_text)) + if not headers: + return [], [] + + # 找 target section + target_text = None + for i, m in enumerate(headers): + if m.group(1) == target_section: + start = m.end() + end = headers[i + 1].start() if i + 1 < len(headers) else len(full_text) + target_text = full_text[start:end].strip() + break + if not target_text: + return [], [] + + # 解析 4 列 (调出代码 调出名称 调入代码 调入名称) + add_rows, remove_rows = [], [] + in_data = False + last_side = None # 处理 2 列 (只有一侧) 的情况 + for ln in target_text.split("\n"): + ln = ln.strip() + if not ln: + continue + # 表头行跳过 + if re.search(r"证券代码|股票代码|调出名单|调入名单|指数代码|换出|换入", ln): + in_data = True + if "调出名单" in ln and "调入名单" not in ln: + last_side = "remove" + elif "调入名单" in ln and "调出名单" not in ln: + last_side = "add" + else: + last_side = None + continue + if not in_data: + continue + # 4 列 + m4 = re.match(r"^(\d{6})\s+(\S+?)\s+(\d{6})\s+(\S+)", ln) + if m4: + rc, rn, ac, an = m4.group(1), m4.group(2), m4.group(3), m4.group(4) + remove_rows.append({"code": rc, "code_name": rn}) + add_rows.append({"code": ac, "code_name": an}) + last_side = None + continue + # 2 列 (单侧) + m2 = re.match(r"^(\d{6})\s+(\S+)", ln) + if m2: + r = {"code": m2.group(1), "code_name": m2.group(2)} + if last_side == "add": + add_rows.append(r) + elif last_side == "remove": + remove_rows.append(r) + return add_rows, remove_rows + + +# ======================== 4. 主流程 ======================== +def process_notice(nid: int, detail_cache: Path, file_cache: Path, + target_index_code: str, target_section: str) -> List[dict]: + """处理单个公告 -> 返回 records 列表 + record: {updateDate, index_code, code, code_name, adjust_type, notice_id, source} + """ + detail = fetch_detail(nid, detail_cache) + if not detail or not detail.get("data"): + log.warning(f" {nid}: detail empty") + return [] + data = detail["data"] + publish_date = (data.get("publishDate") or "").split("T")[0] + title = data.get("title") or "" + + urls = extract_file_urls(data) + files = download_notice_files(nid, publish_date, urls, file_cache) + if not files: + log.warning(f" {nid} {publish_date}: no files; title={title}") + return [] + + records = [] + for f in files: + ext = f.suffix.lower() + try: + if ext == ".pdf": + add, rem = parse_pdf_adjustments(f, target_section=target_section) + elif ext in (".xlsx", ".xls"): + add, rem = parse_xlsx_adjustments(f, target_index_code=target_index_code) + else: + continue + except Exception as e: + log.error(f" parse {f.name} ERR: {e}") + continue + for r in add: + records.append({ + "updateDate": publish_date, "index_code": target_index_code, + "code": r["code"], "code_name": r["code_name"], + "adjust_type": "add", "notice_id": nid, "source": f.name, + }) + for r in rem: + records.append({ + "updateDate": publish_date, "index_code": target_index_code, + "code": r["code"], "code_name": r["code_name"], + "adjust_type": "remove", "notice_id": nid, "source": f.name, + }) + log.info(f" {nid} {publish_date}: parsed add={sum(1 for r in records if r['adjust_type']=='add')}, " + f"remove={sum(1 for r in records if r['adjust_type']=='remove')}, title={title}") + return records + + +def main(): + parser = argparse.ArgumentParser() + parser.add_argument("--out-dir", default=str(OUT_DIR)) + parser.add_argument("--cache-dir", default=str(CACHE_DIR)) + parser.add_argument("--refresh-list", action="store_true", help="强制重新拉全量公告列表") + parser.add_argument("--only", choices=["1000", "2000", "both"], default="both") + args = parser.parse_args() + + out_dir = Path(args.out_dir) + out_dir.mkdir(parents=True, exist_ok=True) + cache_dir = Path(args.cache_dir) + detail_cache = cache_dir / "detail" + file_cache = cache_dir / "files" + + # 1. 拉/缓存全量列表 + notices = fetch_all_notices(cache_dir / "all_notices.json", force=args.refresh_list) + + # ========== CSI 1000 ========== + if args.only in ("1000", "both"): + log.info("\n========== CSI 1000 (000852) ==========") + # 过滤 + 附加已知 id + filtered = filter_csi1000_notices(notices) + log.info(f"filtered CSI 1000 adjustment notices: {len(filtered)}") + + all_ids = sorted({x["id"] for x in filtered} | set(CSI1000_REGULAR_IDS) | set(CSI1000_TEMP_IDS)) + log.info(f"total CSI 1000 notice ids to process: {len(all_ids)}") + + records_1000 = [] + for nid in all_ids: + try: + recs = process_notice(nid, detail_cache, file_cache, "000852", "中证1000") + records_1000.extend(recs) + except Exception as e: + log.error(f" {nid} FAILED: {e}") + + # 加 initial 集合 (launch 2014-09 -> 2014-09-18 id=2998, 无初始样本 xlsx, 跳过) + # 加 current (akshare 现快照) + try: + import akshare as ak + log.info("拉 akshare 中证1000 当前快照...") + df = ak.index_stock_cons_csindex(symbol="000852") + for _, row in df.iterrows(): + code = str(row.iloc[0]).zfill(6) if str(row.iloc[0]).isdigit() else str(row.iloc[0]) + if not code.isdigit() or len(code) != 6: + continue + records_1000.append({ + "updateDate": "current", "index_code": "000852", + "code": code, "code_name": str(row.iloc[1]) if df.shape[1] > 1 else "", + "adjust_type": "current", "notice_id": 0, "source": "akshare.index_stock_cons_csindex", + }) + log.info(f" akshare current: {sum(1 for r in records_1000 if r['adjust_type']=='current')} stocks") + except Exception as e: + log.warning(f"akshare 当前快照拉取失败 (非致命): {e}") + + df_1000 = pd.DataFrame(records_1000, columns=[ + "updateDate", "index_code", "code", "code_name", "adjust_type", "notice_id", "source"]) + out_1000 = out_dir / "000852_announce_union.parquet" + df_1000.to_parquet(out_1000, index=False) + distinct_1000 = df_1000["code"].nunique() + n_add = (df_1000["adjust_type"] == "add").sum() + n_rem = (df_1000["adjust_type"] == "remove").sum() + n_cur = (df_1000["adjust_type"] == "current").sum() + log.info(f"\nCSI 1000 result: {out_1000}") + log.info(f" total records: {len(df_1000)} (add={n_add}, remove={n_rem}, current={n_cur})") + log.info(f" distinct codes (ever-included): {distinct_1000}") + log.info(f" date range: {df_1000['updateDate'].min()} ~ {df_1000['updateDate'].max()}") + log.info(f" distinct notice_ids: {df_1000['notice_id'].nunique()}") + + # ========== CSI 2000 ========== + if args.only in ("2000", "both"): + log.info("\n========== CSI 2000 (932000) ==========") + log.warning("GAP: csindex 公告系统未发布任何 CSI 2000 样本调整公告") + log.warning(" 仅可从 2023-08-10 launch xlsx 获取初始 2000 只样本") + records_2000 = [] + + # launch xlsx (initial) + try: + detail = fetch_detail(CSI2000_LAUNCH_ID, detail_cache) + if detail.get("data"): + data = detail["data"] + publish_date = (data.get("publishDate") or "2023-08-10").split("T")[0] + urls = extract_file_urls(data) + files = download_notice_files(CSI2000_LAUNCH_ID, publish_date, urls, file_cache) + for f in files: + if f.suffix.lower() != ".xlsx": + continue + wb = openpyxl.load_workbook(f, data_only=True) + for sn in wb.sheetnames: + ws = wb[sn] + rows = list(ws.iter_rows(values_only=True)) + if not rows: + continue + header = rows[0] + # 找 code 列 (通常第 1 列 或 命名"证券代码"/"成分券代码") + for row in rows[1:]: + if not row: + continue + code = _norm_code(row[0] if len(row) > 0 else None) + if not code: + continue + name = str(row[1]).strip() if len(row) > 1 and row[1] else "" + records_2000.append({ + "updateDate": publish_date, "index_code": "932000", + "code": code, "code_name": name, + "adjust_type": "initial", "notice_id": CSI2000_LAUNCH_ID, "source": f.name, + }) + break # 只用第一个 xlsx + except Exception as e: + log.error(f"CSI 2000 launch xlsx ERR: {e}") + + # current snapshot via akshare + try: + import akshare as ak + log.info("拉 akshare 中证2000 当前快照...") + df = ak.index_stock_cons_csindex(symbol="932000") + for _, row in df.iterrows(): + code = str(row.iloc[0]).zfill(6) if str(row.iloc[0]).isdigit() else str(row.iloc[0]) + if not code.isdigit() or len(code) != 6: + continue + records_2000.append({ + "updateDate": "current", "index_code": "932000", + "code": code, "code_name": str(row.iloc[1]) if df.shape[1] > 1 else "", + "adjust_type": "current", "notice_id": 0, "source": "akshare.index_stock_cons_csindex", + }) + log.info(f" akshare current: {sum(1 for r in records_2000 if r['adjust_type']=='current')} stocks") + except Exception as e: + log.warning(f"akshare CSI 2000 当前快照失败: {e}") + + df_2000 = pd.DataFrame(records_2000, columns=[ + "updateDate", "index_code", "code", "code_name", "adjust_type", "notice_id", "source"]) + out_2000 = out_dir / "932000_announce_union.parquet" + df_2000.to_parquet(out_2000, index=False) + distinct_2000 = df_2000["code"].nunique() + n_init = (df_2000["adjust_type"] == "initial").sum() + n_cur = (df_2000["adjust_type"] == "current").sum() + log.info(f"\nCSI 2000 result: {out_2000}") + log.info(f" total records: {len(df_2000)} (initial={n_init}, current={n_cur})") + log.info(f" distinct codes: {distinct_2000}") + log.info(f" GAP: 无调整公告,初始集 ∪ 当前集 (中间调整无记录)") + + # 总结 + log.info("\n========== DONE ==========") + log.info(f"OUT_DIR: {out_dir.resolve()}") + for f in sorted(out_dir.glob("*_announce_union.parquet")): + log.info(f" {f.name}: {f.stat().st_size} bytes") + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/scripts/data_platform/probe_constituent.py b/scripts/data_platform/probe_constituent.py new file mode 100644 index 0000000..f7ad2d9 --- /dev/null +++ b/scripts/data_platform/probe_constituent.py @@ -0,0 +1,37 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +"""probe_constituent.py — 单元3 前置: 探查成份股两源 schema (写 migrate 前必须看清)。 + +1. bs_index_constituent (DB, baostock 300/500/50 历史): cols/rows/distinct index_code/抽样 +2. data/index_const_hist/*.parquet (akshare cni 深证 union): 文件列表 + 每个 shape/cols/抽样 +""" +import glob +import os +import sqlite3 + +import pandas as pd + +DB = r"C:\sanguo_vnpy_v2\data\quant_trading.db" +HIST = r"C:\sanguo_vnpy_v2\data\index_const_hist" + +c = sqlite3.connect(DB, timeout=60) +c.execute("PRAGMA busy_timeout = 60000") +cols = [r[1] for r in c.execute("PRAGMA table_info(bs_index_constituent)")] +print("[bs_index_constituent] cols:", cols) +print(" rows:", c.execute("SELECT COUNT(*) FROM bs_index_constituent").fetchone()[0]) +idx = [r[0] for r in c.execute( + "SELECT DISTINCT index_code FROM bs_index_constituent ORDER BY index_code")] +print(" distinct index_code:", idx) +print(" distinct updateDate count:", c.execute( + "SELECT COUNT(DISTINCT updateDate) FROM bs_index_constituent").fetchone()[0]) +print(" sample rows:", c.execute( + "SELECT * FROM bs_index_constituent LIMIT 3").fetchall()) +c.close() + +print("\n[index_const_hist parquets]") +files = sorted(glob.glob(os.path.join(HIST, "*.parquet"))) +print("files:", [os.path.basename(f) for f in files]) +for f in files: + df = pd.read_parquet(f) + print(f" {os.path.basename(f)}: shape={df.shape} cols={list(df.columns)}") + print(f" head:\n{df.head(2).to_string()}") diff --git a/scripts/data_platform/probe_dbbardata_unique.py b/scripts/data_platform/probe_dbbardata_unique.py new file mode 100644 index 0000000..dd2aa72 --- /dev/null +++ b/scripts/data_platform/probe_dbbardata_unique.py @@ -0,0 +1,32 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +"""probe_dbbardata_unique.py — 单元4 merge 前置: 查 dbbardata UNIQUE 约束 + verify staging。 + +INSERT OR REPLACE 依赖 dbbardata 有 UNIQUE(symbol,exchange,datetime,interval) 才能去重, +否则插重复。vnpy DbBarData 通常有 UniqueConstraint, 此处确认。 +""" +import sqlite3 + +DB = r"C:\sanguo_vnpy_v2\data\quant_trading.db" +c = sqlite3.connect(DB, timeout=120) +c.execute("PRAGMA busy_timeout = 120000") + +print("dbbardata sql:", c.execute( + "SELECT sql FROM sqlite_master WHERE name='dbbardata'").fetchone()) +print("\ndbbardata indexes:") +for idx in c.execute("PRAGMA index_list('dbbardata')").fetchall(): + cols = c.execute(f"PRAGMA index_info('{idx[1]}')").fetchall() + print(f" {idx} cols={cols}") + +print("\n--- staging verify ---") +print("staging rows:", c.execute( + "SELECT COUNT(*) FROM dbbardata_staging_daily").fetchone()[0]) +print("staging distinct symbol:", c.execute( + "SELECT COUNT(DISTINCT symbol) FROM dbbardata_staging_daily").fetchone()[0]) +for sym, label in [("000005", "退市"), ("000023", "退市"), + ("600519", "在市"), ("510300", "ETF")]: + r = c.execute( + "SELECT COUNT(*), MIN(datetime), MAX(datetime) " + "FROM dbbardata_staging_daily WHERE symbol=?", (sym,)).fetchone() + print(f" {sym}({label}): {r}") +c.close() diff --git a/scripts/data_platform/register_schtasks.ps1 b/scripts/data_platform/register_schtasks.ps1 new file mode 100644 index 0000000..c9dd92f --- /dev/null +++ b/scripts/data_platform/register_schtasks.ps1 @@ -0,0 +1,10 @@ +# register_schtasks.ps1 — 注册方案A 核心 2 个日增量 schtask (bs-eod 18:05, xt-eod 18:40) +# akshare 三表(季度低频) / 成份股(月度) 留 disabled, 后续单独处理 +schtasks /create /tn sanguo-bs-eod /tr "powershell -ExecutionPolicy Bypass -File C:\sanguo_vnpy_v2\scripts\data_platform\bs_eod_wrapper.ps1" /sc daily /st 18:05 /ru SYSTEM /rl HIGHEST /f +schtasks /create /tn sanguo-xt-eod /tr "powershell -ExecutionPolicy Bypass -File C:\sanguo_vnpy_v2\scripts\data_platform\xt_eod_wrapper.ps1" /sc daily /st 18:40 /ru SYSTEM /rl HIGHEST /f +Write-Output "=== VERIFY ===" +foreach ($t in @('sanguo-bs-eod', 'sanguo-xt-eod', 'sanguo-daily-update', 'sanguo-bs-akshare', 'sanguo-bs-daily-increment', 'sanguo-index-hist')) { + $q = schtasks /query /tn $t /fo list 2>&1 | Out-String + $status = ($q -split "`n" | Where-Object { $_ -match 'Status' } | Select-Object -First 1).Trim() + Write-Output (" {0,-28} {1}" -f $t, $status) +} diff --git a/scripts/data_platform/stop_all_data_schtasks.ps1 b/scripts/data_platform/stop_all_data_schtasks.ps1 new file mode 100644 index 0000000..13378b1 --- /dev/null +++ b/scripts/data_platform/stop_all_data_schtasks.ps1 @@ -0,0 +1,29 @@ +# stop_all_data_schtasks.ps1 — schtasks.exe 版(可靠):停+禁用+杀残留数据脚本 +$tasks = @('sanguo-daily-update', 'sanguo-bs-akshare', 'sanguo-bs-daily-increment', 'sanguo-index-hist') +Write-Output "=== END (停正在跑) + DISABLE ===" +foreach ($t in $tasks) { + $e = (schtasks /end /tn $t 2>&1) -join ' ' + $d = (schtasks /change /tn $t /disable 2>&1) -join ' ' + Write-Output (" {0,-28} end+disable done" -f $t) +} +Start-Sleep 3 + +Write-Output "" +Write-Output "=== 杀残留数据脚本 python (精确匹配命令行, 不碰常驻/回测) ===" +$dataProcs = Get-CimInstance Win32_Process -Filter "Name='python.exe'" | + Where-Object { $_.CommandLine -match 'akshare_static|validate_import|daily_update_xtdata|audit_data_layout|import_vnpy|minute_incr|_run_daily|diag_daily' } +if ($dataProcs) { + foreach ($p in $dataProcs) { + $cmd = if ($p.CommandLine) { $p.CommandLine.Substring(0, [Math]::Min(60, $p.CommandLine.Length)) } else { '' } + Stop-Process -Id $p.ProcessId -Force -ErrorAction SilentlyContinue + Write-Output (" killed pid={0} {1}" -f $p.ProcessId, $cmd) + } +} else { Write-Output " 无残留数据脚本 python" } + +Write-Output "" +Write-Output "=== VERIFY schtask 状态 (应=Disabled) ===" +foreach ($t in $tasks) { + $q = schtasks /query /tn $t /fo list 2>&1 | Out-String + $status = ($q -split "`n" | Where-Object { $_ -match 'Status' } | Select-Object -First 1).Trim() + Write-Output (" {0,-28} {1}" -f $t, $status) +} diff --git a/scripts/data_platform/verify_backup.py b/scripts/data_platform/verify_backup.py new file mode 100644 index 0000000..7670b9e --- /dev/null +++ b/scripts/data_platform/verify_backup.py @@ -0,0 +1,21 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +"""verify_backup.py — 快速验证 .bak 可读(连库读表清单+小表 COUNT, 非全表 integrity)。 + +sqlite3 .backup 是原子一致机制, .bak 28.4GB 完整即一致。本脚本确认 .bak 能连+表在=回滚可用。 +""" +import os +import sqlite3 + +BAK = r"C:\sanguo_vnpy_v2\data\quant_trading.db.bak_20260722" +print("bak size=%.2f GB" % (os.path.getsize(BAK) / 1024 ** 3)) +c = sqlite3.connect(BAK, timeout=60) +c.execute("PRAGMA busy_timeout = 60000") +tabs = [r[0] for r in c.execute( + "SELECT name FROM sqlite_master WHERE type='table' ORDER BY name")] +print("tables:", tabs) +print("backtest_stats:", c.execute("SELECT COUNT(*) FROM backtest_stats").fetchone()[0]) +print("bs_adjust_factor:", c.execute("SELECT COUNT(*) FROM bs_adjust_factor").fetchone()[0]) +print("bs_index_constituent:", c.execute("SELECT COUNT(*) FROM bs_index_constituent").fetchone()[0]) +c.close() +print("BAK READABLE OK") diff --git a/scripts/data_platform/verify_config.py b/scripts/data_platform/verify_config.py new file mode 100644 index 0000000..bf4589a --- /dev/null +++ b/scripts/data_platform/verify_config.py @@ -0,0 +1,24 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +"""verify_config.py — 单元2 验证: VPS datareader 能否读到本地全量日线。 + +打印 SANGUO_DATA_ROOT env + load_config 结果 + read_parquet_daily 600519。 +据此判断: env 覆盖 vs yaml 直改 哪个生效, daily_dir 是否指 qfq(全量)。 +""" +import os +import sys + +sys.path.insert(0, r"C:\sanguo_vnpy_v2") +print("SANGUO_DATA_ROOT:", repr(os.environ.get("SANGUO_DATA_ROOT"))) + +from sanguo_data.config import load_config +from sanguo_data.datareader import read_parquet_daily + +cfg = load_config(r"C:\sanguo_vnpy_v2\config\data_platform.yaml") +print("daily_dir:", cfg.data_paths.get("daily_dir")) +print("vnpy_db:", cfg.data_paths.get("vnpy_db")) + +bars = read_parquet_daily("600519", "2026-07-01", "2026-07-22", cfg) +print("600519 bars:", len(bars), + "last:", bars[-1].datetime.date() if bars else "NONE") +print("VERIFY OK" if bars else "VERIFY WARN: 0 bars") diff --git a/scripts/data_platform/verify_e2e.py b/scripts/data_platform/verify_e2e.py new file mode 100644 index 0000000..13ccb60 --- /dev/null +++ b/scripts/data_platform/verify_e2e.py @@ -0,0 +1,27 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +"""verify_e2e.py — 收尾 E2E: 回测入口 read_db_daily 读 dbbardata 验证。 + +方案A 数据层落地后, 确认回测能读到: +- 在市股(600519) 日线到 2026-07-22 +- 退市股(000005) 日线(治幸存者偏差, 迁移前 dbbardata 没有) +- ETF(510300) 日线(xtata 源保留) +""" +import sys +sys.path.insert(0, r"C:\sanguo_vnpy_v2") +from sanguo_data.config import load_config +from sanguo_data.datareader import read_db_daily + +cfg = load_config(r"C:\sanguo_vnpy_v2\config\data_platform.yaml") +print("vnpy_db:", cfg.data_paths["vnpy_db"]) + +for sym, label, start, end in [ + ("600519", "在市", "2026-07-01", "2026-07-22"), + ("000005", "退市(治偏差)", "2024-01-01", "2024-04-30"), + ("510300", "ETF", "2026-07-01", "2026-07-22"), +]: + bars = read_db_daily(sym, start, end, cfg) + last = bars[-1].datetime.date() if bars else "NONE" + print(f" {sym}({label}): bars={len(bars)} last={last}") + +print("E2E DBBARDATA OK") diff --git a/scripts/data_platform/xt_eod.py b/scripts/data_platform/xt_eod.py new file mode 100644 index 0000000..c781062 --- /dev/null +++ b/scripts/data_platform/xt_eod.py @@ -0,0 +1,130 @@ +#!/usr/bin/env python3 +# -*- coding: utf-8 -*- +"""xt_eod.py — sanguo-xt-eod (方案A schtask 18:40): xtata ETF/基金 EOD 增量 -> dbbardata('d')。 + +baostock 只取 type=1 股票, 不覆盖 ETF/基金 -> xtata 独占 (spec §14)。 +个股日线由 bs_eod (baostock) 灌, 此处不碰个股, 避免 dbbardata 两源冲突。 + +- universe = 沪深ETF ∪ 沪深基金 +- download_history_data2 批量 paced -> 本地缓存 +- get_market_data_ex raw(dividend_type=none) -> dbbardata('d') INSERT OR REPLACE +- volume 手->股 (×100, 与 daily_update_xtdata 同口径) +- 无限流, 单进程 download 不并发 + +用法: python xt_eod.py [--limit N] [--dry-run] +""" +import argparse +import datetime as dt +import sqlite3 +import time + +from xtquant import xtdata as xd +import pandas as pd + +DB = r"C:\sanguo_vnpy_v2\data\quant_trading.db" +LOOKBACK = int(__import__("os").environ.get("LOOKBACK_DAYS", "30")) +T0 = time.time() + + +def log(m): + print(f"[XT-EOD {time.time()-T0:.0f}s] {m}", flush=True) + + +def prefix_of(sym): + return "sh" if sym[:2] in ("51", "56", "58", "50") else ("sh" if sym[:2] == "60" else "sz") + + +def exc_of(sym): + return "SSE" if sym[:2] in ("51", "56", "58", "50", "60", "68") else "SZSE" + + +def main(): + ap = argparse.ArgumentParser() + ap.add_argument("--limit", type=int, default=0) + ap.add_argument("--dry-run", action="store_true") + args = ap.parse_args() + + end = dt.datetime.now().strftime("%Y%m%d") + start = (dt.datetime.now() - dt.timedelta(days=LOOKBACK)).strftime("%Y%m%d") + log(f"start window={start}~{end}") + + u = list(set( + (xd.get_stock_list_in_sector("沪深ETF") or []) + + (xd.get_stock_list_in_sector("沪深基金") or []) + )) + if not u: + log("FATAL empty universe (miniQMT 未连?)") + return 2 + if args.limit: + u = u[:args.limit] + log(f"universe ETF+基金: {len(u)}") + + # download paced + BATCH = 200 + for i in range(0, len(u), BATCH): + try: + xd.download_history_data2(u[i:i+BATCH], "1d", start, end, lambda d, p: None) + except Exception as e: + log(f"dl batch@{i} err: {e}") + time.sleep(1.0) + log("download done") + + conn = sqlite3.connect(DB, timeout=60) + conn.execute("PRAGMA busy_timeout = 60000") + conn.execute("PRAGMA journal_mode = WAL") + + ok = fail = empty = rows = 0 + conn.execute("BEGIN") + try: + for i, code in enumerate(u): + sym = code.split(".")[0] + try: + r = xd.get_market_data_ex([], [code], period="1d", start_time=start, + end_time=end, dividend_type="none") + df = r.get(code) if r else None + if df is None or not len(df): + empty += 1 + continue + db = pd.DataFrame({ + "symbol": sym, + "exchange": exc_of(sym), + "datetime": [str(idx)[:4]+"-"+str(idx)[4:6]+"-"+str(idx)[6:8]+" 00:00:00" + for idx in df.index], + "interval": "d", + "volume": (df["volume"].astype(float).values * 100), + "turnover": df["amount"].astype(float).values, + "open_interest": 0.0, + "open_price": df["open"].astype(float).values, + "high_price": df["high"].astype(float).values, + "low_price": df["low"].astype(float).values, + "close_price": df["close"].astype(float).values, + }) + if not args.dry_run: + conn.executemany( + "INSERT OR REPLACE INTO dbbardata " + "(symbol,exchange,datetime,interval,volume,turnover,open_interest," + "open_price,high_price,low_price,close_price) VALUES (?,?,?,?,?,?,?,?,?,?,?)", + db.itertuples(index=False, name=None)) + rows += len(db) + ok += 1 + except Exception as e: + fail += 1 + if fail <= 5: + log(f"{code} err: {e}") + if (i+1) % 200 == 0: + log(f"进度 {i+1}/{len(u)} ok={ok} empty={empty} fail={fail} rows={rows}") + conn.execute("COMMIT") + except Exception as e: + conn.execute("ROLLBACK") + log(f"FATAL rollback: {e}") + conn.close() + return 1 + conn.close() + + log(f"DONE ok={ok} empty={empty} fail={fail} rows={rows}" + f"{' [DRY-RUN]' if args.dry_run else ''}") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/scripts/data_platform/xt_eod_wrapper.ps1 b/scripts/data_platform/xt_eod_wrapper.ps1 new file mode 100644 index 0000000..89edbb6 --- /dev/null +++ b/scripts/data_platform/xt_eod_wrapper.ps1 @@ -0,0 +1,9 @@ +# xt_eod_wrapper.ps1 — sanguo-xt-eod schtask wrapper (18:40 xtata ETF/基金 EOD 增量) +# 方案A: ETF/基金(baostock 盲区) -> dbbardata('d') +Set-Location C:\sanguo_vnpy_v2 +$ts = Get-Date -Format 'yyyyMMdd_HHmmss' +$logDir = 'C:\sanguo_vnpy_v2\data\migration_logs' +if (-not (Test-Path $logDir)) { New-Item -ItemType Directory -Path $logDir -Force | Out-Null } +$log = Join-Path $logDir "xt_eod_$ts.txt" +C:\Python310\python.exe -X utf8 C:\sanguo_vnpy_v2\scripts\data_platform\xt_eod.py *>> $log +exit $LASTEXITCODE