""" 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