main.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
超短线量化选股程序 - 主程序入口（插件化架构）
交易日14:30选股，T+1卖出，大模型主导规律挖掘，全样本回测
选股器核心调度 + 策略插件化架构，支持随时增删策略
"""

import argparse
import sys
import os
import logging
from pathlib import Path
from dotenv import load_dotenv

# 添加项目根目录到Python路径
sys.path.append(str(Path(__file__).parent))

# 加载环境变量
load_dotenv()

# 配置日志
from src.config_manager import config_manager
LOG_PATH = config_manager.get('system.log_path')
Path(LOG_PATH).parent.mkdir(exist_ok=True)

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
    handlers=[
        logging.FileHandler(LOG_PATH, encoding='utf-8'),
        logging.StreamHandler(sys.stdout)
    ]
)
logger = logging.getLogger(__name__)

def main():
    """主程序入口"""
    parser = argparse.ArgumentParser(
        description="超短线量化选股程序（插件化架构）",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
            运行模式：
              无启动参数   默认以控制台模式运行
              -web        网页端运行模式
              -gui        桌面GUI运行模式
              
            控制台运行模式可用命令：
              --init      初始化项目（创建目录、配置文件等）
              --select    手动触发选股（默认在14:30自动触发）
              --backtest  手动触发回测
              --optimize  手动触发参数优化
              --schedule  启动调度服务（自动选股和回测）
              --test      一键测试程序功能
              --update    手动更新历史数据
              
            选股命令可用参数：
              --strategies  策略列表，用逗号分隔（可选，默认：vcp,model）
              
            网页端运行模式可用参数：
              --host      网页服务绑定的主机地址（默认：0.0.0.0）
              --port      网页服务绑定的端口（默认：5000）
            """
    )
    
    # 运行模式选择
    parser.add_argument("-web", action="store_true", help="网页端运行模式")
    parser.add_argument("-gui", action="store_true", help="桌面GUI运行模式")
    
    # 控制台模式参数
    parser.add_argument("--init", action="store_true", help="初始化项目")
    parser.add_argument("--select", action="store_true", help="手动触发选股")
    parser.add_argument("--backtest", action="store_true", help="手动触发回测")
    parser.add_argument("--optimize", action="store_true", help="手动触发参数优化")
    parser.add_argument("--schedule", action="store_true", help="启动调度服务")
    parser.add_argument("--test", action="store_true", help="一键测试程序功能")
    parser.add_argument("--update", action="store_true", help="手动更新历史数据")
    parser.add_argument("--strategies", type=str, default="vcp,model", 
                        help="策略列表，用逗号分隔（默认：vcp,model）")
    
    # 网页模式参数
    parser.add_argument("--host", type=str, default=config_manager.get('web.host'), 
                        help="网页服务绑定的主机地址")
    parser.add_argument("--port", type=int, default=config_manager.get('web.port'), 
                        help="网页服务绑定的端口")
    
    args = parser.parse_args()
    
    # 确定运行模式
    runner_name = _determine_runner_name(args)
    
    # 创建项目目录
    create_project_directories()
    
    # 初始化配置
    _init_config()
    
    # 运行程序
    from src.runner_manager import runner_manager
    runner_manager.run(runner_name, args)

def _determine_runner_name(args):
    """确定运行模式"""
    # 检查是否指定了运行模式
    if args.web:
        return 'web'
    elif args.gui:
        return 'gui'
    else:
        # 默认以控制台模式运行
        return 'console'

def create_project_directories():
    """创建项目目录结构"""
    from src.utils import ensure_dir_exists
    
    directories = [
        "data",
        "logs",
        "results/selections",
        "results/backtests",
        "results/optimization",
        "results/models"
    ]
    
    for directory in directories:
        ensure_dir_exists(directory)
        logger.info(f"目录已创建或已存在：{directory}")

def _init_config():
    """初始化配置"""
    from src.config_manager import config_manager
    
    # 验证配置
    if not config_manager.validate_config():
        logger.error("配置验证失败，程序无法运行")
        sys.exit(1)
    
    logger.info("配置初始化完成")

def init_project():
    """初始化项目"""
    logger.info("开始初始化项目...")
    
    # 检查环境变量文件
    from pathlib import Path
    env_path = Path(".env")
    if not env_path.exists():
        logger.warning(".env配置文件不存在，请复制.env.example并填写配置")
        return
    
    # 检查依赖是否安装
    try:
        import tushare
        import pandas
        import numpy
        import sqlite3
        import requests
        import bs4
        import apscheduler
        import dotenv
        import openai
        import volcengine
        logger.info("依赖检查通过")
    except ImportError as e:
        logger.error(f"依赖检查失败：{e}")
        logger.error("请运行 pip install -r requirements.txt 安装依赖")
        return
    
    # 初始化数据库
    from src.data_manager import DataManager
    data_manager = DataManager()
    data_manager.init_database()
    logger.info("数据库初始化完成")
    
    # 下载历史数据（首次运行）
    logger.info("首次运行，开始下载历史数据...")
    try:
        data_manager.update_history_data()
        logger.info("历史数据下载完成")
    except Exception as e:
        logger.error(f"历史数据下载失败：{e}")
    
    logger.info("项目初始化完成")

def run_selector(args):
    """手动触发选股"""
    logger.info("手动触发选股...")
    
    # 解析策略列表
    strategy_names = args.strategies.split(",")
    logger.info(f"使用策略：{', '.join(strategy_names)}")
    
    from src.stock_selector import StockSelector
    selector = StockSelector()
    
    try:
        result = selector.run(strategy_names)
        logger.info(f"选股完成，共选出 {len(result)} 只股票")
        logger.info(f"选股结果已保存到：{selector.result_dir}")
    except Exception as e:
        logger.error(f"选股失败：{e}")
        return
    
    # 保存选股结果
    selector.save_results(result, strategy_name="-".join(strategy_names))
    logger.info("选股结果保存完成")
    selector.print_results(result, "-".join(strategy_names))

def run_backtester():
    """手动触发回测"""
    logger.info("手动触发回测...")
    
    from src.backtester import Backtester
    backtester = Backtester()
    
    try:
        report = backtester.run()
        logger.info("回测完成")
        logger.info(f"胜率：{report['win_rate']:.2%}")
        logger.info(f"平均单只收益率：{report['avg_return']:.2%}")
        logger.info(f"累计收益率：{report['total_return']:.2%}")
        logger.info(f"最大回撤：{report['max_drawdown']:.2%}")
        logger.info(f"盈亏比：{report['profit_loss_ratio']:.2f}")
        logger.info(f"每日平均收益率：{report['avg_daily_return']:.2%}")
    except Exception as e:
        logger.error(f"回测失败：{e}")
        return
    
    # 保存回测报告
    backtester.save_report(report)
    logger.info("回测报告保存完成")
    backtester.print_report(report)

def run_optimizer():
    """手动触发参数优化"""
    logger.info("手动触发参数优化...")
    
    from src.parameter_optimizer import ParameterOptimizer
    optimizer = ParameterOptimizer()
    
    try:
        best_params = optimizer.run()
        logger.info("参数优化完成")
        logger.info(f"最优参数：{best_params}")
    except Exception as e:
        logger.error(f"参数优化失败：{e}")
        return
    
    # 保存最优参数
    optimizer.save_best_parameters(best_params)
    logger.info("最优参数保存完成")

def start_scheduler():
    """启动调度服务"""
    logger.info("启动调度服务...")
    
    from src.scheduler import Scheduler
    scheduler = Scheduler()
    scheduler.start()

def start_web_server(host="0.0.0.0", port=5000):
    """启动网页服务"""
    logger.info("启动网页服务...")
    
    try:
        from web.app import WebServer
        server = WebServer(host=host, port=port)
        server.run()
    except Exception as e:
        logger.error(f"启动网页服务失败：{e}")

def run_test():
    """一键测试程序功能"""
    logger.info("开始一键测试...")
    
    try:
        # 测试数据管理模块
        logger.info("测试数据管理模块...")
        from src.data_manager import DataManager
        data_manager = DataManager()
        data_manager.test_connection()
        logger.info("数据管理模块测试通过")
        
        # 测试策略管理模块
        logger.info("测试策略管理模块...")
        from src.strategy_manager import strategy_manager
        strategies = strategy_manager.list_strategies()
        logger.info(f"策略数量：{len(strategies)}")
        for strategy in strategies:
            logger.info(f"策略：{strategy['name']} - {strategy['description']}")
        logger.info("策略管理模块测试通过")
        
        # 测试选股模块
        logger.info("测试选股模块...")
        from src.stock_selector import StockSelector
        selector = StockSelector()
        selector.test_selector(['vcp', 'model'])
        logger.info("选股模块测试通过")
        
        # 测试回测模块
        logger.info("测试回测模块...")
        from src.backtester import Backtester
        backtester = Backtester()
        backtester.test_backtest()
        logger.info("回测模块测试通过")
        
        # 测试参数优化模块
        logger.info("测试参数优化模块...")
        from src.parameter_optimizer import ParameterOptimizer
        optimizer = ParameterOptimizer()
        optimizer.test_optimization()
        logger.info("参数优化模块测试通过")
        
        logger.info("所有模块测试通过")
    except Exception as e:
        logger.error(f"测试失败：{e}")

def update_data():
    """手动更新历史数据"""
    logger.info("开始更新历史数据...")
    
    from src.data_manager import DataManager
    data_manager = DataManager()
    
    try:
        data_manager.update_history_data()
        logger.info("历史数据更新完成")
    except Exception as e:
        logger.error(f"历史数据更新失败：{e}")

if __name__ == "__main__":
    main()