From 235098c0453cc341f714368f73087636d78d9142 Mon Sep 17 00:00:00 2001 From: "Vasily.onl" Date: Fri, 23 May 2025 16:41:08 +0800 Subject: [PATCH] Add strategy management system with multiple trading strategies - Introduced a new strategies module containing the StrategyManager class to orchestrate multiple trading strategies. - Implemented StrategyBase and StrategySignal as foundational components for strategy development. - Added DefaultStrategy for meta-trend analysis and BBRSStrategy for Bollinger Bands + RSI trading. - Enhanced documentation to provide clear usage examples and configuration guidelines for the new system. - Established a modular architecture to support future strategy additions and improvements. --- cycles/strategies/__init__.py | 40 +++ cycles/strategies/base.py | 162 +++++++++++ cycles/strategies/bbrs_strategy.py | 316 +++++++++++++++++++++ cycles/strategies/default_strategy.py | 219 +++++++++++++++ cycles/strategies/manager.py | 381 ++++++++++++++++++++++++++ docs/strategy_manager.md | 330 ++++++++++++++++++++++ 6 files changed, 1448 insertions(+) create mode 100644 cycles/strategies/__init__.py create mode 100644 cycles/strategies/base.py create mode 100644 cycles/strategies/bbrs_strategy.py create mode 100644 cycles/strategies/default_strategy.py create mode 100644 cycles/strategies/manager.py create mode 100644 docs/strategy_manager.md diff --git a/cycles/strategies/__init__.py b/cycles/strategies/__init__.py new file mode 100644 index 0000000..9360697 --- /dev/null +++ b/cycles/strategies/__init__.py @@ -0,0 +1,40 @@ +""" +Strategies Module + +This module contains the strategy management system for trading strategies. +It provides a flexible framework for implementing, combining, and managing multiple trading strategies. + +Components: +- StrategyBase: Abstract base class for all strategies +- DefaultStrategy: Meta-trend based strategy +- BBRSStrategy: Bollinger Bands + RSI strategy +- StrategyManager: Orchestrates multiple strategies +- StrategySignal: Represents trading signals with confidence levels + +Usage: + from cycles.strategies import StrategyManager, create_strategy_manager + + # Create strategy manager from config + strategy_manager = create_strategy_manager(config) + + # Or create individual strategies + from cycles.strategies import DefaultStrategy, BBRSStrategy + default_strategy = DefaultStrategy(weight=1.0, params={}) +""" + +from .base import StrategyBase, StrategySignal +from .default_strategy import DefaultStrategy +from .bbrs_strategy import BBRSStrategy +from .manager import StrategyManager, create_strategy_manager + +__all__ = [ + 'StrategyBase', + 'StrategySignal', + 'DefaultStrategy', + 'BBRSStrategy', + 'StrategyManager', + 'create_strategy_manager' +] + +__version__ = '1.0.0' +__author__ = 'TCP Cycles Team' \ No newline at end of file diff --git a/cycles/strategies/base.py b/cycles/strategies/base.py new file mode 100644 index 0000000..3aca909 --- /dev/null +++ b/cycles/strategies/base.py @@ -0,0 +1,162 @@ +""" +Base classes for the strategy management system. + +This module contains the fundamental building blocks for all trading strategies: +- StrategySignal: Represents trading signals with confidence and metadata +- StrategyBase: Abstract base class that all strategies must inherit from +""" + +from abc import ABC, abstractmethod +from typing import Dict, Optional + + +class StrategySignal: + """ + Represents a trading signal from a strategy. + + A signal encapsulates the strategy's recommendation along with confidence + level, optional price target, and additional metadata. + + Attributes: + signal_type (str): Type of signal - "ENTRY", "EXIT", or "HOLD" + confidence (float): Confidence level from 0.0 to 1.0 + price (Optional[float]): Optional specific price for the signal + metadata (Dict): Additional signal data and context + + Example: + # Entry signal with high confidence + signal = StrategySignal("ENTRY", confidence=0.8) + + # Exit signal with stop loss price + signal = StrategySignal("EXIT", confidence=1.0, price=50000, + metadata={"type": "STOP_LOSS"}) + """ + + def __init__(self, signal_type: str, confidence: float = 1.0, + price: Optional[float] = None, metadata: Optional[Dict] = None): + """ + Initialize a strategy signal. + + Args: + signal_type: Type of signal ("ENTRY", "EXIT", "HOLD") + confidence: Confidence level (0.0 to 1.0) + price: Optional specific price for the signal + metadata: Additional signal data and context + """ + self.signal_type = signal_type + self.confidence = max(0.0, min(1.0, confidence)) # Clamp to [0,1] + self.price = price + self.metadata = metadata or {} + + def __repr__(self) -> str: + """String representation of the signal.""" + return (f"StrategySignal(type={self.signal_type}, " + f"confidence={self.confidence:.2f}, " + f"price={self.price}, metadata={self.metadata})") + + +class StrategyBase(ABC): + """ + Abstract base class for all trading strategies. + + This class defines the interface that all strategies must implement: + - initialize(): Setup strategy with backtester data + - get_entry_signal(): Generate entry signals + - get_exit_signal(): Generate exit signals + - get_confidence(): Optional confidence calculation + + Attributes: + name (str): Strategy name + weight (float): Strategy weight for combination + params (Dict): Strategy parameters + initialized (bool): Whether strategy has been initialized + + Example: + class MyStrategy(StrategyBase): + def initialize(self, backtester): + # Setup strategy indicators + self.initialized = True + + def get_entry_signal(self, backtester, df_index): + # Return StrategySignal based on analysis + if should_enter: + return StrategySignal("ENTRY", confidence=0.7) + return StrategySignal("HOLD", confidence=0.0) + """ + + def __init__(self, name: str, weight: float = 1.0, params: Optional[Dict] = None): + """ + Initialize the strategy base. + + Args: + name: Strategy name/identifier + weight: Strategy weight for combination (default: 1.0) + params: Strategy-specific parameters + """ + self.name = name + self.weight = weight + self.params = params or {} + self.initialized = False + + @abstractmethod + def initialize(self, backtester) -> None: + """ + Initialize strategy with backtester data. + + This method is called once before backtesting begins. + Strategies should setup indicators, validate data, and + set self.initialized = True when complete. + + Args: + backtester: Backtest instance with data and configuration + """ + pass + + @abstractmethod + def get_entry_signal(self, backtester, df_index: int) -> StrategySignal: + """ + Generate entry signal for the given data index. + + Args: + backtester: Backtest instance with current state + df_index: Current index in the dataframe + + Returns: + StrategySignal: Entry signal with confidence level + """ + pass + + @abstractmethod + def get_exit_signal(self, backtester, df_index: int) -> StrategySignal: + """ + Generate exit signal for the given data index. + + Args: + backtester: Backtest instance with current state + df_index: Current index in the dataframe + + Returns: + StrategySignal: Exit signal with confidence level + """ + pass + + def get_confidence(self, backtester, df_index: int) -> float: + """ + Get strategy confidence for the current market state. + + Default implementation returns 1.0. Strategies can override + this to provide dynamic confidence based on market conditions. + + Args: + backtester: Backtest instance with current state + df_index: Current index in the dataframe + + Returns: + float: Confidence level (0.0 to 1.0) + """ + return 1.0 + + def __repr__(self) -> str: + """String representation of the strategy.""" + return (f"{self.__class__.__name__}(name={self.name}, " + f"weight={self.weight}, initialized={self.initialized})") \ No newline at end of file diff --git a/cycles/strategies/bbrs_strategy.py b/cycles/strategies/bbrs_strategy.py new file mode 100644 index 0000000..309f427 --- /dev/null +++ b/cycles/strategies/bbrs_strategy.py @@ -0,0 +1,316 @@ +""" +Bollinger Bands + RSI Strategy (BBRS) + +This module implements a sophisticated trading strategy that combines Bollinger Bands +and RSI indicators with market regime detection. The strategy adapts its parameters +based on whether the market is trending or moving sideways. + +Key Features: +- Dynamic parameter adjustment based on market regime +- Bollinger Band squeeze detection +- RSI overbought/oversold conditions +- Market regime-specific thresholds +""" + +import pandas as pd +import numpy as np +import logging +from typing import Tuple, Optional + +from .base import StrategyBase, StrategySignal + + +class BBRSStrategy(StrategyBase): + """ + Bollinger Bands + RSI Strategy implementation. + + This strategy uses Bollinger Bands and RSI indicators with market regime detection + to generate trading signals. It adapts its parameters based on whether the market + is in a trending or sideways regime. + + Parameters: + bb_width (float): Bollinger Band width threshold (default: 0.05) + bb_period (int): Bollinger Band period (default: 20) + rsi_period (int): RSI calculation period (default: 14) + trending_rsi_threshold (list): RSI thresholds for trending market [low, high] + trending_bb_multiplier (float): BB multiplier for trending market + sideways_rsi_threshold (list): RSI thresholds for sideways market [low, high] + sideways_bb_multiplier (float): BB multiplier for sideways market + strategy_name (str): Strategy implementation name + SqueezeStrategy (bool): Enable squeeze strategy + stop_loss_pct (float): Stop loss percentage (default: 0.05) + + Example: + params = { + "bb_width": 0.05, + "bb_period": 20, + "rsi_period": 14, + "strategy_name": "MarketRegimeStrategy" + } + strategy = BBRSStrategy(weight=1.0, params=params) + """ + + def __init__(self, weight: float = 1.0, params: Optional[dict] = None): + """ + Initialize the BBRS strategy. + + Args: + weight: Strategy weight for combination (default: 1.0) + params: Strategy parameters for Bollinger Bands and RSI + """ + super().__init__("bbrs", weight, params) + + def initialize(self, backtester) -> None: + """ + Initialize BBRS strategy with signal processing. + + Sets up the strategy by: + 1. Initializing empty signal series + 2. Running the BBRS strategy processing if original data is available + 3. Resampling signals from 15-minute to 1-minute resolution + + Args: + backtester: Backtest instance with OHLCV data + """ + # Initialize empty signal series + backtester.strategies["buy_signals"] = pd.Series(False, index=range(len(backtester.df))) + backtester.strategies["sell_signals"] = pd.Series(False, index=range(len(backtester.df))) + backtester.strategies["stop_loss_pct"] = self.params.get("stop_loss_pct", 0.05) + + # Run strategy processing if original data is available + if hasattr(backtester, 'original_df'): + self._run_strategy_processing(backtester) + + self.initialized = True + + def _run_strategy_processing(self, backtester) -> None: + """ + Run the actual BBRS strategy processing. + + Uses the Strategy class from cycles.Analysis.strategies to process + the original dataframe and generate buy/sell signals based on + Bollinger Bands, RSI, and market regime detection. + + Args: + backtester: Backtest instance with original_df attribute + """ + from cycles.Analysis.strategies import Strategy + + # Configure strategy parameters with defaults + config_strategy = { + "bb_width": self.params.get("bb_width", 0.05), + "bb_period": self.params.get("bb_period", 20), + "rsi_period": self.params.get("rsi_period", 14), + "trending": { + "rsi_threshold": self.params.get("trending_rsi_threshold", [30, 70]), + "bb_std_dev_multiplier": self.params.get("trending_bb_multiplier", 2.5), + }, + "sideways": { + "rsi_threshold": self.params.get("sideways_rsi_threshold", [40, 60]), + "bb_std_dev_multiplier": self.params.get("sideways_bb_multiplier", 1.8), + }, + "strategy_name": self.params.get("strategy_name", "MarketRegimeStrategy"), + "SqueezeStrategy": self.params.get("SqueezeStrategy", True) + } + + # Run strategy processing + strategy = Strategy(config=config_strategy, logging=logging) + processed_data = strategy.run(backtester.original_df, config_strategy["strategy_name"]) + + # Store processed data for plotting and analysis + backtester.processed_data = processed_data + + if processed_data.empty: + # If strategy processing failed, keep empty signals + return + + # Extract signals from processed data + buy_signals_raw = processed_data.get('BuySignal', pd.Series(False, index=processed_data.index)).astype(bool) + sell_signals_raw = processed_data.get('SellSignal', pd.Series(False, index=processed_data.index)).astype(bool) + + # Resample signals from 15-minute to 1-minute resolution + self._resample_signals_to_1min(backtester, buy_signals_raw, sell_signals_raw) + + def _resample_signals_to_1min(self, backtester, buy_signals_raw, sell_signals_raw) -> None: + """ + Resample signals from 15-minute to 1-minute resolution. + + Takes the 15-minute signals and maps them to 1-minute timestamps + using forward-fill to maintain signal consistency. + + Args: + backtester: Backtest instance + buy_signals_raw: Raw buy signals from strategy processing + sell_signals_raw: Raw sell signals from strategy processing + """ + # Get the DatetimeIndex from the original 1-minute data + original_datetime_index = backtester.original_df.index + + # Reindex signals from 15-minute to 1-minute resolution using forward-fill + buy_signals_1min = buy_signals_raw.reindex(original_datetime_index, method='ffill').fillna(False) + sell_signals_1min = sell_signals_raw.reindex(original_datetime_index, method='ffill').fillna(False) + + # Convert to integer index to match backtest DataFrame + buy_condition = pd.Series(buy_signals_1min.values, index=range(len(buy_signals_1min))) + sell_condition = pd.Series(sell_signals_1min.values, index=range(len(sell_signals_1min))) + + # Ensure same length as backtest DataFrame + if len(buy_condition) != len(backtester.df): + target_length = len(backtester.df) + if len(buy_condition) > target_length: + # Truncate if longer + buy_condition = buy_condition[:target_length] + sell_condition = sell_condition[:target_length] + else: + # Pad with False if shorter + buy_values = buy_condition.values + sell_values = sell_condition.values + buy_values = np.pad(buy_values, (0, target_length - len(buy_values)), constant_values=False) + sell_values = np.pad(sell_values, (0, target_length - len(sell_values)), constant_values=False) + buy_condition = pd.Series(buy_values, index=range(target_length)) + sell_condition = pd.Series(sell_values, index=range(target_length)) + + # Store the resampled signals + backtester.strategies["buy_signals"] = buy_condition + backtester.strategies["sell_signals"] = sell_condition + + def get_entry_signal(self, backtester, df_index: int) -> StrategySignal: + """ + Generate entry signal based on BBRS buy signals. + + Entry occurs when the BBRS strategy processing has generated + a buy signal based on Bollinger Bands and RSI conditions. + + Args: + backtester: Backtest instance with current state + df_index: Current index in the dataframe + + Returns: + StrategySignal: Entry signal if buy condition met, hold otherwise + """ + if not self.initialized: + return StrategySignal("HOLD", confidence=0.0) + + if df_index >= len(backtester.strategies["buy_signals"]): + return StrategySignal("HOLD", confidence=0.0) + + if backtester.strategies["buy_signals"].iloc[df_index]: + # High confidence for BBRS buy signals + return StrategySignal("ENTRY", confidence=1.0) + + return StrategySignal("HOLD", confidence=0.0) + + def get_exit_signal(self, backtester, df_index: int) -> StrategySignal: + """ + Generate exit signal based on BBRS sell signals or stop loss. + + Exit occurs when: + 1. BBRS strategy generates a sell signal + 2. Stop loss is triggered based on price movement + + Args: + backtester: Backtest instance with current state + df_index: Current index in the dataframe + + Returns: + StrategySignal: Exit signal with type and price, or hold signal + """ + if not self.initialized: + return StrategySignal("HOLD", confidence=0.0) + + if df_index >= len(backtester.strategies["sell_signals"]): + return StrategySignal("HOLD", confidence=0.0) + + # Check for sell signal + if backtester.strategies["sell_signals"].iloc[df_index]: + return StrategySignal("EXIT", confidence=1.0, + metadata={"type": "SELL_SIGNAL"}) + + # Check for stop loss + stop_loss_result, sell_price = self._check_stop_loss(backtester) + if stop_loss_result: + return StrategySignal("EXIT", confidence=1.0, price=sell_price, + metadata={"type": "STOP_LOSS"}) + + return StrategySignal("HOLD", confidence=0.0) + + def get_confidence(self, backtester, df_index: int) -> float: + """ + Get strategy confidence based on signal strength and market conditions. + + Confidence is higher when signals are present and market conditions + are favorable for the BBRS strategy. + + Args: + backtester: Backtest instance with current state + df_index: Current index in the dataframe + + Returns: + float: Confidence level (0.0 to 1.0) + """ + if not self.initialized: + return 0.0 + + # Check if we have processed data for confidence calculation + if hasattr(backtester, 'processed_data') and not backtester.processed_data.empty: + # Could analyze RSI levels, BB position, etc. for dynamic confidence + # For now, return high confidence when signals are present + if (df_index < len(backtester.strategies["buy_signals"]) and + backtester.strategies["buy_signals"].iloc[df_index]): + return 1.0 + elif (df_index < len(backtester.strategies["sell_signals"]) and + backtester.strategies["sell_signals"].iloc[df_index]): + return 1.0 + + # Moderate confidence during neutral periods + return 0.5 + + def _check_stop_loss(self, backtester) -> Tuple[bool, Optional[float]]: + """ + Check if stop loss is triggered using BBRS-specific logic. + + Similar to default strategy but uses BBRS-specific stop loss percentage + and can be enhanced with additional BBRS-specific exit conditions. + + Args: + backtester: Backtest instance with current trade state + + Returns: + Tuple[bool, Optional[float]]: (stop_loss_triggered, sell_price) + """ + # Calculate stop loss price + stop_price = backtester.entry_price * (1 - backtester.strategies["stop_loss_pct"]) + + # Get minute-level data for precise stop loss checking + min1_df = backtester.original_df if hasattr(backtester, 'original_df') else backtester.min1_df + min1_index = min1_df.index + + # Find data range from entry to current time + start_candidates = min1_index[min1_index >= backtester.entry_time] + if len(start_candidates) == 0: + return False, None + + backtester.current_trade_min1_start_idx = start_candidates[0] + end_candidates = min1_index[min1_index <= backtester.current_date] + + if len(end_candidates) == 0: + return False, None + + backtester.current_min1_end_idx = end_candidates[-1] + + # Check if any candle in the range triggered stop loss + min1_slice = min1_df.loc[backtester.current_trade_min1_start_idx:backtester.current_min1_end_idx] + + if (min1_slice['low'] <= stop_price).any(): + # Find the first candle that triggered stop loss + stop_candle = min1_slice[min1_slice['low'] <= stop_price].iloc[0] + + # Use open price if it gapped below stop, otherwise use stop price + if stop_candle['open'] < stop_price: + sell_price = stop_candle['open'] + else: + sell_price = stop_price + + return True, sell_price + + return False, None \ No newline at end of file diff --git a/cycles/strategies/default_strategy.py b/cycles/strategies/default_strategy.py new file mode 100644 index 0000000..65fae0b --- /dev/null +++ b/cycles/strategies/default_strategy.py @@ -0,0 +1,219 @@ +""" +Default Meta-Trend Strategy + +This module implements the default trading strategy based on meta-trend analysis +using multiple Supertrend indicators. The strategy enters when trends align +and exits on trend reversal or stop loss. + +The meta-trend is calculated by comparing three Supertrend indicators: +- Entry: When meta-trend changes from != 1 to == 1 +- Exit: When meta-trend changes to -1 or stop loss is triggered +""" + +import numpy as np +from typing import Tuple, Optional + +from .base import StrategyBase, StrategySignal + + +class DefaultStrategy(StrategyBase): + """ + Default meta-trend strategy implementation. + + This strategy uses multiple Supertrend indicators to determine market direction. + It generates entry signals when all three Supertrend indicators align in an + upward direction, and exit signals when they reverse or stop loss is triggered. + + Parameters: + stop_loss_pct (float): Stop loss percentage (default: 0.03) + + Example: + strategy = DefaultStrategy(weight=1.0, params={"stop_loss_pct": 0.05}) + """ + + def __init__(self, weight: float = 1.0, params: Optional[dict] = None): + """ + Initialize the default strategy. + + Args: + weight: Strategy weight for combination (default: 1.0) + params: Strategy parameters including stop_loss_pct + """ + super().__init__("default", weight, params) + + def initialize(self, backtester) -> None: + """ + Initialize meta trend calculation using Supertrend indicators. + + Calculates the meta-trend by comparing three Supertrend indicators. + When all three agree on direction, meta-trend follows that direction. + Otherwise, meta-trend is neutral (0). + + Args: + backtester: Backtest instance with OHLCV data + """ + from cycles.Analysis.supertrend import Supertrends + + # Calculate Supertrend indicators + supertrends = Supertrends(backtester.df, verbose=False) + supertrend_results_list = supertrends.calculate_supertrend_indicators() + + # Extract trend arrays from each Supertrend + trends = [st['results']['trend'] for st in supertrend_results_list] + trends_arr = np.stack(trends, axis=1) + + # Calculate meta-trend: all three must agree for direction signal + meta_trend = np.where( + (trends_arr[:,0] == trends_arr[:,1]) & (trends_arr[:,1] == trends_arr[:,2]), + trends_arr[:,0], + 0 # Neutral when trends don't agree + ) + + # Store in backtester for access during trading + backtester.strategies["meta_trend"] = meta_trend + backtester.strategies["stop_loss_pct"] = self.params.get("stop_loss_pct", 0.03) + + self.initialized = True + + def get_entry_signal(self, backtester, df_index: int) -> StrategySignal: + """ + Generate entry signal based on meta-trend direction change. + + Entry occurs when meta-trend changes from != 1 to == 1, indicating + all Supertrend indicators now agree on upward direction. + + Args: + backtester: Backtest instance with current state + df_index: Current index in the dataframe + + Returns: + StrategySignal: Entry signal if trend aligns, hold signal otherwise + """ + if not self.initialized: + return StrategySignal("HOLD", 0.0) + + if df_index < 1: + return StrategySignal("HOLD", 0.0) + + # Check for meta-trend entry condition + prev_trend = backtester.strategies["meta_trend"][df_index - 1] + curr_trend = backtester.strategies["meta_trend"][df_index] + + if prev_trend != 1 and curr_trend == 1: + # Strong confidence when all indicators align for entry + return StrategySignal("ENTRY", confidence=1.0) + + return StrategySignal("HOLD", confidence=0.0) + + def get_exit_signal(self, backtester, df_index: int) -> StrategySignal: + """ + Generate exit signal based on meta-trend reversal or stop loss. + + Exit occurs when: + 1. Meta-trend changes to -1 (trend reversal) + 2. Stop loss is triggered based on price movement + + Args: + backtester: Backtest instance with current state + df_index: Current index in the dataframe + + Returns: + StrategySignal: Exit signal with type and price, or hold signal + """ + if not self.initialized: + return StrategySignal("HOLD", 0.0) + + if df_index < 1: + return StrategySignal("HOLD", 0.0) + + # Check for meta-trend exit signal + prev_trend = backtester.strategies["meta_trend"][df_index - 1] + curr_trend = backtester.strategies["meta_trend"][df_index] + + if prev_trend != 1 and curr_trend == -1: + return StrategySignal("EXIT", confidence=1.0, + metadata={"type": "META_TREND_EXIT_SIGNAL"}) + + # Check for stop loss + stop_loss_result, sell_price = self._check_stop_loss(backtester) + if stop_loss_result: + return StrategySignal("EXIT", confidence=1.0, price=sell_price, + metadata={"type": "STOP_LOSS"}) + + return StrategySignal("HOLD", confidence=0.0) + + def get_confidence(self, backtester, df_index: int) -> float: + """ + Get strategy confidence based on meta-trend strength. + + Higher confidence when meta-trend is strongly directional, + lower confidence during neutral periods. + + Args: + backtester: Backtest instance with current state + df_index: Current index in the dataframe + + Returns: + float: Confidence level (0.0 to 1.0) + """ + if not self.initialized or df_index >= len(backtester.strategies["meta_trend"]): + return 0.0 + + curr_trend = backtester.strategies["meta_trend"][df_index] + + # High confidence for strong directional signals + if curr_trend == 1 or curr_trend == -1: + return 1.0 + + # Low confidence for neutral trend + return 0.3 + + def _check_stop_loss(self, backtester) -> Tuple[bool, Optional[float]]: + """ + Check if stop loss is triggered based on price movement. + + Calculates stop loss price and checks if any candle since entry + has triggered the stop loss condition. + + Args: + backtester: Backtest instance with current trade state + + Returns: + Tuple[bool, Optional[float]]: (stop_loss_triggered, sell_price) + """ + # Calculate stop loss price + stop_price = backtester.entry_price * (1 - backtester.strategies["stop_loss_pct"]) + + # Get minute-level data for precise stop loss checking + min1_df = backtester.original_df if hasattr(backtester, 'original_df') else backtester.min1_df + min1_index = min1_df.index + + # Find data range from entry to current time + start_candidates = min1_index[min1_index >= backtester.entry_time] + if len(start_candidates) == 0: + return False, None + + backtester.current_trade_min1_start_idx = start_candidates[0] + end_candidates = min1_index[min1_index <= backtester.current_date] + + if len(end_candidates) == 0: + return False, None + + backtester.current_min1_end_idx = end_candidates[-1] + + # Check if any candle in the range triggered stop loss + min1_slice = min1_df.loc[backtester.current_trade_min1_start_idx:backtester.current_min1_end_idx] + + if (min1_slice['low'] <= stop_price).any(): + # Find the first candle that triggered stop loss + stop_candle = min1_slice[min1_slice['low'] <= stop_price].iloc[0] + + # Use open price if it gapped below stop, otherwise use stop price + if stop_candle['open'] < stop_price: + sell_price = stop_candle['open'] + else: + sell_price = stop_price + + return True, sell_price + + return False, None \ No newline at end of file diff --git a/cycles/strategies/manager.py b/cycles/strategies/manager.py new file mode 100644 index 0000000..8cc3ed7 --- /dev/null +++ b/cycles/strategies/manager.py @@ -0,0 +1,381 @@ +""" +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 + + +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)) + else: + raise ValueError(f"Unknown strategy: {name}. " + f"Available strategies: default, bbrs") + + 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. + + Args: + backtester: Backtest instance with OHLCV data + """ + for strategy in self.strategies: + try: + strategy.initialize(backtester) + logging.info(f"Initialized strategy: {strategy.name}") + 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") + + 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, and combination rules + """ + return { + "strategies": [ + { + "name": strategy.name, + "weight": strategy.weight, + "params": strategy.params, + "initialized": strategy.initialized + } + for strategy in self.strategies + ], + "combination_rules": self.combination_rules, + "total_strategies": len(self.strategies), + "initialized": self.initialized + } + + 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) \ No newline at end of file diff --git a/docs/strategy_manager.md b/docs/strategy_manager.md new file mode 100644 index 0000000..4aca4c3 --- /dev/null +++ b/docs/strategy_manager.md @@ -0,0 +1,330 @@ +# TCP Cycles Strategy Management System + +The Strategy Manager system provides a flexible framework for implementing, combining, and managing multiple trading strategies within the TCP Cycles project. + +## 🏗️ Architecture + +### Module Structure +``` +cycles/ +├── strategies/ # Strategy management module +│ ├── __init__.py # Module exports and version info +│ ├── base.py # Base classes (StrategyBase, StrategySignal) +│ ├── default_strategy.py # Meta-trend strategy implementation +│ ├── bbrs_strategy.py # Bollinger Bands + RSI strategy +│ └── manager.py # StrategyManager and orchestration +├── Analysis/ # Technical analysis tools +├── utils/ # Utility functions +├── backtest.py # Backtesting engine +└── charts.py # Charting and visualization +``` + +### Core Components + +1. **`StrategyBase`**: Abstract base class that all strategies inherit from +2. **`StrategySignal`**: Represents trading signals with confidence levels and metadata +3. **`DefaultStrategy`**: Implementation of the meta-trend strategy using Supertrend indicators +4. **`BBRSStrategy`**: Implementation of the Bollinger Bands + RSI strategy with market regime detection +5. **`StrategyManager`**: Orchestrates multiple strategies and combines their signals + +### Signal Types +- **`"ENTRY"`**: Strategy suggests entering a position +- **`"EXIT"`**: Strategy suggests exiting a position +- **`"HOLD"`**: Strategy suggests no action + +## 📋 Configuration + +### Single Strategy Configuration + +**Default Strategy Only:** +```json +{ + "start_date": "2025-03-01", + "stop_date": "2025-03-15", + "initial_usd": 10000, + "timeframes": ["15min"], + "stop_loss_pcts": [0.03, 0.05], + "strategies": [ + { + "name": "default", + "weight": 1.0, + "params": {} + } + ], + "combination_rules": { + "entry": "any", + "exit": "any", + "min_confidence": 0.5 + } +} +``` + +**BBRS Strategy Only:** +```json +{ + "strategies": [ + { + "name": "bbrs", + "weight": 1.0, + "params": { + "bb_width": 0.05, + "bb_period": 20, + "rsi_period": 14, + "strategy_name": "MarketRegimeStrategy" + } + } + ] +} +``` + +### Multiple Strategy Configuration + +```json +{ + "strategies": [ + { + "name": "default", + "weight": 0.6, + "params": {} + }, + { + "name": "bbrs", + "weight": 0.4, + "params": { + "bb_width": 0.05, + "strategy_name": "MarketRegimeStrategy" + } + } + ], + "combination_rules": { + "entry": "weighted_consensus", + "exit": "any", + "min_confidence": 0.6 + } +} +``` + +## 🔧 Combination Rules + +### Entry Signal Combination Methods + +- **`"any"`**: Enter if ANY strategy signals entry above min_confidence +- **`"all"`**: Enter only if ALL strategies signal entry above min_confidence +- **`"majority"`**: Enter if more than 50% of strategies signal entry +- **`"weighted_consensus"`**: Enter based on weighted average confidence + +### Exit Signal 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: STOP_LOSS > SELL_SIGNAL > others + +### Parameters + +- **`min_confidence`**: Minimum confidence threshold (0.0 to 1.0) +- **`weight`**: Strategy weight for weighted calculations + +## 🚀 Usage Examples + +### Running with Default Strategy +```bash +python main.py config_default.json +``` + +### Running with BBRS Strategy +```bash +python main.py config_bbrs.json +``` + +### Running with Combined Strategies +```bash +python main.py config_combined.json +``` + +### Running without Config (Interactive) +```bash +python main.py +``` + +### Programmatic Usage +```python +from cycles.strategies import create_strategy_manager + +# Create strategy manager from config +config = { + "strategies": [ + {"name": "default", "weight": 0.7, "params": {}}, + {"name": "bbrs", "weight": 0.3, "params": {"bb_width": 0.05}} + ], + "combination_rules": { + "entry": "weighted_consensus", + "exit": "any", + "min_confidence": 0.6 + } +} + +strategy_manager = create_strategy_manager(config) +``` + +## ⚙️ Strategy Parameters + +### Default Strategy Parameters +- **`stop_loss_pct`**: Stop loss percentage (default: 0.03) + +### BBRS Strategy Parameters +- **`bb_width`**: Bollinger Band width (default: 0.05) +- **`bb_period`**: Bollinger Band period (default: 20) +- **`rsi_period`**: RSI period (default: 14) +- **`trending_rsi_threshold`**: RSI thresholds for trending market [low, high] +- **`trending_bb_multiplier`**: BB multiplier for trending market +- **`sideways_rsi_threshold`**: RSI thresholds for sideways market [low, high] +- **`sideways_bb_multiplier`**: BB multiplier for sideways market +- **`strategy_name`**: Strategy implementation name +- **`SqueezeStrategy`**: Enable squeeze strategy (boolean) +- **`stop_loss_pct`**: Stop loss percentage (default: 0.05) + +## 🔌 Adding New Strategies + +### 1. Create Strategy Class + +Create a new file in `cycles/strategies/` (e.g., `my_strategy.py`): + +```python +from .base import StrategyBase, StrategySignal + +class MyStrategy(StrategyBase): + def __init__(self, weight=1.0, params=None): + super().__init__("my_strategy", weight, params) + + def initialize(self, backtester): + # Initialize your strategy indicators + self.initialized = True + + def get_entry_signal(self, backtester, df_index): + # Implement entry logic + if entry_condition: + return StrategySignal("ENTRY", confidence=0.8) + return StrategySignal("HOLD", confidence=0.0) + + def get_exit_signal(self, backtester, df_index): + # Implement exit logic + if exit_condition: + return StrategySignal("EXIT", confidence=1.0, + metadata={"type": "MY_EXIT"}) + return StrategySignal("HOLD", confidence=0.0) +``` + +### 2. Register Strategy + +Update `cycles/strategies/manager.py` in the `_load_strategies` method: + +```python +elif name == "my_strategy": + from .my_strategy import MyStrategy + strategies.append(MyStrategy(weight, params)) +``` + +### 3. Export Strategy + +Update `cycles/strategies/__init__.py`: + +```python +from .my_strategy import MyStrategy + +__all__ = [ + # ... existing exports ... + 'MyStrategy' +] +``` + +## 📊 Performance Features + +### Strategy Analysis +- Individual strategy performance tracking +- Combined strategy performance metrics +- Signal quality analysis +- Confidence level monitoring + +### Plotting Support +- Automatic chart generation for BBRS strategies +- Meta-trend visualization for default strategy +- Combined signal overlays +- Performance comparison charts + +## 🔄 Backward Compatibility + +The system maintains full backward compatibility: +- ✅ Existing code using single strategies works unchanged +- ✅ Legacy strategy functions are preserved in main.py +- ✅ Default behavior matches original implementation +- ✅ Gradual migration path available + +## 📚 Best Practices + +### 1. **Risk Management** +- Use `"any"` exit rule for faster risk exits +- Set appropriate stop loss percentages per strategy +- Monitor combined drawdown vs individual strategies + +### 2. **Signal Quality** +- Set appropriate `min_confidence` based on strategy reliability +- Test individual strategies thoroughly before combining +- Monitor signal frequency and quality + +### 3. **Weight Distribution** +- Balance strategy weights based on historical performance +- Consider strategy correlation when setting weights +- Regularly rebalance based on changing market conditions + +### 4. **Testing & Validation** +- Backtest individual strategies first +- Test combinations on historical data +- Validate on out-of-sample data + +### 5. **Monitoring** +- Log strategy initialization and errors +- Track individual vs combined performance +- Monitor signal generation frequency + +## 🔍 Troubleshooting + +### Strategy Not Found Error +``` +ValueError: Unknown strategy: my_strategy +``` +**Solution**: Ensure strategy is registered in `manager.py` `_load_strategies` method + +### No Signals Generated +**Possible Causes**: +- Strategy initialization failed +- Data insufficient for strategy requirements +- `min_confidence` threshold too high + +**Solution**: Check logs, verify data, adjust confidence threshold + +### Poor Combined Performance +**Analysis Steps**: +1. Review individual strategy performance +2. Check strategy correlation and overlap +3. Adjust weights and combination rules +4. Consider market regime compatibility + +### Import Errors +``` +ImportError: cannot import name 'StrategyManager' +``` +**Solution**: Use correct import path: `from cycles.strategies import StrategyManager` + +## 📞 Support + +For issues, feature requests, or contributions: +1. Check existing documentation and examples +2. Review troubleshooting section +3. Examine configuration files for proper syntax +4. Ensure all dependencies are installed + +--- + +**Version**: 1.0.0 +**Last Updated**: January 2025 +**TCP Cycles Project** \ No newline at end of file