""" Random Strategy for Testing This strategy generates random entry and exit signals for testing the strategy system. It's useful for verifying that the strategy framework is working correctly. """ import random import logging from typing import Dict, List, Optional import pandas as pd from .base import StrategyBase, StrategySignal logger = logging.getLogger(__name__) class RandomStrategy(StrategyBase): """ Random signal generator strategy for testing. This strategy generates random entry and exit signals with configurable probability and confidence levels. It's designed to test the strategy framework and signal processing system. Parameters: entry_probability: Probability of generating an entry signal (0.0-1.0) exit_probability: Probability of generating an exit signal (0.0-1.0) min_confidence: Minimum confidence level for signals max_confidence: Maximum confidence level for signals timeframe: Timeframe to operate on (default: "1min") signal_frequency: How often to generate signals (every N bars) """ def __init__(self, weight: float = 1.0, params: Optional[Dict] = None): """Initialize the random strategy.""" super().__init__("random", weight, params) # Strategy parameters with defaults self.entry_probability = self.params.get("entry_probability", 0.05) # 5% chance per bar self.exit_probability = self.params.get("exit_probability", 0.1) # 10% chance per bar self.min_confidence = self.params.get("min_confidence", 0.6) self.max_confidence = self.params.get("max_confidence", 0.9) self.timeframe = self.params.get("timeframe", "1min") self.signal_frequency = self.params.get("signal_frequency", 1) # Every bar # Internal state self.bar_count = 0 self.last_signal_bar = -1 self.last_processed_timestamp = None # Track last processed timestamp to avoid duplicates logger.info(f"RandomStrategy initialized with entry_prob={self.entry_probability}, " f"exit_prob={self.exit_probability}, timeframe={self.timeframe}") def get_timeframes(self) -> List[str]: """Return required timeframes for this strategy.""" return [self.timeframe, "1min"] # Always include 1min for precision def initialize(self, backtester) -> None: """Initialize strategy with backtester data.""" try: logger.info(f"RandomStrategy: Starting initialization...") # Resample data to required timeframes self._resample_data(backtester.original_df) # Get primary timeframe data self.df = self.get_primary_timeframe_data() if self.df is None or self.df.empty: raise ValueError(f"No data available for timeframe {self.timeframe}") # Reset internal state self.bar_count = 0 self.last_signal_bar = -1 self.initialized = True logger.info(f"RandomStrategy initialized with {len(self.df)} bars on {self.timeframe}") logger.info(f"RandomStrategy: Data range from {self.df.index[0]} to {self.df.index[-1]}") except Exception as e: logger.error(f"Failed to initialize RandomStrategy: {e}") logger.error(f"RandomStrategy: backtester.original_df shape: {backtester.original_df.shape if hasattr(backtester, 'original_df') else 'No original_df'}") raise def get_entry_signal(self, backtester, df_index: int) -> StrategySignal: """Generate random entry signals.""" if not self.initialized: logger.warning(f"RandomStrategy: get_entry_signal called but not initialized") return StrategySignal("HOLD", 0.0) try: # Get current timestamp to avoid duplicate signals current_timestamp = None if hasattr(backtester, 'original_df') and not backtester.original_df.empty: current_timestamp = backtester.original_df.index[-1] # Skip if we already processed this timestamp if current_timestamp and self.last_processed_timestamp == current_timestamp: return StrategySignal("HOLD", 0.0) self.bar_count += 1 # Debug logging every 10 bars if self.bar_count % 10 == 0: logger.info(f"RandomStrategy: Processing bar {self.bar_count}, df_index={df_index}, timestamp={current_timestamp}") # Check if we should generate a signal based on frequency if (self.bar_count - self.last_signal_bar) < self.signal_frequency: return StrategySignal("HOLD", 0.0) # Generate random entry signal random_value = random.random() if random_value < self.entry_probability: confidence = random.uniform(self.min_confidence, self.max_confidence) self.last_signal_bar = self.bar_count self.last_processed_timestamp = current_timestamp # Update last processed timestamp # Get current price from backtester's original data (more reliable) try: if hasattr(backtester, 'original_df') and not backtester.original_df.empty: # Use the last available price from the original data current_price = backtester.original_df['close'].iloc[-1] elif hasattr(backtester, 'df') and not backtester.df.empty: # Fallback to backtester's main dataframe current_price = backtester.df['close'].iloc[min(df_index, len(backtester.df)-1)] else: # Last resort: use our internal dataframe current_price = self.df.iloc[min(df_index, len(self.df)-1)]['close'] except (IndexError, KeyError) as e: logger.warning(f"RandomStrategy: Error getting current price: {e}, using fallback") current_price = self.df.iloc[-1]['close'] if not self.df.empty else 50000.0 logger.info(f"RandomStrategy: Generated ENTRY signal at bar {self.bar_count}, " f"price=${current_price:.2f}, confidence={confidence:.2f}, random_value={random_value:.3f}") return StrategySignal( "ENTRY", confidence=confidence, price=current_price, metadata={ "strategy": "random", "bar_count": self.bar_count, "timeframe": self.timeframe } ) # Update timestamp even if no signal generated if current_timestamp: self.last_processed_timestamp = current_timestamp return StrategySignal("HOLD", 0.0) except Exception as e: logger.error(f"RandomStrategy entry signal error: {e}") return StrategySignal("HOLD", 0.0) def get_exit_signal(self, backtester, df_index: int) -> StrategySignal: """Generate random exit signals.""" if not self.initialized: return StrategySignal("HOLD", 0.0) try: # Only generate exit signals if we have an open position # This is handled by the strategy trader, but we can add logic here # Generate random exit signal if random.random() < self.exit_probability: confidence = random.uniform(self.min_confidence, self.max_confidence) # Get current price from backtester's original data (more reliable) try: if hasattr(backtester, 'original_df') and not backtester.original_df.empty: # Use the last available price from the original data current_price = backtester.original_df['close'].iloc[-1] elif hasattr(backtester, 'df') and not backtester.df.empty: # Fallback to backtester's main dataframe current_price = backtester.df['close'].iloc[min(df_index, len(backtester.df)-1)] else: # Last resort: use our internal dataframe current_price = self.df.iloc[min(df_index, len(self.df)-1)]['close'] except (IndexError, KeyError) as e: logger.warning(f"RandomStrategy: Error getting current price for exit: {e}, using fallback") current_price = self.df.iloc[-1]['close'] if not self.df.empty else 50000.0 # Randomly choose exit type exit_types = ["SELL_SIGNAL", "TAKE_PROFIT", "STOP_LOSS"] exit_type = random.choice(exit_types) logger.info(f"RandomStrategy: Generated EXIT signal at bar {self.bar_count}, " f"price=${current_price:.2f}, confidence={confidence:.2f}, type={exit_type}") return StrategySignal( "EXIT", confidence=confidence, price=current_price, metadata={ "type": exit_type, "strategy": "random", "bar_count": self.bar_count, "timeframe": self.timeframe } ) return StrategySignal("HOLD", 0.0) except Exception as e: logger.error(f"RandomStrategy exit signal error: {e}") return StrategySignal("HOLD", 0.0) def get_confidence(self, backtester, df_index: int) -> float: """Return random confidence level.""" return random.uniform(self.min_confidence, self.max_confidence) def __repr__(self) -> str: """String representation of the strategy.""" return (f"RandomStrategy(entry_prob={self.entry_probability}, " f"exit_prob={self.exit_probability}, timeframe={self.timeframe})")