runner_manager.py
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
运行模式管理模块
负责程序运行模式的管理和切换
支持三种运行模式：控制台运行模式、网页端运行模式、桌面GUI运行模式
"""
import os
import sys
import logging
import argparse
from pathlib import Path
from typing import Dict, List, Tuple, Any, Optional
# 配置日志
logger = logging.getLogger(__name__)
class RunnerManager:
"""运行模式管理类"""
_instance = None
def __new__(cls, *args, **kwargs):
"""单例模式创建实例"""
if cls._instance is None:
cls._instance = super().__new__(cls)
cls._instance._initialized = False
return cls._instance
def __init__(self):
"""初始化运行模式管理器"""
if self._initialized:
return
self._initialized = True
self._runners = {}
self._current_runner = None
self._load_runners()
def _load_runners(self):
"""加载所有运行模式"""
try:
# 运行模式列表
self._runners = {
'console': ConsoleRunner(),
'web': WebRunner(),
'gui': GUIRunner()
}
logger.info(f"共加载 {len(self._runners)} 种运行模式")
except Exception as e:
logger.error(f"运行模式加载失败：{e}")
def get_runner(self, runner_name: str) -> Optional[Any]:
"""获取指定名称的运行模式实例"""
try:
if runner_name not in self._runners:
logger.error(f"未找到指定运行模式：{runner_name}")
return None
logger.info(f"运行模式获取成功：{runner_name}")
return self._runners[runner_name]
except Exception as e:
logger.error(f"运行模式获取失败：{e}")
return None
def list_runners(self) -> List[Dict[str, Any]]:
"""列出所有可用的运行模式"""
runners = []
for runner_name, runner_instance in self._runners.items():
runners.append({
'name': runner_name,
'class': runner_instance.__class__.__name__,
'description': self._get_runner_description(runner_instance)
})
return runners
def _get_runner_description(self, runner_instance: Any) -> str:
"""获取运行模式实例的描述信息"""
if runner_instance.__class__.__doc__:
# 去除文档字符串中的换行和缩进
return runner_instance.__class__.__doc__.strip().split('\n')[0]
else:
return '未指定描述'
def run(self, runner_name: str, args: argparse.Namespace = None) -> None:
"""运行指定的运行模式"""
try:
runner = self.get_runner(runner_name)
if not runner:
return
# 保存当前运行模式
self._current_runner = runner
# 运行模式
runner.run(args)
except Exception as e:
logger.error(f"运行模式启动失败：{runner_name} - {e}")
def get_current_runner(self) -> Optional[Any]:
"""获取当前运行模式实例"""
return self._current_runner
def stop(self) -> None:
"""停止当前运行模式"""
try:
if self._current_runner:
self._current_runner.stop()
logger.info("运行模式停止成功")
else:
logger.warning("当前没有正在运行的模式")
except Exception as e:
logger.error(f"运行模式停止失败：{e}")
class BaseRunner:
"""运行模式基类"""
def run(self, args: argparse.Namespace = None) -> None:
"""运行方法（子类必须实现）"""
raise NotImplementedError("子类必须实现run方法")
def stop(self) -> None:
"""停止方法（可选实现）"""
pass
class ConsoleRunner(BaseRunner):
"""控制台运行模式"""
def run(self, args: argparse.Namespace = None) -> None:
"""运行控制台模式"""
try:
logger.info("启动控制台运行模式")
# 初始化策略管理器
from src.strategy_manager import strategy_manager
# 根据命令参数执行相应操作
if args.init:
self._init_project()
elif args.select:
self._run_selection(args)
elif args.backtest:
self._run_backtest(args)
elif args.optimize:
self._run_optimization(args)
elif args.schedule:
self._start_scheduler()
elif args.test:
self._run_test()
elif args.update:
self._update_data()
else:
logger.warning("未指定操作，程序将退出")
except Exception as e:
logger.error(f"控制台模式运行失败：{e}")
def _init_project(self) -> None:
"""初始化项目"""
try:
logger.info("开始初始化项目...")
from main import init_project
init_project()
except Exception as e:
logger.error(f"项目初始化失败：{e}")
def _run_selection(self, args: argparse.Namespace) -> None:
"""运行选股"""
try:
# 解析策略列表
strategy_names = args.strategies.split(",")
logger.info(f"使用策略：{', '.join(strategy_names)}")
from src.stock_selector import StockSelector
selector = StockSelector()
result = selector.run(strategy_names)
if result is not None and not result.empty:
strategy_name = "-".join(strategy_names)
selector.save_results(result, strategy_name=strategy_name)
selector.print_results(result, strategy_name=strategy_name)
else:
logger.warning("未选出符合条件的股票")
except Exception as e:
logger.error(f"选股失败：{e}")
def _run_backtest(self, args: argparse.Namespace) -> None:
"""运行回测"""
try:
logger.info("开始回测...")
from src.backtester import Backtester
backtester = Backtester()
report = backtester.run()
backtester.save_report(report)
backtester.print_report(report)
except Exception as e:
logger.error(f"回测失败：{e}")
def _run_optimization(self, args: argparse.Namespace) -> None:
"""运行参数优化"""
try:
logger.info("开始参数优化...")
from src.parameter_optimizer import ParameterOptimizer
optimizer = ParameterOptimizer()
best_params = optimizer.run()
optimizer.save_best_parameters(best_params)
except Exception as e:
logger.error(f"参数优化失败：{e}")
def _start_scheduler(self) -> None:
"""启动调度器"""
try:
logger.info("开始调度服务...")
from src.scheduler import Scheduler
scheduler = Scheduler()
scheduler.start()
except Exception as e:
logger.error(f"调度服务启动失败：{e}")
def _run_test(self) -> None:
"""运行测试"""
try:
logger.info("开始一键测试...")
from main import run_test
run_test()
except Exception as e:
logger.error(f"测试失败：{e}")
def _update_data(self) -> None:
"""更新数据"""
try:
logger.info("开始更新历史数据...")
from main import update_data
update_data()
except Exception as e:
logger.error(f"数据更新失败：{e}")
def stop(self) -> None:
"""停止控制台模式"""
logger.info("控制台模式停止")
class WebRunner(BaseRunner):
"""网页端运行模式"""
def run(self, args: argparse.Namespace = None) -> None:
"""运行网页端模式"""
try:
logger.info("启动网页端运行模式")
from web.app import WebServer
# 获取网页服务配置
host = args.host if args and hasattr(args, 'host') else '0.0.0.0'
port = args.port if args and hasattr(args, 'port') else 5000
logger.info(f"网页服务地址：http://{host}:{port}")
server = WebServer(host=host, port=port)
server.run()
except Exception as e:
logger.error(f"网页端模式运行失败：{e}")
def stop(self) -> None:
"""停止网页端模式"""
logger.info("网页端模式停止")
class GUIRunner(BaseRunner):
"""桌面GUI运行模式"""
def run(self, args: argparse.Namespace = None) -> None:
"""运行桌面GUI模式"""
try:
logger.info("启动桌面GUI运行模式")
# 检查是否安装了Tkinter
try:
import tkinter
from tkinter import ttk
except ImportError as e:
logger.error(f"桌面GUI模式需要Tkinter库，请安装：{e}")
return
logger.warning("桌面GUI模式尚未完全实现，仅显示基础界面")
self._create_gui()
except Exception as e:
logger.error(f"桌面GUI模式运行失败：{e}")
def _create_gui(self):
"""创建GUI界面"""
try:
import tkinter as tk
from tkinter import ttk
from tkinter import messagebox
# 创建主窗口
root = tk.Tk()
root.title("超短线量化选股程序")
root.geometry("800x600")
# 创建菜单栏
menubar = tk.Menu(root)
# 文件菜单
file_menu = tk.Menu(menubar, tearoff=0)
file_menu.add_command(label="退出", command=root.quit)
menubar.add_cascade(label="文件", menu=file_menu)
# 操作菜单
action_menu = tk.Menu(menubar, tearoff=0)
action_menu.add_command(label="选股", command=self._run_selection_gui)
action_menu.add_command(label="回测", command=self._run_backtest_gui)
action_menu.add_command(label="参数优化", command=self._run_optimization_gui)
action_menu.add_command(label="更新数据", command=self._update_data_gui)
menubar.add_cascade(label="操作", menu=action_menu)
# 帮助菜单
help_menu = tk.Menu(menubar, tearoff=0)
help_menu.add_command(label="关于", command=self._show_about)
menubar.add_cascade(label="帮助", menu=help_menu)
root.config(menu=menubar)
# 创建标签
label = ttk.Label(root, text="超短线量化选股程序", font=("Arial", 16))
label.pack(pady=20)
# 创建状态标签
status_label = ttk.Label(root, text="就绪")
status_label.pack(pady=10)
# 运行主事件循环
root.mainloop()
except Exception as e:
logger.error(f"GUI界面创建失败：{e}")
def _run_selection_gui(self):
"""运行选股"""
logger.info("选股操作被调用")
def _run_backtest_gui(self):
"""运行回测"""
logger.info("回测操作被调用")
def _run_optimization_gui(self):
"""运行参数优化"""
logger.info("参数优化操作被调用")
def _update_data_gui(self):
"""更新数据"""
logger.info("数据更新操作被调用")
def _show_about(self):
"""显示关于信息"""
import tkinter as tk
from tkinter import messagebox
messagebox.showinfo("关于", "超短线量化选股程序\n版本：1.0.0")
def stop(self) -> None:
"""停止桌面GUI模式"""
logger.info("桌面GUI模式停止")
# 创建运行模式管理器实例
runner_manager = RunnerManager()