# source: https://raw.githubusercontent.com/Chovus13/strategije-gemini/be26911545b88a6934fdb4cee1bfc9daae6fbd62/JohnWickSniperStrategy.py import numpy as np import pandas as pd import pandas_ta as pta from pandas import DataFrame import talib.abstract as ta from freqtrade.strategy import IStrategy from freqtrade.strategy import CategoricalParameter, DecimalParameter, IntParameter from datetime import datetime import logging logger = logging.getLogger(__name__) class Github_Chovus13_strategije_gemini__JohnWickSniperStrategy__20250612_015949(IStrategy): """ V1 verzija scalping strategije za Freqtrade na 15m timeframe-u. Koristi 5m i 1m kandele za potvrdu trenda, Hammer/Reverse Hammer sveće, Donchian+ADX za LONG, Bollinger+RSI za SHORT. """ # Parametri strategije timeframe = "15m" informative_timeframes = ["5m", "1m"] minimal_roi = {"0": 0.02} # 2% ROI stoploss = -0.015 # Početni fiksni stop-loss trailing_stop = True trailing_stop_positive = 0.01 trailing_stop_positive_offset = 0.015 # Optimizirani parametri donchian_period = IntParameter(10, 30, default=20, space="buy") adx_period = IntParameter(10, 20, default=14, space="buy") adx_threshold = DecimalParameter(20, 40, default=25, space="buy") bb_period = IntParameter(10, 30, default=20, space="sell") bb_std = DecimalParameter(1.5, 3.0, default=2.0, space="sell") rsi_period = IntParameter(10, 20, default=14, space="sell") rsi_sell = DecimalParameter(60, 80, default=70, space="sell") atr_period = IntParameter(10, 20, default=14, space="buy_sell") volume_spike_factor = DecimalParameter(1.5, 3.0, default=2.0, space="buy_sell") # Leverage parametri leverage_num = IntParameter(1, 10, default=3, space="protection") margin_mode = CategoricalParameter(['isolated', 'cross'], default='isolated', space="protection") # Helper funkcija za Heikin Ashi sveće def heikin_ashi(self, dataframe: DataFrame) -> DataFrame: ha_df = DataFrame(index=dataframe.index) ha_df['ha_close'] = (dataframe['open'] + dataframe['high'] + dataframe['low'] + dataframe['close']) / 4 for i in range(len(dataframe)): if i == 0: ha_df.at[i, 'ha_open'] = ((dataframe.at[i, 'open'] + dataframe.at[i, 'close']) / 2) else: ha_df.at[i, 'ha_open'] = ((ha_df.at[i - 1, 'ha_open'] + ha_df.at[i - 1, 'ha_close']) / 2) ha_df['ha_high'] = ha_df[['ha_open', 'ha_close']].join(dataframe['high']).max(axis=1) ha_df['ha_low'] = ha_df[['ha_open', 'ha_close']].join(dataframe['low']).min(axis=1) return ha_df def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: # --- KORAK 1: Indikatori na informativnim timeframe-ovima ('5m', '1m') --- for timeframe_inf in self.informative_timeframes: informative = self.dp.get_pair_dataframe(pair=metadata['pair'], timeframe=timeframe_inf) if informative.empty: continue ha_informative = self.heikin_ashi(informative) informative['ha_hammer'] = ha_informative.apply( lambda x: 1 if (x['ha_high'] - x['ha_low']) > 3 * abs(x['ha_open'] - x['ha_close']) and (x['ha_close'] - x['ha_low']) / (0.001 + x['ha_high'] - x['ha_low']) > 0.6 and (x['ha_open'] - x['ha_low']) / (0.001 + x['ha_high'] - x['ha_low']) > 0.6 else 0, axis=1 ) informative['ha_reverse_hammer'] = ha_informative.apply( lambda x: 1 if (x['ha_high'] - x['ha_low']) > 3 * abs(x['ha_open'] - x['ha_close']) and (x['ha_high'] - x['ha_close']) / (0.001 + x['ha_high'] - x['ha_low']) > 0.6 and (x['ha_high'] - x['ha_open']) / (0.001 + x['ha_high'] - x['ha_low']) > 0.6 else 0, axis=1 ) informative['rsi'] = ta.RSI(informative) informative['volume_spike'] = (informative['volume'] > informative['volume'].rolling(20).mean() * 2).astype(int) informative.rename(columns={ 'ha_hammer': f'ha_hammer_{timeframe_inf}', 'ha_reverse_hammer': f'ha_reverse_hammer_{timeframe_inf}', 'rsi': f'rsi_{timeframe_inf}', 'volume_spike': f'volume_spike_{timeframe_inf}' }, inplace=True) dataframe = pd.merge_asof( dataframe, informative[[ 'date', f'ha_hammer_{timeframe_inf}', f'ha_reverse_hammer_{timeframe_inf}', f'rsi_{timeframe_inf}', f'volume_spike_{timeframe_inf}' ]], on='date', direction='backward' ) # --- KORAK 2: Indikatori na glavnom timeframe-u ('15m') --- bollinger = ta.BBANDS(dataframe, timeperiod=20, nbdevup=2.0, nbdevdn=2.0, matype=0) dataframe['bb_lowerband'] = bollinger['lowerband'] dataframe['bb_middleband'] = bollinger['middleband'] dataframe['bb_upperband'] = bollinger['upperband'] donchian = pta.donchian(high=dataframe['high'], low=dataframe['low'], close=dataframe['close'], length=self.donchian_period.value) upper_col = f'DCU_{self.donchian_period.value}' lower_col = f'DCL_{self.donchian_period.value}' if upper_col in donchian.columns and lower_col in donchian.columns: dataframe['dc_upper'] = donchian[upper_col] dataframe['dc_lower'] = donchian[lower_col] else: dataframe['dc_upper'] = float('NaN') dataframe['dc_lower'] = float('NaN') dataframe['atr'] = ta.ATR(dataframe) # <<< EVO GA DODAT OVDE >>> try: dataframe['adx'] = ta.ADX(dataframe, timeperiod=self.adx_period.value) except Exception: dataframe['adx'] = float('NaN') dataframe['rsi'] = ta.RSI(dataframe) ha_dataframe = self.heikin_ashi(dataframe) dataframe['ha_hammer'] = ha_dataframe.apply(lambda x: 1 if (x['ha_high'] - x['ha_low']) > 3 * abs(x['ha_open'] - x['ha_close']) and (x['ha_close'] - x['ha_low']) / (0.001 + x['ha_high'] - x['ha_low']) > 0.6 and (x['ha_open'] - x['ha_low']) / (0.001 + x['ha_high'] - x['ha_low']) > 0.6 else 0, axis=1) dataframe['ha_reverse_hammer'] = ha_dataframe.apply(lambda x: 1 if (x['ha_high'] - x['ha_low']) > 3 * abs(x['ha_open'] - x['ha_close']) and (x['ha_high'] - x['ha_close']) / (0.001 + x['ha_high'] - x['ha_low']) > 0.6 and (x['ha_high'] - x['ha_open']) / (0.001 + x['ha_high'] - x['ha_low']) > 0.6 else 0, axis=1) return dataframe def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """ Uslovi za ulaz u LONG i SHORT pozicije. """ # LONG: Breakout + Volume Spike dataframe.loc[ ( (dataframe['close'] > dataframe['dc_upper']) & (dataframe['adx'] > self.adx_threshold.value) & (dataframe['hammer']) & (dataframe['hammer_5m'] | dataframe['hammer_1m']) & (dataframe['volume_spike']) ), 'enter_long'] = 1 # SHORT: Mean Reversion + Volume Spike dataframe.loc[ ( (dataframe['close'] > dataframe['bb_upper']) & (dataframe['rsi'] > self.rsi_sell.value) & (dataframe['reverse_hammer']) & (dataframe['reverse_hammer_5m'] | dataframe['reverse_hammer_1m']) & (dataframe['volume_spike']) ), 'enter_short'] = 1 return dataframe def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """ Uslovi za izlaz iz LONG i SHORT pozicija. """ # Izlaz iz LONG-a dataframe.loc[ ( (dataframe['reverse_hammer']) & (dataframe['reverse_hammer_5m'] | dataframe['reverse_hammer_1m']) ), 'exit_long'] = 1 # Izlaz iz SHORT-a dataframe.loc[ ( (dataframe['hammer']) & (dataframe['hammer_5m'] | dataframe['hammer_1m']) ), 'exit_short'] = 1 return dataframe def custom_stoploss(self, pair: str, trade: 'Trade', current_time: datetime, current_rate: float, current_profit: float, **kwargs) -> float: dataframe, _ = self.dp.get_analyzed_dataframe(pair, self.timeframe) last_candle = dataframe.iloc[-1].squeeze() # Stop loss na osnovu ATR-a # Vrednost se ne ažurira ako je profit veći od neke vrednosti (trailing) if current_profit < 0.1: # npr. ne pomeraj ako je profit > 10% # Uzimamo ATR vrednost iz poslednje sveće stoploss_atr = last_candle['atr'] * 2 # Množilac ATR-a je dobar kandidat za optimizaciju # Postavljamo stop loss ispod cene ulaska za vrednost ATR-a return trade.open_rate - stoploss_atr # Vrati postojeći stop loss ako se uslov ne ispuni return -1 # -1 znači "ne menjaj stop loss" def leverage(self, pair: str, current_time: 'datetime', current_rate: float, proposed_leverage: float, max_leverage: float, side: str, **kwargs) -> float: """Postavlja leverage na 3x.""" return 3.0 def confirm_trade_entry(self, pair: str, order_type: str, amount: float, rate: float, time_in_force: str, current_time: 'datetime', **kwargs) -> bool: """Provjerava valjanost ulaza u trejd.""" try: return True except Exception as e: print(f"Greška u confirm_trade_entry: {e}") return False