# 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 from typing import Optional, Union from freqtrade.strategy import ( IStrategy, Trade, Order, PairLocks, informative, merge_informative_pair, stoploss_from_open, stoploss_from_absolute, ) # -------------------------------- # Add your lib to import here import talib.abstract as ta import freqtrade.vendor.qtpylib.indicators as qtpylib class KrakenSwingStrategy(IStrategy): """ Kraken-optimized swing trading strategy for BTC/USD and ETH/USD pairs. Designed for single $1000 positions targeting 4-6% minimum moves to overcome Kraken's 0.8% round-trip fee structure. Based on research from Kraken Trading Pairs Analysis and Implementation Roadmap documents. """ # Strategy interface version - allow new iterations of the strategy interface. INTERFACE_VERSION = 3 # Optimal timeframe for swing trading on Kraken timeframe = '1h' # Can this strategy go short? can_short: bool = False # Minimal ROI designed for strategy - targeting 4-6% minimum moves minimal_roi = { "0": 0.06, # 6% immediate target "40": 0.05, # 5% after 40 minutes "80": 0.04, # 4% after 80 minutes (minimum for fee efficiency) "120": 0.01 # 1% after 2 hours (safety exit) } # Optimal stoploss for single position strategy stoploss = -0.03 # 3% maximum loss per trade ($30 on $1000 position) # Trailing stoploss trailing_stop = False # Hyperoptable parameters buy_rsi_period = 14 buy_rsi_value = 35 sell_rsi_value = 70 # Buy/Sell signal optimization spaces buy_bb_lower_offset = 0.02 # 2% below lower BB sell_bb_upper_offset = 0.02 # 2% above upper BB def informative_pairs(self): """ Define additional, informative pair/interval combinations to be cached from the exchange. """ return [] def populate_indicators(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """ Adds several different TA indicators to the given DataFrame Performance Note: For preprocessing, consider using new columns via: dataframe['ema12'] = ta.EMA(dataframe, timeperiod=12) """ # RSI - Primary momentum indicator dataframe['rsi'] = ta.RSI(dataframe, timeperiod=self.buy_rsi_period) # Bollinger Bands - Volatility and mean reversion bollinger = qtpylib.bollinger_bands(qtpylib.typical_price(dataframe), window=20, stds=2) dataframe['bb_lowerband'] = bollinger['lower'] dataframe['bb_middleband'] = bollinger['mid'] dataframe['bb_upperband'] = bollinger['upper'] dataframe["bb_percent"] = ( (dataframe["close"] - dataframe["bb_lowerband"]) / (dataframe["bb_upperband"] - dataframe["bb_lowerband"]) ) dataframe["bb_width"] = ( (dataframe["bb_upperband"] - dataframe["bb_lowerband"]) / dataframe["bb_middleband"] ) # MACD - Trend confirmation macd = ta.MACD(dataframe) dataframe['macd'] = macd['macd'] dataframe['macdsignal'] = macd['macdsignal'] dataframe['macdhist'] = macd['macdhist'] # EMA - Trend direction dataframe['ema12'] = ta.EMA(dataframe, timeperiod=12) dataframe['ema26'] = ta.EMA(dataframe, timeperiod=26) # Volume indicators dataframe['volume_sma'] = ta.SMA(dataframe['volume'], timeperiod=20) # ADX - Trend strength (for filtering weak signals) dataframe['adx'] = ta.ADX(dataframe) return dataframe def populate_entry_trend(self, dataframe: DataFrame, metadata: dict) -> DataFrame: """ Based on TA indicators, populates the entry signal for the given dataframe Entry Criteria for 4-6% swing moves: 1. RSI oversold but recovering (momentum building) 2. Price near lower Bollinger Band (mean reversion setup) 3. MACD showing bullish divergence 4. Strong trend confirmation from EMA 5. Above average volume (institutional interest) """ conditions = [] # RSI recovery from oversold conditions.append( (dataframe['rsi'] > self.buy_rsi_value) & (dataframe['rsi'].shift(1) <= self.buy_rsi_value) ) # Price bouncing off lower Bollinger Band conditions.append( (dataframe['close'] <= dataframe['bb_lowerband'] * (1 + self.buy_bb_lower_offset)) & (dataframe['close'] > dataframe['bb_lowerband'] * (1 - self.buy_bb_lower_offset)) ) # MACD bullish signal conditions.append( (dataframe['macd'] > dataframe['macdsignal']) & (dataframe['macdhist'] > 0) ) # EMA trend confirmation (short above long) conditions.append(dataframe['ema12'] > dataframe['ema26']) # Volume confirmation (above average) conditions.append(dataframe['volume'] > dataframe['volume_sma']) # ADX shows strong trend (above 25) conditions.append(dataframe['adx'] > 25) # Bollinger Band width shows sufficient volatility for 4-6% moves conditions.append(dataframe['bb_width'] > 0.04) # 4% minimum width 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 Exit Criteria for profit-taking: 1. RSI overbought (momentum exhausted) 2. Price approaching upper Bollinger Band 3. MACD showing bearish divergence 4. Volume declining (profit-taking phase) """ conditions = [] # RSI overbought conditions.append(dataframe['rsi'] > self.sell_rsi_value) # Price at upper Bollinger Band conditions.append( dataframe['close'] >= dataframe['bb_upperband'] * (1 - self.sell_bb_upper_offset) ) # MACD bearish signal conditions.append( (dataframe['macd'] < dataframe['macdsignal']) | (dataframe['macdhist'] < 0) ) # Volume declining (optional exit signal) volume_declining = dataframe['volume'] < dataframe['volume'].shift(1) 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 for Kraken optimization - Maintains 3% maximum loss - No trailing stop to avoid Kraken API rate limit issues """ return self.stoploss def custom_sell(self, pair: str, trade: 'Trade', current_time: datetime, current_rate: float, current_profit: float, **kwargs) -> Optional[Union[str, bool]]: """ Custom sell logic for profit optimization on Kraken Ensures we hit minimum 4% profit targets to overcome fees """ # Force exit if we hit 6% profit (optimal target) if current_profit >= 0.06: return 'profit_target_6pct' # Consider exit if we hit 4% minimum and other conditions met if current_profit >= 0.04: # Check if momentum is declining (basic check) return None # Let normal exit signals handle this return None def leverage(self, pair: str, current_time: datetime, current_rate: float, proposed_leverage: float, max_leverage: float, entry_tag: Optional[str], side: str, **kwargs) -> float: """ Customize leverage for each new trade. Only called when margin trading is enabled. For Kraken spot trading, this returns 1.0 (no leverage) """ return 1.0 def reduce(function, iterable, initializer=None): """Python reduce function for combining conditions""" it = iter(iterable) if initializer is None: value = next(it) else: value = initializer for element in it: value = function(value, element) return value