Add RandomStrategy implementation and update strategy manager

cycles/strategies/__init__.py

@@ -25,6 +25,7 @@ Usage:
 from .base import StrategyBase, StrategySignal
 from .default_strategy import DefaultStrategy
 from .bbrs_strategy import BBRSStrategy
+from .random_strategy import RandomStrategy
 from .manager import StrategyManager, create_strategy_manager
 
 __all__ = [
@@ -32,6 +33,7 @@ __all__ = [
     'StrategySignal', 
     'DefaultStrategy',
     'BBRSStrategy',
+    'RandomStrategy',
     'StrategyManager',
     'create_strategy_manager'
 ]

cycles/strategies/manager.py

@@ -15,6 +15,7 @@ import logging
 from .base import StrategyBase, StrategySignal
 from .default_strategy import DefaultStrategy
 from .bbrs_strategy import BBRSStrategy
+from .random_strategy import RandomStrategy
 
 
 class StrategyManager:
@@ -88,9 +89,11 @@ class StrategyManager:
                 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")
+                               f"Available strategies: default, bbrs, random")
         
         return strategies
     

cycles/strategies/random_strategy.py

@@ -0,0 +1,218 @@
+"""
+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})")