claude_dev
a372544045
- raw_redownload 加断点续传(exists/skip已存在, 扩范围重下覆盖) - baostock_download: 15min双源(qfq+raw)下载器, NAS容器跑, 限速防封 - full_deploy_5yr.sh: 无人值守日线双源pipeline(qfq→raw→rsync→验证) - verify_dual_source: 容器内双源部署验证(fetch_day/iter_bars/除权日) 实测: 日线双源5yr 29600文件×2(rsync NAS), 浦发除权日 raw-5.9%/qfq-0.7%, 容器内 fetch_day/iter_bars/全市场抽检5只全通过. 15min沪深300 baostock限流121只.
65 lines
2.7 KiB
Python
65 lines
2.7 KiB
Python
#!/usr/bin/env python3
|
||
"""双源（raw/qfq）部署验证：容器内读 NAS data_platform.yaml 配置的双源，
|
||
确认 fetch_day（C-S3 实走接口）+ iter_bars 在 raw_dir/qfq_dir 正确路由，
|
||
且除权日 raw 有缺口 / qfq 平滑（分红除权准确）。
|
||
|
||
用法（容器）：
|
||
docker run --rm -v /volume1/stock:/volume1/stock \
|
||
-v /volume1/homes/admin/.sanguo_projects/sanguo_vnpy_v2:/app \
|
||
--entrypoint sh sanguo_vnpy_v2:with-sqlite \
|
||
-c "cd /app && python scripts/verify_dual_source.py"
|
||
"""
|
||
import sys
|
||
sys.path.insert(0, "/app")
|
||
from sanguo_data.config import load_config, find_config_path
|
||
from sanguo_trader.data_source import fetch_day, iter_bars
|
||
|
||
cfg = load_config(find_config_path())
|
||
print(f"config: {find_config_path()}")
|
||
print(f"data_paths: raw_dir={cfg.data_paths.get('raw_dir')} qfq_dir={cfg.data_paths.get('qfq_dir')}")
|
||
|
||
OK = True
|
||
|
||
# 1) fetch_day（C-S3 实走接口）除权日双源
|
||
print("\n=== fetch_day 除权日双源（浦发 2022-07-21 除权）===")
|
||
res = {}
|
||
for adj in ["qfq", "raw"]:
|
||
b = fetch_day("600000", "2022-07-21", "d", adj, cfg)
|
||
bp = fetch_day("600000", "2022-07-20", "d", adj, cfg)
|
||
if b is None or bp is None:
|
||
print(f" ❌ {adj}: fetch_day 返回 None（数据缺失或路由失败）")
|
||
OK = False
|
||
continue
|
||
drop = ((b.close_price / bp.close_price) - 1) * 100
|
||
res[adj] = drop
|
||
print(f" {adj}: 07-20={bp.close_price:.4f} → 07-21={b.close_price:.4f} ({drop:+.1f}%)")
|
||
|
||
# 除权日 raw 应有明显缺口（< -3%），qfq 平滑（> -3%）
|
||
if "raw" in res and "qfq" in res:
|
||
if res["raw"] < -3 and res["qfq"] > -3:
|
||
print(" ✅ 除权处理正确：raw 含缺口（撮合真实价）/ qfq 平滑（信号准）")
|
||
else:
|
||
print(f" ⚠️ 除权差异不明显 raw={res['raw']:.1f}% qfq={res['qfq']:.1f}%")
|
||
|
||
# 2) iter_bars cross-section（symbols=list）
|
||
print("\n=== iter_bars cross-section（4 天）===")
|
||
for adj in ["qfq", "raw"]:
|
||
days = list(iter_bars(["600000"], "2022-07-19", "2022-07-22", "d", adj, cfg))
|
||
if not days:
|
||
print(f" ❌ {adj}: iter_bars 0 days")
|
||
OK = False
|
||
else:
|
||
last = days[-1]
|
||
print(f" {adj}: {len(days)} days, 末日={last[0]} close={last[1]['600000'].close_price:.4f}")
|
||
|
||
# 3) 全市场可读性抽检（5 只）
|
||
print("\n=== 全市场抽检（5 只 fetch_day）===")
|
||
for sym in ["600000", "000001", "000858", "300750", "688981"]:
|
||
b = fetch_day(sym, "2026-07-07", "d", "qfq", cfg)
|
||
print(f" {sym}: {'✅ close='+format(b.close_price,'.4f') if b else '❌ None'}")
|
||
if b is None:
|
||
OK = False
|
||
|
||
print("\n" + ("✅ 双源部署验证通过" if OK else "❌ 双源验证有失败项"))
|
||
sys.exit(0 if OK else 1)
|