""" Strategy Factory Module for Strategy Management This module provides strategy calculation and signal generation capabilities designed to work with the TCP Trading Platform's technical indicators and market data. IMPORTANT: Strategy-Indicator Integration - Strategies consume pre-calculated technical indicators - Uses BaseStrategy abstract class pattern for consistency - Supports dynamic strategy loading and registration - Follows right-aligned timestamp convention - Integrates with existing database and configuration systems """ from typing import Dict, List, Optional, Any import pandas as pd from data.common.data_types import OHLCVCandle from data.common.indicators import TechnicalIndicators from .base import BaseStrategy from .data_types import StrategyResult from .utils import create_indicator_key from .implementations import ( EMAStrategy, RSIStrategy, MACDStrategy ) class StrategyFactory: """ Strategy factory for creating and managing trading strategies. This class provides strategy instantiation, calculation orchestration, and signal generation. It integrates with the TechnicalIndicators system to provide pre-calculated indicator data to strategies. STRATEGY-INDICATOR INTEGRATION: - Strategies declare required indicators via get_required_indicators() - Factory pre-calculates all required indicators - Strategies receive indicator data as dictionary of DataFrames - Results maintain original timestamp alignment """ def __init__(self, logger=None): """ Initialize strategy factory. Args: logger: Optional logger instance """ self.logger = logger self.technical_indicators = TechnicalIndicators(logger) # Registry of available strategies (will be populated as strategies are implemented) self._strategy_registry = { 'ema_crossover': EMAStrategy, 'rsi': RSIStrategy, 'macd': MACDStrategy } if self.logger: self.logger.info("StrategyFactory: Initialized strategy factory") def register_strategy(self, name: str, strategy_class: type) -> None: """ Register a new strategy class in the factory. Args: name: Strategy name identifier strategy_class: Strategy class (must inherit from BaseStrategy) """ if not issubclass(strategy_class, BaseStrategy): raise ValueError(f"Strategy class {strategy_class} must inherit from BaseStrategy") self._strategy_registry[name] = strategy_class if self.logger: self.logger.info(f"StrategyFactory: Registered strategy '{name}'") def get_available_strategies(self) -> List[str]: """ Get list of available strategy names. Returns: List of registered strategy names """ return list(self._strategy_registry.keys()) def create_strategy(self, strategy_name: str) -> Optional[BaseStrategy]: """ Create a strategy instance by name. Args: strategy_name: Name of the strategy to create Returns: Strategy instance or None if strategy not found """ strategy_class = self._strategy_registry.get(strategy_name) if not strategy_class: if self.logger: self.logger.error(f"Unknown strategy: {strategy_name}") return None try: return strategy_class(logger=self.logger) except Exception as e: if self.logger: self.logger.error(f"Error creating strategy {strategy_name}: {e}") return None def calculate_strategy_signals(self, strategy_name: str, df: pd.DataFrame, strategy_config: Dict[str, Any]) -> List[StrategyResult]: """ Calculate signals for a specific strategy. Args: strategy_name: Name of the strategy to execute df: DataFrame with OHLCV data strategy_config: Strategy-specific configuration parameters Returns: List of strategy results with signals """ # Create strategy instance strategy = self.create_strategy(strategy_name) if not strategy: return [] try: # Get required indicators for this strategy required_indicators = strategy.get_required_indicators() # Pre-calculate all required indicators indicators_data = self._calculate_required_indicators(df, required_indicators) # Calculate strategy signals results = strategy.calculate(df, indicators_data, **strategy_config) return results except Exception as e: if self.logger: self.logger.error(f"Error calculating strategy {strategy_name}: {e}") return [] # def calculate_multiple_strategies(self, df: pd.DataFrame, # strategies_config: Dict[str, Dict[str, Any]]) -> Dict[str, List[StrategyResult]]: # """ # Calculate signals for multiple strategies efficiently. # # Args: # df: DataFrame with OHLCV data # strategies_config: Configuration for strategies to calculate # Example: { # 'ema_cross_1': {'strategy': 'ema_crossover', 'fast_period': 12, 'slow_period': 26}, # 'rsi_momentum': {'strategy': 'rsi', 'period': 14, 'oversold': 30, 'overbought': 70} # } # # Returns: # Dictionary mapping strategy instance names to their results # """ # results = {} # # for strategy_instance_name, config in strategies_config.items(): # strategy_name = config.get('strategy') # if not strategy_name: # if self.logger: # self.logger.warning(f"No strategy specified for {strategy_instance_name}") # results[strategy_instance_name] = [] # continue # # # Extract strategy parameters (exclude 'strategy' key) # strategy_params = {k: v for k, v in config.items() if k != 'strategy'} # # try: # strategy_results = self.calculate_strategy_signals( # strategy_name, df, strategy_params # ) # results[strategy_instance_name] = strategy_results # # except Exception as e: # if self.logger: # self.logger.error(f"Error calculating strategy {strategy_instance_name}: {e}") # results[strategy_instance_name] = [] # # return results def _calculate_required_indicators(self, df: pd.DataFrame, required_indicators: List[Dict[str, Any]]) -> Dict[str, pd.DataFrame]: """ Pre-calculate all indicators required by a strategy. Args: df: DataFrame with OHLCV data required_indicators: List of indicator configurations Returns: Dictionary of indicator DataFrames keyed by indicator name """ indicators_data = {} for indicator_config in required_indicators: indicator_type = indicator_config.get('type') if not indicator_type: continue # Create a unique key for this indicator configuration indicator_key = self._create_indicator_key(indicator_config) try: # Calculate the indicator using TechnicalIndicators indicator_result = self.technical_indicators.calculate( indicator_type, df, **{k: v for k, v in indicator_config.items() if k != 'type'} ) if indicator_result is not None and not indicator_result.empty: indicators_data[indicator_key] = indicator_result else: if self.logger: self.logger.warning(f"Empty result for indicator: {indicator_key}") indicators_data[indicator_key] = pd.DataFrame() except Exception as e: if self.logger: self.logger.error(f"Error calculating indicator {indicator_key}: {e}") indicators_data[indicator_key] = pd.DataFrame() return indicators_data def _create_indicator_key(self, indicator_config: Dict[str, Any]) -> str: """ Create a unique key for an indicator configuration. Delegates to shared utility function to ensure consistency. Args: indicator_config: Indicator configuration Returns: Unique string key for this indicator configuration """ return create_indicator_key(indicator_config)