From d499c5b8d02de6bfa20e8e26cd5ef21f8834ab61 Mon Sep 17 00:00:00 2001 From: Ajasra Date: Sun, 25 May 2025 18:42:47 +0800 Subject: [PATCH] Add RandomStrategy implementation and update strategy manager --- cycles/strategies/__init__.py | 2 + cycles/strategies/manager.py | 5 +- cycles/strategies/random_strategy.py | 218 +++++++++++++++++++++++++++ 3 files changed, 224 insertions(+), 1 deletion(-) create mode 100644 cycles/strategies/random_strategy.py diff --git a/cycles/strategies/__init__.py b/cycles/strategies/__init__.py index 9360697..febb3d5 100644 --- a/cycles/strategies/__init__.py +++ b/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' ] diff --git a/cycles/strategies/manager.py b/cycles/strategies/manager.py index 6684541..9bf5547 100644 --- a/cycles/strategies/manager.py +++ b/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 diff --git a/cycles/strategies/random_strategy.py b/cycles/strategies/random_strategy.py new file mode 100644 index 0000000..ab4430f --- /dev/null +++ b/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})") \ No newline at end of file