""" Strategy Manager This module contains the StrategyManager class that orchestrates multiple trading strategies and combines their signals using configurable aggregation rules. The StrategyManager supports various combination methods for entry and exit signals: - Entry: any, all, majority, weighted_consensus - Exit: any, all, priority (with stop loss prioritization) """ from typing import Dict, List, Tuple, Optional import logging from .base import StrategyBase, StrategySignal from .default_strategy import DefaultStrategy from .bbrs_strategy import BBRSStrategy from .random_strategy import RandomStrategy class StrategyManager: """ Manages multiple strategies and combines their signals. The StrategyManager loads multiple strategies from configuration, initializes them with backtester data, and combines their signals using configurable aggregation rules. Attributes: strategies (List[StrategyBase]): List of loaded strategies combination_rules (Dict): Rules for combining signals initialized (bool): Whether manager has been initialized Example: config = { "strategies": [ {"name": "default", "weight": 0.6, "params": {}}, {"name": "bbrs", "weight": 0.4, "params": {"bb_width": 0.05}} ], "combination_rules": { "entry": "weighted_consensus", "exit": "any", "min_confidence": 0.6 } } manager = StrategyManager(config["strategies"], config["combination_rules"]) """ def __init__(self, strategies_config: List[Dict], combination_rules: Optional[Dict] = None): """ Initialize the strategy manager. Args: strategies_config: List of strategy configurations combination_rules: Rules for combining signals """ self.strategies = self._load_strategies(strategies_config) self.combination_rules = combination_rules or { "entry": "weighted_consensus", "exit": "any", "min_confidence": 0.5 } self.initialized = False def _load_strategies(self, strategies_config: List[Dict]) -> List[StrategyBase]: """ Load strategies from configuration. Creates strategy instances based on configuration and registers them with the manager. Supports extensible strategy registration. Args: strategies_config: List of strategy configurations Returns: List[StrategyBase]: List of instantiated strategies Raises: ValueError: If unknown strategy name is specified """ strategies = [] for config in strategies_config: name = config.get("name", "").lower() weight = config.get("weight", 1.0) params = config.get("params", {}) if name == "default": strategies.append(DefaultStrategy(weight, params)) elif name == "bbrs": strategies.append(BBRSStrategy(weight, params)) elif name == "random": strategies.append(RandomStrategy(weight, params)) else: raise ValueError(f"Unknown strategy: {name}. " f"Available strategies: default, bbrs, random") return strategies def initialize(self, backtester) -> None: """ Initialize all strategies with backtester data. Calls the initialize method on each strategy, allowing them to set up indicators, validate data, and prepare for trading. Each strategy will handle its own timeframe resampling. Args: backtester: Backtest instance with OHLCV data """ for strategy in self.strategies: try: strategy.initialize(backtester) # Log strategy timeframe information timeframes = strategy.get_timeframes() logging.info(f"Initialized strategy: {strategy.name} with timeframes: {timeframes}") except Exception as e: logging.error(f"Failed to initialize strategy {strategy.name}: {e}") raise self.initialized = True logging.info(f"Strategy manager initialized with {len(self.strategies)} strategies") # Log summary of all timeframes being used all_timeframes = set() for strategy in self.strategies: all_timeframes.update(strategy.get_timeframes()) logging.info(f"Total unique timeframes in use: {sorted(all_timeframes)}") def get_entry_signal(self, backtester, df_index: int) -> bool: """ Get combined entry signal from all strategies. Collects entry signals from all strategies and combines them according to the configured combination rules. Args: backtester: Backtest instance with current state df_index: Current index in the dataframe Returns: bool: True if combined signal suggests entry, False otherwise """ if not self.initialized: return False # Collect signals from all strategies signals = {} for strategy in self.strategies: try: signal = strategy.get_entry_signal(backtester, df_index) signals[strategy.name] = { "signal": signal, "weight": strategy.weight, "confidence": signal.confidence } except Exception as e: logging.warning(f"Strategy {strategy.name} entry signal failed: {e}") signals[strategy.name] = { "signal": StrategySignal("HOLD", 0.0), "weight": strategy.weight, "confidence": 0.0 } return self._combine_entry_signals(signals) def get_exit_signal(self, backtester, df_index: int) -> Tuple[Optional[str], Optional[float]]: """ Get combined exit signal from all strategies. Collects exit signals from all strategies and combines them according to the configured combination rules. Args: backtester: Backtest instance with current state df_index: Current index in the dataframe Returns: Tuple[Optional[str], Optional[float]]: (exit_type, exit_price) or (None, None) """ if not self.initialized: return None, None # Collect signals from all strategies signals = {} for strategy in self.strategies: try: signal = strategy.get_exit_signal(backtester, df_index) signals[strategy.name] = { "signal": signal, "weight": strategy.weight, "confidence": signal.confidence } except Exception as e: logging.warning(f"Strategy {strategy.name} exit signal failed: {e}") signals[strategy.name] = { "signal": StrategySignal("HOLD", 0.0), "weight": strategy.weight, "confidence": 0.0 } return self._combine_exit_signals(signals) def _combine_entry_signals(self, signals: Dict) -> bool: """ Combine entry signals based on combination rules. Supports multiple combination methods: - any: Enter if ANY strategy signals entry - all: Enter only if ALL strategies signal entry - majority: Enter if majority of strategies signal entry - weighted_consensus: Enter based on weighted average confidence Args: signals: Dictionary of strategy signals with weights and confidence Returns: bool: Combined entry decision """ method = self.combination_rules.get("entry", "weighted_consensus") min_confidence = self.combination_rules.get("min_confidence", 0.5) # Filter for entry signals above minimum confidence entry_signals = [ s for s in signals.values() if s["signal"].signal_type == "ENTRY" and s["signal"].confidence >= min_confidence ] if not entry_signals: return False if method == "any": # Enter if any strategy signals entry return len(entry_signals) > 0 elif method == "all": # Enter only if all strategies signal entry return len(entry_signals) == len(self.strategies) elif method == "majority": # Enter if majority of strategies signal entry return len(entry_signals) > len(self.strategies) / 2 elif method == "weighted_consensus": # Enter based on weighted average confidence total_weight = sum(s["weight"] for s in entry_signals) if total_weight == 0: return False weighted_confidence = sum( s["signal"].confidence * s["weight"] for s in entry_signals ) / total_weight return weighted_confidence >= min_confidence else: logging.warning(f"Unknown entry combination method: {method}, using 'any'") return len(entry_signals) > 0 def _combine_exit_signals(self, signals: Dict) -> Tuple[Optional[str], Optional[float]]: """ Combine exit signals based on combination rules. Supports multiple combination methods: - any: Exit if ANY strategy signals exit (recommended for risk management) - all: Exit only if ALL strategies agree on exit - priority: Exit based on priority order (STOP_LOSS > SELL_SIGNAL > others) Args: signals: Dictionary of strategy signals with weights and confidence Returns: Tuple[Optional[str], Optional[float]]: (exit_type, exit_price) or (None, None) """ method = self.combination_rules.get("exit", "any") # Filter for exit signals exit_signals = [ s for s in signals.values() if s["signal"].signal_type == "EXIT" ] if not exit_signals: return None, None if method == "any": # Exit if any strategy signals exit (first one found) for signal_data in exit_signals: signal = signal_data["signal"] exit_type = signal.metadata.get("type", "EXIT") return exit_type, signal.price elif method == "all": # Exit only if all strategies agree on exit if len(exit_signals) == len(self.strategies): signal = exit_signals[0]["signal"] exit_type = signal.metadata.get("type", "EXIT") return exit_type, signal.price elif method == "priority": # Priority order: STOP_LOSS > SELL_SIGNAL > others stop_loss_signals = [ s for s in exit_signals if s["signal"].metadata.get("type") == "STOP_LOSS" ] if stop_loss_signals: signal = stop_loss_signals[0]["signal"] return "STOP_LOSS", signal.price sell_signals = [ s for s in exit_signals if s["signal"].metadata.get("type") == "SELL_SIGNAL" ] if sell_signals: signal = sell_signals[0]["signal"] return "SELL_SIGNAL", signal.price # Return first available exit signal signal = exit_signals[0]["signal"] exit_type = signal.metadata.get("type", "EXIT") return exit_type, signal.price else: logging.warning(f"Unknown exit combination method: {method}, using 'any'") # Fallback to 'any' method signal = exit_signals[0]["signal"] exit_type = signal.metadata.get("type", "EXIT") return exit_type, signal.price return None, None def get_strategy_summary(self) -> Dict: """ Get summary of loaded strategies and their configuration. Returns: Dict: Summary of strategies, weights, combination rules, and timeframes """ return { "strategies": [ { "name": strategy.name, "weight": strategy.weight, "params": strategy.params, "timeframes": strategy.get_timeframes(), "initialized": strategy.initialized } for strategy in self.strategies ], "combination_rules": self.combination_rules, "total_strategies": len(self.strategies), "initialized": self.initialized, "all_timeframes": list(set().union(*[strategy.get_timeframes() for strategy in self.strategies])) } def __repr__(self) -> str: """String representation of the strategy manager.""" strategy_names = [s.name for s in self.strategies] return (f"StrategyManager(strategies={strategy_names}, " f"initialized={self.initialized})") def create_strategy_manager(config: Dict) -> StrategyManager: """ Factory function to create StrategyManager from configuration. Provides a convenient way to create a StrategyManager instance from a configuration dictionary. Args: config: Configuration dictionary with strategies and combination_rules Returns: StrategyManager: Configured strategy manager instance Example: config = { "strategies": [ {"name": "default", "weight": 1.0, "params": {}} ], "combination_rules": { "entry": "any", "exit": "any" } } manager = create_strategy_manager(config) """ strategies_config = config.get("strategies", []) combination_rules = config.get("combination_rules", {}) if not strategies_config: raise ValueError("No strategies specified in configuration") return StrategyManager(strategies_config, combination_rules)