diff --git a/docs/deployment/data-download.md b/docs/deployment/data-download.md new file mode 100644 index 0000000..6e336a5 --- /dev/null +++ b/docs/deployment/data-download.md @@ -0,0 +1,79 @@ +# A 股数据下载（v2 维护） + +> 维护：Main Agent · 2026-07-07 +> v1 数据下载脚本（`~/.openclaw/sanguo_projects/sanguo_vnpy/data_platform/`）移植到 v2， +> 改 SSH 模式（免 SMB 挂载/密码/macFUSE），Mac launchd 定时（替代 crontab）。 + +## 架构 + +``` +Mac Mini（常开） + launchd 每日 15:30 + → run_daily_update.sh + 1. rsync 拉 NAS 现有 → Mac 本地（/tmp/stock_dl，增量；首次慢后续快） + 2. 跑 v1 daily_all_update.py（多源 fallback + 增量 + 失败率熔断） + 3. rsync 推 本地 → NAS（/volume1/stock） + ↕ SSH key 免密（sanguo-nas，不依赖 SMB 挂载/密码） +NAS /volume1/stock（日线/15min parquet + vnpy DB） +``` + +不挂载、不要密码、不装 macFUSE，用 `sanguo-nas` SSH key（`~/.ssh/config`）。 + +## 脚本（`v2/scripts/data_platform/`） + +从 v1 `data_platform/` 复制（11 个脚本）。关键改动： + +- **`daily_all_update.py`** — v1 全市场增量（日线 + 15min，多源 fallback：东财 4s + BaoStock + 腾讯，失败率熔断 >80% 终止） + - `STOCK_MOUNT` env：路径根（默认 `~/stock_mount` SMB 挂载点；SSH 模式包装脚本设 `/tmp/stock_dl`） + - `STOCK_LIMIT` env：限制股票数（前 N，**验证用**，默认 0=全市场） +- **`run_daily_update.sh`** — 包装（rsync 拉 + v1 + 推），SSH 模式入口 + - env：`STOCK_LIMIT=N` / `SKIP_PULL=1`（验证跳过拉） +- 其他 v1 脚本：`backfill_15min_baostock.py`（15min 回补）/ `realtime.py`（实时三源）/ `validator.py`（校验）等 + +## 定时（launchd，替代 crontab） + +`~/Library/LaunchAgents/com.sanguo.data-update.plist` — 每日 15:30 跑 `run_daily_update.sh`。 + +```bash +launchctl load ~/Library/LaunchAgents/com.sanguo.data-update.plist +launchctl unload ~/Library/LaunchAgents/com.sanguo.data-update.plist +launchctl list | grep sanguo.data-update +``` + +**为何不用 crontab**：macOS crontab 写（`crontab file`）需 Full Disk Access，Claude/终端无 FDA 时写操作卡死（读 OK）。launchd 用户级 plist（`~/Library/LaunchAgents/`）不需 FDA，更稳。Mac 原生推荐方式。 + +## 验证（2026-07-07） + +``` +STOCK_LIMIT=2 STOCK_MOUNT=/tmp/stock_dl_test python3 daily_all_update.py --skip-15min +→ updated: 2, records: 24（06-19~07-07），000001 拉到 2026-07-07（今天），21.8s ✅ +``` + +增量逻辑：v1 读本地现有 parquet 最后日期（`get_daily_last_date`）→ 拉 last+1 ~ 今天。**本地必须有现有 parquet**（rsync 拉或之前数据），否则 skip。 + +## v1 crontab 取消 + +v1 crontab `30 15 * * 1-5 .../sanguo_vnpy/data_platform/daily_update.sh` 取消——用 v2 launchd 替代。 + +> macOS crontab 写卡（FDA），`crontab -e` 手动去那行，或给终端 Full Disk Access。 + +v1 crontab 若残留无害（v1 脚本跑时 NAS 未挂载会 `ERROR: NAS未挂载，跳过更新`）。 + +## 全市场补全（首次） + +`run_daily_update.sh` 首次跑：rsync 拉全市场现有（5264 parquet，几分钟）+ v1 增量（全市场 × 东财 4s 限频 ≈ 数小时，夜间）+ 推。 + +手动触发补全： +```bash +cd v2/scripts/data_platform +./run_daily_update.sh # 全量（日线+15min），夜间跑 +./run_daily_update.sh --skip-15min # 只日线 +``` + +## raw 双数据源（task #79，后续） + +当前 parquet 是前复权/后复权（成交价显示失真，如浦发 110 元 vs 现价 ~10）。raw 不复权： +- BaoStock `adjustflag="3"`（v1 `backfill_15min_baostock.py:131` 已用 raw 下 15min） +- akshare `adjust=""` + +task #79 接入 raw 双源（撮合/涨跌停/均价用 raw，策略信号用 qfq）。 diff --git a/scripts/data_platform/backfill_15min_baostock.py b/scripts/data_platform/backfill_15min_baostock.py new file mode 100644 index 0000000..8bd88e7 --- /dev/null +++ b/scripts/data_platform/backfill_15min_baostock.py @@ -0,0 +1,317 @@ +#!/usr/bin/env python3 +""" +BaoStock 15分钟线历史回补脚本 (v1.1) + +功能：用BaoStock免费数据源全量重建全市场15分钟K线历史数据 +策略： + - 全量重建：BaoStock获取完整历史，完全替换旧parquet + - 旧文件备份到 backup/ 目录 + - 增量模式：已有BaoStock重建过的文件自动跳过（检查标记） + +BaoStock特点： + - 无反爬限制，0.35s/只 + - 不复权数据（adjustflag=3） + - 分钟线从1999年起，我们取2024-01-01起 + +用法： + python3 backfill_15min_baostock.py # 全量回补 + python3 backfill_15min_baostock.py --start 20200101 # 指定起始日期 + python3 backfill_15min_baostock.py --limit 10 # 只处理前10只（测试） + python3 backfill_15min_baostock.py --force # 强制重建（覆盖已有BaoStock数据） + +变更记录： + v1.0 (2026-05-05) 赵云创建 + v1.1 (2026-05-05) 修复：旧新浪数据含后复权值，改为全量重建模式 +""" + +import argparse +import json +import logging +import os +import shutil +import sys +import time +from datetime import datetime +from pathlib import Path +from typing import List, Optional, Tuple + +import baostock as bs +import pandas as pd + +# ======================== 配置 ======================== + +NAS_ROOT = Path("/Volumes/stock") +MINUTE_15_DIR = NAS_ROOT / "minute_kline" / "15min" +BACKUP_DIR = MINUTE_15_DIR / "backup_sina" # 旧新浪数据备份 +LOG_DIR = NAS_ROOT / "logs" / "daily_update" +PROGRESS_DIR = LOG_DIR / "progress" +ALL_STOCKS_FILE = NAS_ROOT / "A股数据" / "stock_info" / "stock_basic_info_raw_20260326_113530.csv" + +# BaoStock配置 +BS_START_DATE = "2024-01-01" +BS_INTERVAL = 0.4 # 请求间隔秒 +BS_MAX_RETRIES = 2 +PROGRESS_SAVE_EVERY = 500 + +# ======================== 日志 ======================== + +def setup_logging(): + LOG_DIR.mkdir(parents=True, exist_ok=True) + ts = datetime.now().strftime("%Y%m%d_%H%M%S") + # 日志写本地避免SMB阻塞 + log_file = Path(f"/tmp/backfill_15min_{ts}.log") + + logging.basicConfig( + level=logging.INFO, + format="%(asctime)s %(levelname)s %(message)s", + handlers=[ + logging.FileHandler(log_file, encoding="utf-8"), + logging.StreamHandler(), + ], + ) + return logging.getLogger(__name__), log_file + +logger, LOG_FILE = setup_logging() + + +# ======================== 工具函数 ======================== + +def get_all_codes() -> List[str]: + df = pd.read_csv(ALL_STOCKS_FILE) + for col in ["代码", "code", "股票代码"]: + if col in df.columns: + return [str(c).zfill(6) for c in df[col].tolist()] + raise ValueError(f"找不到代码列: {list(df.columns)}") + + +def code_to_baostock(code: str) -> Tuple[str, str]: + """6位代码 → (BaoStock格式, parquet前缀)""" + if code.startswith(("6", "68", "51")): + return f"sh.{code}", "sh" + else: + return f"sz.{code}", "sz" + + +def is_backfilled(parquet_path: Path) -> bool: + """检查文件是否已经被BaoStock回补过""" + # 用标记文件判断 + marker = parquet_path.parent / f".{parquet_path.stem}.baostock" + return marker.exists() + + +def load_progress() -> set: + progress_file = PROGRESS_DIR / "backfill_15min_progress.json" + if progress_file.exists(): + try: + return set(json.loads(progress_file.read_text()).get("done", [])) + except Exception: + pass + return set() + + +def save_progress(done_set: set): + PROGRESS_DIR.mkdir(parents=True, exist_ok=True) + progress_file = PROGRESS_DIR / "backfill_15min_progress.json" + progress_file.write_text(json.dumps({ + "done": sorted(done_set), + "ts": datetime.now().isoformat(), + }, ensure_ascii=False)) + + +# ======================== 核心逻辑 ======================== + +def fetch_bs_15min(bs_code: str, start_date: str, end_date: str) -> Optional[pd.DataFrame]: + """从BaoStock获取15min不复权数据""" + rs = bs.query_history_k_data_plus( + bs_code, + "date,time,code,open,high,low,close,volume,amount,adjustflag", + start_date=start_date, + end_date=end_date, + frequency="15", + adjustflag="3", # 不复权 + ) + + if rs.error_code != "0": + logger.debug("BaoStock %s 错误: %s %s", bs_code, rs.error_code, rs.error_msg) + return None + + rows = [] + while rs.next(): + rows.append(rs.get_row_data()) + + if not rows: + return None + + df = pd.DataFrame(rows, columns=["date", "time", "code", "open", "high", "low", + "close", "volume", "amount", "adjustflag"]) + + # 转换时间格式: 20260428094500000 → 2026-04-28 09:45:00 + df["day"] = df["time"].apply(lambda t: f"{t[:4]}-{t[4:6]}-{t[6:8]} {t[8:10]}:{t[10:12]}:00") + + # 数值转换 + for col in ["open", "high", "low", "close"]: + df[col] = pd.to_numeric(df[col], errors="coerce") + df["volume"] = df["volume"].astype(str) + df["amount"] = df["amount"].astype(str) + + # 保留与parquet一致的列 + df = df[["day", "open", "high", "low", "close", "volume", "amount"]] + + # 过滤无效数据 + df = df.dropna(subset=["open", "high", "low", "close"]) + bad_ohlc = (df["high"] < df[["open", "close"]].max(axis=1)) | \ + (df["low"] > df[["open", "close"]].min(axis=1)) + if bad_ohlc.any(): + df = df[~bad_ohlc] + + return df if not df.empty else None + + +def backfill_one(code: str, start_date: str, end_date: str, force: bool = False) -> Tuple[str, int]: + """ + 全量重建单只股票的15min历史 + 返回: (status, total_rows) + """ + bs_code, prefix = code_to_baostock(code) + parquet_path = MINUTE_15_DIR / f"{prefix}{code}_15min.parquet" + + # 已回补过的跳过 + if not force and is_backfilled(parquet_path): + return "skipped", 0 + + # 获取BaoStock数据 + df_new = None + for attempt in range(BS_MAX_RETRIES): + try: + df_new = fetch_bs_15min(bs_code, start_date, end_date) + if df_new is not None and len(df_new) > 0: + break + except Exception as e: + logger.debug("backfill %s 重试%d: %s", code, attempt + 1, e) + time.sleep(1) + + if df_new is None or df_new.empty: + return "failed", 0 + + # 备份旧文件 + if parquet_path.exists(): + BACKUP_DIR.mkdir(parents=True, exist_ok=True) + backup_path = BACKUP_DIR / parquet_path.name + if not backup_path.exists(): # 不覆盖已有备份 + shutil.copy2(str(parquet_path), str(backup_path)) + + # 写入新文件 + try: + df_new = df_new.sort_values("day").reset_index(drop=True) + df_new.to_parquet(parquet_path, index=False) + # 写标记文件 + marker = parquet_path.parent / f".{parquet_path.stem}.baostock" + marker.write_text(datetime.now().isoformat()) + return "ok", len(df_new) + except Exception as e: + logger.error("写入 %s 失败: %s", code, e) + return "failed", 0 + + +# ======================== 主流程 ======================== + +def main(): + parser = argparse.ArgumentParser(description="BaoStock 15min历史回补（全量重建）") + parser.add_argument("--start", default=BS_START_DATE, help="回补起始日期 (YYYYMMDD)") + parser.add_argument("--end", default="", help="结束日期 (默认今天)") + parser.add_argument("--codes", help="指定股票代码，逗号分隔") + parser.add_argument("--limit", type=int, default=0, help="限制处理数量（测试用）") + parser.add_argument("--force", action="store_true", help="强制重建（覆盖已有BaoStock数据）") + args = parser.parse_args() + + start_date = f"{args.start[:4]}-{args.start[4:6]}-{args.start[6:8]}" + end_date = f"{args.end[:4]}-{args.end[4:6]}-{args.end[6:8]}" if args.end else datetime.now().strftime("%Y-%m-%d") + + if not NAS_ROOT.exists(): + logger.error("❌ NAS未挂载") + sys.exit(1) + + # 登录BaoStock + lg = bs.login() + if lg.error_code != "0": + logger.error("❌ BaoStock登录失败: %s", lg.error_msg) + sys.exit(1) + logger.info("✅ BaoStock登录成功") + + # 获取股票列表 + if args.codes: + codes = [c.strip() for c in args.codes.split(",")] + else: + codes = get_all_codes() + + if args.limit > 0: + codes = codes[:args.limit] + + logger.info("=" * 60) + logger.info("BaoStock 15min全量重建开始") + logger.info(f" 股票数: {len(codes)}") + logger.info(f" 日期范围: {start_date} ~ {end_date}") + logger.info(f" 数据目录: {MINUTE_15_DIR}") + logger.info(f" 旧数据备份: {BACKUP_DIR}") + logger.info(f" 日志文件: {LOG_FILE}") + + # 进度恢复 + done_set = load_progress() + if args.force: + todo = codes # 强制模式不跳过 + else: + todo = [c for c in codes if c not in done_set] + logger.info(f" 待处理: {len(todo)}（已完成: {len(done_set)}）") + + stats = {"ok": 0, "skipped": 0, "failed": 0, "rows": 0} + t_start = time.time() + RELOGIN_EVERY = 400 # 每400只重新登录BaoStock，防止连接断开 + + for i, code in enumerate(todo): + # 定期重新登录保持连接 + if i > 0 and i % RELOGIN_EVERY == 0: + bs.logout() + time.sleep(2) + lg = bs.login() + if lg.error_code != "0": + logger.error("BaoStock重连失败: %s，等待30秒", lg.error_msg) + time.sleep(30) + lg = bs.login() + logger.info("BaoStock重连 @ %d/%d", i, len(todo)) + + try: + status, total_rows = backfill_one(code, start_date, end_date, args.force) + except Exception as e: + status, total_rows = "failed", 0 + logger.debug("backfill %s 异常: %s", code, e) + + stats[status] = stats.get(status, 0) + 1 + if status == "ok": + stats["rows"] += total_rows + + done_set.add(code) + + if (i + 1) % PROGRESS_SAVE_EVERY == 0: + save_progress(done_set) + elapsed = time.time() - t_start + logger.info("进度: %d/%d ok=%d skipped=%d failed=%d rows=%d (%.0f秒)", + i + 1, len(todo), stats["ok"], stats["skipped"], + stats["failed"], stats["rows"], elapsed) + + # 频率控制 + if i < len(todo) - 1: + time.sleep(BS_INTERVAL) + + # 保存最终进度 + save_progress(done_set) + + bs.logout() + + elapsed = time.time() - t_start + logger.info("=" * 60) + logger.info("✅ 回补完成，耗时 %.1f 秒", elapsed) + logger.info("统计: %s", json.dumps(stats, ensure_ascii=False)) + + +if __name__ == "__main__": + main() diff --git a/scripts/data_platform/daily_all_update.py b/scripts/data_platform/daily_all_update.py new file mode 100644 index 0000000..97e8c94 --- /dev/null +++ b/scripts/data_platform/daily_all_update.py @@ -0,0 +1,1103 @@ +#!/usr/bin/env python3 +""" +全市场每日增量更新 - 日线 + 15分钟线 (v2.0) + +功能： + 1. 日线：多源fallback → 更新Parquet + 本地vnpy DB + 2. 15分钟线：多源fallback → 增量合并Parquet + 本地vnpy DB + 3. 本地DB构建完成后mv原子重命名到NAS + +数据源降级链（按质量排序，BaoStock T+1延迟已考虑）： + 日线增量（当天实时）：东方财富(实时,4s限频) → BaoStock(T+1,无反爬) → 腾讯(amount有时0) + 15min增量（当天实时）：东方财富(实时7周) → BaoStock(T+1,无反爬) → 新浪(已挂,保留) + +设计原则： + - 多源fallback：按质量排序，成功即用，失败试下一个 + - 增量更新，不重复下载 + - vnpy DB本地构建 → mv原子重命名到NAS（避免SMB锁） + - 失败率检测（滑动窗口100只，>80%失败则终止） + - DB轮转备份（保留7天） + +用法： + python3 daily_all_update.py # 全量更新（日线+15min） + python3 daily_all_update.py --skip-daily # 只更新15min + python3 daily_all_update.py --skip-15min # 只更新日线 + +变更记录： + v1.0 (2026-05-03) - 初始版本 + v1.1 (2026-05-03) - 司马懿评审：interval→15m, 严格增量, 进度文件, 源检测, DB备份 + v1.2 (2026-05-05) - 东方财富集成：日线主源切换东方财富 + v2.0 (2026-05-06) - 重大架构变更（司马懿+姜维评审通过）： + - BaoStock替代所有主源，多源fallback机制 + - vnpy DB写入改为本地构建+mv原子重命名（解决SMB锁） + - interval统一1m（vnpy 4.x Interval.MINUTE硬约束） + - 日线跨年写入修复 + - 进度文件加日期 + - overview增量更新（不做全表聚合） + - 失败率检测替代固定次数暂停 + - 东方财富当天实时+BaoStock T+1补全 +""" + +import argparse +import json +import os +import re +import shutil +import sqlite3 +import sys +import time +import logging +import random +import urllib.request +import urllib.error +from datetime import datetime, timedelta +from pathlib import Path +from typing import Optional, List, Tuple, Callable +from collections import deque + +import pandas as pd + +try: + import baostock as bs + HAS_BAOSTOCK = True +except ImportError: + HAS_BAOSTOCK = False + +try: + import requests as _requests + HAS_REQUESTS = True +except ImportError: + HAS_REQUESTS = False + +# ======================== 配置 ======================== + +# STOCK_MOUNT: NAS 挂载点（SMB 模式）或本地 rsync 同步目录（SSH 模式，包装脚本设） +STOCK_MOUNT = Path(os.environ.get("STOCK_MOUNT", "/Users/chufeng/stock_mount")) +LOG_DIR = STOCK_MOUNT / "logs" / "daily_update" +DAILY_DIR = STOCK_MOUNT / "A股数据" / "日线数据" / "daily" +MINUTE_15_DIR = STOCK_MOUNT / "minute_kline" / "15min" +VNPY_DB_PATH = STOCK_MOUNT / "sanguo_vnpy" / "data" / "quant_trading.db" +LOCAL_DB_PATH = Path("/tmp/quant_trading_new.db") +ALL_STOCKS_FILE = STOCK_MOUNT / "A股数据" / "stock_info" / "stock_basic_info_raw_20260326_113530.csv" +PROGRESS_DIR = STOCK_MOUNT / "logs" / "daily_update" / "progress" +# 验证用：STOCK_LIMIT 限制处理的股票数（前 N 只），默认 0 = 全市场 +STOCK_LIMIT = int(os.environ.get("STOCK_LIMIT", "0")) + +REQUEST_INTERVAL_EM = 4.0 # 东方财富：4s + 随机抖动±1s +REQUEST_INTERVAL_SINA = 0.3 # 新浪（已挂，保留） +REQUEST_INTERVAL_BS = 0.0 # BaoStock：无需限频 +EM_JITTER = 1.0 # 东方财富随机抖动范围(±秒) +MAX_RETRIES = 3 +# 失败率检测：滑动窗口 +GLOBAL_FAIL_WINDOW = 100 # 最近N只 +GLOBAL_FAIL_THRESHOLD = 0.8 # 失败率阈值 +# DB +DB_BACKUP_KEEP_DAYS = 7 +BATCH_SIZE = 50000 +# vnpy interval +# interval='1m' — vnpy 4.x Interval.MINUTE硬约束，实际存储15分钟线 +INTERVAL_MINUTE = "1m" +INTERVAL_DAILY = "d" + +HEADERS = {"User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7)"} +HEADERS_EM = { + "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36", + "Referer": "https://quote.eastmoney.com/", + "Accept": "*/*", + "Accept-Language": "zh-CN,zh;q=0.9", +} + + +def setup_logging(): + LOG_DIR.mkdir(parents=True, exist_ok=True) + ts = datetime.now().strftime("%Y%m%d_%H%M%S") + log_file = LOG_DIR / f"update_{ts}.log" + logging.basicConfig( + level=logging.INFO, + format="%(asctime)s %(levelname)s %(message)s", + handlers=[ + logging.FileHandler(log_file, encoding="utf-8"), + logging.StreamHandler(), + ], + ) + return logging.getLogger(__name__) + + +logger = setup_logging() + + +def _make_opener(): + return urllib.request.build_opener(urllib.request.ProxyHandler({})) + + +# ======================== 工具函数 ======================== + +def get_market_prefix(code: str) -> Tuple[str, str]: + """返回 (prefix, clean_code)：sh/sz + 6位数字""" + code = re.sub(r"[^0-9]", "", code).zfill(6) + if code.startswith(("60", "68", "51")): + return "sh", code + return "sz", code + + +def code_to_baostock(code: str) -> str: + """纯6位代码转BaoStock格式：sh.600000""" + prefix, clean = get_market_prefix(code) + return f"{prefix}.{clean}" + + +def get_all_codes() -> List[str]: + df = pd.read_csv(ALL_STOCKS_FILE) + for col in ["代码", "code", "股票代码"]: + if col in df.columns: + codes = [str(c).zfill(6) for c in df[col].tolist()] + return codes[:STOCK_LIMIT] if STOCK_LIMIT else codes + raise ValueError(f"找不到代码列: {list(df.columns)}") + + +def nas_mounted() -> bool: + return DAILY_DIR.exists() and MINUTE_15_DIR.exists() + + +# ======================== DB备份 ======================== + +def rotate_db_backup(): + """轮转备份NAS vnpy DB，保留最近N天""" + backup_dir = VNPY_DB_PATH.parent + today = datetime.now().strftime("%Y%m%d") + backup_file = backup_dir / f"quant_trading_{today}.db.bak" + + if backup_file.exists(): + logger.info("DB今日已备份: %s", backup_file) + return + + logger.info("开始DB备份: %s → %s", VNPY_DB_PATH.name, backup_file.name) + try: + shutil.copy2(str(VNPY_DB_PATH), str(backup_file)) + logger.info("✅ DB备份完成 (%.1f MB)", backup_file.stat().st_size / 1024 / 1024) + except Exception as e: + logger.error("❌ DB备份失败: %s", e) + return + + cutoff = datetime.now() - timedelta(days=DB_BACKUP_KEEP_DAYS) + for f in backup_dir.glob("quant_trading_*.db.bak"): + try: + date_str = f.stem.split("_")[-1] + file_date = datetime.strptime(date_str, "%Y%m%d") + if file_date < cutoff: + f.unlink() + logger.info("清理过期备份: %s", f.name) + except (ValueError, OSError): + pass + + +# ======================== 进度文件 ======================== + +def load_progress(name: str) -> set: + """加载进度文件（v2.0：带日期）""" + today = datetime.now().strftime("%Y%m%d") + progress_file = PROGRESS_DIR / f"{name}_{today}_progress.json" + PROGRESS_DIR.mkdir(parents=True, exist_ok=True) + if progress_file.exists(): + try: + return set(json.loads(progress_file.read_text()).get("done", [])) + except Exception: + pass + return set() + + +def save_progress(name: str, done_set: set): + """保存进度文件（v2.0：带日期）""" + today = datetime.now().strftime("%Y%m%d") + progress_file = PROGRESS_DIR / f"{name}_{today}_progress.json" + progress_file.write_text(json.dumps({ + "done": sorted(list(done_set)), + "ts": datetime.now().isoformat(), + })) + + +# ======================== 失败率检测（v2.0） ======================== + +class SourceHealthMonitor: + """滑动窗口失败率检测：最近N只中失败率>T则判定源不可用""" + + def __init__(self, window: int = GLOBAL_FAIL_WINDOW, threshold: float = GLOBAL_FAIL_THRESHOLD): + self.window = window + self.threshold = threshold + self.history = deque(maxlen=window) + + def report(self, code: str, failed: bool) -> bool: + """ + 报告单只结果，返回True=源健康，False=源不可用应终止 + """ + self.history.append(1 if failed else 0) + if len(self.history) >= 20: # 至少20只才判断 + fail_rate = sum(self.history) / len(self.history) + if fail_rate >= self.threshold: + logger.error( + "⚠️ 源不可用检测触发：最近%d只失败率 %.0f%%，终止更新", + len(self.history), fail_rate * 100, + ) + return False + return True + + +# ======================== 数据源：BaoStock ======================== + +def fetch_baostock_daily(code: str, start_date: str, end_date: str) -> Optional[pd.DataFrame]: + """BaoStock日线：全量历史，无反爬，amount真实，T+1延迟""" + if not HAS_BAOSTOCK: + return None + bs_code = code_to_baostock(code) + try: + rs = bs.query_history_k_data_plus( + bs_code, + "date,open,high,low,close,volume,amount", + start_date=start_date.replace("-", ""), + end_date=end_date.replace("-", ""), + frequency="d", + adjustflag="2", + ) + rows = [] + while (rs.error_code == "0") and rs.next(): + rows.append(rs.get_row_data()) + if not rows: + return None + df = pd.DataFrame(rows, columns=["date", "open", "high", "low", "close", "volume", "amount"]) + for c in ["open", "high", "low", "close", "volume", "amount"]: + df[c] = pd.to_numeric(df[c], errors="coerce") + df = df.dropna(subset=["close"]) + if df.empty: + return None + return df + except Exception as e: + logger.debug("BaoStock日线失败 %s: %s", code, e) + return None + + +def fetch_baostock_15min(code: str, start_date: str, end_date: str) -> Optional[pd.DataFrame]: + """BaoStock 15min：全量历史，无反爬，amount真实，T+1延迟""" + if not HAS_BAOSTOCK: + return None + bs_code = code_to_baostock(code) + try: + rs = bs.query_history_k_data_plus( + bs_code, + "date,time,open,high,low,close,volume,amount", + start_date=start_date.replace("-", ""), + end_date=end_date.replace("-", ""), + frequency="15", + adjustflag="2", + ) + rows = [] + while (rs.error_code == "0") and rs.next(): + rows.append(rs.get_row_data()) + if not rows: + return None + # BaoStock返回: [date, time(YYYYMMDDHHMMSSSSS), open, high, low, close, volume, amount] + df = pd.DataFrame(rows) + # 构造day列：YYYY-MM-DD HH:MM:SS + df.columns = ["date", "time", "open", "high", "low", "close", "volume", "amount"][:len(df.columns)] + df["day"] = df["time"].apply(lambda t: f"{t[:4]}-{t[4:6]}-{t[6:8]} {t[8:10]}:{t[10:12]}:00") + for c in ["open", "high", "low", "close", "volume", "amount"]: + df[c] = pd.to_numeric(df[c], errors="coerce") + df = df.dropna(subset=["close"]) + if df.empty: + return None + return df[["day", "open", "high", "low", "close", "volume", "amount"]] + except Exception as e: + logger.debug("BaoStock 15min失败 %s: %s", code, e) + return None + + +# ======================== 数据源：东方财富 ======================== + +def _get_em_secid(code: str) -> str: + if code.startswith(("60", "68", "51")): + return f"1.{code}" + return f"0.{code}" + + +def _parse_em_klines(klines: list) -> Optional[pd.DataFrame]: + """解析东方财富K线数据（日线和15min通用）""" + if not klines: + return None + rows = [] + for line in klines: + parts = line.split(",") + if len(parts) < 7: + continue + rows.append({ + "date": parts[0], + "open": float(parts[1]), + "close": float(parts[2]), + "high": float(parts[3]), + "low": float(parts[4]), + "volume": float(parts[5]), + "amount": float(parts[6]), + }) + if not rows: + return None + return pd.DataFrame(rows) + + +def fetch_eastmoney_daily(code: str, start_date: str, end_date: str) -> Optional[pd.DataFrame]: + """东方财富日线：当天实时，amount真实，4s限频""" + if not HAS_REQUESTS: + return None + secid = _get_em_secid(code) + ts = str(int(time.time() * 1000)) + url = ( + f"https://push2his.eastmoney.com/api/qt/stock/kline/get?" + f"secid={secid}&klt=101&fqt=1&" + f"beg={start_date.replace('-', '')}&end={end_date.replace('-', '')}&" + f"fields1=f1,f2,f3,f4,f5,f6,f7,f8&" + f"fields2=f51,f52,f53,f54,f55,f56,f57,f58,f59,f60,f61&" + f"ut=b2884a393a59ad64002292a3e90d46a5&lmt=10000&" + f"cb=jQuery_em_{ts}&_={ts}" + ) + session = _requests.Session() + session.trust_env = False + try: + r = session.get(url, headers=HEADERS_EM, timeout=15, verify=False) + if r.status_code != 200: + return None + text = r.text + data = json.loads(text[text.index("(") + 1:text.rindex(")")]) + if data.get("rc") != 0: + return None + klines = data.get("data", {}).get("klines", []) + df = _parse_em_klines(klines) + if df is None: + return None + df["date"] = pd.to_datetime(df["date"]).dt.strftime("%Y-%m-%d") + mask = (df["date"] >= start_date) & (df["date"] <= end_date) + result = df.loc[mask, ["date", "open", "high", "low", "close", "volume", "amount"]] + return result if not result.empty else None + except Exception as e: + logger.debug("东方财富日线失败 %s: %s", code, e) + return None + + +def fetch_eastmoney_15min(code: str, start_date: str, end_date: str) -> Optional[pd.DataFrame]: + """东方财富15min：当天实时，约7周历史，4s限频""" + if not HAS_REQUESTS: + return None + secid = _get_em_secid(code) + ts = str(int(time.time() * 1000)) + url = ( + f"https://push2his.eastmoney.com/api/qt/stock/kline/get?" + f"secid={secid}&klt=15&fqt=1&" + f"beg={start_date.replace('-', '')}&end={end_date.replace('-', '')}&" + f"fields1=f1,f2,f3,f4,f5,f6,f7,f8&" + f"fields2=f51,f52,f53,f54,f55,f56,f57,f58,f59,f60,f61&" + f"ut=b2884a393a59ad64002292a3e90d46a5&lmt=100000&" + f"cb=jQuery_em_{ts}&_={ts}" + ) + session = _requests.Session() + session.trust_env = False + try: + r = session.get(url, headers=HEADERS_EM, timeout=15, verify=False) + if r.status_code != 200: + return None + text = r.text + data = json.loads(text[text.index("(") + 1:text.rindex(")")]) + if data.get("rc") != 0: + return None + klines = data.get("data", {}).get("klines", []) + if not klines: + return None + rows = [] + for line in klines: + parts = line.split(",") + if len(parts) < 7: + continue + rows.append({ + "day": parts[0], + "open": float(parts[1]), + "close": float(parts[2]), + "high": float(parts[3]), + "low": float(parts[4]), + "volume": float(parts[5]), + "amount": float(parts[6]), + }) + if not rows: + return None + df = pd.DataFrame(rows) + # 转day格式：东方财富返回 "2026-04-30 15:00" 或 "2026-04-30" + df["day"] = df["day"].apply(lambda d: d if " " in str(d) else f"{d} 00:00:00") + # 补全秒 + df["day"] = df["day"].apply(lambda d: d if d.count(":") == 2 else d + ":00") + for c in ["open", "high", "low", "close", "volume", "amount"]: + df[c] = pd.to_numeric(df[c], errors="coerce") + df = df.dropna(subset=["close"]) + if df.empty: + return None + return df[["day", "open", "high", "low", "close", "volume", "amount"]] + except Exception as e: + logger.debug("东方财富15min失败 %s: %s", code, e) + return None + + +# ======================== 数据源：腾讯 ======================== + +def fetch_tencent_daily(code: str, start_date: str, end_date: str) -> Optional[pd.DataFrame]: + """腾讯日线：amount有时为0""" + prefix, clean = get_market_prefix(code) + tq = f"{prefix}{clean}" + days = (pd.Timestamp(end_date) - pd.Timestamp(start_date)).days + 10 + url = f"https://web.ifzq.gtimg.cn/appstock/app/fqkline/get?param={tq},day,{start_date},,{days}," + opener = _make_opener() + try: + req = urllib.request.Request(url, headers=HEADERS) + with opener.open(req, timeout=10) as r: + raw = r.read().decode("utf-8", errors="replace") + data = json.loads(raw) + d = data.get("data") + if not isinstance(d, dict): + return None + klines = d.get(tq, {}).get("day", []) + if not klines: + return None + df = pd.DataFrame(klines) + ncols = len(df.columns) + if ncols >= 7: + df.columns = ["date", "open", "close", "high", "low", "volume", "amount"][:ncols] + else: + df.columns = ["date", "open", "close", "high", "low", "volume"][:ncols] + df["amount"] = 0.0 + if "amount" not in df.columns: + df["amount"] = 0.0 + for c in ["open", "close", "high", "low", "volume", "amount"]: + df[c] = pd.to_numeric(df[c], errors="coerce").fillna(0) + df["date"] = pd.to_datetime(df["date"]).dt.strftime("%Y-%m-%d") + mask = (df["date"] >= start_date) & (df["date"] <= end_date) + result = df.loc[mask, ["date", "open", "high", "low", "close", "volume", "amount"]] + return result if not result.empty else None + except Exception as e: + logger.debug("腾讯日线失败 %s: %s", code, e) + return None + + +# ======================== 数据源：新浪（已挂，保留代码） ======================== + +def try_sina_15min(symbol: str, datalen: int = 800) -> Optional[pd.DataFrame]: + """新浪15分钟K线API（当前已挂，保留作为fallback）""" + url = ( + f"https://quotes.sina.cn/cn/api/jsonp_v2.php/var%20=min15_{symbol}=/" + f"CN_MarketDataService.getKLineData?symbol={symbol}&scale=15&ma=no&datalen={datalen}" + ) + opener = _make_opener() + try: + req = urllib.request.Request(url, headers=HEADERS) + with opener.open(req, timeout=15) as r: + raw = r.read().decode("utf-8", errors="replace") + m = re.search(r"\((\[.*\])\)", raw, re.DOTALL) + if not m: + return None + data = json.loads(m.group(1)) + if not data: + return None + df = pd.DataFrame(data) + cols = ["day", "open", "high", "low", "close", "volume", "amount"] + for c in cols: + if c not in df.columns: + return None + return df[cols] + except Exception: + return None + + +# ======================== Fallback机制 ======================== + +def fetch_with_fallback( + sources: List[Tuple[str, Callable, float]], + code: str, + start_date: str, + end_date: str, + is_daily: bool = True, +) -> Tuple[Optional[pd.DataFrame], str]: + """ + 多源fallback获取数据 + sources: [(name, fetch_fn, interval_seconds), ...] + 返回: (DataFrame或None, 使用的源名) + """ + for name, fetch_fn, interval in sources: + if interval > 0: + jitter = (hash(code + name) % 200 - 100) / 100.0 * (interval * 0.1) + time.sleep(max(0, interval + jitter)) + data = None + for attempt in range(MAX_RETRIES): + try: + if is_daily: + data = fetch_fn(code, start_date, end_date) + else: + data = fetch_fn(code, start_date, end_date) + if data is not None and len(data) > 0: + return data, name + except Exception: + pass + if attempt < MAX_RETRIES - 1: + time.sleep(1) + return None, "" + + +# 日线数据源（当天实时优先） +SOURCES_DAILY = [ + ("eastmoney", fetch_eastmoney_daily, REQUEST_INTERVAL_EM), + # ("baostock", fetch_baostock_daily, REQUEST_INTERVAL_BS), # 2026-05-15 暂禁：BaoStock无超时导致进程卡死 + ("tencent", fetch_tencent_daily, 0), +] + +# 15min数据源（当天实时优先） +SOURCES_15MIN = [ + ("eastmoney", fetch_eastmoney_15min, REQUEST_INTERVAL_EM), + # ("baostock", fetch_baostock_15min, REQUEST_INTERVAL_BS), # 2026-05-15 暂禁：BaoStock无超时导致进程卡死 + ("sina", lambda code, s, e: try_sina_15min(f"{get_market_prefix(code)[0]}{get_market_prefix(code)[1]}"), + REQUEST_INTERVAL_SINA), +] + + +# ======================== 日线更新 ======================== + +def get_daily_last_date(code: str) -> str: + """获取日线Parquet中最后日期""" + prefix, clean = get_market_prefix(code) + for year in range(datetime.now().year, 2009, -1): + fpath = DAILY_DIR / str(year) / f"{prefix}{clean}_daily.parquet" + if fpath.exists(): + try: + df = pd.read_parquet(fpath, columns=["date"]) + if not df.empty: + return str(df["date"].max())[:10] + except Exception: + pass + return "" + + +def update_daily_parquet(code: str, new_data: pd.DataFrame) -> int: + """增量写入日线Parquet（v2.0：按数据日期分年目录）""" + prefix, clean = get_market_prefix(code) + new_data = new_data.copy() + new_data["date"] = new_data["date"].astype(str) + + total_new = 0 + for yr in new_data["date"].str[:4].unique(): + year_data = new_data[new_data["date"].str[:4] == yr].copy() + parquet_path = DAILY_DIR / str(yr) / f"{prefix}{clean}_daily.parquet" + + if parquet_path.exists(): + existing = pd.read_parquet(parquet_path) + existing["date"] = existing["date"].astype(str) + combined = pd.concat([existing, year_data], ignore_index=True) + combined = combined.drop_duplicates(subset=["date"], keep="last") + combined = combined.sort_values("date").reset_index(drop=True) + else: + parquet_path.parent.mkdir(parents=True, exist_ok=True) + combined = year_data + + tmp = parquet_path.with_suffix(".tmp") + combined.to_parquet(tmp, index=False) + tmp.rename(parquet_path) + total_new += len(year_data) + + return total_new + + +def run_daily_update(codes: List[str], local_conn: sqlite3.Connection) -> dict: + """日线增量更新""" + logger.info("=" * 60) + logger.info("日线增量更新开始，共 %d 只", len(codes)) + today = datetime.now().strftime("%Y-%m-%d") + + stats = {"updated": 0, "skipped": 0, "failed": 0, "records": 0, "db_records": 0} + all_db_values = [] + + done_set = load_progress("daily") + todo = [c for c in codes if c not in done_set] + logger.info("待更新: %d（已完成: %d）", len(todo), len(done_set)) + + health = SourceHealthMonitor() + + for i, code in enumerate(todo): + last_date = get_daily_last_date(code) + if not last_date: + logger.debug("日线无历史数据，跳过（需先全量导入）: %s", code) + stats["skipped"] += 1 + done_set.add(code) + continue + next_day = (pd.Timestamp(last_date) + timedelta(days=1)).strftime("%Y-%m-%d") + if next_day > today: + stats["skipped"] += 1 + done_set.add(code) + continue + + data, source = fetch_with_fallback(SOURCES_DAILY, code, next_day, today, is_daily=True) + failed = (data is None or data.empty) + + if not health.report(code, failed): + logger.error("❌ 日线所有源不可用，终止日线更新") + stats["source_aborted"] = True + break + + if failed: + stats["failed"] += 1 + # 注意：失败不加done_set，下次运行可重试 + if (i + 1) % 500 == 0: + logger.info("日线进度: %d/%d updated=%d failed=%d", i + 1, len(todo), stats["updated"], stats["failed"]) + save_progress("daily", done_set) + continue + + # 校验 + for c in ["open", "high", "low", "close"]: + data[c] = pd.to_numeric(data[c], errors="coerce") + if (data[["close", "open"]] <= 0).any().any(): + stats["failed"] += 1 + # 校验失败也不加done_set + continue + + try: + n = update_daily_parquet(code, data) + stats["updated"] += 1 + stats["records"] += n + + # 收集vnpy DB数据 + prefix, clean = get_market_prefix(code) + exchange = "SSE" if prefix == "sh" else "SZSE" + for _, row in data.iterrows(): + all_db_values.append(( + clean, exchange, str(row["date"]), INTERVAL_DAILY, + float(row.get("volume", 0)), float(row.get("amount", 0)), 0.0, + float(row.get("open", 0)), float(row.get("high", 0)), + float(row.get("low", 0)), float(row.get("close", 0)), + )) + stats["db_records"] += len(data) + except Exception as e: + stats["failed"] += 1 + logger.warning("日线写入失败 %s: %s", code, e) + + done_set.add(code) + + if (i + 1) % 500 == 0: + logger.info("日线进度: %d/%d updated=%d failed=%d src=%s", + i + 1, len(todo), stats["updated"], stats["failed"], source) + save_progress("daily", done_set) + + # 写入本地DB + if all_db_values: + _write_local_db(local_conn, all_db_values, "日线") + + save_progress("daily", done_set) + if stats.get("source_aborted"): + pass + logger.info("日线完成: %s", json.dumps(stats, ensure_ascii=False)) + return stats + + +# ======================== 15分钟线更新 ======================== + +def get_15min_last_date(parquet_path: Path) -> str: + """获取15min Parquet中最后一条时间戳""" + if not parquet_path.exists(): + return "" + try: + df = pd.read_parquet(parquet_path, columns=["day"]) + if not df.empty: + return str(df["day"].max()) + except Exception: + pass + return "" + + +def fetch_15min_with_fallback(code: str, start_date: str, end_date: str) -> Tuple[Optional[pd.DataFrame], str]: + """15min多源fallback（特殊处理新浪接口不同）""" + prefix, clean = get_market_prefix(code) + symbol = f"{prefix}{clean}" + + # 源1：东方财富 + jitter = (hash(code) % 200 - 100) / 100.0 * EM_JITTER + time.sleep(max(0, REQUEST_INTERVAL_EM + jitter)) + for attempt in range(MAX_RETRIES): + try: + data = fetch_eastmoney_15min(code, start_date, end_date) + if data is not None and len(data) > 0: + return data, "eastmoney" + except Exception: + pass + if attempt < MAX_RETRIES - 1: + time.sleep(1) + + # 源2：BaoStock + for attempt in range(MAX_RETRIES): + try: + data = fetch_baostock_15min(code, start_date, end_date) + if data is not None and len(data) > 0: + return data, "baostock" + except Exception: + pass + if attempt < MAX_RETRIES - 1: + time.sleep(0.5) + + # 源3：新浪（已挂，保留） + time.sleep(REQUEST_INTERVAL_SINA) + for attempt in range(MAX_RETRIES): + try: + data = try_sina_15min(symbol) + if data is not None and len(data) > 0: + return data, "sina" + except Exception: + pass + if attempt < MAX_RETRIES - 1: + time.sleep(0.5) + + return None, "" + + +def run_15min_update(codes: List[str], local_conn: sqlite3.Connection) -> dict: + """15分钟线增量更新""" + logger.info("=" * 60) + logger.info("15分钟线增量更新开始，共 %d 只", len(codes)) + + stats = {"updated": 0, "skipped": 0, "failed": 0, "records": 0, "db_records": 0} + all_db_values = [] + + done_set = load_progress("15min") + todo = [c for c in codes if c not in done_set] + logger.info("待更新: %d（已完成: %d）", len(todo), len(done_set)) + + health = SourceHealthMonitor() + + for i, code in enumerate(todo): + prefix, clean = get_market_prefix(code) + parquet_path = MINUTE_15_DIR / f"{prefix}{clean}_15min.parquet" + + # 获取最后日期，决定增量范围 + last_date = get_15min_last_date(parquet_path) + if last_date: + next_dt = (pd.Timestamp(last_date) + timedelta(minutes=15)).strftime("%Y-%m-%d") + else: + next_dt = "2024-01-02" + today = datetime.now().strftime("%Y-%m-%d") + + if next_dt > today: + stats["skipped"] += 1 + done_set.add(code) + continue + + # 多源fallback + df_new, source = fetch_15min_with_fallback(code, next_dt, today) + failed = (df_new is None or (hasattr(df_new, 'empty') and df_new.empty)) + + if not health.report(code, failed): + logger.error("❌ 15min所有源不可用，终止15min更新") + stats["source_aborted"] = True + break + + if failed: + stats["failed"] += 1 + # 注意：失败不加done_set，下次运行可重试 + if (i + 1) % 500 == 0: + logger.info("15min进度: %d/%d updated=%d failed=%d", i + 1, len(todo), stats["updated"], stats["failed"]) + save_progress("15min", done_set) + continue + + # 数据校验 + for col in ["open", "high", "low", "close"]: + df_new[col] = pd.to_numeric(df_new[col], errors="coerce") + df_new["volume"] = pd.to_numeric(df_new["volume"], errors="coerce").fillna(0) + df_new["amount"] = pd.to_numeric(df_new["amount"], errors="coerce").fillna(0) + df_new["day"] = df_new["day"].astype(str) + + bad = (df_new[["close", "open"]] <= 0).any(axis=1) + if bad.any(): + df_new = df_new[~bad] + if df_new.empty: + stats["failed"] += 1 + # 校验失败也不加done_set + continue + + # 增量合并 + if last_date: + df_increment = df_new[df_new["day"] > last_date].copy() + if df_increment.empty: + stats["skipped"] += 1 + done_set.add(code) + continue + + existing = pd.read_parquet(parquet_path) + existing["day"] = existing["day"].astype(str) + # 统一类型：旧数据volume/amount可能是str，新数据是float + for col in ["volume", "amount", "open", "high", "low", "close"]: + if col in existing.columns: + existing[col] = pd.to_numeric(existing[col], errors="coerce").fillna(0) + if col in df_increment.columns: + df_increment[col] = pd.to_numeric(df_increment[col], errors="coerce").fillna(0) + combined = pd.concat([existing, df_increment], ignore_index=True) + combined = combined.sort_values("day").reset_index(drop=True) + new_rows = len(df_increment) + else: + df_increment = df_new + combined = df_new.sort_values("day").reset_index(drop=True) + new_rows = len(df_new) + + # 原子写入Parquet + tmp = parquet_path.with_suffix(".tmp") + combined.to_parquet(tmp, index=False) + tmp.rename(parquet_path) + + stats["updated"] += 1 + stats["records"] += new_rows + + # 收集vnpy DB数据 + # interval='1m' — vnpy 4.x Interval.MINUTE硬约束，实际存储15分钟线 + exchange = "SSE" if prefix == "sh" else "SZSE" + for _, row in df_increment.iterrows(): + all_db_values.append(( + clean, exchange, str(row["day"]), + INTERVAL_MINUTE, # interval='1m' — vnpy 4.x Interval.MINUTE硬约束，实际存储15分钟线 + float(row.get("volume", 0)), float(row.get("amount", 0)), 0.0, + float(row.get("open", 0)), float(row.get("high", 0)), + float(row.get("low", 0)), float(row.get("close", 0)), + )) + stats["db_records"] += len(df_increment) + + done_set.add(code) + + if (i + 1) % 500 == 0: + logger.info("15min进度: %d/%d updated=%d failed=%d src=%s", + i + 1, len(todo), stats["updated"], stats["failed"], source) + save_progress("15min", done_set) + + # 写入本地DB + if all_db_values: + _write_local_db(local_conn, all_db_values, "15min") + + save_progress("15min", done_set) + logger.info("15min完成: %s", json.dumps(stats, ensure_ascii=False)) + return stats + + +# ======================== 本地vnpy DB写入（v2.0） ======================== + +def init_local_db() -> sqlite3.Connection: + """初始化本地vnpy DB：从NAS复制或创建新的""" + local_path = str(LOCAL_DB_PATH) + + if LOCAL_DB_PATH.exists(): + logger.info("使用已有本地DB: %s", local_path) + conn = sqlite3.connect(local_path, timeout=30) + return conn + + # 从NAS复制 + if VNPY_DB_PATH.exists(): + logger.info("从NAS复制DB到本地: %s → %s", VNPY_DB_PATH, local_path) + shutil.copy2(str(VNPY_DB_PATH), local_path) + conn = sqlite3.connect(local_path, timeout=30) + else: + logger.info("创建新本地DB: %s", local_path) + conn = sqlite3.connect(local_path, timeout=30) + c = conn.cursor() + c.execute("""CREATE TABLE IF NOT EXISTS dbbardata ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + symbol TEXT NOT NULL, + exchange TEXT NOT NULL, + datetime TEXT NOT NULL, + interval TEXT NOT NULL, + volume REAL DEFAULT 0, + turnover REAL DEFAULT 0, + open_interest REAL DEFAULT 0, + open_price REAL, + high_price REAL, + low_price REAL, + close_price REAL, + UNIQUE(symbol, exchange, datetime, interval) + )""") + c.execute("""CREATE TABLE IF NOT EXISTS dbbaroverview ( + symbol TEXT NOT NULL, + exchange TEXT NOT NULL, + interval TEXT NOT NULL, + count INTEGER, + start TEXT, + end TEXT, + UNIQUE(symbol, exchange, interval) + )""") + conn.commit() + + c = conn.cursor() + c.execute("PRAGMA journal_mode=WAL") + c.execute("PRAGMA synchronous=NORMAL") + conn.commit() + return conn + + +def _write_local_db(conn: sqlite3.Connection, values: list, label: str): + """批量写入本地vnpy DB""" + logger.info("写入本地DB [%s]: %d 条记录", label, len(values)) + c = conn.cursor() + for i in range(0, len(values), BATCH_SIZE): + c.executemany( + """INSERT OR REPLACE INTO dbbardata + (symbol,exchange,datetime,interval,volume,turnover,open_interest, + open_price,high_price,low_price,close_price) + VALUES (?,?,?,?,?,?,?,?,?,?,?)""", + values[i:i + BATCH_SIZE], + ) + conn.commit() + logger.info("✅ 本地DB [%s] 写入完成: %d条", label, len(values)) + + # 增量更新overview（v2.0：只更新本次涉及的symbol） + _update_overview_incremental(conn, values) + + +def _update_overview_incremental(conn: sqlite3.Connection, values: list): + """增量更新overview：只更新本次涉及的(symbol, exchange, interval)""" + c = conn.cursor() + affected = set((v[0], v[1], v[3]) for v in values) # symbol, exchange, interval + for sym, exc, ivl in affected: + c.execute( + """INSERT OR REPLACE INTO dbbaroverview (symbol, exchange, interval, count, start, end) + SELECT ?, ?, ?, + COUNT(*), + MIN(datetime), + MAX(datetime) + FROM dbbardata + WHERE symbol=? AND exchange=? AND interval=?""", + (sym, exc, ivl, sym, exc, ivl), + ) + conn.commit() + logger.info(" overview增量更新: %d 组", len(affected)) + + +def sync_db_to_nas(): + """本地DB → NAS mv原子重命名（v2.0）""" + if not LOCAL_DB_PATH.exists(): + logger.warning("本地DB不存在，跳过同步") + return + + local_size = LOCAL_DB_PATH.stat().st_size / 1024 / 1024 + logger.info("同步DB到NAS: %.1f MB", local_size) + + nas_path = str(VNPY_DB_PATH) + new_path = nas_path + ".new" + old_path = nas_path + ".old" + + # 1. 复制到NAS .new文件 + logger.info(" 复制到 %s", new_path) + shutil.copy2(str(LOCAL_DB_PATH), new_path) + + # 2. 原子重命名：old → old备份，current → old，new → current + # 在NAS同一文件系统上rename是原子的 + try: + if os.path.exists(old_path): + os.remove(old_path) + if os.path.exists(nas_path): + os.rename(nas_path, old_path) + os.rename(new_path, nas_path) + logger.info("✅ DB同步完成（mv原子重命名）") + # 注意：.old备份不立即清理，下次sync时清理（保留回退路径） + except Exception as e: + logger.error("❌ DB同步失败: %s，尝试恢复", e) + # 恢复：把old改回来 + if os.path.exists(old_path) and not os.path.exists(nas_path): + os.rename(old_path, nas_path) + logger.info(" 已恢复旧DB") + + +# ======================== 告警与报告 ======================== + +def check_failure_rate(stats: dict, label: str) -> bool: + """检查失败率""" + total = stats.get("updated", 0) + stats.get("failed", 0) + stats.get("skipped", 0) + failed = stats.get("failed", 0) + if total == 0: + return False + rate = failed / total + if rate > 0.05: + logger.error("❌ [%s] 失败率 %.1f%% (%d/%d)", label, rate * 100, failed, total) + return True + if stats.get("source_aborted"): + logger.error("❌ [%s] 源不可用导致终止", label) + return True + return False + + +# ======================== 主入口 ======================== + +def main(): + parser = argparse.ArgumentParser(description="全市场每日增量更新 v2.0") + parser.add_argument("--skip-daily", action="store_true", help="跳过日线更新") + parser.add_argument("--skip-15min", action="store_true", help="跳过15分钟线更新") + parser.add_argument("--fresh-db", action="store_true", help="强制从NAS重新复制DB") + args = parser.parse_args() + + if not nas_mounted(): + logger.error("❌ NAS未挂载，退出") + sys.exit(1) + + codes = get_all_codes() + logger.info("全市场股票数: %d", len(codes)) + logger.info("更新时间: %s", datetime.now().isoformat()) + logger.info("版本: v2.0 (多源fallback + 本地DB构建)") + + # 如果强制刷新或本地DB不存在，先删掉旧的 + if args.fresh_db and LOCAL_DB_PATH.exists(): + LOCAL_DB_PATH.unlink() + logger.info("已删除旧本地DB（--fresh-db）") + + # BaoStock login（如果可用） + if HAS_BAOSTOCK: + lg = bs.login() + logger.info("BaoStock login: %s", lg.error_msg) + + t_start = time.time() + report = {} + has_alert = False + + # DB备份 + rotate_db_backup() + + # 初始化本地DB + local_conn = init_local_db() + + try: + if not args.skip_daily: + report["daily"] = run_daily_update(codes, local_conn) + if check_failure_rate(report["daily"], "日线"): + has_alert = True + + if not args.skip_15min: + report["15min"] = run_15min_update(codes, local_conn) + if check_failure_rate(report["15min"], "15min"): + has_alert = True + finally: + # 关闭本地DB连接 + local_conn.close() + + # 同步DB到NAS + sync_db_to_nas() + + # BaoStock logout + if HAS_BAOSTOCK: + bs.logout() + logger.info("BaoStock logout") + + elapsed = time.time() - t_start + report["elapsed_sec"] = round(elapsed, 1) + report["has_alert"] = has_alert + + logger.info("=" * 60) + if has_alert: + logger.error("⚠️ 本次更新存在异常，请检查日志") + else: + logger.info("✅ 全部完成，耗时 %.1f 秒", elapsed) + logger.info(json.dumps(report, ensure_ascii=False, indent=2)) + + # 报告文件 + report_file = LOG_DIR / f"report_{datetime.now().strftime('%Y%m%d')}.json" + report_file.write_text(json.dumps(report, ensure_ascii=False, indent=2)) + + return report + + +if __name__ == "__main__": + main() diff --git a/scripts/data_platform/daily_update.sh b/scripts/data_platform/daily_update.sh new file mode 100755 index 0000000..79fe84a --- /dev/null +++ b/scripts/data_platform/daily_update.sh @@ -0,0 +1,26 @@ +#!/bin/bash +# 每日增量更新 wrapper +# 由 crontab 在每交易日15:30触发 + +set -e + +LOG_DIR="$HOME/.openclaw/sanguo_projects/sanguo_vnpy/data_platform/logs" +mkdir -p "$LOG_DIR" + +LOG_FILE="$LOG_DIR/update_$(date +%Y%m%d_%H%M%S).log" + +echo "=== $(date) 开始每日增量更新 ===" >> "$LOG_FILE" + +# 确保NAS已挂载 +if [ ! -d "/Volumes/stock/A股数据/日线数据/daily" ]; then + echo "ERROR: NAS未挂载，跳过更新" >> "$LOG_FILE" + exit 1 +fi + +cd "$HOME/.openclaw/sanguo_projects/sanguo_vnpy/data_platform" +python3 updater.py >> "$LOG_FILE" 2>&1 + +echo "=== $(date) 更新完成 ===" >> "$LOG_FILE" + +# 清理30天前的日志 +find "$LOG_DIR" -name "update_*.log" -mtime +30 -delete 2>/dev/null || true diff --git a/scripts/data_platform/daily_update_all.sh b/scripts/data_platform/daily_update_all.sh new file mode 100755 index 0000000..7b08aa9 --- /dev/null +++ b/scripts/data_platform/daily_update_all.sh @@ -0,0 +1,26 @@ +#!/bin/bash +# 每日全市场增量更新 - 日线 + 15分钟线 +# 由 OpenClaw cron 或 system crontab 在每交易日15:35触发 + +set -e + +# 确保NAS已挂载 +if [ ! -d "/Volumes/stock/A股数据" ]; then + echo "ERROR: NAS未挂载" >&2 + exit 1 +fi + +SCRIPT_DIR="$(cd "$(dirname "$0")" && pwd)" +LOG_DIR="/Volumes/stock/logs/daily_update" +mkdir -p "$LOG_DIR" + +echo "=== $(date) 开始每日增量更新 ===" + +cd "$SCRIPT_DIR" +python3 daily_all_update.py + +echo "=== $(date) 更新完成 ===" + +# 清理30天前的日志 +find "$LOG_DIR" -name "update_*.log" -mtime +30 -delete 2>/dev/null || true +find "$LOG_DIR" -name "report_*.json" -mtime +30 -delete 2>/dev/null || true diff --git a/scripts/data_platform/download_minute.py b/scripts/data_platform/download_minute.py new file mode 100644 index 0000000..3123aa2 --- /dev/null +++ b/scripts/data_platform/download_minute.py @@ -0,0 +1,333 @@ +#!/usr/bin/env python3 +""" +15分钟线数据下载脚本 + +数据源降级链： + 1. 新浪财经15分钟K线API（有真实amount，800条/次） + 2. 腾讯 minute/query + 聚合为15分钟（仅当天数据，amount为估算） + +功能： + - 支持HS300 / 全市场下载 + - 增量下载（追加新数据，不覆盖已有） + - 断点续传（JSON进度文件） + - 限频保护（0.3s间隔 + 重试） + - 与已有84只Parquet格式完全一致（7列，end-of-bar时间戳） + - 数据校验（价格>0, OHLC一致性） + +用法： + python3 download_minute.py --scope hs300 + python3 download_minute.py --scope all + python3 download_minute.py --codes 000001 600519 + python3 download_minute.py --scope hs300 --resume +""" + +import argparse +import json +import re +import sys +import time +import logging +import urllib.request +import urllib.error +from datetime import datetime +from pathlib import Path +from typing import Optional, List, Tuple + +import pandas as pd + +logging.basicConfig( + level=logging.INFO, + format="%(asctime)s %(levelname)s %(message)s", +) +logger = logging.getLogger(__name__) + +# --- 配置 --- +OUTPUT_DIR = Path("/Volumes/stock/minute_kline/15min") +PROGRESS_FILE = OUTPUT_DIR / "download_progress.json" +REQUEST_INTERVAL = 0.3 +MAX_RETRIES = 3 +CONSECUTIVE_FAIL_PAUSE = 60 +MAX_CONSECUTIVE_FAILS = 5 +HS300_FILE = Path("/Volumes/stock/A股数据/stock_info/hs300_constituents_latest.csv") +ALL_STOCKS_FILE = Path("/Volumes/stock/A股数据/stock_info/stock_basic_info_raw_20260326_113530.csv") + +HEADERS = {"User-Agent": "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7)"} + + +def _make_opener(): + """创建无代理opener，避免akshare代理污染""" + return urllib.request.build_opener(urllib.request.ProxyHandler({})) + + +# --- 新浪15分钟K线API（主源） --- +def try_sina_15min(symbol: str, datalen: int = 800) -> Optional[pd.DataFrame]: + """ + 新浪财经15分钟K线API + symbol: sz000001 或 sh600519 + datalen: 返回条数（最大约800） + 返回: DataFrame(day, open, high, low, close, volume, amount) 或 None + """ + url = ( + f"https://quotes.sina.cn/cn/api/jsonp_v2.php/var%20=min15_{symbol}=/" + f"CN_MarketDataService.getKLineData?symbol={symbol}&scale=15&ma=no&datalen={datalen}" + ) + try: + opener = _make_opener() + req = urllib.request.Request(url, headers=HEADERS) + with opener.open(req, timeout=15) as r: + raw = r.read().decode("utf-8", errors="replace") + m = re.search(r'\((\[.*\])\)', raw, re.DOTALL) + if not m: + return None + data = json.loads(m.group(1)) + if not data: + return None + df = pd.DataFrame(data) + # 确保列顺序 + cols = ["day", "open", "high", "low", "close", "volume", "amount"] + for c in cols: + if c not in df.columns: + return None + return df[cols] + except Exception as e: + logger.debug("新浪15min失败 %s: %s", symbol, e) + return None + + +# --- 腾讯 minute/query + 聚合（备源，仅当天） --- +def try_minute_query_aggregate(symbol: str, date: str) -> Optional[pd.DataFrame]: + """ + 腾讯minute/query API，返回1分钟线，聚合为15分钟线 + symbol: sz000001 + date: 20260502 + """ + url = f"http://web.ifzq.gtimg.cn/appstock/app/minute/query?code={symbol}" + try: + opener = _make_opener() + req = urllib.request.Request(url, headers=HEADERS) + with opener.open(req, timeout=10) as r: + data = json.loads(r.read()) + minute_data = data.get("data", {}).get(symbol, {}).get("data", {}).get("data", []) + if not minute_data: + return None + one_min = [] + for line in minute_data: + parts = line.split() + if len(parts) >= 4: + hhmm = parts[0] + one_min.append({ + "time": f"{date[:4]}-{date[4:6]}-{date[6:8]} {hhmm[:2]}:{hhmm[2:]}:00", + "price": float(parts[1]), + "vol": float(parts[2]), + "amount": float(parts[3]), + }) + if not one_min: + return None + return _aggregate_1m_to_15m(pd.DataFrame(one_min)) + except Exception as e: + logger.debug("minute_query失败 %s: %s", symbol, e) + return None + + +def _aggregate_1m_to_15m(df: pd.DataFrame) -> pd.DataFrame: + """1分钟线聚合为15分钟线（end-of-bar时间戳）""" + df["time"] = pd.to_datetime(df["time"]) + # end-of-bar对齐：已有84只数据用K线结束时间（09:45, 10:00...） + df["group"] = df["time"].dt.floor("15min") + pd.Timedelta(minutes=15) + + agg = df.groupby("group").agg( + open=("price", "first"), + high=("price", "max"), + low=("price", "min"), + close=("price", "last"), + volume=("vol", "sum"), + amount=("amount", "last"), + ).reset_index() + + return pd.DataFrame({ + "day": agg["group"].dt.strftime("%Y-%m-%d %H:%M:%S"), + "open": agg["open"], + "high": agg["high"], + "low": agg["low"], + "close": agg["close"], + "volume": agg["volume"].astype(str), + "amount": agg["amount"].astype(str), + }) + + +# --- 下载主流程 --- +def get_market_prefix(code: str) -> Tuple[str, str]: + code = re.sub(r"[^0-9]", "", code).zfill(6) + if code.startswith(("60", "68", "51")): + return "sh", code + return "sz", code + + +def download_single(code: str) -> Tuple[Optional[pd.DataFrame], str]: + """下载单只股票15分钟线，返回(df, source)""" + prefix, clean = get_market_prefix(code) + symbol = f"{prefix}{clean}" + + # 主源：新浪15分钟线 + df = try_sina_15min(symbol) + if df is not None and len(df) > 0: + return df, "sina_15min" + + # 备源：minute/query + 聚合 + today = datetime.now().strftime("%Y%m%d") + df = try_minute_query_aggregate(symbol, today) + if df is not None and len(df) > 0: + return df, "minute_query" + + return None, "failed" + + +def download_with_increment(code: str, output_dir: Path) -> Tuple[str, int]: + """增量下载单只股票""" + prefix, clean = get_market_prefix(code) + filename = f"{prefix}{clean}_15min.parquet" + parquet_path = output_dir / filename + + df_new, source = download_single(code) + if df_new is None: + return "failed", 0 + + # 数据校验 + for col in ["open", "high", "low", "close"]: + df_new[col] = pd.to_numeric(df_new[col], errors="coerce") + df_new["volume"] = pd.to_numeric(df_new["volume"], errors="coerce").fillna(0) + df_new["amount"] = pd.to_numeric(df_new["amount"], errors="coerce").fillna(0) + + # 价格>0 + bad_zero = (df_new[["close", "open"]] <= 0).any(axis=1) + if bad_zero.any(): + logger.warning("价格<=0 %s: %d条", code, bad_zero.sum()) + df_new = df_new[~bad_zero] + # OHLC一致性 + bad_ohlc = (df_new["high"] < df_new[["open", "close"]].max(axis=1)) | \ + (df_new["low"] > df_new[["open", "close"]].min(axis=1)) + if bad_ohlc.any(): + logger.warning("OHLC异常 %s: %d条", code, bad_ohlc.sum()) + df_new = df_new[~bad_ohlc] + if df_new.empty: + return "failed", 0 + + # 转回object类型与已有数据兼容 + df_new["volume"] = df_new["volume"].astype(str) + df_new["amount"] = df_new["amount"].astype(str) + + if parquet_path.exists(): + existing = pd.read_parquet(parquet_path) + combined = pd.concat([existing, df_new], ignore_index=True) + combined = combined.drop_duplicates(subset=["day"], keep="last") + combined = combined.sort_values("day").reset_index(drop=True) + else: + combined = df_new + + # 原子写入 + tmp_path = parquet_path.with_suffix(".tmp") + combined.to_parquet(tmp_path, index=False) + tmp_path.rename(parquet_path) + + return f"ok({source})", len(df_new) + + +# --- 断点续传 --- +def load_progress() -> dict: + if PROGRESS_FILE.exists(): + return json.loads(PROGRESS_FILE.read_text()) + return {"completed": [], "failed": [], "last_update": ""} + + +def save_progress(progress: dict): + progress["last_update"] = datetime.now().isoformat() + PROGRESS_FILE.write_text(json.dumps(progress, ensure_ascii=False, indent=2)) + + +# --- 股票列表 --- +def get_stock_list(scope: str) -> List[str]: + if scope == "hs300": + df = pd.read_csv(HS300_FILE) + for col in ["成分券代码", "代码", "code"]: + if col in df.columns: + return [str(c).zfill(6) for c in df[col].tolist()] + raise ValueError(f"HS300文件中找不到代码列，现有列: {list(df.columns)}") + if scope == "all": + df = pd.read_csv(ALL_STOCKS_FILE) + for col in ["代码", "code", "股票代码"]: + if col in df.columns: + return [str(c).zfill(6) for c in df[col].tolist()] + raise ValueError(f"全市场文件中找不到代码列，现有列: {list(df.columns)}") + raise ValueError(f"Unknown scope: {scope}") + + +# --- 主入口 --- +def main(): + parser = argparse.ArgumentParser(description="15分钟线数据下载") + parser.add_argument("--scope", choices=["hs300", "all"], help="下载范围") + parser.add_argument("--codes", nargs="+", help="指定股票代码") + parser.add_argument("--resume", action="store_true", help="断点续传") + parser.add_argument("--output-dir", default=str(OUTPUT_DIR), help="输出目录") + args = parser.parse_args() + + output_dir = Path(args.output_dir) + output_dir.mkdir(parents=True, exist_ok=True) + + if args.codes: + codes = args.codes + elif args.scope: + codes = get_stock_list(args.scope) + else: + parser.error("必须指定 --scope 或 --codes") + + progress = load_progress() if args.resume else {"completed": [], "failed": []} + skip_set = set(progress["completed"]) + todo = [c for c in codes if c not in skip_set] + logger.info("股票总数: %d, 已完成: %d, 待下载: %d", len(codes), len(skip_set), len(todo)) + + t_start = time.time() + ok_count = 0 + fail_count = 0 + consecutive_fails = 0 + + for i, code in enumerate(todo): + if i > 0: + time.sleep(REQUEST_INTERVAL) + + status = "failed" + rows = 0 + for attempt in range(MAX_RETRIES): + try: + status, rows = download_with_increment(code, output_dir) + if status != "failed": + break + except Exception as e: + logger.warning(" 重试 %d/%d: %s", attempt + 1, MAX_RETRIES, e) + time.sleep(1) + + if status != "failed": + ok_count += 1 + consecutive_fails = 0 + progress["completed"].append(code) + logger.info("[%d/%d] %s: %s (%d rows)", i + 1, len(todo), code, status, rows) + else: + fail_count += 1 + consecutive_fails += 1 + progress["failed"].append(code) + logger.warning("[%d/%d] %s: FAILED", i + 1, len(todo), code) + if consecutive_fails >= MAX_CONSECUTIVE_FAILS: + logger.error("连续失败 %d 次，暂停 %d 秒", consecutive_fails, CONSECUTIVE_FAIL_PAUSE) + time.sleep(CONSECUTIVE_FAIL_PAUSE) + consecutive_fails = 0 + + if (i + 1) % 50 == 0: + save_progress(progress) + + save_progress(progress) + elapsed = time.time() - t_start + logger.info("=" * 50) + logger.info("下载完成: 成功 %d, 失败 %d, 耗时 %.1f 秒", ok_count, fail_count, elapsed) + + +if __name__ == "__main__": + main() diff --git a/scripts/data_platform/fallback.py b/scripts/data_platform/fallback.py new file mode 100644 index 0000000..b8df5a1 --- /dev/null +++ b/scripts/data_platform/fallback.py @@ -0,0 +1,104 @@ +#!/usr/bin/env python3 +"""多源降级管理器 - 日线(akshare→腾讯) + 实时(新浪→东财→腾讯)""" +import pandas as pd +import urllib.request +import json +import logging +from datetime import datetime, timedelta +from typing import Optional + +logging.basicConfig(level=logging.INFO, format='%(asctime)s %(levelname)s %(message)s') +logger = logging.getLogger(__name__) + + +class FallbackManager: + def __init__(self): + self._source_used = "" + + def get_source_used(self) -> str: + return self._source_used + + def get_daily(self, symbol: str, start_date: str, end_date: str) -> pd.DataFrame: + """获取日线数据，降级链：akshare → 腾讯K线""" + # 1. akshare + try: + df = self._fetch_akshare_daily(symbol, start_date, end_date) + if df is not None and not df.empty: + self._source_used = "akshare" + return df + except Exception as e: + logger.warning(f"akshare日线失败 {symbol}: {e}") + + # 2. 腾讯K线 + try: + df = self._fetch_tencent_daily(symbol, start_date, end_date) + if df is not None and not df.empty: + self._source_used = "tencent_kline" + return df + except Exception as e: + logger.warning(f"腾讯K线失败 {symbol}: {e}") + + raise RuntimeError(f"所有日线数据源失败: {symbol} {start_date}~{end_date}") + + def get_realtime(self, symbol: str) -> dict: + """获取实时行情，降级链：新浪→东财→腾讯""" + from realtime import get_realtime_quote + result = get_realtime_quote(symbol) + self._source_used = result.get("source", "unknown") + return result + + def _fetch_akshare_daily(self, symbol: str, start_date: str, end_date: str) -> Optional[pd.DataFrame]: + import akshare as ak + code = symbol.replace("SH", "").replace("SZ", "").replace("sh", "").replace("sz", "") + s = start_date.replace("-", "") + e = end_date.replace("-", "") + df = ak.stock_zh_a_hist(symbol=code, period="daily", start_date=s, end_date=e, adjust="") + if df is None or df.empty: + return None + df = df.rename(columns={"日期": "date", "开盘": "open", "收盘": "close", + "最高": "high", "最低": "low", "成交量": "volume", + "成交额": "amount"}) + df["date"] = pd.to_datetime(df["date"]).dt.strftime("%Y-%m-%d") + for c in ["open", "high", "low", "close", "volume", "amount"]: + df[c] = pd.to_numeric(df[c], errors="coerce").fillna(0) + return df[["date", "open", "high", "low", "close", "volume", "amount"]] + + def _fetch_tencent_daily(self, symbol: str, start_date: str, end_date: str) -> Optional[pd.DataFrame]: + """腾讯K线API获取日线""" + code = symbol.replace("SH", "").replace("SZ", "").replace("sh", "").replace("sz", "") + if code.startswith(("6", "5", "1")): + prefix = "sh" + else: + prefix = "sz" + tq_symbol = f"{prefix}{code}" + + days = (datetime.strptime(end_date, "%Y-%m-%d") - datetime.strptime(start_date, "%Y-%m-%d")).days + 10 + url = f"https://web.ifzq.gtimg.cn/appstock/app/fqkline/get?param={tq_symbol},day,{start_date},,{days}," + try: + import urllib.request, json as _json + opener = urllib.request.build_opener(urllib.request.ProxyHandler({})) + req = urllib.request.Request(url, headers={"User-Agent": "Mozilla/5.0"}) + with opener.open(req, timeout=10) as r: + resp = _json.loads(r.read()) + d = resp.get("data") + if not isinstance(d, dict): + return None + klines = d.get(tq_symbol, {}).get("day", []) + if not klines: + return None + df = pd.DataFrame(klines) + ncols = len(df.columns) + if ncols >= 7: + df.columns = ["date", "open", "close", "high", "low", "volume", "amount"][:ncols] + else: + df.columns = ["date", "open", "close", "high", "low", "volume"][:ncols] + if "amount" not in df.columns: + df["amount"] = 0.0 + for c in ["open", "close", "high", "low", "volume", "amount"]: + df[c] = pd.to_numeric(df[c], errors="coerce").fillna(0) + df["date"] = pd.to_datetime(df["date"]).dt.strftime("%Y-%m-%d") + mask = (df["date"] >= start_date) & (df["date"] <= end_date) + return df.loc[mask, ["date", "open", "high", "low", "close", "volume", "amount"]] + except Exception as e: + logger.warning(f"腾讯K线请求失败: {e}") + return None diff --git a/scripts/data_platform/import_vnpy_daily.py b/scripts/data_platform/import_vnpy_daily.py new file mode 100644 index 0000000..8fb70be --- /dev/null +++ b/scripts/data_platform/import_vnpy_daily.py @@ -0,0 +1,172 @@ +#!/usr/bin/env python3 +""" +将NAS日线Parquet数据导入vnpy SQLite数据库 +用法: python3 import_vnpy_daily.py [--start-year 2013] [--dry-run] +""" +import sqlite3 +import pandas as pd +import os +import re +import sys +import time +from pathlib import Path + +DB_PATH = os.environ.get('VNPY_DB_PATH', '/tmp/quant_trading_import.db') +DAILY_DIR = '/Volumes/stock/A股数据/日线数据/daily/' + +BATCH_SIZE = 50000 # 每批插入行数 + + +def parse_filename(filename): + """sh600519_daily.parquet → ('600519', 'SSE')""" + m = re.match(r'(sh|sz)(\d{6})_daily\.parquet', filename) + if not m: + return None, None + prefix, code = m.groups() + exchange = 'SSE' if prefix == 'sh' else 'SZSE' + return code, exchange + + +def import_year(conn, year, dry_run=False): + """导入指定年份的所有日线数据""" + year_dir = Path(DAILY_DIR) / str(year) + if not year_dir.exists(): + print(f' ⚠️ {year} 目录不存在') + return 0, 0, 0 + + files = sorted(year_dir.glob('*.parquet')) + if not files: + print(f' ⚠️ {year} 无parquet文件') + return 0, 0, 0 + + c = conn.cursor() + imported_files = 0 + imported_rows = 0 + failed = 0 + + # 收集所有数据 + all_values = [] + + for f in files: + code, exchange = parse_filename(f.name) + if code is None: + failed += 1 + continue + + try: + df = pd.read_parquet(f, columns=['date', 'open', 'high', 'low', 'close', 'volume', 'amount']) + if df.empty: + continue + + for _, row in df.iterrows(): + all_values.append(( + code, + exchange, + str(row['date']), + 'd', + float(row['volume']) if pd.notna(row['volume']) else 0.0, + float(row['amount']) if pd.notna(row['amount']) else 0.0, + 0.0, # open_interest + float(row['open']) if pd.notna(row['open']) else 0.0, + float(row['high']) if pd.notna(row['high']) else 0.0, + float(row['low']) if pd.notna(row['low']) else 0.0, + float(row['close']) if pd.notna(row['close']) else 0.0, + )) + + imported_files += 1 + imported_rows += len(df) + + except Exception as e: + failed += 1 + if failed <= 3: + print(f' ❌ {f.name}: {e}') + + if dry_run: + print(f' [DRY RUN] Would insert {len(all_values)} rows from {imported_files} files') + return imported_files, imported_rows, failed + + # 批量插入 + if all_values: + for i in range(0, len(all_values), BATCH_SIZE): + batch = all_values[i:i+BATCH_SIZE] + c.executemany(''' + INSERT OR REPLACE INTO dbbardata + (symbol, exchange, datetime, interval, volume, turnover, open_interest, + open_price, high_price, low_price, close_price) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + ''', batch) + conn.commit() + + print(f' ✅ {year}: {imported_files} files, {imported_rows} rows inserted, {failed} failed') + return imported_files, imported_rows, failed + + +def update_overview(conn): + """更新 dbbaroverview 汇总表""" + c = conn.cursor() + c.execute(''' + INSERT OR REPLACE INTO dbbaroverview (symbol, exchange, interval, count, start, end) + SELECT symbol, exchange, interval, + COUNT(*) as count, + MIN(datetime) as start, + MAX(datetime) as end + FROM dbbardata + GROUP BY symbol, exchange, interval + ''') + conn.commit() + c.execute('SELECT COUNT(*) FROM dbbaroverview') + print(f' Overview: {c.fetchone()[0]} entries') + + +def main(): + start_year = 2013 # 默认从2013开始（2010-2012已导入） + dry_run = '--dry-run' in sys.argv + + for i, arg in enumerate(sys.argv): + if arg == '--start-year' and i+1 < len(sys.argv): + start_year = int(sys.argv[i+1]) + + print(f'=== vnpy Daily Import ===') + print(f'DB: {DB_PATH}') + print(f'Start year: {start_year}') + print(f'Dry run: {dry_run}') + print() + + conn = sqlite3.connect(DB_PATH, timeout=60) + + total_files = 0 + total_rows = 0 + total_failed = 0 + start_time = time.time() + + for year in range(start_year, 2027): + t0 = time.time() + f, r, fail = import_year(conn, year, dry_run) + t1 = time.time() + total_files += f + total_rows += r + total_failed += fail + print(f' ({t1-t0:.1f}s, total: {total_rows} rows)') + + if not dry_run: + print('\nUpdating overview...') + update_overview(conn) + + elapsed = time.time() - start_time + print(f'\n=== Summary ===') + print(f'Files: {total_files}') + print(f'Rows: {total_rows}') + print(f'Failed: {total_failed}') + print(f'Time: {elapsed:.1f}s ({elapsed/60:.1f}min)') + + # Final verification + c = conn.cursor() + c.execute('SELECT COUNT(*) FROM dbbardata') + print(f'DB total rows: {c.fetchone()[0]}') + + conn.close() + print('✅ Done') + + +if __name__ == '__main__': + main() diff --git a/scripts/data_platform/import_vnpy_daily_fast.py b/scripts/data_platform/import_vnpy_daily_fast.py new file mode 100644 index 0000000..f97caab --- /dev/null +++ b/scripts/data_platform/import_vnpy_daily_fast.py @@ -0,0 +1,126 @@ +#!/usr/bin/env python3 +""" +高效导入NAS日线Parquet到vnpy SQLite DB +用pandas向量化代替逐行迭代，速度快10x+ +""" +import sqlite3 +import pandas as pd +import numpy as np +import os +import re +import sys +import time +from pathlib import Path + +DB_PATH = '/tmp/quant_trading_import.db' +DAILY_DIR = '/Volumes/stock/A股数据/日线数据/daily/' + + +def parse_filename(filename): + m = re.match(r'(sh|sz)(\d{6})_daily\.parquet', filename) + if not m: + return None, None + prefix, code = m.groups() + return code, 'SSE' if prefix == 'sh' else 'SZSE' + + +def import_year(conn, year): + year_dir = Path(DAILY_DIR) / str(year) + if not year_dir.exists(): + return 0, 0 + + files = sorted(year_dir.glob('*.parquet')) + if not files: + return 0, 0 + + c = conn.cursor() + all_dfs = [] + + for f in files: + code, exchange = parse_filename(f.name) + if code is None: + continue + try: + df = pd.read_parquet(f, columns=['date', 'open', 'high', 'low', 'close', 'volume', 'amount']) + if df.empty: + continue + df['symbol'] = code + df['exchange'] = exchange + all_dfs.append(df) + except Exception: + pass + + if not all_dfs: + return 0, 0 + + combined = pd.concat(all_dfs, ignore_index=True) + + # Vectorized conversion + combined['datetime'] = combined['date'].astype(str) + combined['interval'] = 'd' + combined['open_interest'] = 0.0 + combined = combined.rename(columns={ + 'open': 'open_price', 'high': 'high_price', + 'low': 'low_price', 'close': 'close_price', 'amount': 'turnover' + }) + + # Fill NaN + for col in ['volume', 'turnover', 'open_price', 'high_price', 'low_price', 'close_price']: + combined[col] = combined[col].fillna(0.0).astype(float) + + values = combined[['symbol','exchange','datetime','interval','volume','turnover', + 'open_interest','open_price','high_price','low_price','close_price' + ]].values.tolist() + + # Batch insert + BATCH = 50000 + for i in range(0, len(values), BATCH): + c.executemany('''INSERT OR REPLACE INTO dbbardata + (symbol,exchange,datetime,interval,volume,turnover,open_interest, + open_price,high_price,low_price,close_price) + VALUES (?,?,?,?,?,?,?,?,?,?,?)''', values[i:i+BATCH]) + conn.commit() + + return len(all_dfs), len(combined) + + +def main(): + start_year = 2017 + for i, arg in enumerate(sys.argv): + if arg == '--start-year' and i+1 < len(sys.argv): + start_year = int(sys.argv[i+1]) + + print(f'Importing from {start_year} to local DB: {DB_PATH}') + + conn = sqlite3.connect(DB_PATH) + total_rows = 0 + t_start = time.time() + + for year in range(start_year, 2027): + t0 = time.time() + files, rows = import_year(conn, year) + t1 = time.time() + total_rows += rows + print(f'{year}: {files} files, {rows} rows ({t1-t0:.1f}s) total={total_rows}') + + elapsed = time.time() - t_start + + # Update overview + c = conn.cursor() + c.execute('''INSERT OR REPLACE INTO dbbaroverview (symbol,exchange,interval,count,start,end) + SELECT symbol,exchange,interval,COUNT(*),MIN(datetime),MAX(datetime) + FROM dbbardata GROUP BY symbol,exchange,interval''') + conn.commit() + + c.execute('SELECT COUNT(*) FROM dbbardata') + final = c.fetchone()[0] + c.execute('SELECT COUNT(*) FROM dbbaroverview') + overview = c.fetchone()[0] + + print(f'\nDone in {elapsed:.1f}s ({elapsed/60:.1f}min)') + print(f'Total rows: {final}, Overview entries: {overview}') + conn.close() + + +if __name__ == '__main__': + main() diff --git a/scripts/data_platform/import_vnpy_minute.py b/scripts/data_platform/import_vnpy_minute.py new file mode 100644 index 0000000..a0033dd --- /dev/null +++ b/scripts/data_platform/import_vnpy_minute.py @@ -0,0 +1,205 @@ +#!/usr/bin/env python3 +""" +导入15分钟线Parquet到vnpy SQLite DB + +复用P1导入逻辑（pandas向量化+批量INSERT OR REPLACE） +与 import_vnpy_daily_fast.py 的区别： + - interval = '15m'（而非 'd'） + - datetime已是 "YYYY-MM-DD HH:MM:SS" 格式 + - 从单个parquet文件导入（非按年份目录） + - volume和amount是object类型需转float + +用法： + python3 import_vnpy_minute.py --scope hs300 + python3 import_vnpy_minute.py --scope all + python3 import_vnpy_minute.py --codes 000001 600519 + python3 import_vnpy_minute.py --db /path/to/quant_trading.db +""" + +import sqlite3 +import re +import sys +import time +import argparse +import logging +from pathlib import Path + +import pandas as pd + +logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s") +logger = logging.getLogger(__name__) + +# --- 配置 --- +MINUTE_DIR = Path("/Volumes/stock/minute_kline/15min") +DB_PATH = "/tmp/quant_trading_import.db" +INTERVAL = "1m" # vnpy 4.x Interval.MINUTE.value = '1m', 统一用1m存储分钟线 +BATCH_SIZE = 50000 + +HS300_FILE = Path("/Volumes/stock/A股数据/stock_info/hs300_constituents_latest.csv") +ALL_STOCKS_FILE = Path("/Volumes/stock/A股数据/stock_info/stock_basic_info_raw_20260326_113530.csv") + + +def parse_filename(filename: str): + """解析文件名: sz000001_15min.parquet -> (code, exchange)""" + m = re.match(r"(sh|sz)(\d{6})_15min\.parquet", filename) + if not m: + return None, None + prefix, code = m.groups() + return code, "SSE" if prefix == "sh" else "SZSE" + + +def import_file(conn, filepath: Path) -> int: + """导入单个Parquet文件，返回导入行数""" + code, exchange = parse_filename(filepath.name) + if code is None: + return 0 + + try: + df = pd.read_parquet(filepath) + if df.empty: + return 0 + except Exception as e: + logger.warning("读取失败 %s: %s", filepath.name, e) + return 0 + + # 构建导入数据 + df = df.rename(columns={ + "open": "open_price", + "high": "high_price", + "low": "low_price", + "close": "close_price", + "amount": "turnover", + }) + + df["symbol"] = code + df["exchange"] = exchange + df["interval"] = INTERVAL + df["open_interest"] = 0.0 + df["datetime"] = df["day"].astype(str) + + # 类型转换（volume/amount可能是object） + for col in ["volume", "turnover", "open_price", "high_price", "low_price", "close_price"]: + df[col] = pd.to_numeric(df[col], errors="coerce") + # 丢弃NaN行（价格/成交量为NaN说明原始数据异常） + na_before = len(df) + df = df.dropna(subset=["open_price", "close_price"]) + if len(df) < na_before: + logger.warning("%s 丢弃 %d 条NaN行", filepath.name, na_before - len(df)) + df["volume"] = df["volume"].fillna(0.0) + df["turnover"] = df["turnover"].fillna(0.0) + + values = df[[ + "symbol", "exchange", "datetime", "interval", "volume", "turnover", + "open_interest", "open_price", "high_price", "low_price", "close_price" + ]].values.tolist() + + c = conn.cursor() + for i in range(0, len(values), BATCH_SIZE): + c.executemany( + """INSERT OR REPLACE INTO dbbardata + (symbol, exchange, datetime, interval, volume, turnover, open_interest, + open_price, high_price, low_price, close_price) + VALUES (?,?,?,?,?,?,?,?,?,?,?)""", + values[i : i + BATCH_SIZE], + ) + conn.commit() + return len(values) + + +def get_stock_list(scope: str): + """获取股票代码列表""" + if scope == "hs300": + df = pd.read_csv(HS300_FILE) + for col in ["成分券代码", "代码", "code"]: + if col in df.columns: + return [str(c).zfill(6) for c in df[col].tolist()] + raise ValueError(f"HS300文件找不到代码列: {list(df.columns)}") + + if scope == "all": + df = pd.read_csv(ALL_STOCKS_FILE) + for col in ["代码", "code", "股票代码"]: + if col in df.columns: + return [str(c).zfill(6) for c in df[col].tolist()] + raise ValueError(f"全市场文件找不到代码列: {list(df.columns)}") + + raise ValueError(f"Unknown scope: {scope}") + + +def main(): + parser = argparse.ArgumentParser(description="导入15分钟线到vnpy DB") + parser.add_argument("--scope", choices=["hs300", "all"], help="导入范围") + parser.add_argument("--codes", nargs="+", help="指定股票代码") + parser.add_argument("--db", default=DB_PATH, help="SQLite DB路径") + parser.add_argument("--minute-dir", default=str(MINUTE_DIR), help="分钟线Parquet目录") + args = parser.parse_args() + + minute_dir = Path(args.minute_dir) + if not minute_dir.exists(): + logger.error("分钟线目录不存在: %s", minute_dir) + sys.exit(1) + + # 获取代码列表 + if args.codes: + codes = set(args.codes) + elif args.scope: + codes = set(get_stock_list(args.scope)) + else: + parser.error("必须指定 --scope 或 --codes") + + # 匹配文件 + files = sorted(minute_dir.glob("*_15min.parquet")) + matched = [] + for f in files: + code, _ = parse_filename(f.name) + if code and code in codes: + matched.append(f) + + logger.info("匹配文件: %d / %d", len(matched), len(files)) + + if not matched: + logger.error("没有匹配的Parquet文件") + sys.exit(1) + + # 导入 + conn = sqlite3.connect(args.db) + total_rows = 0 + t_start = time.time() + + for i, f in enumerate(matched): + t0 = time.time() + rows = import_file(conn, f) + t1 = time.time() + total_rows += rows + logger.info("[%d/%d] %s: %d rows (%.1fs) total=%d", i + 1, len(matched), f.name, rows, t1 - t0, total_rows) + + # 更新 overview + c = conn.cursor() + c.execute( + """INSERT OR REPLACE INTO dbbaroverview (symbol, exchange, interval, count, start, end) + SELECT symbol, exchange, interval, COUNT(*), MIN(datetime), MAX(datetime) + FROM dbbardata GROUP BY symbol, exchange, interval""" + ) + conn.commit() + + # 统计 + c.execute("SELECT COUNT(*) FROM dbbardata WHERE interval = ?", (INTERVAL,)) + minute_rows = c.fetchone()[0] + c.execute("SELECT COUNT(*) FROM dbbardata") + all_rows = c.fetchone()[0] + c.execute("SELECT COUNT(*) FROM dbbaroverview") + overview_count = c.fetchone()[0] + + elapsed = time.time() - t_start + conn.close() + + logger.info("=" * 50) + logger.info("导入完成") + logger.info("15分钟线: %d 行", minute_rows) + logger.info("总数据量: %d 行 (含日线)", all_rows) + logger.info("Overview: %d 条", overview_count) + logger.info("耗时: %.1f 秒 (%.1f 分钟)", elapsed, elapsed / 60) + logger.info("DB路径: %s", args.db) + + +if __name__ == "__main__": + main() diff --git a/scripts/data_platform/realtime.py b/scripts/data_platform/realtime.py new file mode 100644 index 0000000..29152ee --- /dev/null +++ b/scripts/data_platform/realtime.py @@ -0,0 +1,159 @@ +#!/usr/bin/env python3 +"""实时行情三源降级 - 新浪→东财→腾讯""" +import urllib.request +import json +import re +import logging +from datetime import datetime +from typing import Optional + +logging.basicConfig(level=logging.INFO, format='%(asctime)s %(levelname)s %(message)s') +logger = logging.getLogger(__name__) + +HEADERS_SINA = { + "User-Agent": "Mozilla/5.0", "Referer": "https://finance.sina.com.cn", + "Accept-Language": "zh-CN,zh;q=0.9" +} +HEADERS_EM = {"Referer": "https://www.eastmoney.com"} +FETCHED_AT = datetime.now().strftime("%Y-%m-%d %H:%M:%S") + + +def _fetch_url(url: str, headers: dict = None, timeout: int = 10) -> str: + req = urllib.request.Request(url, headers=headers or {}) + try: + with urllib.request.urlopen(req, timeout=timeout) as r: + charset = "gbk" if "sina" in url or "sinajs" in url else "utf-8" + return r.read().decode(charset, errors="replace") + except Exception: + return "" + + +def _parse_sina(raw: str, symbol: str) -> Optional[dict]: + m = re.search(r'"([^"]*)"', raw) + if not m: + return None + parts = m.group(1).split(",") + if len(parts) < 32: + return None + try: + prev_close = float(parts[2]) if parts[2] else 0 + current = float(parts[3]) if parts[3] else 0 + return { + "symbol": symbol, "name": parts[0], + "current": round(current, 2), "prev_close": round(prev_close, 2), + "open": round(float(parts[1]), 2) if parts[1] else 0, + "high": round(float(parts[4]), 2) if parts[4] else 0, + "low": round(float(parts[5]), 2) if parts[5] else 0, + "volume": int(float(parts[8])) if parts[8] else 0, + "amount": round(float(parts[9]), 2) if parts[9] else 0, + "timestamp": f"{parts[30]} {parts[31]}" if len(parts) > 31 else "", + "source": "sina", "fetched_at": FETCHED_AT, + } + except (ValueError, IndexError): + return None + + +def _parse_tencent(raw: str, symbol: str) -> Optional[dict]: + m = re.search(r'"([^"]*)"', raw) + if not m: + return None + fields = m.group(1).split("~") + if len(fields) < 35: + return None + try: + current = float(fields[3]) + prev_close = float(fields[4]) + if current <= 0: + return None + return { + "symbol": symbol, "name": fields[1], + "current": round(current, 2), "prev_close": round(prev_close, 2), + "open": round(float(fields[5]), 2), + "high": round(float(fields[33]), 2) if fields[33] else 0, + "low": round(float(fields[34]), 2) if fields[34] else 0, + "volume": int(float(fields[6])) if fields[6] else 0, + "amount": round(float(fields[37]) * 10000, 2) if fields[37] else 0, + "timestamp": fields[30][:8] + " " + fields[30][8:] if fields[30] else "", + "source": "tencent", "fetched_at": FETCHED_AT, + } + except (ValueError, IndexError): + return None + + +def _parse_eastmoney(raw: str, symbol: str) -> Optional[dict]: + try: + obj = json.loads(raw) + d = obj.get("data", {}) or {} + if not d.get("f43"): + return None + return { + "symbol": symbol, "name": d.get("f58", ""), + "current": round(d["f43"] / 100, 2), + "prev_close": round(d["f60"] / 100, 2), + "open": round(d["f46"] / 100, 2), + "high": round(d["f44"] / 100, 2), + "low": round(d["f45"] / 100, 2), + "volume": d.get("f47", 0), + "amount": round(d.get("f48", 0) / 1e8, 2), + "timestamp": "", + "source": "eastmoney", "fetched_at": FETCHED_AT, + } + except Exception: + return None + + +def _get_prefix(code: str) -> str: + code = re.sub(r"[^0-9]", "", code) + if code.startswith(("60", "68", "51", "58", "11")): + return "sh", code + return "sz", code + + +def _em_secid(code: str) -> str: + code = re.sub(r"[^0-9]", "", code) + m = 1 if code.startswith(("60", "68")) else 0 + return f"{m}.{code}" + + +def get_realtime_quote(code: str) -> dict: + """获取实时行情，三源降级：新浪→东财→腾讯""" + prefix, clean = _get_prefix(code) + symbol = f"{prefix}{clean}" + + # 1. 新浪 + raw = _fetch_url(f"http://hq.sinajs.cn/list={symbol}", HEADERS_SINA) + if raw: + data = _parse_sina(raw, symbol) + if data and data["current"] > 0: + logger.info(f"新浪成功: {symbol} = {data['current']}") + return data + + # 2. 东财 + secid = _em_secid(code) + raw = _fetch_url( + f"http://push2.eastmoney.com/api/qt/stock/get?secid={secid}" + f"&fields=f43,f44,f45,f46,f47,f48,f57,f58,f60,f169,f170", + HEADERS_EM) + if raw: + data = _parse_eastmoney(raw, symbol) + if data and data["current"] > 0: + logger.info(f"东财成功: {symbol} = {data['current']}") + return data + + # 3. 腾讯 + raw = _fetch_url(f"http://qt.gtimg.cn/q={symbol}") + if raw: + data = _parse_tencent(raw, symbol) + if data and data["current"] > 0: + logger.info(f"腾讯成功: {symbol} = {data['current']}") + return data + + return {"error": f"所有数据源均无法获取 {code}", "symbol": symbol, "fetched_at": FETCHED_AT} + + +if __name__ == "__main__": + import sys + code = sys.argv[1] if len(sys.argv) > 1 else "600519" + result = get_realtime_quote(code) + for k, v in result.items(): + print(f" {k}: {v}") diff --git a/scripts/data_platform/run_daily_update.sh b/scripts/data_platform/run_daily_update.sh new file mode 100755 index 0000000..c2e042d --- /dev/null +++ b/scripts/data_platform/run_daily_update.sh @@ -0,0 +1,36 @@ +#!/bin/bash +# v2 数据下载包装（SSH 模式：rsync NAS↔本地 + 跑 v1 daily_all_update） +# 不依赖 SMB 挂载，用 sanguo-nas SSH key 免密。 +# 用法：./run_daily_update.sh [--skip-daily|--skip-15min] +# env: +# STOCK_LIMIT=N 限制股票数（验证用，默认 0=全市场） +# SKIP_PULL=1 跳过 rsync 拉（验证用；生产必须拉，v1 才能增量） +set -e +NAS=sanguo-nas +LOCAL=${STOCK_MOUNT:-/tmp/stock_dl} +export STOCK_MOUNT=$LOCAL +SCRIPT_DIR="$(cd "$(dirname "$0")" && pwd)" + +mkdir -p "$LOCAL/logs/daily_update/progress" \ + "$LOCAL/A股数据/日线数据/daily" \ + "$LOCAL/minute_kline/15min" \ + "$LOCAL/A股数据/stock_info" \ + "$LOCAL/sanguo_vnpy/data" + +if [ "${SKIP_PULL:-0}" != "1" ]; then + echo "=== 1. rsync 拉 NAS 现有 → 本地（增量，首次慢后续快）===" + rsync -az "$NAS:/volume1/stock/A股数据/stock_info/" "$LOCAL/A股数据/stock_info/" || echo " stock_info pull warn" + rsync -az "$NAS:/volume1/stock/A股数据/日线数据/daily/" "$LOCAL/A股数据/日线数据/daily/" || echo " daily pull warn" + rsync -az "$NAS:/volume1/stock/minute_kline/15min/" "$LOCAL/minute_kline/15min/" || echo " 15min pull warn" + rsync -az "$NAS:/volume1/stock/sanguo_vnpy/data/" "$LOCAL/sanguo_vnpy/data/" || echo " vnpy data pull warn" +fi + +echo "=== 2. 跑 v1 增量更新（STOCK_MOUNT=$LOCAL STOCK_LIMIT=${STOCK_LIMIT:-0}）===" +cd "$SCRIPT_DIR" +python3 daily_all_update.py "$@" + +echo "=== 3. rsync 推 本地 → NAS ===" +rsync -az "$LOCAL/A股数据/日线数据/daily/" "$NAS:/volume1/stock/A股数据/日线数据/daily/" +rsync -az "$LOCAL/minute_kline/15min/" "$NAS:/volume1/stock/minute_kline/15min/" +rsync -az "$LOCAL/sanguo_vnpy/data/" "$NAS:/volume1/stock/sanguo_vnpy/data/" +echo "=== 完成 $(date) ===" diff --git a/scripts/data_platform/updater.py b/scripts/data_platform/updater.py new file mode 100644 index 0000000..a002026 --- /dev/null +++ b/scripts/data_platform/updater.py @@ -0,0 +1,196 @@ +#!/usr/bin/env python3 +"""增量更新 - 补vnpy DB增量数据（腾讯K线主源） +Parquet双写暂缓（需要处理date类型兼容），优先补DB数据 +""" +import os +import re +import sys +import json +import sqlite3 +import shutil +import logging +import time +import pandas as pd +from pathlib import Path +from datetime import datetime + +logging.basicConfig(level=logging.INFO, format='%(asctime)s %(levelname)s %(message)s') +logger = logging.getLogger(__name__) + +DAILY_DIR = "/Volumes/stock/A股数据/日线数据/daily/" +VNPY_DB_PATH = "/Volumes/stock/sanguo_vnpy/data/quant_trading.db" +LOCAL_DB_TMP = "/tmp/quant_trading_updater.db" +BATCH_SIZE = 50000 + + +def parse_filename(filename): + m = re.match(r'(sh|sz)(\d{6})_daily\.parquet', filename) + if not m: + return None, None + return m.group(2), 'SSE' if m.group(1) == 'sh' else 'SZSE' + + +def get_all_symbols(): + """扫描最新年份目录获取所有股票代码""" + latest_year = max(d.name for d in Path(DAILY_DIR).iterdir() if d.is_dir() and d.name.isdigit()) + symbols = [] + for f in (Path(DAILY_DIR) / latest_year).glob('*.parquet'): + code, exchange = parse_filename(f.name) + if code: + symbols.append((code, exchange)) + return symbols + + +def get_last_date(code: str, exchange: str) -> str: + """获取某只股票在NAS Parquet中的最后日期""" + prefix = 'sh' if exchange == 'SSE' else 'sz' + for year_dir in sorted(Path(DAILY_DIR).iterdir(), reverse=True): + if not year_dir.is_dir() or not year_dir.name.isdigit(): + continue + fpath = year_dir / f"{prefix}{code}_daily.parquet" + if fpath.exists(): + try: + df = pd.read_parquet(fpath, columns=['date']) + if not df.empty: + last = df['date'].max() + return str(last)[:10] + except Exception: + pass + return "" + + +def fetch_tencent_daily(code: str, start_date: str, end_date: str): + """腾讯K线API获取日线增量数据""" + import urllib.request + import json as _json + + prefix = 'sh' if code.startswith(('6', '5', '1')) else 'sz' + tq = f"{prefix}{code}" + days = (pd.Timestamp(end_date) - pd.Timestamp(start_date)).days + 10 + url = f"https://web.ifzq.gtimg.cn/appstock/app/fqkline/get?param={tq},day,{start_date},,{days}," + + proxy_handler = urllib.request.ProxyHandler({}) + opener = urllib.request.build_opener(proxy_handler) + req = urllib.request.Request(url, headers={"User-Agent": "Mozilla/5.0"}) + resp = opener.open(req, timeout=10) + raw = resp.read().decode('utf-8', errors='replace') + data = _json.loads(raw) + d = data.get('data') + if not isinstance(d, dict): + return None + klines = d.get(tq, {}).get('day', []) + if not klines: + return None + + # kline format: [date, open, close, high, low, volume] or 7 cols with amount + df = pd.DataFrame(klines) + ncols = len(df.columns) + if ncols >= 7: + df.columns = ['date', 'open', 'close', 'high', 'low', 'volume', 'amount'][:ncols] + else: + df.columns = ['date', 'open', 'close', 'high', 'low', 'volume'][:ncols] + if 'amount' not in df.columns: + df['amount'] = 0.0 + for c in ['open', 'close', 'high', 'low', 'volume', 'amount']: + df[c] = pd.to_numeric(df[c], errors='coerce').fillna(0) + df['date'] = pd.to_datetime(df['date']).dt.strftime('%Y-%m-%d') + mask = (df['date'] >= start_date) & (df['date'] <= end_date) + result = df.loc[mask, ['date', 'open', 'high', 'low', 'close', 'volume', 'amount']] + return result if not result.empty else None + + +def main(): + today = datetime.now().strftime("%Y-%m-%d") + + logger.info(f"=== vnpy DB增量更新开始 {today} ===") + + # 获取所有股票 + symbols = get_all_symbols() + logger.info(f"扫描到 {len(symbols)} 只股票") + + updated = 0 + skipped = 0 + failed = 0 + new_records = 0 + all_db_values = [] + + for i, (code, exchange) in enumerate(symbols): + last_date = get_last_date(code, exchange) + if not last_date: + skipped += 1 + continue + + next_day = (pd.Timestamp(last_date) + pd.Timedelta(days=1)).strftime("%Y-%m-%d") + if next_day > today: + skipped += 1 + continue + + data = fetch_tencent_daily(code, next_day, today) + if data is None or data.empty: + skipped += 1 + continue + + # 简单校验 + if (data[['open', 'high', 'low', 'close']] <= 0).any().any(): + logger.warning(f"{code} 有非正价格，跳过") + failed += 1 + continue + + # 收集vnpy DB数据 + for _, row in data.iterrows(): + all_db_values.append(( + code, exchange, str(row['date']), 'd', + float(row.get('volume', 0)), float(row.get('amount', 0)), 0.0, + float(row.get('open', 0)), float(row.get('high', 0)), + float(row.get('low', 0)), float(row.get('close', 0)), + )) + + new_records += len(data) + updated += 1 + + if (i + 1) % 500 == 0: + logger.info(f"进度: {i+1}/{len(symbols)} updated={updated} skipped={skipped} failed={failed} records={new_records}") + + time.sleep(0.3) + + # 写vnpy DB - 直接在NAS上追加增量（避免1.4GB全量copy） + if all_db_values: + logger.info(f"写入vnpy DB: {len(all_db_values)} 条记录") + try: + # 方式1: 直接在NAS DB上INSERT OR REPLACE（增量数据量小，SMB可接受） + conn = sqlite3.connect(VNPY_DB_PATH, timeout=120) + c = conn.cursor() + c.execute("PRAGMA journal_mode=WAL") + for j in range(0, len(all_db_values), BATCH_SIZE): + c.executemany('''INSERT OR REPLACE INTO dbbardata + (symbol,exchange,datetime,interval,volume,turnover,open_interest, + open_price,high_price,low_price,close_price) + VALUES (?,?,?,?,?,?,?,?,?,?,?)''', all_db_values[j:j+BATCH_SIZE]) + conn.commit() + logger.info(f" 写入批次 {j//BATCH_SIZE + 1}/{(len(all_db_values)-1)//BATCH_SIZE + 1}") + + # 重建overview + c.execute('''INSERT OR REPLACE INTO dbbaroverview (symbol,exchange,interval,count,start,end) + SELECT symbol,exchange,interval,COUNT(*),MIN(datetime),MAX(datetime) + FROM dbbardata GROUP BY symbol,exchange,interval''') + conn.commit() + conn.close() + logger.info("✅ vnpy DB更新完成") + except Exception as e: + logger.error(f"❌ vnpy DB更新失败: {e}") + + report = { + "date": today, + "total_symbols": len(symbols), + "updated": updated, + "skipped": skipped, + "failed": failed, + "new_records": new_records, + } + logger.info(f"=== 更新完成 ===") + logger.info(json.dumps(report, ensure_ascii=False, indent=2)) + return report + + +if __name__ == "__main__": + main() diff --git a/scripts/data_platform/validator.py b/scripts/data_platform/validator.py new file mode 100644 index 0000000..55581f0 --- /dev/null +++ b/scripts/data_platform/validator.py @@ -0,0 +1,110 @@ +#!/usr/bin/env python3 +"""数据校验层 - V1 7条fatal规则""" +import pandas as pd +from datetime import datetime +from typing import List, Tuple + + +class ValidationResult: + def __init__(self): + self.passed = True + self.fatal_errors: List[str] = [] + self.warnings: List[str] = [] + self.checked_rows = 0 + self.failed_rows = 0 + + def __repr__(self): + return (f"ValidationResult(passed={self.passed}, " + f"fatal={len(self.fatal_errors)}, warnings={len(self.warnings)}, " + f"rows={self.checked_rows}, failed={self.failed_rows})") + + def to_dict(self): + return { + "passed": self.passed, + "fatal_errors": self.fatal_errors, + "warnings": self.warnings, + "checked_rows": self.checked_rows, + "failed_rows": self.failed_rows, + } + + +class DataValidator: + """数据校验器 - V1 7条fatal规则""" + + def validate(self, df: pd.DataFrame, data_type: str = "daily") -> ValidationResult: + result = ValidationResult() + if df is None or df.empty: + result.fatal_errors.append("数据为空") + result.passed = False + return result + result.checked_rows = len(df) + + if data_type == "daily": + self._validate_daily(df, result) + elif data_type == "realtime": + self._validate_realtime(df, result) + return result + + def validate_realtime_dict(self, data: dict) -> ValidationResult: + """校验单条实时行情""" + result = ValidationResult() + result.checked_rows = 1 + errors = [] + # R1: 价格>0 + if not data or data.get("current", 0) <= 0: + errors.append("R1: current价格<=0") + if data.get("prev_close", 0) <= 0: + errors.append("R1: prev_close<=0") + # R7: 必须携带source和fetched_at + if not data.get("source"): + errors.append("R7: 缺少source字段") + if not data.get("fetched_at"): + errors.append("R7: 缺少fetched_at字段") + if errors: + result.fatal_errors = errors + result.passed = False + result.failed_rows = 1 + return result + + def _validate_daily(self, df: pd.DataFrame, result: ValidationResult): + today = datetime.now().strftime("%Y-%m-%d") + + for idx, row in df.iterrows(): + row_errors = [] + # D1: 价格>0 + for col in ["close", "open", "high", "low"]: + val = row.get(col, 0) + if pd.isna(val) or float(val) <= 0: + row_errors.append(f"D1: {col}<=0 (row {idx})") + break + + # D2: OHLC一致性 + o, h, l, c = float(row.get("open", 0)), float(row.get("high", 0)), \ + float(row.get("low", 0)), float(row.get("close", 0)) + if o > 0 and c > 0: + if h < max(o, c) or l > min(o, c): + row_errors.append(f"D2: OHLC不一致 (row {idx}, o={o} h={h} l={l} c={c})") + + # D3: volume >= 0 + vol = row.get("volume", 0) + if pd.notna(vol) and float(vol) < 0: + row_errors.append(f"D3: volume<0 (row {idx})") + + # D7: 非未来日期 + dt = str(row.get("date", row.get("datetime", "")))[:10] + if dt > today: + row_errors.append(f"D7: 未来日期 {dt} (row {idx})") + + if row_errors: + result.fatal_errors.extend(row_errors) + result.failed_rows += 1 + + # D6: 日期不重复 (check after all rows) + date_col = "date" if "date" in df.columns else "datetime" + if date_col in df.columns: + dupes = df[df.duplicated(subset=[date_col], keep=False)] + if not dupes.empty and len(df) > 1: + result.fatal_errors.append(f"D6: {len(dupes)}条重复日期") + + if result.fatal_errors: + result.passed = False