feat(data): v2 数据下载(SSH模式 STOCK_MOUNT+rsync) + launchd定时 + 文档

- v1 data_platform 脚本复制到 v2/scripts/data_platform/ 维护
- daily_all_update.py: 路径用 STOCK_MOUNT env(SSH模式) + STOCK_LIMIT(验证)
- run_daily_update.sh: rsync拉NAS→本地 + v1增量 + 推NAS(SSH key免密,不挂载)
- launchd com.sanguo.data-update 每日15:30(替代crontab,macOS FDA限制)
- 验证: STOCK_LIMIT=2 updated=2 records=24 拉到当天
- 文档 docs/deployment/data-download.md

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）。

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()

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

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

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()

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

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()

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()

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()

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}")

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) ==="

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()

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