42 lines
1.7 KiB
Python
42 lines
1.7 KiB
Python
"""端到端冒烟：验证 read_db_daily 能读 NAS quant_trading.db（真实数据）。
|
|
|
|
在临时容器内跑（docker run --rm，挂载 /volume1/stock + /app）。
|
|
探测 vnpy 4.4.0 database 配置 + 验证数据读取。
|
|
"""
|
|
import sys
|
|
sys.path.insert(0, "/app")
|
|
|
|
print("=== Step 1: 探测 vnpy 4.4.0 database 配置 ===")
|
|
from vnpy.trader.setting import SETTINGS
|
|
db_keys = {k: v for k, v in SETTINGS.items() if "database" in k.lower() or "sql" in k.lower()}
|
|
print("database 相关 SETTINGS:", db_keys)
|
|
|
|
SETTINGS["database.driver"] = "sqlite"
|
|
SETTINGS["database.database"] = "/volume1/stock/sanguo_vnpy/data/quant_trading.db"
|
|
print(f"已设置: driver=sqlite, database={SETTINGS['database.database']}")
|
|
|
|
print("\n=== Step 2: 加载 sanguo_data 配置 ===")
|
|
from sanguo_data.config import load_config
|
|
cfg = load_config("config/data_platform.yaml")
|
|
print(f"cfg vnpy_db: {cfg.data_paths.get('vnpy_db')}")
|
|
|
|
print("\n=== Step 3: read_db_daily 读 600000 ===")
|
|
try:
|
|
from sanguo_data.datareader import read_db_daily
|
|
bars = read_db_daily("600000", "2026-01-01", "2026-06-30", cfg)
|
|
print(f"RESULT: 读取 {len(bars)} 条 BarData")
|
|
if bars:
|
|
b = bars[0]
|
|
print(f"FIRST: symbol={b.symbol} dt={b.datetime} close={b.close_price} vol={b.volume}")
|
|
else:
|
|
print("WARN: 0 条——可能 vnpy database 路径配置不对，或数据不在该 symbol/区间")
|
|
except Exception as e:
|
|
import traceback
|
|
traceback.print_exc()
|
|
print(f"FAIL: {type(e).__name__}: {e}")
|
|
|
|
print("\n=== Step 4: 直接探测 NAS db 文件 ===")
|
|
import os
|
|
db_path = "/volume1/stock/sanguo_vnpy/data/quant_trading.db"
|
|
print(f"{db_path} exists: {os.path.exists(db_path)}, size: {os.path.getsize(db_path) if os.path.exists(db_path) else 'N/A'}")
|