# region R-đava kôd® - KobacPro - Profesionalna, Asimetrična Strategija # --- Fajl: KobacPro.py --- from functools import reduce from pandas import DataFrame import logging import pandas as pd import talib.abstract as ta from freqtrade.strategy import IStrategy, IntParameter, DecimalParameter from datetime import datetime logger = logging.getLogger(__name__) class KobacPro(IStrategy): """ Profesionalna verzija strategije sa potpuno odvojenom i asimetričnom logikom za LONG i SHORT. Fokus na robusnosti i realnim tržišnim uslovima. Spremna za Hyperopt od nule. """ INTERFACE_VERSION = 3 can_short = True trading_mode = "futures" margin_mode = "isolated" stoploss = -0.25 # Realan fiksni stoploss, ATR je naša glavna zaštita minimal_roi = {"0": 1} use_custom_stoploss = True # --- Optimizabilni parametri --- # Zajednički parametar leverage_num = IntParameter(2, 5, default=3, space='buy', optimize=False) # --- LONG ULAZNI PARAMETRI: Breakout strategija (space='buy') --- long_donchian_period = IntParameter(20, 50, default=27, space='buy', optimize=True) long_adx_period = IntParameter(14, 30, default=26, space='buy', optimize=True) long_adx_level = IntParameter(20, 35, default=31, space='buy', optimize=True) # --- SHORT ULAZNI PARAMETRI: Mean Reversion strategija (space='sell') --- short_bb_period = IntParameter(15, 30, default=20, space='sell', optimize=True) short_bb_stddev = DecimalParameter(2.0, 3.5, default=2.5, decimals=1, space='sell', optimize=True) short_rsi_period = IntParameter(14, 30, default=21, space='sell', optimize=True) short_rsi_level = IntParameter(68, 85, default=75, space='sell', optimize=True) # --- IZLAZNI PARAMETRI (zajednički za obe strane) --- exit_ema_period = IntParameter(5, 20, default=10, space='sell', optimize=True) # --- STOP-LOSS PARAMETRI --- stoploss_atr_period = IntParameter(10, 20, default=14, space='stoploss', optimize=True) stoploss_atr_multiplier = DecimalParameter(2.5, 5.0, default=4.0, decimals=1, space='stoploss', optimize=True) def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: # Long Indikatori (Donchian Channels za breakout) dataframe['donchian_upper'] = dataframe['high'].rolling(self.long_donchian_period.value).max().shift(1) dataframe['long_adx'] = ta.ADX(dataframe, self.long_adx_period.value) # Short Indikatori (Bollinger Bands i RSI za preokret) bollinger_short = ta.BBANDS(dataframe, timeperiod=self.short_bb_period.value, nbdevup=self.short_bb_stddev.value, nbdevdn=self.short_bb_stddev.value) dataframe['short_bb_upperband'] = bollinger_short['upperband'] dataframe['short_rsi'] = ta.RSI(dataframe, self.short_rsi_period.value) # Exit Indikator dataframe['exit_ema'] = ta.EMA(dataframe, self.exit_ema_period.value) # Stoploss Indikator dataframe['atr'] = ta.ATR(dataframe, timeperiod=self.stoploss_atr_period.value) return dataframe def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: # --- LONG LOGIKA: Kupujemo proboj (Breakout) --- long_conditions = { 'breakout': (dataframe['close'] > dataframe['donchian_upper']), 'trend_strong': (dataframe['long_adx'] > self.long_adx_level.value), 'volume_ok': (dataframe['volume'] > 0) } # --- SHORT LOGIKA: Short-ujemo euforiju (Mean Reversion) --- short_conditions = { 'price_overextended': (dataframe['close'] > dataframe['short_bb_upperband']), 'rsi_overbought': (dataframe['short_rsi'] > self.short_rsi_level.value), 'volume_ok': (dataframe['volume'] > 0) } # Postavljanje signala dataframe.loc[reduce(lambda x, y: x & y, long_conditions.values()), ['enter_long', 'enter_tag']] = (1, 'long_breakout') dataframe.loc[reduce(lambda x, y: x & y, short_conditions.values()), ['enter_short', 'enter_tag']] = (1, 'short_mean_reversion') return dataframe def populate_exit_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: # Za LONG, izlazimo kada cena padne ispod brze EMA linije dataframe.loc[dataframe['close'] < dataframe['exit_ema'], ['exit_long', 'exit_tag']] = (1, 'exit_ema_cross') # Za SHORT, takođe izlazimo kada cena padne ispod brze EMA linije dataframe.loc[dataframe['close'] < dataframe['exit_ema'], ['exit_short', 'exit_tag']] = (1, 'exit_ema_cross') 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) # Proveravamo da li je stoploss već postavljen. Ako nije, postavljamo inicijalni ATR stop. if trade.stop_loss == self.stoploss: atr_multiplier = self.stoploss_atr_multiplier.value # Uzimamo ATR sa sveće na kojoj je trejd otvoren za veću preciznost try: entry_candle = dataframe.loc[trade.open_date_utc] atr_at_entry = entry_candle['atr'] except KeyError: # Fallback ako ne možemo naći tačnu sveću atr_at_entry = dataframe.iloc[-1]['atr'] if trade.is_short: initial_stop = current_rate + (atr_at_entry * atr_multiplier) else: initial_stop = current_rate - (atr_at_entry * atr_multiplier) logger.info(f"ULAZNI STOPLOSS za {pair}: Postavljam početni ATR stop na {initial_stop:.5f}") return initial_stop # TRAILING LOGIKA: Pomeramo stop-loss samo ako je profitabilan i ako je novi stop bolji if current_profit > 0.01: # Počinjemo da pratimo tek posle 1% profita atr_multiplier = self.stoploss_atr_multiplier.value last_atr = dataframe.iloc[-1]['atr'] if trade.is_short: new_stop_price = current_rate + (last_atr * atr_multiplier) # Pomeramo samo naniže if new_stop_price < trade.stop_loss: logger.info(f"ATR TRAILING (SHORT) za {pair}: Pomeram stop sa {trade.stop_loss:.5f} na {new_stop_price:.5f}") return new_stop_price else: new_stop_price = current_rate - (last_atr * atr_multiplier) # Pomeramo samo naviše if new_stop_price > trade.stop_loss: logger.info(f"ATR TRAILING (LONG) za {pair}: Pomeram stop sa {trade.stop_loss:.5f} na {new_stop_price:.5f}") return new_stop_price # Ako nema uslova za pomeranje, vraćamo trenutni stop-loss return trade.stop_loss def leverage(self, pair: str, current_time: datetime, current_rate: float, proposed_leverage: float, max_leverage: float, side: str, **kwargs) -> float: return self.leverage_num.value def confirm_trade_entry(self, pair: str, order_type: str, amount: float, rate: float, time_in_force: str, current_time: datetime, entry_tag: str, side: str, **kwargs) -> bool: logger.info(f"✅ Potvrda ulaza: {pair} ({side}) na ceni {rate:.5f}") return True #endregion