strategy_manager.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
策略管理模块 - 负责策略的加载、管理和切换
与选股器核心调度器配合，实现策略的完全解耦
"""

import os
import logging
import importlib
import sys
from pathlib import Path
from typing import Dict, List, Any, Optional

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

# 配置日志
logger = logging.getLogger(__name__)

class StrategyManager:
    """策略管理类 - 单例模式"""
    
    _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._strategies = {}
        self._load_strategies()
    
    def _load_strategies(self):
        """加载所有选股策略"""
        try:
            # 策略模块目录
            strategy_dir = Path(__file__).parent / "strategies"
            
            # 遍历策略目录，加载所有策略类
            for strategy_file in strategy_dir.glob("*_strategy.py"):
                if strategy_file.name == "base_strategy.py":
                    continue
                    
                try:
                    # 导入策略模块
                    module_name = f"src.strategies.{strategy_file.stem}"
                    module = importlib.import_module(module_name)
                    
                    # 获取策略类（类名应与文件名对应，根据策略文件的实际类名）
                    # 对于 vcp_strategy.py，类名是 VCPStrategy
                    if strategy_file.stem == "vcp_strategy":
                        strategy_class_name = "VCPStrategy"
                    else:
                        strategy_class_name = ''.join([word.capitalize() for word in strategy_file.stem.split("_")])
                    
                    if hasattr(module, strategy_class_name):
                        strategy_class = getattr(module, strategy_class_name)
                        
                        # 创建策略实例
                        strategy_name = strategy_file.stem.split("_")[0]
                        strategy_instance = strategy_class()
                        
                        # 注册策略
                        self._strategies[strategy_name] = strategy_instance
                        logger.info(f"策略加载成功: {strategy_name} - {strategy_instance.get_description()}")
                    else:
                        logger.warning(f"策略模块 {strategy_file.name} 中未找到策略类 {strategy_class_name}")
                
                except Exception as e:
                    logger.error(f"策略加载失败: {strategy_file.name} - {e}")
                    continue
            
            logger.info(f"共加载 {len(self._strategies)} 个选股策略")
            
        except Exception as e:
            logger.error(f"策略加载失败: {e}")
    
    def get_strategy(self, strategy_name: str) -> Optional[Any]:
        """
        获取指定名称的选股策略实例
        
        参数:
            strategy_name: 策略名称
            
        返回:
            Optional[BaseStrategy]: 策略实例或None
        """
        try:
            if strategy_name not in self._strategies:
                logger.error(f"未找到指定策略: {strategy_name}")
                return None
            
            logger.debug(f"策略获取成功: {strategy_name}")
            return self._strategies[strategy_name]
            
        except Exception as e:
            logger.error(f"策略获取失败: {strategy_name} - {e}")
            return None
    
    def list_strategies(self) -> List[Dict[str, Any]]:
        """
        列出所有可用的选股策略
        
        返回:
            List[Dict[str, Any]]: 策略信息列表
        """
        strategies = []
        
        for strategy_name, strategy_instance in self._strategies.items():
            strategies.append({
                'name': strategy_name,
                'class': strategy_instance.__class__.__name__,
                'description': strategy_instance.get_description(),
                'params': strategy_instance.get_params()
            })
        
        return strategies
    
    def register_strategy(self, strategy_name: str, strategy_instance: Any) -> bool:
        """
        注册新的选股策略
        
        参数:
            strategy_name: 策略名称
            strategy_instance: 策略实例
            
        返回:
            bool: 注册是否成功
        """
        try:
            if strategy_name in self._strategies:
                logger.warning(f"策略已存在: {strategy_name}，将被覆盖")
            
            self._strategies[strategy_name] = strategy_instance
            logger.info(f"策略注册成功: {strategy_name} - {strategy_instance.get_description()}")
            return True
            
        except Exception as e:
            logger.error(f"策略注册失败: {e}")
            return False
    
    def unregister_strategy(self, strategy_name: str) -> bool:
        """
        取消注册选股策略
        
        参数:
            strategy_name: 策略名称
            
        返回:
            bool: 取消注册是否成功
        """
        try:
            if strategy_name in self._strategies:
                del self._strategies[strategy_name]
                logger.info(f"策略取消注册成功: {strategy_name}")
                return True
            else:
                logger.warning(f"未找到指定策略: {strategy_name}")
                return False
            
        except Exception as e:
            logger.error(f"策略取消注册失败: {e}")
            return False
    
    def set_strategy_params(self, strategy_name: str, params: Dict[str, Any]) -> bool:
        """
        设置策略参数
        
        参数:
            strategy_name: 策略名称
            params: 参数配置
            
        返回:
            bool: 是否成功
        """
        try:
            strategy = self.get_strategy(strategy_name)
            if strategy:
                strategy.set_params(params)
                return True
            return False
            
        except Exception as e:
            logger.error(f"策略参数设置失败: {strategy_name} - {e}")
            return False
    
    def get_strategy_params(self, strategy_name: str) -> Dict[str, Any]:
        """
        获取策略参数
        
        参数:
            strategy_name: 策略名称
            
        返回:
            Dict[str, Any]: 参数配置
        """
        try:
            strategy = self.get_strategy(strategy_name)
            if strategy:
                return strategy.get_params()
            return {}
            
        except Exception as e:
            logger.error(f"策略参数获取失败: {strategy_name} - {e}")
            return {}

# 创建策略管理器实例（单例）
strategy_manager = StrategyManager()