cfdaily
160 lines
5.2 KiB
Python
Executable File
160 lines
5.2 KiB
Python
Executable File
#!/usr/bin/env python3
|
||
"""
|
||
简单检查赵云将军本地数据中的 510300.SSE 标的
|
||
"""
|
||
|
||
import sqlite3
|
||
import os
|
||
import sys
|
||
|
||
def find_vnpy_database():
|
||
"""查找vn.py数据库文件"""
|
||
db_paths = [
|
||
'/Users/chufeng/.openclaw/workspace-zhaoyun/sanguo_quant_live/zhaoyun-data/data/running_data/database_test.db',
|
||
'/Users/chufeng/.openclaw/workspace-zhaoyun/zhaoyun-data/data/database_test.db',
|
||
'/Users/chufeng/.openclaw/memory/zhaoyun-data.sqlite',
|
||
]
|
||
|
||
existing_dbs = []
|
||
for path in db_paths:
|
||
if os.path.exists(path):
|
||
existing_dbs.append(path)
|
||
print(f"✅ 找到数据库: {path}")
|
||
size = os.path.getsize(path) / (1024*1024)
|
||
print(f" 文件大小: {size:.2f} MB")
|
||
else:
|
||
print(f"❌ 不存在: {path}")
|
||
|
||
return existing_dbs
|
||
|
||
def check_symbol_in_db(db_path, symbol):
|
||
"""检查数据库中是否存在指定标的"""
|
||
print(f"\n🔍 检查数据库 {db_path} 中的 {symbol}...")
|
||
|
||
try:
|
||
conn = sqlite3.connect(db_path)
|
||
cursor = conn.cursor()
|
||
|
||
# 列出所有表
|
||
cursor.execute("SELECT name FROM sqlite_master WHERE type='table';")
|
||
tables = cursor.fetchall()
|
||
table_names = [t[0] for t in tables]
|
||
print(f"📊 数据库中的表: {table_names}")
|
||
|
||
# 检查常见表名
|
||
found = False
|
||
for table_name in ['dbbardata', 'bar_data', 'daily_data', '1d', 'daily', 'bar']:
|
||
if table_name not in table_names:
|
||
continue
|
||
|
||
print(f"\n🔍 检查表 {table_name}...")
|
||
|
||
# 获取总行数
|
||
cursor.execute(f"SELECT COUNT(*) FROM {table_name};")
|
||
total_rows = cursor.fetchone()[0]
|
||
print(f" 总行数: {total_rows}")
|
||
|
||
# 检查symbol是否存在
|
||
cursor.execute(f"SELECT COUNT(*) FROM {table_name} WHERE symbol = ?", (symbol,))
|
||
count = cursor.fetchone()[0]
|
||
|
||
if count > 0:
|
||
print(f" ✅ 找到标的: {symbol}")
|
||
print(f" 数据行数: {count}")
|
||
|
||
# 获取时间范围
|
||
try:
|
||
cursor.execute(f"SELECT MIN(datetime), MAX(datetime) FROM {table_name} WHERE symbol = ?", (symbol,))
|
||
min_dt, max_dt = cursor.fetchone()
|
||
print(f" 时间范围: {min_dt} -> {max_dt}")
|
||
except Exception:
|
||
pass
|
||
|
||
# 获取列名
|
||
cursor.execute(f"PRAGMA table_info({table_name})")
|
||
columns = [col[1] for col in cursor.fetchall()]
|
||
print(f" 列名: {columns}")
|
||
|
||
# 看看是否有其他标的
|
||
cursor.execute(f"SELECT DISTINCT symbol FROM {table_name} LIMIT 5")
|
||
symbols = [s[0] for s in cursor.fetchall()]
|
||
print(f" 其他标的: {symbols}")
|
||
|
||
found = True
|
||
break
|
||
else:
|
||
print(f" ❌ 未找到标的 {symbol}")
|
||
|
||
conn.close()
|
||
return found
|
||
|
||
except Exception as e:
|
||
print(f"❌ 连接数据库出错: {e}")
|
||
return False
|
||
|
||
def check_data_paths():
|
||
"""检查可能的数据路径"""
|
||
print("\n📂 检查可能的数据路径...")
|
||
possible_paths = [
|
||
'/Users/chufeng/.openclaw/workspace-zhaoyun/sanguo_quant_live/zhaoyun-data/',
|
||
'/Users/chufeng/.openclaw/workspace-zhaoyun/zhaoyun-data/',
|
||
'/Users/chufeng/.openclaw/sanguo_projects/sanguo_quant_live/zhaoyun-data/',
|
||
]
|
||
|
||
for path in possible_paths:
|
||
if os.path.exists(path):
|
||
print(f"✅ 存在: {path}")
|
||
files = os.listdir(path)
|
||
print(f" 文件数: {len(files)}")
|
||
else:
|
||
print(f"❌ 不存在: {path}")
|
||
|
||
def main():
|
||
"""主函数"""
|
||
print("🚀 简单检查 510300.SSE 标的数据")
|
||
print("="*60)
|
||
|
||
symbol_to_check = "510300.SSE"
|
||
print(f"目标标的: {symbol_to_check}")
|
||
|
||
# 检查数据路径
|
||
check_data_paths()
|
||
|
||
# 查找数据库
|
||
dbs = find_vnpy_database()
|
||
|
||
if not dbs:
|
||
print("\n❌ 未找到任何数据库文件")
|
||
print("\n📋 需要检查:")
|
||
print("1. 赵云将军是否已经下载数据")
|
||
print("2. 数据是否已经转换到vn.py格式")
|
||
print("3. 数据路径配置是否正确")
|
||
return False
|
||
|
||
# 检查每个数据库
|
||
found = False
|
||
for db in dbs:
|
||
if check_symbol_in_db(db, symbol_to_check):
|
||
found = True
|
||
break
|
||
|
||
# 总结
|
||
print("\n" + "="*60)
|
||
print("检查结果:")
|
||
|
||
if found:
|
||
print(f"✅ 找到 {symbol_to_check} 数据")
|
||
print("数据存在，可以配置路径")
|
||
else:
|
||
print(f"❌ 未找到 {symbol_to_check} 数据")
|
||
print("\n📋 可能的原因:")
|
||
print("1. 数据还未下载到本地")
|
||
print("2. 数据还未转换为vn.py格式")
|
||
print("3. 标的名称格式不正确（可能是其他格式，比如 510300.XSHG）")
|
||
print("4. 数据路径配置错误")
|
||
|
||
print("="*60)
|
||
return found
|
||
|
||
if __name__ == "__main__":
|
||
main() |