""" 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, List 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. The strategy works best on 15-minute timeframes but can be configured for other timeframes. Parameters: stop_loss_pct (float): Stop loss percentage (default: 0.03) timeframe (str): Preferred timeframe for analysis (default: "15min") Example: strategy = DefaultStrategy(weight=1.0, params={"stop_loss_pct": 0.05, "timeframe": "15min"}) """ 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 and timeframe """ super().__init__("default", weight, params) def get_timeframes(self) -> List[str]: """ Get the timeframes required by the default strategy. The default strategy works on a single timeframe (typically 15min) but also needs 1min data for precise stop-loss execution. Returns: List[str]: List containing primary timeframe and 1min for stop-loss """ primary_timeframe = self.params.get("timeframe", "15min") # Always include 1min for stop-loss precision, avoid duplicates timeframes = [primary_timeframe] if primary_timeframe != "1min": timeframes.append("1min") return timeframes 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 # First, resample the original 1-minute data to required timeframes self._resample_data(backtester.original_df) # Get the primary timeframe data for strategy calculations primary_timeframe = self.get_timeframes()[0] strategy_data = self.get_data_for_timeframe(primary_timeframe) # Calculate Supertrend indicators on the primary timeframe supertrends = Supertrends(strategy_data, 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 # Note: backtester.df should now be using our primary timeframe backtester.strategies["meta_trend"] = meta_trend backtester.strategies["stop_loss_pct"] = self.params.get("stop_loss_pct", 0.03) backtester.strategies["primary_timeframe"] = primary_timeframe 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 primary timeframe 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 primary timeframe 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 using 1-minute data for precision 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 primary timeframe 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. Uses 1-minute data for precise stop loss checking regardless of the primary timeframe used for strategy signals. 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"]) # Use 1-minute data for precise stop loss checking min1_data = self.get_data_for_timeframe("1min") if min1_data is None: # Fallback to original_df if 1min timeframe not available min1_data = backtester.original_df if hasattr(backtester, 'original_df') else backtester.min1_df min1_index = min1_data.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_data.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