cfdaily
120 lines
3.7 KiB
Python
Executable File
120 lines
3.7 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""
|
|
诊断exchange提取问题：检查数据库中510300.SSE的数据
|
|
"""
|
|
|
|
import sys
|
|
import types
|
|
|
|
# 兼容性模块
|
|
print("🔧 [DIAG] 加载vnpy.app兼容性模块...")
|
|
vnpy_app_module = types.ModuleType('vnpy.app')
|
|
sys.modules['vnpy.app'] = vnpy_app_module
|
|
submodules = ['cta_strategy', 'cta_backtester', 'data_manager']
|
|
for name in submodules:
|
|
full_name = f'vnpy.app.{name}'
|
|
submodule = types.ModuleType(full_name)
|
|
sys.modules[full_name] = submodule
|
|
setattr(vnpy_app_module, name, submodule)
|
|
|
|
from vnpy_ctastrategy import CtaTemplate, CtaStrategyApp
|
|
from vnpy_ctastrategy import (
|
|
CtaTemplate,
|
|
StopOrder,
|
|
TickData,
|
|
BarData,
|
|
TradeData,
|
|
OrderData,
|
|
BarGenerator,
|
|
ArrayManager,
|
|
)
|
|
from vnpy.trader.constant import Direction, Offset, Interval, Exchange
|
|
|
|
sys.modules['vnpy.app.cta_strategy'].CtaTemplate = CtaTemplate
|
|
sys.modules['vnpy.app.cta_strategy'].CtaStrategyApp = CtaStrategyApp
|
|
vnpy_app_module.CtaTemplate = CtaTemplate
|
|
vnpy_app_module.CtaStrategyApp = CtaStrategyApp
|
|
|
|
from vnpy_ctabacktester import BacktesterEngine
|
|
sys.modules['vnpy.app.cta_backtester'].BacktesterEngine = BacktesterEngine
|
|
vnpy_app_module.BacktesterEngine = BacktesterEngine
|
|
|
|
print("✅ [DIAG] vnpy.app兼容性模块加载完成！")
|
|
|
|
from vnpy.event import EventEngine
|
|
from vnpy.trader.engine import MainEngine
|
|
from vnpy.trader.database import get_database
|
|
from datetime import datetime
|
|
import traceback
|
|
|
|
def parse_date(date_val):
|
|
"""解析日期"""
|
|
date_ts = float(date_val)
|
|
date_int = int(date_ts)
|
|
s = str(date_int)
|
|
|
|
if len(s) == 8:
|
|
year = int(s[:4])
|
|
month = int(s[4:6])
|
|
day = int(s[6:8])
|
|
return datetime(year, month, day)
|
|
elif len(s) >= 10:
|
|
return datetime.fromtimestamp(date_int)
|
|
else:
|
|
year = int(s[:4])
|
|
month = int(s[4:6])
|
|
day = int(s[6:8])
|
|
return datetime(year, month, day)
|
|
|
|
if __name__ == '__main__':
|
|
print("\n=== 诊断exchange数据加载问题 ===")
|
|
|
|
# 测试日期解析
|
|
start_ts = 1609459200
|
|
end_ts = 1772515200
|
|
start_dt = parse_date(start_ts)
|
|
end_dt = parse_date(end_ts)
|
|
|
|
print(f"\n📅 时间范围:")
|
|
print(f" start: {start_ts} -> {start_dt}")
|
|
print(f" end: {end_ts} -> {end_dt}")
|
|
|
|
# 连接数据库
|
|
db = get_database()
|
|
print(f"\n💾 数据库连接成功: {type(db)}")
|
|
|
|
# 查看所有表
|
|
print(f"\n🔍 查看数据库中的bar数据统计:")
|
|
|
|
# 尝试不同的symbol和exchange组合
|
|
test_cases = [
|
|
("510300", "SSE"),
|
|
("510300.SSE", "SSE"),
|
|
("510300", "XSHG"),
|
|
("510300.SSE", "XSHG"),
|
|
]
|
|
|
|
for symbol_str, exchange_str in test_cases:
|
|
try:
|
|
exchange = Exchange(exchange_str)
|
|
print(f"\n▶️ 测试: symbol={symbol_str}, exchange={exchange}")
|
|
bars = db.load_bar_data(symbol_str, exchange, Interval.DAILY, start_dt, end_dt)
|
|
print(f" ✅ 查询成功，共 {len(bars)} 条bar数据")
|
|
if len(bars) > 0:
|
|
print(f" 第一条: {bars[0].datetime}, close={bars[0].close_price}")
|
|
print(f" 最后一条: {bars[-1].datetime}, close={bars[-1].close_price}")
|
|
except Exception as e:
|
|
print(f" ❌ 查询失败: {e}")
|
|
|
|
# 尝试获取所有标的信息
|
|
print(f"\n🔍 尝试获取所有bar数据统计:")
|
|
try:
|
|
# vnpy_sqlite 应该有 count_bar_data 方法
|
|
if hasattr(db, 'count_bar_data'):
|
|
total = db.count_bar_data()
|
|
print(f" 数据库总共有 {total} 条bar数据")
|
|
except Exception as e:
|
|
print(f" 无法获取总数: {e}")
|
|
|
|
print("\n=== 诊断完成 ===")
|