# pragma pylint: disable=missing-docstring, invalid-name, pointless-string-statement # flake8: noqa: F401 # isort: skip_file # --- Do not remove these libs --- import numpy as np import pandas as pd from pandas import DataFrame from datetime import datetime, timedelta from typing import Optional, Union from freqtrade.strategy import (BooleanParameter, CategoricalParameter, DecimalParameter, IStrategy, IntParameter, merge_informative_pair) # -------------------------------- # Add your lib to import here import talib.abstract as ta from freqtrade.persistence import Trade from functools import reduce class VolumeTrabar_v2(IStrategy): """ Strategy berdasarkan indikator Volume Trabar.id dari TradingView Konsep: - Menggunakan multiple Bollinger Bands dengan periode berbeda - Menghitung rank berdasarkan kondisi oversold/overbought - Rank negatif = oversold (buy signal) - Rank positif = overbought (sell signal) - Multi-timeframe analysis untuk konfirmasi signal """ # Strategy interface version - allow new iterations of the strategy interface. # Check the documentation or the Sample strategy to get the latest version. INTERFACE_VERSION = 3 # Optimal timeframe for the strategy. timeframe = '5m' # Can this strategy go short? can_short: bool = False # Minimal ROI designed for the strategy. minimal_roi = { "0": 0.08, "30": 0.05, "60": 0.03, "120": 0.01 } # Optimal stoploss designed for the strategy. stoploss = -0.06 # Trailing stoploss trailing_stop = True trailing_only_offset_is_reached = True trailing_stop_positive = 0.01 trailing_stop_positive_offset = 0.025 # Run "populate_indicators()" only for new candle. process_only_new_candles = True # These values can be overridden in the config. use_exit_signal = True exit_profit_only = False ignore_roi_if_entry_signal = False # Number of candles the strategy requires before producing valid signals startup_candle_count: int = 500 # Strategy parameters - Calculation Settings len1 = IntParameter(10, 20, default=14, space="buy", optimize=True) len2 = IntParameter(35, 50, default=42, space="buy", optimize=True) len3 = IntParameter(150, 200, default=168, space="buy", optimize=True) len4 = IntParameter(300, 400, default=336, space="buy", optimize=False) len5 = IntParameter(600, 750, default=672, space="buy", optimize=False) len6 = IntParameter(1200, 1400, default=1344, space="buy", optimize=False) len7 = IntParameter(1900, 2100, default=2016, space="buy", optimize=False) len8 = IntParameter(3900, 4100, default=4032, space="buy", optimize=False) # Standard deviation multiplier std_mult = DecimalParameter(1.5, 3.0, default=2.0, space="buy", optimize=True) # MA Type ma_type = CategoricalParameter(["SMA", "EMA"], default="SMA", space="buy", optimize=False) # Thresholds short_term_threshold = IntParameter(1, 3, default=1, space="buy", optimize=False) mid_term_threshold = IntParameter(3, 6, default=4, space="buy", optimize=True) long_term_threshold = IntParameter(5, 8, default=6, space="buy", optimize=True) # Entry/Exit thresholds buy_rank_threshold = IntParameter(-10, -3, default=-5, space="buy", optimize=True) sell_rank_threshold = IntParameter(3, 10, default=5, space="sell", optimize=True) # Multi-timeframe ready signal enable_mtf_ready = BooleanParameter(default=True, space="buy", optimize=False) ready_threshold = IntParameter(3, 7, default=5, space="buy", optimize=True) # Informative timeframes for MTF analysis (must be >= base timeframe) informative_timeframes = ['15m', '30m', '1h'] def informative_pairs(self): """ Define additional, informative pair/interval combinations to be cached from the exchange. """ pairs = self.dp.current_whitelist() informative_pairs = [] for tf in self.informative_timeframes: informative_pairs.extend([(pair, tf) for pair in pairs]) return informative_pairs def calculate_ma(self, series: pd.Series, length: int, ma_type: str) -> pd.Series: """Calculate moving average based on type""" if ma_type == "SMA": return series.rolling(window=length).mean() elif ma_type == "EMA": return series.ewm(span=length, adjust=False).mean() else: # RMA (same as EMA for simplicity) return series.ewm(span=length, adjust=False).mean() def check_bands(self, dataframe: DataFrame, length: int, mult: float, ma_type: str): """ Check if price is outside Bollinger Bands Returns oversold and overbought conditions """ ma = self.calculate_ma(dataframe['close'], length, ma_type) std = dataframe['close'].rolling(window=length).std() lower_band = ma - (std * mult) upper_band = ma + (std * mult) is_oversold = (dataframe['close'] < lower_band) & (~pd.isna(ma)) & (~pd.isna(std)) is_overbought = (dataframe['close'] > upper_band) & (~pd.isna(ma)) & (~pd.isna(std)) return is_oversold, is_overbought def calculate_rank(self, dataframe: DataFrame) -> pd.Series: """ Calculate the volume rank based on multiple Bollinger Bands Negative rank = oversold (buy signal) Positive rank = overbought (sell signal) """ ma_type_val = self.ma_type.value std_mult_val = self.std_mult.value # Check all timeframes oversold1, overbought1 = self.check_bands(dataframe, self.len1.value, std_mult_val, ma_type_val) oversold2, overbought2 = self.check_bands(dataframe, self.len2.value, std_mult_val, ma_type_val) oversold3, overbought3 = self.check_bands(dataframe, self.len3.value, std_mult_val, ma_type_val) oversold4, overbought4 = self.check_bands(dataframe, self.len4.value, std_mult_val, ma_type_val) oversold5, overbought5 = self.check_bands(dataframe, self.len5.value, std_mult_val, ma_type_val) oversold6, overbought6 = self.check_bands(dataframe, self.len6.value, std_mult_val, ma_type_val) oversold7, overbought7 = self.check_bands(dataframe, self.len7.value, std_mult_val, ma_type_val) oversold8, overbought8 = self.check_bands(dataframe, self.len8.value, std_mult_val, ma_type_val) # Calculate oversold rank (sequential check) oversold_rank = pd.Series(0, index=dataframe.index) condition = oversold1 & oversold2 oversold_rank = np.where(condition, 2, oversold_rank) condition = condition & oversold3 oversold_rank = np.where(condition, 3, oversold_rank) condition = condition & oversold4 oversold_rank = np.where(condition, 4, oversold_rank) condition = condition & oversold5 oversold_rank = np.where(condition, 5, oversold_rank) condition = condition & oversold6 oversold_rank = np.where(condition, 6, oversold_rank) condition = condition & oversold7 oversold_rank = np.where(condition, 7, oversold_rank) condition = condition & oversold8 oversold_rank = np.where(condition, 8, oversold_rank) # Calculate overbought rank (sequential check) overbought_rank = pd.Series(0, index=dataframe.index) condition = overbought1 & overbought2 overbought_rank = np.where(condition, 2, overbought_rank) condition = condition & overbought3 overbought_rank = np.where(condition, 3, overbought_rank) condition = condition & overbought4 overbought_rank = np.where(condition, 4, overbought_rank) condition = condition & overbought5 overbought_rank = np.where(condition, 5, overbought_rank) condition = condition & overbought6 overbought_rank = np.where(condition, 6, overbought_rank) condition = condition & overbought7 overbought_rank = np.where(condition, 7, overbought_rank) condition = condition & overbought8 overbought_rank = np.where(condition, 8, overbought_rank) # Final rank calculation rank = pd.Series(0, index=dataframe.index) # If only oversold, rank is negative rank = np.where((oversold_rank > 0) & (overbought_rank == 0), -oversold_rank, rank) # If only overbought, rank is positive rank = np.where((overbought_rank > 0) & (oversold_rank == 0), overbought_rank, rank) # If both, choose the stronger one rank = np.where((oversold_rank > overbought_rank) & (oversold_rank > 0) & (overbought_rank > 0), -oversold_rank, rank) rank = np.where((overbought_rank > oversold_rank) & (oversold_rank > 0) & (overbought_rank > 0), overbought_rank, rank) return pd.Series(rank, index=dataframe.index) def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """ Populate indicators that will be used in the Buy, Sell, and Exit strategy """ # Calculate main rank for current timeframe dataframe['rank'] = self.calculate_rank(dataframe) # Add informative timeframes for MTF analysis for tf in self.informative_timeframes: informative = self.dp.get_pair_dataframe(pair=metadata['pair'], timeframe=tf) informative['rank'] = self.calculate_rank(informative) # Merge informative data dataframe = merge_informative_pair(dataframe, informative, self.timeframe, tf, ffill=True) # MTF Ready Signal if self.enable_mtf_ready.value and len(self.informative_timeframes) >= 3: tf1 = self.informative_timeframes[0] tf2 = self.informative_timeframes[1] tf3 = self.informative_timeframes[2] # Ready Buy: semua MTF rank <= -ready_threshold dataframe['mtf_ready_buy'] = ( (dataframe[f'rank_{tf1}'] <= -self.ready_threshold.value) & (dataframe[f'rank_{tf2}'] <= -self.ready_threshold.value) & (dataframe[f'rank_{tf3}'] <= -self.ready_threshold.value) ) # Ready Sell: semua MTF rank >= ready_threshold dataframe['mtf_ready_sell'] = ( (dataframe[f'rank_{tf1}'] >= self.ready_threshold.value) & (dataframe[f'rank_{tf2}'] >= self.ready_threshold.value) & (dataframe[f'rank_{tf3}'] >= self.ready_threshold.value) ) else: dataframe['mtf_ready_buy'] = False dataframe['mtf_ready_sell'] = False # Add volume for confirmation dataframe['volume_ma'] = dataframe['volume'].rolling(window=20).mean() return dataframe def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """ Based on TA indicators, populates the entry signal for the given dataframe """ conditions = [] # Basic condition: rank is below buy threshold (oversold) conditions.append(dataframe['rank'] <= self.buy_rank_threshold.value) # Volume confirmation conditions.append(dataframe['volume'] > 0) # Optional: MTF Ready confirmation if self.enable_mtf_ready.value: # Either MTF ready buy OR strong rank on current timeframe conditions.append( (dataframe['mtf_ready_buy'] == True) | (dataframe['rank'] <= (self.buy_rank_threshold.value - 2)) ) if conditions: dataframe.loc[ reduce(lambda x, y: x & y, conditions), 'enter_long'] = 1 return dataframe def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """ Based on TA indicators, populates the exit signal for the given dataframe """ conditions = [] # Basic condition: rank is above sell threshold (overbought) conditions.append(dataframe['rank'] >= self.sell_rank_threshold.value) # Optional: MTF Ready confirmation if self.enable_mtf_ready.value: # Either MTF ready sell OR strong rank on current timeframe conditions.append( (dataframe['mtf_ready_sell'] == True) | (dataframe['rank'] >= (self.sell_rank_threshold.value + 2)) ) if conditions: dataframe.loc[ reduce(lambda x, y: x & y, conditions), 'exit_long'] = 1 return dataframe def custom_stoploss(self, pair: str, trade: 'Trade', current_time: datetime, current_rate: float, current_profit: float, **kwargs) -> float: """ Custom stoploss logic, returning the new distance relative to current_rate """ # After 20 minutes, allow stoploss to be closer if current_time - trade.open_date_utc > timedelta(minutes=20): return -0.05 return self.stoploss