run.py

#!/usr/bin/env python3
"""
AI Quant 项目直接运行程序
"""

import sys
import os
import logging
import argparse

# 添加项目根目录到路径
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))

# 配置日志
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
    handlers=[
        logging.FileHandler("aiquant.log"),
        logging.StreamHandler()
    ]
)

logger = logging.getLogger("AIQuant.run")


def check_dependencies():
    """
    检查项目依赖是否已安装
    """
    logger.info("检查项目依赖...")

    required_packages = [
        "vnpy", "numpy", "pandas", "plotly", "pyqtgraph", "PySide6", "zmq",
        "qdarkstyle", "talib", "tqdm", "tzlocal", "loguru", "deap", "nbformat",
        "akshare", "tushare", "ollama"
    ]

    missing_packages = []
    for package in required_packages:
        try:
            __import__(package)
            logger.info(f"✅ 依赖包 {package} 已安装")
        except ImportError:
            logger.error(f"❌ 依赖包 {package} 未安装")
            missing_packages.append(package)

    if missing_packages:
        logger.error(f"缺失依赖包: {', '.join(missing_packages)}")
        logger.error("请运行以下命令安装依赖:")
        logger.error("pip install -r requirements.txt --break-system-packages")
        return False

    logger.info("✅ 所有项目依赖已安装")
    return True


def show_project_info():
    """
    显示项目信息
    """
    logger.info("=" * 50)
    logger.info("AI Quant - A 股量化交易框架")
    logger.info("=" * 50)
    logger.info(f"版本: {__version__}")
    logger.info(f"项目路径: {os.path.abspath('.')}")
    logger.info(f"Python 版本: {sys.version}")
    logger.info("=" * 50)


def run_simple_test():
    """
    运行简单测试
    """
    logger.info("运行简单测试...")

    from core.api.zs_sec import ZSSecConfig

    # 测试招商证券接口配置
    config = ZSSecConfig(
        server_address="180.168.146.187",
        port=10100,
        account="123456",
        password="123456",
        broker_code="0000"
    )

    if config.validate():
        logger.info("✅ 招商证券接口配置验证成功")
    else:
        logger.error("❌ 招商证券接口配置验证失败")

    from app.strategy.single_stock import SingleStockAnalyzer
    from service.data.mock_data import generate_random_kline_data

    # 测试单票量化策略
    analyzer = SingleStockAnalyzer()

    # 创建测试数据（使用模拟数据生成模块）
    data = generate_random_kline_data(
        "000001.SZ", "2023-01-01", "2023-12-31", "d")

    # 分析股票
    result = analyzer.analyze("000001.SZ", data)

    logger.info(f"✅ 单票量化策略分析成功:")
    logger.info(f"   {result}")
    logger.info("=" * 50)


def main():
    """
    主入口函数
    """
    global __version__
    __version__ = "1.0.0"

    # 解析命令行参数
    parser = argparse.ArgumentParser(
        description="AI Quant 项目直接运行程序"
    )
    parser.add_argument(
        "--check",
        action="store_true",
        help="检查项目依赖"
    )
    parser.add_argument(
        "--test",
        action="store_true",
        help="运行简单测试"
    )
    parser.add_argument(
        "--info",
        action="store_true",
        help="显示项目信息"
    )
    parser.add_argument(
        "--console",
        action="store_true",
        help="启动控制台模式"
    )
    parser.add_argument(
        "--gui",
        action="store_true",
        help="启动图形界面模式"
    )

    args = parser.parse_args()

    logger.info("AI Quant 项目直接运行程序启动")

    # 检查依赖
    if args.check or not (args.test or args.info or args.console or args.gui):
        if not check_dependencies():
            return 1

    # 显示项目信息
    if args.info or not (args.test or args.console or args.gui):
        show_project_info()

    # 运行测试
    if args.test or not (args.console or args.gui):
        run_simple_test()

    # 启动控制台模式
    if args.console:
        logger.info("启动控制台模式...")
        from app.console import run_console
        return run_console()

    # 启动图形界面模式
    if args.gui:
        logger.info("启动图形界面模式...")
        from app.main import run_gui
        return run_gui()

    # 如果没有指定任何模式，显示帮助信息
    if not (args.check or args.test or args.info or args.console or args.gui):
        parser.print_help()
        return 0

    return 0


if __name__ == "__main__":
    try:
        sys.exit(main())
    except KeyboardInterrupt:
        logger.info("程序被用户中断")
        sys.exit(0)
    except Exception as e:
        logger.error(f"程序运行失败: {e}")
        logger.exception("详细错误信息:")
        sys.exit(1)