""" Meta Supertrend strategy implementation. """ import numpy as np import pandas as pd from engine.market import MarketType from strategies.base import BaseStrategy from .indicators import SuperTrendIndicator class MetaSupertrendStrategy(BaseStrategy): """ Meta Supertrend Strategy using 3 Supertrend indicators. Enters long when all 3 Supertrends are bullish. Enters short when all 3 Supertrends are bearish. Designed for perpetual futures with leverage and short-selling support. """ # Market configuration default_market_type = MarketType.PERPETUAL default_leverage = 5 # Risk management parameters default_sl_stop = 0.02 # 2% stop loss default_sl_trail = True # Trailing stop enabled default_exit_on_bearish_flip = False # Rely on SL/TP, not bearish flip def run( self, close: pd.Series, high: pd.Series = None, low: pd.Series = None, period1: int = 10, multiplier1: float = 3.0, period2: int = 11, multiplier2: float = 2.0, period3: int = 12, multiplier3: float = 1.0, exit_on_bearish_flip: bool = None, enable_short: bool = True, **kwargs ) -> tuple[pd.DataFrame, pd.DataFrame, pd.DataFrame, pd.DataFrame]: # 1. Validation & Setup if exit_on_bearish_flip is None: exit_on_bearish_flip = self.default_exit_on_bearish_flip if high is None or low is None: raise ValueError("MetaSupertrendStrategy requires High and Low prices.") # 2. Calculate Supertrends t1, t2, t3 = self._calculate_supertrends( high, low, close, period1, multiplier1, period2, multiplier2, period3, multiplier3 ) # 3. Meta Signals bullish, bearish = self._calculate_meta_signals(t1, t2, t3, close) # 4. Generate Entry/Exit Signals return self._generate_signals(bullish, bearish, exit_on_bearish_flip, enable_short) def _calculate_supertrends( self, high, low, close, p1, m1, p2, m2, p3, m3 ): """Run the 3 Supertrend indicators.""" # Pass NumPy arrays explicitly to avoid Numba typing errors h_vals = high.values l_vals = low.values c_vals = close.values def run_st(p, m): st = SuperTrendIndicator.run(h_vals, l_vals, c_vals, period=p, multiplier=m) trend = st.trend if isinstance(trend, pd.DataFrame): trend.index = close.index if trend.shape[1] == 1: trend = trend.iloc[:, 0] elif isinstance(trend, pd.Series): trend.index = close.index return trend t1 = run_st(p1, m1) t2 = run_st(p2, m2) t3 = run_st(p3, m3) return t1, t2, t3 def _calculate_meta_signals(self, t1, t2, t3, close_series): """Combine 3 Supertrends into boolean Bullish/Bearish signals.""" # Use NumPy broadcasting t1_vals = t1.values if isinstance(t1, pd.DataFrame) else t1.values.reshape(-1, 1) # Force column vectors for broadcasting if scalar result t2_vals = t2.values.reshape(-1, 1) t3_vals = t3.values.reshape(-1, 1) # Boolean logic on numpy arrays (1 = Bull, -1 = Bear) bullish_vals = (t1_vals == 1) & (t2_vals == 1) & (t3_vals == 1) bearish_vals = (t1_vals == -1) & (t2_vals == -1) & (t3_vals == -1) # Reconstruct Pandas objects if isinstance(t1, pd.DataFrame): bullish = pd.DataFrame(bullish_vals, index=t1.index, columns=t1.columns) bearish = pd.DataFrame(bearish_vals, index=t1.index, columns=t1.columns) else: bullish = pd.Series(bullish_vals.flatten(), index=t1.index) bearish = pd.Series(bearish_vals.flatten(), index=t1.index) return bullish, bearish def _generate_signals( self, bullish, bearish, exit_on_bearish_flip, enable_short ): """Generate long/short entry/exit signals based on meta trend.""" # Long Entries: Change from Not Bullish to Bullish prev_bullish = bullish.shift(1).fillna(False) long_entries = bullish & (~prev_bullish) # Long Exits if exit_on_bearish_flip: prev_bearish = bearish.shift(1).fillna(False) long_exits = bearish & (~prev_bearish) else: long_exits = BaseStrategy.create_empty_signals(long_entries) # Short signals if enable_short: prev_bearish = bearish.shift(1).fillna(False) short_entries = bearish & (~prev_bearish) if exit_on_bearish_flip: short_exits = bullish & (~prev_bullish) else: short_exits = BaseStrategy.create_empty_signals(long_entries) else: short_entries = BaseStrategy.create_empty_signals(long_entries) short_exits = BaseStrategy.create_empty_signals(long_entries) return long_entries, long_exits, short_entries, short_exits