256 lines
8.4 KiB
Python
256 lines
8.4 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
测试数据库集成和策略存储功能
|
||
"""
|
||
|
||
import sys
|
||
from pathlib import Path
|
||
from datetime import datetime, date
|
||
|
||
# 添加src目录到路径
|
||
current_dir = Path(__file__).parent
|
||
src_dir = current_dir / "src"
|
||
sys.path.insert(0, str(src_dir))
|
||
|
||
from loguru import logger
|
||
from src.database.database_manager import DatabaseManager
|
||
from src.utils.config_loader import ConfigLoader
|
||
from src.data.data_fetcher import ADataFetcher
|
||
from src.utils.notification import NotificationManager
|
||
from src.strategy.kline_pattern_strategy import KLinePatternStrategy
|
||
|
||
|
||
def test_database_operations():
|
||
"""测试数据库基本操作"""
|
||
logger.info("🗄️ 测试数据库基本操作...")
|
||
|
||
try:
|
||
# 初始化数据库管理器
|
||
db_manager = DatabaseManager()
|
||
|
||
# 测试策略统计
|
||
strategy_stats = db_manager.get_strategy_stats()
|
||
logger.info(f"📊 策略统计记录数: {len(strategy_stats)}")
|
||
|
||
# 测试最新信号
|
||
latest_signals = db_manager.get_latest_signals(limit=10)
|
||
logger.info(f"📈 最新信号记录数: {len(latest_signals)}")
|
||
|
||
# 测试日期范围查询
|
||
start_date = date.today()
|
||
signals_by_date = db_manager.get_signals_by_date_range(start_date)
|
||
logger.info(f"🗓️ 今日信号记录数: {len(signals_by_date)}")
|
||
|
||
# 测试回踩提醒
|
||
pullback_alerts = db_manager.get_pullback_alerts(days=7)
|
||
logger.info(f"⚠️ 最近7天回踩提醒: {len(pullback_alerts)}")
|
||
|
||
logger.info("✅ 数据库基本操作测试完成")
|
||
return True
|
||
|
||
except Exception as e:
|
||
logger.error(f"❌ 数据库操作测试失败: {e}")
|
||
return False
|
||
|
||
|
||
def test_strategy_integration():
|
||
"""测试策略与数据库集成"""
|
||
logger.info("🔄 测试策略与数据库集成...")
|
||
|
||
try:
|
||
# 初始化组件
|
||
config_loader = ConfigLoader()
|
||
config = config_loader.load_config()
|
||
|
||
data_fetcher = ADataFetcher()
|
||
notification_manager = NotificationManager(config.get('notification', {}))
|
||
db_manager = DatabaseManager()
|
||
|
||
# 初始化策略（自动创建数据库记录）
|
||
kline_config = config.get('strategy', {}).get('kline_pattern', {})
|
||
strategy = KLinePatternStrategy(
|
||
data_fetcher=data_fetcher,
|
||
notification_manager=notification_manager,
|
||
config=kline_config,
|
||
db_manager=db_manager
|
||
)
|
||
|
||
logger.info(f"📋 策略ID: {strategy.strategy_id}")
|
||
logger.info(f"📝 策略名称: {strategy.strategy_name}")
|
||
|
||
# 测试分析单只股票（会自动保存到数据库）
|
||
test_stock = "000001.SZ"
|
||
logger.info(f"🔍 测试分析股票: {test_stock}")
|
||
|
||
stock_results = strategy.analyze_stock(test_stock, days=30)
|
||
total_signals = sum(len(signals) for signals in stock_results.values())
|
||
|
||
logger.info(f"📊 分析结果: {total_signals} 个信号")
|
||
|
||
# 验证数据库中的记录
|
||
latest_signals = db_manager.get_latest_signals(strategy_name=strategy.strategy_name, limit=10)
|
||
logger.info(f"💾 数据库中最新信号数: {len(latest_signals)}")
|
||
|
||
logger.info("✅ 策略与数据库集成测试完成")
|
||
return True
|
||
|
||
except Exception as e:
|
||
logger.error(f"❌ 策略集成测试失败: {e}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
return False
|
||
|
||
|
||
def test_scan_market_with_database():
|
||
"""测试市场扫描与数据库存储"""
|
||
logger.info("🌍 测试市场扫描与数据库存储...")
|
||
|
||
try:
|
||
# 初始化组件
|
||
config_loader = ConfigLoader()
|
||
config = config_loader.load_config()
|
||
|
||
data_fetcher = ADataFetcher()
|
||
notification_manager = NotificationManager(config.get('notification', {}))
|
||
db_manager = DatabaseManager()
|
||
|
||
# 初始化策略
|
||
kline_config = config.get('strategy', {}).get('kline_pattern', {})
|
||
strategy = KLinePatternStrategy(
|
||
data_fetcher=data_fetcher,
|
||
notification_manager=notification_manager,
|
||
config=kline_config,
|
||
db_manager=db_manager
|
||
)
|
||
|
||
# 小规模市场扫描测试（限制5只股票）
|
||
logger.info("🔍 开始小规模市场扫描测试...")
|
||
test_stocks = ["000001.SZ", "000002.SZ", "600000.SH", "600036.SH", "000858.SZ"]
|
||
|
||
results = strategy.scan_market(
|
||
stock_list=test_stocks,
|
||
max_stocks=5,
|
||
use_hot_stocks=False
|
||
)
|
||
|
||
total_signals = sum(
|
||
sum(len(signals) for signals in stock_results.values())
|
||
for stock_results in results.values()
|
||
)
|
||
|
||
logger.info(f"📊 扫描完成: 发现 {total_signals} 个信号")
|
||
|
||
# 验证数据库存储
|
||
recent_signals = db_manager.get_latest_signals(
|
||
strategy_name=strategy.strategy_name,
|
||
limit=50
|
||
)
|
||
logger.info(f"💾 数据库中存储的信号数: {len(recent_signals)}")
|
||
|
||
# 显示最新的几个信号
|
||
if not recent_signals.empty:
|
||
logger.info("📋 最新信号示例:")
|
||
for i, signal in recent_signals.head(3).iterrows():
|
||
logger.info(f" {signal['stock_code']}({signal['stock_name']}) - {signal['breakout_price']:.2f}元")
|
||
|
||
logger.info("✅ 市场扫描与数据库存储测试完成")
|
||
return True
|
||
|
||
except Exception as e:
|
||
logger.error(f"❌ 市场扫描测试失败: {e}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
return False
|
||
|
||
|
||
def test_database_queries():
|
||
"""测试数据库查询功能"""
|
||
logger.info("🔍 测试数据库查询功能...")
|
||
|
||
try:
|
||
db_manager = DatabaseManager()
|
||
|
||
# 测试策略统计
|
||
strategy_stats = db_manager.get_strategy_stats()
|
||
if not strategy_stats.empty:
|
||
logger.info("📊 策略统计:")
|
||
for _, stat in strategy_stats.iterrows():
|
||
logger.info(f" {stat['strategy_name']}: {stat['total_signals']}个信号, {stat['unique_stocks']}只股票")
|
||
|
||
# 测试按日期查询
|
||
today = date.today()
|
||
today_signals = db_manager.get_signals_by_date_range(today, today)
|
||
logger.info(f"📅 今日信号数: {len(today_signals)}")
|
||
|
||
# 测试获取策略ID
|
||
strategy_id = db_manager.get_strategy_id("K线形态策略")
|
||
logger.info(f"🆔 K线形态策略ID: {strategy_id}")
|
||
|
||
logger.info("✅ 数据库查询功能测试完成")
|
||
return True
|
||
|
||
except Exception as e:
|
||
logger.error(f"❌ 数据库查询测试失败: {e}")
|
||
return False
|
||
|
||
|
||
def main():
|
||
"""主测试函数"""
|
||
logger.remove()
|
||
logger.add(sys.stdout, level="INFO", format="{time:HH:mm:ss} | {level} | {message}")
|
||
|
||
print("=" * 70)
|
||
print("🧪 A股量化交易系统 - 数据库集成测试")
|
||
print("=" * 70)
|
||
|
||
test_results = []
|
||
|
||
# 运行测试
|
||
tests = [
|
||
("数据库基本操作", test_database_operations),
|
||
("策略与数据库集成", test_strategy_integration),
|
||
("数据库查询功能", test_database_queries),
|
||
("市场扫描与存储", test_scan_market_with_database),
|
||
]
|
||
|
||
for test_name, test_func in tests:
|
||
logger.info(f"\n🚀 开始测试: {test_name}")
|
||
try:
|
||
result = test_func()
|
||
test_results.append((test_name, result))
|
||
if result:
|
||
logger.info(f"✅ {test_name} 测试通过")
|
||
else:
|
||
logger.error(f"❌ {test_name} 测试失败")
|
||
except Exception as e:
|
||
logger.error(f"❌ {test_name} 测试异常: {e}")
|
||
test_results.append((test_name, False))
|
||
|
||
# 输出测试结果
|
||
print("\n" + "=" * 70)
|
||
print("📊 测试结果汇总:")
|
||
print("=" * 70)
|
||
|
||
passed = 0
|
||
total = len(test_results)
|
||
|
||
for test_name, result in test_results:
|
||
status = "✅ 通过" if result else "❌ 失败"
|
||
print(f" {test_name}: {status}")
|
||
if result:
|
||
passed += 1
|
||
|
||
print(f"\n🎯 总计: {passed}/{total} 个测试通过")
|
||
|
||
if passed == total:
|
||
print("🎉 所有测试都通过了！数据库集成功能正常工作。")
|
||
print("🌐 现在可以启动Web界面查看数据:")
|
||
print(" cd web && python app.py")
|
||
else:
|
||
print("⚠️ 部分测试失败，请检查错误信息并修复问题。")
|
||
|
||
print("=" * 70)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main() |